new file: ali_oss.py
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:
铭坤
73
app/ali_oss.py
Normal file
73
app/ali_oss.py
Normal 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
45
app/app.py
Normal 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
261
app/app_common.py
Normal 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
46
app/config.py
Normal 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
92
app/coze.py
Normal 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
261
app/generate_api.py
Normal 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
68
app/html_crypto.py
Normal 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
344
app/main.py
Normal 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()
|
||||
1
app/web_source/admin.html
Normal file
1
app/web_source/admin.html
Normal file
File diff suppressed because one or more lines are too long
1
app/web_source/brand.html
Normal file
1
app/web_source/brand.html
Normal file
File diff suppressed because one or more lines are too long
1
app/web_source/home.html
Normal file
1
app/web_source/home.html
Normal file
File diff suppressed because one or more lines are too long
1
app/web_source/index.html
Normal file
1
app/web_source/index.html
Normal file
File diff suppressed because one or more lines are too long
1
app/web_source/login.html
Normal file
1
app/web_source/login.html
Normal file
File diff suppressed because one or more lines are too long
Reference in New Issue
Block a user