373 lines
14 KiB
Python
373 lines
14 KiB
Python
"""
|
||
基于 pywebview 实现的跨平台 UI，需先登录方可使用
|
||
"""
|
||
from config import cache_path,debug,version,JAVA_API_BASE,JSON_TASK_QUEUE,runing_task,runing_shop
|
||
import datetime
|
||
import json
|
||
import sys
|
||
import shutil
|
||
import os
|
||
from amazon.main import TaskMonitor
|
||
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
|
||
import subprocess
|
||
|
||
|
||
with open("version.txt","w",encoding="utf-8") as file:
|
||
file.write(json.dumps({"version":version}))
|
||
|
||
print(f"""
|
||
========================================================
|
||
版本： {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.bin'):
|
||
"""弹窗选择保存位置，从 url 下载文件并保存。根据文件后缀动态设置保存类型。"""
|
||
if not url or not url.strip():
|
||
return {'success': False, 'error': '下载地址为空'}
|
||
|
||
filename = str(default_filename or 'download.bin').strip() or 'download.bin'
|
||
ext = os.path.splitext(filename)[1].lower()
|
||
|
||
# 根据默认文件名后缀动态设置文件类型，避免固定为 zip 造成误导
|
||
if ext == '.zip':
|
||
file_types = ('ZIP 压缩包 (*.zip)', '所有文件 (*.*)')
|
||
elif ext in ('.xlsx', '.xls'):
|
||
file_types = ('Excel 文件 (*.xlsx;*.xls)', '所有文件 (*.*)')
|
||
elif ext == '.csv':
|
||
file_types = ('CSV 文件 (*.csv)', '所有文件 (*.*)')
|
||
elif ext == '.txt':
|
||
file_types = ('文本文件 (*.txt)', '所有文件 (*.*)')
|
||
elif ext == '.json':
|
||
file_types = ('JSON 文件 (*.json)', '所有文件 (*.*)')
|
||
else:
|
||
file_types = ('所有文件 (*.*)',)
|
||
|
||
result = self._window.create_file_dialog(
|
||
webview.SAVE_DIALOG,
|
||
save_filename=filename,
|
||
file_types=file_types
|
||
)
|
||
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 保存到用户选择的位置。"""
|
||
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 open_external_url(self, url):
|
||
if not url or not str(url).strip():
|
||
return {'success': False, 'error': '链接为空'}
|
||
target = str(url).strip()
|
||
try:
|
||
if sys.platform.startswith('win'):
|
||
os.startfile(target)
|
||
elif sys.platform == 'darwin':
|
||
subprocess.Popen(['open', target])
|
||
else:
|
||
subprocess.Popen(['xdg-open', target])
|
||
return {'success': True}
|
||
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 对象或数组'}
|
||
# if payload.get("type") == "delete-brand-run" and payload.get("data").get("taskId") in runing_task:
|
||
# return {'success': False, 'error': '已存在正在执行的删除品牌任务'}
|
||
# 判断当前店铺是否正在运行中
|
||
if payload.get("data").get("items"):
|
||
shop_name = payload["data"]["items"][0]["shopName"]
|
||
if shop_name in runing_shop:
|
||
return {'success': False, 'error': '当前店铺正在执行中,请等待店铺完成之后重试'}
|
||
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 get_detail_del(self,task_id):
|
||
"""获取正在执行的删除品牌任务详情"""
|
||
task_info = runing_task.get(task_id)
|
||
if not task_info:
|
||
return {'success': False, 'error': '任务不存在'}
|
||
return {'success': True, 'task_info': task_info}
|
||
|
||
def stop_task(self,task_id):
|
||
"""停止正在执行的删除品牌任务"""
|
||
task_info = runing_task.get(task_id)
|
||
if not task_info:
|
||
return {'success': False, 'error': '任务不存在'}
|
||
# 这里可以设置一个标志位，实际的删除品牌任务需要定期检查这个标志位来决定是否停止
|
||
task_info['stop_requested'] = True
|
||
return {'success': True, 'message': '已请求停止任务'}
|
||
|
||
|
||
def start_flask():
|
||
from app import run_app
|
||
run_app(host='127.0.0.1', port=PORT)
|
||
|
||
|
||
def start_task_monitor():
|
||
"""启动任务监控器（在独立线程中运行）"""
|
||
monitor = TaskMonitor()
|
||
try:
|
||
print(f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 任务监控线程已启动")
|
||
monitor.start()
|
||
except Exception:
|
||
import traceback
|
||
print(f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 任务监控线程异常退出: {traceback.format_exc()}")
|
||
|
||
|
||
def main():
|
||
if not os.path.exists(cache_path):
|
||
os.makedirs(cache_path,exist_ok=True)
|
||
|
||
# 启动 Flask 服务
|
||
t = threading.Thread(target=start_flask, daemon=True)
|
||
t.start()
|
||
|
||
# 启动任务监控线程
|
||
monitor_thread = threading.Thread(target=start_task_monitor, daemon=True)
|
||
monitor_thread.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.open_external_url,api.enqueue_json,api.get_detail_del)
|
||
webview.start(
|
||
debug=True,
|
||
storage_path=cache_path,
|
||
private_mode=False
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|