super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

new file: ali_oss.py

Browse Source 
new file:   app.py
	new file:   app/ali_oss.py
	new file:   app/app.py
	new file:   app/app_common.py
	new file:   app/config.py
	new file:   app/coze.py
	new file:   app/generate_api.py
	new file:   app/html_crypto.py
	new file:   app/main.py
	new file:   app/web_source/admin.html
	new file:   app/web_source/brand.html
	new file:   app/web_source/home.html
	new file:   app/web_source/index.html
	new file:   app/web_source/login.html
	new file:   app_common.py
	new file:   blueprints/__init__.py
	new file:   blueprints/__pycache__/__init__.cpython-311.pyc
	new file:   blueprints/__pycache__/__init__.cpython-39.pyc
	new file:   blueprints/__pycache__/admin.cpython-311.pyc
	new file:   blueprints/__pycache__/admin.cpython-39.pyc
	new file:   blueprints/__pycache__/auth.cpython-311.pyc
	new file:   blueprints/__pycache__/auth.cpython-39.pyc
	new file:   blueprints/__pycache__/brand.cpython-311.pyc
	new file:   blueprints/__pycache__/brand.cpython-39.pyc
	new file:   blueprints/__pycache__/image.cpython-311.pyc
	new file:   blueprints/__pycache__/image.cpython-39.pyc
	new file:   blueprints/__pycache__/main.cpython-311.pyc
	new file:   blueprints/__pycache__/main.cpython-39.pyc
	new file:   blueprints/admin.py
	new file:   blueprints/auth.py
	new file:   blueprints/brand.py
	new file:   blueprints/image.py
	new file:   blueprints/main.py
	new file:   brand_spider/__pycache__/main.cpython-39.pyc
	new file:   brand_spider/__pycache__/web_dec.cpython-39.pyc
	new file:   brand_spider/main.py
	new file:   brand_spider/web_dec.py
	new file:   config.py
	new file:   coze.py
	new file:   generate_api.py
	new file:   html_crypto.py
	new file:   main.py
	new file:   static/bg.jpg
	new file:   "static/\345\223\201\347\211\214\346\226\207\346\241\243\346\240\274\345\274\217_\346\250\241\346\235\277.xlsx"
	new file:   "static/\346\250\241\346\235\2772-\344\273\245\346\226\207\344\273\266\345\244\271\346\226\271\345\274\217\344\270\212\344\274\240.zip"
	new file:   tool/.device_id
	new file:   tool/__pycache__/devices.cpython-311.pyc
	new file:   tool/__pycache__/devices.cpython-39.pyc
	new file:   tool/devices.py
This commit is contained in:
铭坤
2026-03-20 11:23:57 +08:00
parent 43d508f527
commit 8a64a6cd9b
45 changed files with 5253 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
app/ali_oss.py Normal file
View File

@@ -0,0 +1,73 @@
import argparse
import base64
import re
import time
import alibabacloud_oss_v2 as oss
import requests
from config import region, endpoint, bucket, file_url_pre, bucket_path
def upload_file(file_content: bytes, key: str):
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
cfg.region = region
cfg.endpoint = endpoint
cfg.retry_max_attempts = 3
client = oss.Client(cfg)
result = client.put_object(
oss.PutObjectRequest(
bucket=bucket, # 存储空间名称
key=key, # 对象名称
body=file_content # 读取文件内容
)
)
# print(result)
return file_url_pre + key
def upload_data_url(data_url: str, prefix: str = "history", key_hint: str = "") -> str:
"""
将 base64 data URL 上传到 OSS，返回图片链接
data_url: data:image/png;base64,xxxx 或 data:image/jpeg;base64,xxxx
prefix: OSS key 前缀
key_hint: 可选后缀避免重名，如 "_0", "_1"
"""
match = re.match(r'data:image/(\w+);base64,(.+)', data_url)
if not match:
raise ValueError('无效的 data URL 格式')
ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg'
file_content = base64.b64decode(match.group(2))
ts = int(time.time() * 1000)
key = f"{bucket_path}{prefix}/{ts}{key_hint}.{ext}"
return upload_file(file_content, key)
def upload_data_urls(data_urls: list, prefix: str = "history") -> list:
"""批量上传 base64 图片到 OSS，返回图片链接列表"""
urls = []
ts = int(time.time() * 1000)
for i, data_url in enumerate(data_urls or []):
if not data_url or not isinstance(data_url, str):
continue
if not data_url.startswith("http"):
match = re.match(r'data:image/(\w+);base64,(.+)', data_url)
if not match:
continue
ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg'
file_content = base64.b64decode(match.group(2))
else:
file_content = requests.get(data_url).content
ext = "png"
key = f"{bucket_path}{prefix}/{ts}_{i}.{ext}"
urls.append(upload_file(file_content, key))
return urls
# 脚本入口，当文件被直接运行时调用main函数
if __name__ == "__main__":
with open("测试图片数据/IMG_2685.JPG", "rb") as f:
file_content = f.read()
upload_file(file_content,key=bucket_path+"test.png")

45
app/app.py Normal file
View File

@@ -0,0 +1,45 @@
"""
卖相AI - Flask 后端
按功能拆分为蓝图：认证(auth)、主页面(main)、管理员(admin)、图片(image)、品牌(brand)
"""
import os
import secrets
from datetime import timedelta
from flask import Flask
from flask_cors import CORS
from app_common import init_db, BASE_DIR
from blueprints.auth import auth_bp
from blueprints.main import main_bp
from blueprints.admin import admin_bp
from blueprints.image import image_bp
from blueprints.brand import brand_bp
def create_app():
app = Flask(__name__, template_folder=BASE_DIR, static_folder=BASE_DIR)
CORS(app)
app.secret_key = os.environ.get('SECRET_KEY', secrets.token_hex(32))
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=7)
# 注册蓝图（不设 url_prefix，保持原有 URL 路径不变，前端无需改动）
app.register_blueprint(auth_bp)
app.register_blueprint(main_bp)
app.register_blueprint(admin_bp)
app.register_blueprint(image_bp)
app.register_blueprint(brand_bp)
return app
app = create_app()
def run_app(host='127.0.0.1', port=5123):
init_db()
app.run(host=host, port=port, threaded=True, use_reloader=False)
if __name__ == '__main__':
run_app()

261
app/app_common.py Normal file
View File

@@ -0,0 +1,261 @@
"""
公共模块：数据库连接、初始化、会话校验、装饰器、模板渲染
供各蓝图复用
"""
import os
import secrets
from datetime import timedelta
from functools import wraps
import pymysql
from flask import request, redirect, url_for, session, jsonify, render_template, render_template_string
from config import mysql_host, mysql_user, mysql_password, mysql_database
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
STATIC_DIR = os.path.join(BASE_DIR, 'static')
def get_db():
return pymysql.connect(
host=mysql_host,
user=mysql_user,
password=mysql_password,
database=mysql_database,
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
def _render_html(template_name: str, **context):
"""读取 HTML 模板：若为加密文件则先解密，再渲染。未加密或解密失败时按明文渲染。"""
path = os.path.join(BASE_DIR, "web_source", template_name)
if not os.path.isfile(path):
return render_template(template_name, **context)
with open(path, "rb") as f:
raw = f.read()
try:
from html_crypto import decrypt
content = decrypt(raw).decode("utf-8")
except Exception:
content = raw.decode("utf-8", errors="replace")
return render_template_string(content, **context)
def _is_session_user_valid():
"""校验 session 中的 user_id 是否在数据库中仍存在；不存在则清除 session 并返回 False"""
uid = session.get('user_id')
if not uid:
return False
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT id FROM users WHERE id = %s", (uid,))
row = cur.fetchone()
conn.close()
if not row:
session.clear()
return False
return True
except Exception:
session.clear()
return False
def _get_current_admin_role():
"""获取当前登录用户的管理角色：super_admin / admin / None（非管理员）"""
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"SELECT id, username, is_admin, role, created_by_id FROM users WHERE id = %s",
(session['user_id'],)
)
row = cur.fetchone()
conn.close()
if not row or not row.get('is_admin'):
return None, None
return row.get('role') or ('super_admin' if row.get('created_by_id') is None else 'admin'), row
except Exception:
return None, None
def login_required(f):
@wraps(f)
def decorated(*args, **kwargs):
if not session.get('user_id') or not _is_session_user_valid():
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
return jsonify({'success': False, 'error': '未登录'}), 401
return redirect(url_for('auth.login'))
return f(*args, **kwargs)
return decorated
def admin_required(f):
@wraps(f)
def decorated(*args, **kwargs):
if not session.get('user_id'):
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
return jsonify({'success': False, 'error': '未登录'}), 401
return redirect(url_for('auth.login'))
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT is_admin, role FROM users WHERE id = %s", (session['user_id'],))
row = cur.fetchone()
conn.close()
if not row or not row.get('is_admin'):
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
return jsonify({'success': False, 'error': '需要管理员权限'}), 403
return redirect(url_for('main.home'))
except Exception as e:
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
return jsonify({'success': False, 'error': str(e)}), 500
return redirect(url_for('main.home'))
return f(*args, **kwargs)
return decorated
def init_db():
"""初始化数据库表，若不存在则创建"""
from werkzeug.security import generate_password_hash
conn = pymysql.connect(
host=mysql_host,
user=mysql_user,
password=mysql_password,
charset='utf8mb4'
)
try:
with conn.cursor() as cur:
cur.execute(f"CREATE DATABASE IF NOT EXISTS `{mysql_database}` DEFAULT CHARSET utf8mb4")
cur.execute(f"USE `{mysql_database}`")
cur.execute("""
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(64) NOT NULL UNIQUE,
password_hash VARCHAR(256) NOT NULL,
is_admin TINYINT(1) DEFAULT 0,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS image_history (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
panel_type VARCHAR(64) DEFAULT '',
original_urls JSON,
params JSON,
result_urls JSON,
long_image_url VARCHAR(1024) DEFAULT NULL,
INDEX idx_user_created (user_id, created_at DESC)
)
""")
try:
cur.execute("ALTER TABLE image_history ADD COLUMN long_image_url VARCHAR(1024) DEFAULT NULL")
except Exception:
pass
try:
cur.execute("ALTER TABLE users ADD COLUMN machine VARCHAR(64) DEFAULT NULL")
except Exception:
pass
try:
cur.execute("ALTER TABLE users ADD COLUMN role VARCHAR(20) DEFAULT 'normal'")
except Exception:
pass
try:
cur.execute("ALTER TABLE users ADD COLUMN created_by_id INT NULL")
except Exception:
pass
cur.execute("""
CREATE TABLE IF NOT EXISTS brand_crawl_tasks (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT NOT NULL,
file_paths JSON NOT NULL,
status VARCHAR(20) DEFAULT 'pending',
task_type TINYINT DEFAULT 1 COMMENT '1=立即执行,2=添加任务执行',
result_paths JSON NULL,
error_message TEXT NULL,
progress_current INT DEFAULT 0,
progress_total INT DEFAULT 0,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
INDEX idx_user_status (user_id, status)
)
""")
try:
cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN progress_current INT DEFAULT 0")
except Exception:
pass
try:
cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN progress_total INT DEFAULT 0")
except Exception:
pass
try:
cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN `desc` VARCHAR(500) NULL COMMENT '上传文件名描述'")
except Exception:
pass
try:
cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN strategy VARCHAR(20) DEFAULT 'Terms' COMMENT '品牌匹配方式: Terms/Simple'")
except Exception:
pass
try:
cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN task_type TINYINT DEFAULT 1 COMMENT '1=立即执行,2=添加任务执行'")
except Exception:
pass
cur.execute("""
CREATE TABLE IF NOT EXISTS columns (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(128) NOT NULL COMMENT '栏目名',
column_key VARCHAR(64) NOT NULL COMMENT '栏目标识',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uk_column_key (column_key)
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS user_column_permission (
user_id INT NOT NULL,
column_id INT NOT NULL,
PRIMARY KEY (user_id, column_id),
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY (column_id) REFERENCES columns(id) ON DELETE CASCADE
)
""")
try:
cur.execute("UPDATE users SET role = 'normal' WHERE (role IS NULL OR role = '') AND (is_admin = 0 OR is_admin IS NULL)")
cur.execute("SELECT MIN(id) AS mid FROM users WHERE is_admin = 1")
row = cur.fetchone()
if row and row.get('mid'):
mid = row['mid']
cur.execute("UPDATE users SET role = 'super_admin' WHERE id = %s", (mid,))
cur.execute("UPDATE users SET role = 'admin' WHERE is_admin = 1 AND id != %s", (mid,))
cur.execute("UPDATE users SET created_by_id = %s WHERE role = 'admin' AND (created_by_id IS NULL)", (mid,))
except Exception:
pass
conn.commit()
finally:
conn.close()
_create_initial_admin()
def _create_initial_admin():
"""若没有任何管理员，则创建默认超级管理员（首次启动时，仅一个）"""
from werkzeug.security import generate_password_hash
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT id FROM users WHERE role = 'super_admin' LIMIT 1")
if cur.fetchone():
conn.close()
return
admin_user = os.environ.get('ADMIN_USER', 'admin')
admin_pwd = os.environ.get('ADMIN_PASSWORD', 'admin123')
pwd_hash = generate_password_hash(admin_pwd, method='pbkdf2:sha256')
cur.execute(
"INSERT INTO users (username, password_hash, is_admin, role) VALUES (%s, %s, 1, 'super_admin')",
(admin_user, pwd_hash)
)
conn.commit()
conn.close()
except Exception:
pass

46
app/config.py Normal file
View File

@@ -0,0 +1,46 @@
base_url = "https://api.coze.cn/v1"
coze_token = "sat_12nW40INoJxArrDXbY4lSCoudbqkOYTcphC99BP2efWyzxmsk4q81WDX3ezWgqZ5"
workflow_id = "7608812635877900322"
STITCH_WORKFLOW_ID = "7608813873483300907"
import os
from dotenv import load_dotenv
load_dotenv()
_base_url = os.getenv("base_url","http://159.75.121.33:15124")
# MySQL 配置
mysql_host = os.getenv("mysql_host")
mysql_user = os.getenv("mysql_user")
mysql_password = os.getenv("mysql_password","WTFrb5y6hNLz6hNy")
mysql_database = os.getenv("mysql_database","aiimage")
proxy_url = os.getenv("proxy_url")
proxy_mode = int(os.getenv("proxy_mode",1))
client_name=os.getenv("client_name") + ".exe"
cache_path = "./user_data"
region = "cn-hangzhou"
endpoint = "oss-cn-hangzhou.aliyuncs.com"
bucket = "nanri-ai-images"
accessKeyId = "LTAI5tNpyvzMNz9f2dHarsm8"
accessKeySecret = "bQSZnFH455i8tzyOgeahJmUzwmhynz"
bucket_path = "nanri-image/"
file_url_pre = f"https://{bucket}.oss-cn-hangzhou.aliyuncs.com/"
os.environ['OSS_ACCESS_KEY_ID'] = accessKeyId
os.environ['OSS_ACCESS_KEY_SECRET'] = accessKeySecret
os.environ['SECRET_KEY'] = "ddffc7c1d02121d9554d7b080b2511b6"
debug = True
version = "1.0.3"
APP_UPDATE_URL = f"{_base_url}/api/version/latest" # 检测更新接口地址，返回 { "file_url": "...", "version": "x.x.x" }
os.environ['APP_VERSION'] = version
os.environ['APP_UPDATE_URL'] = APP_UPDATE_URL

92
app/coze.py Normal file
View File

@@ -0,0 +1,92 @@
import requests
import time
from config import base_url, coze_token
def upload_file(file_path,_coze_token =coze_token):
url = f"{base_url}/files/upload"
headers = {
"Authorization": f"Bearer {_coze_token}"
}
print(file_path)
with open(file_path, 'rb') as f:
files = {
'file': f
}
# with open(f"{time.time()".replace(".","_")+".png","wb") as file:
# file.write(f.read())
# allow_redirects=True 模拟 curl 的 --location，默认即为 True
response = requests.post(url, headers=headers, files=files, allow_redirects=True)
data = response.json()
print("上传图片->>",data)
return data
def workflow_run(workflow_id, parameters,_coze_token = coze_token,is_async=True):
url = f"{base_url}/workflow/run"
headers = {
"Authorization": f"Bearer {_coze_token}",
"Content-Type": "application/json"
}
# 请求体（JSON 格式）
payload = {
"parameters": parameters,
# "parameters": {
# "name": "包包",
# "ratio": "3:4",
# "menu": 3,
# "resolution": "2K",
# "count": 1,
# "desc": "1、一个中国美女手上挎着这个包，2、站在商场内",
# "language": "中文",
# "style": "极简高级"
# },
"workflow_id": workflow_id,
"is_async" : is_async,
}
# 发送 POST 请求（使用 json 参数自动序列化并设置 Content-Type）
response = requests.post(url, headers=headers, json=payload)
data = response.json()
print("图片生成参数->>",payload)
print("图片生成->>",data)
return data
def query_result(workflow_id, execute_id,_coze_token=coze_token):
url = f"{base_url}/workflows/{workflow_id}/run_histories/{execute_id}"
headers = {
"Authorization": f"Bearer {_coze_token}",
"Content-Type": "application/json"
}
response = requests.get(url, headers=headers)
data = response.json()
print(f"【{execute_id}】结果查询->>",data)
return data
if __name__ == '__main__':
from config import workflow_id
parameters = {
"images": [
{
"file_id": "7612667079497613350",
},{
"file_id": "7612667042990768174",
},{
"file_id": "7612667111332577332",
}
]
}
# resp = workflow_run("7607680760402100258",parameters,_coze_token="sat_ZVNLR9Om54A3iMlJWAasHF9kbtDZnR3BjwjAxRwe8x7igEW446y5ROyVlW1UlpVX")
# print(resp)
# resp = query_result("7607680760402100258","7612668958797447187",_coze_token="sat_ZVNLR9Om54A3iMlJWAasHF9kbtDZnR3BjwjAxRwe8x7igEW446y5ROyVlW1UlpVX")
# print(resp)
resp = upload_file("D:\私单交付\maixiang_AI\测试图片数据\IMG_2686.JPG",_coze_token="sat_ZVNLR9Om54A3iMlJWAasHF9kbtDZnR3BjwjAxRwe8x7igEW446y5ROyVlW1UlpVX")
print(resp)

261
app/generate_api.py Normal file
View File

@@ -0,0 +1,261 @@
"""
生成图片 API - 供 pywebview 前端调用
流程：上传图片 -> workflow_run -> 轮询 query_result -> 解析返回图片 URL
"""
import base64
import json
import os
import tempfile
import time
import re
import traceback
import requests
from coze import upload_file, workflow_run, query_result
from config import workflow_id
from ali_oss import upload_data_urls as oss_upload_data_urls
def _data_url_to_temp_file(data_url: str, allow_video: bool = False) -> str:
"""将 base64 data URL 转为临时文件路径"""
# data:image/png;base64,xxxx 或 data:video/mp4;base64,xxxx
if allow_video:
match = re.match(r'data:(?:image|video)/(\w+);base64,(.+)', data_url)
else:
match = re.match(r'data:image/(\w+);base64,(.+)', data_url)
if not match:
raise ValueError('无效的 data URL 格式')
mime = match.group(1).lower()
ext_map = {'png': 'png', 'webp': 'png', 'jpeg': 'jpg', 'jpg': 'jpg', 'mp4': 'mp4', 'webm': 'webm'}
ext = ext_map.get(mime, 'jpg')
data = base64.b64decode(match.group(2))
fd, path = tempfile.mkstemp(suffix=f'.{ext}')
try:
os.write(fd, data)
finally:
os.close(fd)
return path
def _upload_video(video_data_url: str) -> str:
"""上传视频，返回 file_id"""
if not video_data_url or not isinstance(video_data_url, str):
raise ValueError('无效的视频数据')
path = _data_url_to_temp_file(video_data_url, allow_video=True)
try:
resp = upload_file(path)
if resp.get('code') == 0 and resp.get('data', {}).get('id'):
return resp['data']['id']
raise RuntimeError(f'视频上传失败: {resp.get("msg", resp)}')
finally:
try:
os.unlink(path)
except Exception:
pass
def _upload_images(image_data_urls: list) -> list:
"""上传多张图片，返回 file_id 列表"""
file_ids = []
temp_paths = []
try:
for data_url in (image_data_urls or []):
if not data_url or not isinstance(data_url, str):
continue
if data_url.startswith("http"):
fd, path = tempfile.mkstemp(suffix=f'.png')
data_content = requests.get(data_url).content
os.write(fd, data_content)
else:
path = _data_url_to_temp_file(data_url)
temp_paths.append(path)
resp = upload_file(path)
if resp.get('code') == 0 and resp.get('data', {}).get('id'):
file_ids.append(resp['data']['id'])
else:
raise RuntimeError(f'上传失败: {resp.get("msg", resp)}')
return file_ids
finally:
for p in temp_paths:
try:
os.unlink(p)
except Exception:
pass
def _parse_output(output_str: str) -> list:
"""解析 query_result 中的 output，提取图片 URL 列表"""
try:
outer = json.loads(output_str)
inner_str = outer.get('Output', '{}')
inner = json.loads(inner_str)
data_str = inner.get('data', '[]')
data = json.loads(data_str)
urls = data.get("images")
long_image_url = data.get("merged_image_url")
return urls if isinstance(urls, list) else [],long_image_url
except Exception:
return [],None
def _parse_output_text(output_str: str) -> list:
"""解析 query_result 中的 output，提取图片 URL 列表"""
try:
outer = json.loads(output_str)
inner_str = outer.get('Output', '{}')
inner = json.loads(inner_str)
data_str = inner.get('data', '')
if isinstance(data_str,str):
data = json.loads(data_str).get("reverse_prompt")
return [data]
pattern = r'【?图像 \d+】.*?(?=【?图像 \d+】|$)'
segments = re.findall(pattern, data_str, re.DOTALL)
return segments
except Exception:
return ""
def _poll_until_done(wf_id: str, execute_id: str, interval: float =10, timeout: int = 600*2, is_text: bool = False) -> list:
"""轮询直到成功或超时，返回图片 URL 列表"""
start = time.time()
while time.time() - start < timeout:
resp = query_result(wf_id, execute_id)
if resp.get('code') != 0:
raise RuntimeError(f'查询失败: {resp.get("msg", resp)}')
items = resp.get('data') or []
if not items:
time.sleep(interval)
continue
item = items[0]
status = item.get('execute_status', '')
if status == 'Success':
output = item.get('output', '{}')
if is_text:
return _parse_output_text(output)
return _parse_output(output)
if status and status not in ('Running', 'Pending', ''):
raise RuntimeError(f'执行失败: {status}')
time.sleep(interval)
raise RuntimeError('生成超时')
def _res_to_2k(res: str) -> str:
"""统一分辨率为 2K/4K"""
r = (res or '2k').strip().upper()
return '4K' if r == '4K' else '2K'
def generate(params: dict) -> dict:
"""
生成图片
params: {
menu: int, # 1=图片反推 2=图片编辑 3=随机海报 4=克隆海报 5=服饰穿搭
prompt: str, # 用户自定义指令 (menu=1 必填)
ref_images: list, # base64 data URLs - 参考图
video: str, # base64 data URL - 视频 (menu=1 可选，仅支持1个)
model_images: list, # base64 data URLs - 多模特图 (menu!=1，最多5张)
name: str, desc: str, ratio: str, resolution: str, count: int,
language: str, style: str, batch_prompt: list, brand_name: str,
Ingredients: str, activity: str, mode: str, texts: list,
proc_images: list, layout_image: str,
}
返回: { success: bool, urls: list, prompts: list, error: str }
"""
original_urls = []
api_key = (params.get('api_key') or '').strip()
try:
menu = int(params.get('menu', 2))
ref_data = params.get('ref_images') or []
proc_data = params.get('proc_images') or []
layout_data = params.get('layout_image')
video_data = params.get('video')
model_data = params.get('model_images') or []
# menu 1: 反推词，简化参数
if menu == 1:
base_params = {'menu': 1, 'prompt': str(params.get('prompt', '')).strip() or ''}
ref_ids = _upload_images(ref_data) if ref_data else []
if ref_ids:
base_params['ref_images'] = [{'file_id': fid} for fid in ref_ids]
if video_data:
try:
vid = _upload_video(video_data)
base_params['video'] = {'file_id': vid}
except Exception as ve:
return {'success': False, 'urls': [], 'prompts': [], 'error': f'视频上传失败: {ve}'}
base_params = {k: v for k, v in base_params.items() if v is not None and v != ''}
base_params["api_key"] = api_key
resp = workflow_run(workflow_id, base_params)
if resp.get('code') != 0:
return {'success': False, 'urls': [], 'prompts': [], 'error': resp.get('msg', str(resp))}
execute_id = resp.get('execute_id')
if not execute_id:
return {'success': False, 'urls': [], 'prompts': [], 'error': '未返回 execute_id'}
prompts = _poll_until_done(workflow_id, str(execute_id), is_text=True)
# menu 1 返回提示词列表，统一转为字符串
# prompts = [str(x) for x in result_list] if result_list else []
# prompts = [prompts]
return {'success': True, 'urls': [], 'prompts': prompts, 'error': ''}
# menu != 1: 原有逻辑
all_ref = list(ref_data)
if layout_data:
all_ref = [layout_data] + list(ref_data)
all_originals = list(all_ref) + list(proc_data) + list(model_data)
if all_originals:
try:
original_urls = oss_upload_data_urls(all_originals, prefix="originals")
except Exception:
traceback.print_exc()
pass
ref_ids = _upload_images(all_ref) if all_ref else []
proc_ids = _upload_images(proc_data) if proc_data else []
model_ids = _upload_images(model_data) if model_data else []
res = _res_to_2k(params.get('resolution', '2K'))
base_params = {
'name': str(params.get('name', '')).strip(),
'ratio': str(params.get('ratio', '')).strip(),
'menu': menu,
'resolution': res,
'count': int(params.get('count', 1)),
'desc': str(params.get('desc', '')).strip(),
'language': str(params.get('language', '中文')).strip() or '中文',
'style': str(params.get('style', '')).strip(),
'prompt': str(params.get('prompt', '')).strip(),
'batch_prompt': params.get('batch_prompt') or [],
'brand_name': str(params.get('brand_name', '')).strip(),
'Ingredients': str(params.get('Ingredients', '')).strip(),
'activity': str(params.get('activity', '')).strip(),
'mode': str(params.get('mode', '1')).strip() or '1',
'texts': params.get('text') or [],
"api_key" : api_key
}
if ref_ids:
base_params['ref_images'] = [{'file_id': fid} for fid in ref_ids]
if proc_ids:
base_params['proc_images'] = [{'file_id': fid} for fid in proc_ids]
if model_ids:
base_params['model_images'] = [{'file_id': fid} for fid in model_ids]
base_params = {k: v for k, v in base_params.items() if v is not None and v != ''}
resp = workflow_run(workflow_id, base_params)
if resp.get('code') != 0:
return {'success': False, 'urls': [], 'error': resp.get('msg', str(resp))}
execute_id = resp.get('execute_id')
if not execute_id:
return {'success': False, 'urls': [], 'error': '未返回 execute_id'}
urls,long_image_url = _poll_until_done(workflow_id, str(execute_id))
return {'success': True, 'urls': urls, 'original_urls': original_urls, 'error': '',"long_image_url":long_image_url}
except Exception as e:
return {'success': False, 'urls': [], 'prompts': [], 'original_urls': [], 'error': str(e),"long_image_url":""}

68
app/html_crypto.py Normal file
View File

@@ -0,0 +1,68 @@
"""
HTML 模板加密/解密模块
使用 AES (Fernet) 对 HTML 资源进行加密存储，运行时解密后渲染
"""
import base64
import os
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
def _get_fernet_key() -> bytes:
"""从环境变量 HTML_ENCRYPT_KEY 派生 Fernet 密钥，若未设置则使用默认开发密钥"""
raw = os.environ.get(
"HTML_ENCRYPT_KEY",
"maixiang_html_encrypt_default_key_change_in_production",
)
# 使用 PBKDF2 派生 32 字节密钥，再转为 Fernet 所需的 base64url
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=b"maixiang_html_salt",
iterations=100000,
)
key_bytes = kdf.derive(raw.encode("utf-8"))
return base64.urlsafe_b64encode(key_bytes)
def encrypt(plain_data: bytes) -> bytes:
"""加密原始数据，返回密文（bytes）"""
key = _get_fernet_key()
f = Fernet(key)
return f.encrypt(plain_data)
def decrypt(encrypted_data: bytes) -> bytes:
"""解密数据，返回明文（bytes）"""
key = _get_fernet_key()
f = Fernet(key)
return f.decrypt(encrypted_data)
def encrypt_file(path: str) -> None:
"""就地加密文件：读取 UTF-8 内容，加密后写回同一路径"""
with open(path, "rb") as f:
plain = f.read()
encrypted = encrypt(plain)
with open(path, "wb") as f:
f.write(encrypted)
def decrypt_file(path: str) -> None:
"""就地解密文件：读取密文，解密后以 UTF-8 写回同一路径"""
with open(path, "rb") as f:
encrypted = f.read()
plain = decrypt(encrypted)
with open(path, "wb") as f:
f.write(plain)
def is_encrypted(data: bytes) -> bool:
"""简单启发式：Fernet 密文以 b'gAAAAA' 开头（base64 编码后）"""
try:
return data[:7] == b"gAAAAAB"
except Exception:
return False

344
app/main.py Normal file
View File

@@ -0,0 +1,344 @@
"""
基于 pywebview 实现的跨平台 UI，需先登录方可使用
"""
from config import cache_path,debug,version
import datetime
import json
import sys
import os
os.makedirs(cache_path,exist_ok=True)
if not debug:
today = datetime.datetime.now().strftime("%Y_%m_%d")
sys.stdout = open(os.path.join(cache_path,f'{today}.log'), 'a',buffering=1)
sys.stderr = sys.stdout
import webview
import threading
import time
with open("version.txt","w",encoding="utf-8") as file:
file.write(json.dumps({"version":version}))
# 获取当前脚本所在目录
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
APP_URL = "http://127.0.0.1:5123"
PORT = 5123
def generate_images(params):
"""供前端调用的生成图片接口"""
from generate_api import generate
return generate(params)
class WindowAPI:
"""暴露给 JavaScript 的窗口控制 API"""
def __init__(self, window):
self._window = window
self._is_maximized = False
window.events.maximized += lambda _: setattr(self, '_is_maximized', True)
window.events.restored += lambda _: setattr(self, '_is_maximized', False)
def close(self):
sys.exit(1)
self._window.destroy()
def minimize(self):
self._window.minimize()
def maximize(self):
self._window.maximize()
def toggle_maximize(self):
if self._is_maximized:
self._window.restore()
else:
self._window.maximize()
def save_image(self, url_or_data, filename='image.png'):
"""弹窗选择保存位置并下载图片。url_or_data 可为 http(s) 链接或 data: base64"""
import base64
import re
result = self._window.create_file_dialog(
webview.SAVE_DIALOG,
save_filename=filename,
file_types=('图片文件 (*.png;*.jpg;*.jpeg)', '所有文件 (*.*)')
)
if not result:
return {'success': False, 'error': '用户取消'}
path = result[0] if isinstance(result, (list, tuple)) else result
try:
if url_or_data.startswith('data:'):
match = re.match(r'data:image/\w+;base64,(.+)', url_or_data)
if not match:
print(url_or_data)
print('无效的 data URL')
return {'success': False, 'error': '无效的 data URL'}
data = base64.b64decode(match.group(1))
else:
import requests
resp = requests.get(url_or_data, timeout=30)
resp.raise_for_status()
data = resp.content
with open(path, 'wb') as f:
f.write(data)
return {'success': True, 'path': path}
except Exception as e:
return {'success': False, 'error': str(e)}
def save_image_to_folder(self, url_or_data, dir_path, filename='image.png'):
"""将图片保存到指定文件夹。url_or_data 可为 http(s) 链接或 data: base64"""
import base64
import re
path = os.path.join(dir_path, filename)
try:
if url_or_data.startswith('data:'):
match = re.match(r'data:image/\w+;base64,(.+)', url_or_data)
if not match:
return {'success': False, 'error': '无效的 data URL'}
data = base64.b64decode(match.group(1))
else:
import requests
resp = requests.get(url_or_data, timeout=30)
resp.raise_for_status()
data = resp.content
os.makedirs(dir_path, exist_ok=True)
with open(path, 'wb') as f:
f.write(data)
return {'success': True, 'path': path}
except Exception as e:
return {'success': False, 'error': str(e)}
def select_folder(self):
# 此方法会在 JavaScript 中被调用，返回选择的文件夹路径
result = webview.windows[0].create_file_dialog(
webview.FOLDER_DIALOG, # 指定为文件夹对话框
allow_multiple=False, # 是否允许多选
directory='' # 初始目录
)
# create_file_dialog 返回的是列表（多选时）或 None
return result[0] if result else ''
def select_brand_xlsx_files(self):
"""品牌爬虫：选择 .xlsx 文件，允许多选，返回路径列表"""
result = webview.windows[0].create_file_dialog(
webview.OPEN_DIALOG,
allow_multiple=True,
file_types=('Excel 文件 (*.xlsx)', '所有文件 (*.*)')
)
print("选择的excel 文件",result)
if not result:
return []
return result if isinstance(result, list) or isinstance(result, tuple) else [result]
def select_brand_folder(self):
"""品牌爬虫：选择文件夹，返回文件夹路径（前端再请求后端展开该目录下所有 xlsx）"""
result = webview.windows[0].create_file_dialog(
webview.FOLDER_DIALOG,
allow_multiple=False,
directory=''
)
return result[0] if result else ''
def save_file_from_url(self, url, default_filename='download.zip'):
"""弹窗选择保存位置，从 url 下载文件并保存。用于品牌任务结果 zip 等。"""
if not url or not url.strip():
return {'success': False, 'error': '下载地址为空'}
result = self._window.create_file_dialog(
webview.SAVE_DIALOG,
save_filename=default_filename,
file_types=('ZIP 压缩包 (*.zip)', '所有文件 (*.*)')
)
if not result:
return {'success': False, 'error': '用户取消'}
path = result[0] if isinstance(result, (list, tuple)) else result
try:
import requests
resp = requests.get(url.strip(), timeout=120, stream=True)
resp.raise_for_status()
with open(path, 'wb') as f:
for chunk in resp.iter_content(chunk_size=65536):
if chunk:
f.write(chunk)
return {'success': True, 'path': path}
except Exception as e:
return {'success': False, 'error': str(e)}
def save_template_xlsx(self):
"""弹窗选择保存位置，将品牌文档格式模板 xlsx 保存到用户选择的位置。"""
import shutil
template_name = '品牌文档格式_模板.xlsx'
template_path = os.path.join(BASE_DIR, 'static', template_name)
if not os.path.isfile(template_path):
return {'success': False, 'error': '模板文件不存在'}
result = self._window.create_file_dialog(
webview.SAVE_DIALOG,
save_filename=template_name,
file_types=('Excel 文件 (*.xlsx)', '所有文件 (*.*)')
)
if not result:
return {'success': False, 'error': '用户取消'}
path = result[0] if isinstance(result, (list, tuple)) else result
try:
shutil.copy2(template_path, path)
return {'success': True, 'path': path}
except Exception as e:
return {'success': False, 'error': str(e)}
def save_template_zip(self):
"""弹窗选择保存位置，将品牌文档格式模板 xlsx 保存到用户选择的位置。"""
import shutil
template_name = '模板2-以文件夹方式上传.zip'
template_path = os.path.join(BASE_DIR, 'static', template_name)
if not os.path.isfile(template_path):
return {'success': False, 'error': '模板文件不存在'}
result = self._window.create_file_dialog(
webview.SAVE_DIALOG,
save_filename=template_name,
file_types=('ZIP 文件 (*.zip)', '所有文件 (*.*)')
)
if not result:
return {'success': False, 'error': '用户取消'}
path = result[0] if isinstance(result, (list, tuple)) else result
try:
shutil.copy2(template_path, path)
return {'success': True, 'path': path}
except Exception as e:
return {'success': False, 'error': str(e)}
def _select_folder_win32():
"""Windows: 用 ctypes 调用系统文件夹选择对话框，打包成 exe 后也能正常弹窗（不依赖 tkinter）"""
try:
import ctypes
from ctypes import wintypes
BIF_RETURNONLYFSDIRS = 0x00000001
BIF_NEWDIALOGSTYLE = 0x00000040
MAX_PATH = 260
shell32 = ctypes.windll.shell32 # type: ignore[attr-defined]
# BROWSEINFOW
class BROWSEINFOW(ctypes.Structure):
_fields_ = [
("hwndOwner", wintypes.HWND),
("pidlRoot", ctypes.c_void_p),
("pszDisplayName", ctypes.c_wchar_p),
("lpszTitle", ctypes.c_wchar_p),
("ulFlags", wintypes.UINT),
("lpfn", ctypes.c_void_p),
("lParam", wintypes.LPARAM),
("iImage", ctypes.c_int),
]
display_name_buf = ctypes.create_unicode_buffer(MAX_PATH)
bi = BROWSEINFOW()
bi.hwndOwner = 0
bi.pidlRoot = 0
bi.pszDisplayName = ctypes.cast(display_name_buf, ctypes.c_wchar_p)
bi.lpszTitle = "选择保存图片的文件夹"
bi.ulFlags = BIF_RETURNONLYFSDIRS | BIF_NEWDIALOGSTYLE
bi.lpfn = 0
bi.lParam = 0
bi.iImage = 0
pidl = shell32.SHBrowseForFolderW(ctypes.byref(bi))
if not pidl:
return ''
path_buf = ctypes.create_unicode_buffer(MAX_PATH)
if not shell32.SHGetPathFromIDListW(pidl, path_buf):
return ''
ole32 = ctypes.windll.ole32 # type: ignore[attr-defined]
ole32.CoTaskMemFree(pidl)
return path_buf.value or ''
except Exception as e:
print("Windows 文件夹选择失败:", e)
return ''
# def _select_folder_tk():
# """使用 tkinter 选择文件夹（在非 Windows 或作为后备）"""
# try:
#
# root = tkinter.Tk()
# root.withdraw()
# root.attributes('-topmost', True)
# path = filedialog.askdirectory(title='选择保存图片的文件夹')
# try:
# root.destroy()
# except Exception:
# pass
# return path if path else ''
# except Exception as e:
# print("tkinter 选择文件夹失败:", e)
# return ''
def _select_folder_tk():
import tkinter as tk
from tkinter import filedialog
root = tk.Tk()
root.withdraw()
root.attributes('-topmost', True)
# 使用 after 确保 askdirectory 在主循环启动后执行
def ask():
nonlocal path
path = filedialog.askdirectory(title='选择保存图片的文件夹')
root.quit() # 关闭主循环
path = ''
root.after(100, ask)
root.mainloop() # 启动主循环，直到 root.quit() 被调用
try:
root.destroy()
except:
pass
return path
def select_folder():
"""弹窗选择文件夹，返回路径（空字符串表示取消）。Windows 打包 exe 时优先用系统 API 避免 tkinter 不弹窗。"""
# if sys.platform == 'win32':
# path = _select_folder_win32()
# if path:
# return path
return _select_folder_tk()
def start_flask():
from app import run_app
run_app(host='127.0.0.1', port=PORT)
def main():
if not os.path.exists(cache_path):
os.makedirs(cache_path,exist_ok=True)
t = threading.Thread(target=start_flask, daemon=True)
t.start()
# 等待 Flask 启动
time.sleep(2)
window = webview.create_window(
title="南日AI",
url=APP_URL,
width=1400,
height=900,
resizable=True,
min_size=(1200, 700),
background_color="#1a1a1a",
frameless=False,
easy_drag=True
)
api = WindowAPI(window)
window.expose(api.close, api.minimize, api.maximize, api.toggle_maximize, generate_images, api.save_image, api.save_image_to_folder, api.select_folder, api.select_brand_xlsx_files, api.select_brand_folder, api.save_file_from_url, api.save_template_xlsx,api.save_template_zip)
webview.start(
debug=True,
storage_path=cache_path,
private_mode=False
)
if __name__ == "__main__":
main()