d6625762cf
modified: app/ali_oss.py modified: app/app.py new file: app/blueprints/__pycache__/communication.cpython-39.pyc modified: app/blueprints/communication.py modified: app/brand_spider/__pycache__/main.cpython-39.pyc modified: app/config.py modified: app/main.py modified: app/static/bg.jpg modified: app/tool/__pycache__/devices.cpython-39.pyc modified: app/web_source/brand.html modified: app/web_source/templates_backup/brand.html
363 lines
14 KiB
Python
363 lines
14 KiB
Python
"""
|
||
基于 pywebview 实现的跨平台 UI，需先登录方可使用
|
||
"""
|
||
from config import cache_path,debug,version,JAVA_API_BASE,JSON_TASK_QUEUE
|
||
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
|
||
import requests
|
||
|
||
|
||
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 upload_file_to_java(self, file_path, relative_path=None):
|
||
"""按本地路径读取文件并上传到 Java 后端临时目录。"""
|
||
if not file_path or not str(file_path).strip():
|
||
return {'success': False, 'error': '文件路径为空'}
|
||
normalized_path = os.path.abspath(str(file_path).strip())
|
||
if not os.path.isfile(normalized_path):
|
||
return {'success': False, 'error': '文件不存在'}
|
||
try:
|
||
with open(normalized_path, 'rb') as file_obj:
|
||
response = requests.post(
|
||
f"{JAVA_API_BASE}/api/files/upload",
|
||
files={'file': (os.path.basename(normalized_path), file_obj)},
|
||
data={'relativePath': str(relative_path).strip()} if relative_path else None,
|
||
timeout=120,
|
||
)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
if not isinstance(payload, dict):
|
||
return {'success': False, 'error': 'Java 返回格式错误'}
|
||
return payload
|
||
except Exception as e:
|
||
return {'success': False, 'error': str(e)}
|
||
|
||
def enqueue_json(self, data):
|
||
"""保存任务到队列"""
|
||
try:
|
||
print("==============================")
|
||
print(datetime.datetime.now().strftime( "%Y-%m-%d %H:%M:%S"),"调用传入",data)
|
||
print("==================================")
|
||
payload = data
|
||
if isinstance(data, str):
|
||
payload = json.loads(data)
|
||
if not isinstance(payload, (dict, list)):
|
||
return {'success': False, 'error': '仅支持 JSON 对象或数组'}
|
||
JSON_TASK_QUEUE.put(payload)
|
||
return {'success': True, 'queue_size': JSON_TASK_QUEUE.qsize()}
|
||
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():
|
||
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 不弹窗。"""
|
||
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,api.upload_file_to_java,api.enqueue_json)
|
||
webview.start(
|
||
debug=False,
|
||
storage_path=cache_path,
|
||
private_mode=False
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|