super
1018 lines
38 KiB
Python
1018 lines
38 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
import json
|
|
import mimetypes
|
|
import os
|
|
import re
|
|
import socket
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from tkinter import Tk, filedialog
|
|
from typing import Iterable
|
|
from urllib.parse import urljoin
|
|
from uuid import uuid4
|
|
|
|
try:
|
|
import requests
|
|
except ModuleNotFoundError as exc:
|
|
raise SystemExit(
|
|
"缺少依赖 requests。\n"
|
|
"请先执行：python -m pip install -r .\\desktop\\requirements.txt"
|
|
) from exc
|
|
|
|
try:
|
|
import webview
|
|
except ModuleNotFoundError as exc:
|
|
raise SystemExit(
|
|
"缺少依赖 pywebview。\n"
|
|
"请先执行：python -m pip install -r .\\desktop\\requirements.txt"
|
|
) from exc
|
|
|
|
try:
|
|
from openpyxl import load_workbook
|
|
except ModuleNotFoundError as exc:
|
|
raise SystemExit(
|
|
"缺少依赖 openpyxl。\n"
|
|
"请先执行：python -m pip install -r .\\desktop\\requirements.txt"
|
|
) from exc
|
|
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parents[1]
|
|
FRONTEND_DIR = ROOT_DIR / "frontend-vue"
|
|
DIST_DIR = FRONTEND_DIR / "dist"
|
|
INDEX_FILE = DIST_DIR / "index.html"
|
|
PROXY_PREFIXES = ("/api", "/login", "/logout", "/static")
|
|
TEMPLATE_STORE_PATH = ROOT_DIR / "desktop" / "txt_templates.json"
|
|
TXT_TEMPLATES = {
|
|
"uk_offer": {
|
|
"id": "uk_offer",
|
|
"label": "英国 Offer 模板",
|
|
"output_filename": "英国.txt",
|
|
"is_default": True,
|
|
"built_in": True,
|
|
"preamble_lines": ["TemplateType=Offer\tVersion=1.4"],
|
|
"header_columns": [
|
|
"sku",
|
|
"price",
|
|
"quantity",
|
|
"product-id",
|
|
"product-id-type",
|
|
"condition-type",
|
|
"condition-note",
|
|
"ASIN-hint",
|
|
"title",
|
|
"product-tax-code",
|
|
"operation-type",
|
|
"sale-price",
|
|
"sale-start-date",
|
|
"sale-end-date",
|
|
"leadtime-to-ship",
|
|
"launch-date",
|
|
"is-giftwrap-available",
|
|
"is-gift-message-available",
|
|
],
|
|
"blank_header_rows_after_schema": 1,
|
|
"required_source_columns": ["ASIN", "价格"],
|
|
"defaults": {
|
|
"quantity": "100",
|
|
"product-id-type": "ASIN",
|
|
"condition-type": "new",
|
|
"leadtime-to-ship": "4",
|
|
},
|
|
"field_mapping": {
|
|
"price": "价格",
|
|
"product-id": "ASIN",
|
|
},
|
|
}
|
|
}
|
|
|
|
|
|
def find_free_port() -> int:
|
|
with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
|
|
sock.bind(("127.0.0.1", 0))
|
|
return int(sock.getsockname()[1])
|
|
|
|
|
|
def ensure_dist_exists() -> None:
|
|
if not INDEX_FILE.exists():
|
|
raise FileNotFoundError(
|
|
f"未找到前端构建产物：{INDEX_FILE}\n"
|
|
"请先进入 frontend-vue 执行 npm run build。"
|
|
)
|
|
|
|
|
|
def resolve_url(base_url: str, maybe_relative_url: str) -> str:
|
|
if maybe_relative_url.startswith(("http://", "https://")):
|
|
return maybe_relative_url
|
|
return urljoin(base_url.rstrip("/") + "/", maybe_relative_url.lstrip("/"))
|
|
|
|
|
|
def run_file_dialog(fn):
|
|
root = Tk()
|
|
root.withdraw()
|
|
root.attributes("-topmost", True)
|
|
try:
|
|
return fn(root)
|
|
finally:
|
|
root.destroy()
|
|
|
|
|
|
def normalize_cell_text(value: object) -> str:
|
|
if value is None:
|
|
return ""
|
|
text = str(value).replace("\ufeff", "").replace("\u3000", " ")
|
|
text = text.replace("\r\n", " ").replace("\r", " ").replace("\n", " ").replace("\t", " ")
|
|
text = re.sub(r"\s+", " ", text)
|
|
return text.strip()
|
|
|
|
|
|
def build_output_path(output_dir: str, source_path: str) -> str:
|
|
source = Path(source_path)
|
|
candidate = Path(output_dir) / f"{source.stem}_cleaned{source.suffix}"
|
|
index = 2
|
|
while candidate.exists():
|
|
candidate = Path(output_dir) / f"{source.stem}_cleaned_{index}{source.suffix}"
|
|
index += 1
|
|
return str(candidate)
|
|
|
|
|
|
def build_named_output_path(output_dir: str, filename: str) -> str:
|
|
candidate = Path(output_dir) / filename
|
|
if not candidate.exists():
|
|
return str(candidate)
|
|
stem = candidate.stem
|
|
suffix = candidate.suffix
|
|
index = 2
|
|
while True:
|
|
next_candidate = Path(output_dir) / f"{stem}_{index}{suffix}"
|
|
if not next_candidate.exists():
|
|
return str(next_candidate)
|
|
index += 1
|
|
|
|
|
|
def should_keep_id(value: object, keep_integer_ids: bool, keep_underscore_ids: bool) -> bool:
|
|
text = normalize_cell_text(value)
|
|
if not text:
|
|
return False
|
|
if keep_integer_ids and text.isdigit():
|
|
return True
|
|
if keep_underscore_ids and re.fullmatch(r"\d+_\d+", text):
|
|
return True
|
|
return False
|
|
|
|
|
|
def generate_snowflake_like_id() -> int:
|
|
return int(time.time() * 1000)
|
|
|
|
|
|
def load_user_templates() -> tuple[dict[str, dict], str | None]:
|
|
if not TEMPLATE_STORE_PATH.exists():
|
|
return {}, None
|
|
try:
|
|
payload = json.loads(TEMPLATE_STORE_PATH.read_text(encoding="utf-8"))
|
|
templates = payload.get("templates") or {}
|
|
default_template_id = payload.get("default_template_id")
|
|
return templates, default_template_id
|
|
except Exception:
|
|
return {}, None
|
|
|
|
|
|
def save_user_templates(templates: dict[str, dict], default_template_id: str | None) -> None:
|
|
payload = {
|
|
"default_template_id": default_template_id,
|
|
"templates": templates,
|
|
}
|
|
TEMPLATE_STORE_PATH.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def build_effective_templates() -> tuple[dict[str, dict], str]:
|
|
builtins = {key: dict(value) for key, value in TXT_TEMPLATES.items()}
|
|
user_templates, persisted_default = load_user_templates()
|
|
templates = {**builtins, **user_templates}
|
|
default_template_id = persisted_default or next((key for key, value in builtins.items() if value.get("is_default")), "uk_offer")
|
|
for key, value in templates.items():
|
|
value["is_default"] = key == default_template_id
|
|
return templates, default_template_id
|
|
|
|
|
|
def recognize_txt_template(template_text: str, custom_name: str | None = None) -> dict | None:
|
|
lines = [line.rstrip("\n") for line in template_text.splitlines()]
|
|
non_empty_lines = [line for line in lines if line.strip()]
|
|
if len(non_empty_lines) < 2:
|
|
return None
|
|
|
|
first_line = non_empty_lines[0]
|
|
header_line = non_empty_lines[1]
|
|
header_columns = header_line.split("\t")
|
|
uk_template = TXT_TEMPLATES["uk_offer"]
|
|
|
|
if not first_line.startswith("TemplateType=Offer"):
|
|
return None
|
|
if header_columns != uk_template["header_columns"]:
|
|
return None
|
|
|
|
template_id = f"user_{int(time.time() * 1000)}"
|
|
output_filename = custom_name or Path("英国.txt").name
|
|
if not output_filename.endswith(".txt"):
|
|
output_filename = f"{output_filename}.txt"
|
|
|
|
return {
|
|
"id": template_id,
|
|
"label": custom_name or "自定义 Offer 模板",
|
|
"output_filename": output_filename,
|
|
"is_default": False,
|
|
"built_in": False,
|
|
"preamble_lines": [first_line],
|
|
"header_columns": header_columns,
|
|
"blank_header_rows_after_schema": 1,
|
|
"required_source_columns": ["ASIN", "价格"],
|
|
"defaults": dict(uk_template["defaults"]),
|
|
"field_mapping": dict(uk_template["field_mapping"]),
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class DesktopConfig:
|
|
mode: str
|
|
frontend_url: str
|
|
backend_url: str
|
|
host: str
|
|
port: int
|
|
width: int
|
|
height: int
|
|
title: str
|
|
|
|
|
|
class DesktopApi:
|
|
def __init__(self, backend_base_url: str) -> None:
|
|
self.app_base_url = backend_base_url.rstrip("/")
|
|
self.session = requests.Session()
|
|
|
|
def select_brand_xlsx_files(self) -> list[str]:
|
|
files = run_file_dialog(
|
|
lambda _root: filedialog.askopenfilenames(
|
|
title="选择 Excel 文件",
|
|
filetypes=[("Excel files", "*.xlsx")],
|
|
)
|
|
)
|
|
return list(files)
|
|
|
|
def select_clean_xlsx_files(self) -> list[str]:
|
|
files = run_file_dialog(
|
|
lambda _root: filedialog.askopenfilenames(
|
|
title="选择待清洗 Excel 文件",
|
|
filetypes=[("Excel files", "*.xlsx")],
|
|
)
|
|
)
|
|
return list(files)
|
|
|
|
def select_brand_folder(self) -> str | None:
|
|
folder = run_file_dialog(
|
|
lambda _root: filedialog.askdirectory(title="选择文件夹")
|
|
)
|
|
return folder or None
|
|
|
|
def select_clean_folder(self) -> str | None:
|
|
folder = run_file_dialog(
|
|
lambda _root: filedialog.askdirectory(title="选择待清洗文件夹")
|
|
)
|
|
return folder or None
|
|
|
|
def get_excel_headers(self, file_path: str) -> dict:
|
|
try:
|
|
workbook = load_workbook(file_path, read_only=True, data_only=True)
|
|
sheet = workbook[workbook.sheetnames[0]]
|
|
first_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), None)
|
|
workbook.close()
|
|
if not first_row:
|
|
return {"success": False, "error": "Excel 表头为空"}
|
|
|
|
headers: list[str] = []
|
|
seen: set[str] = set()
|
|
for cell in first_row:
|
|
if cell is None:
|
|
continue
|
|
raw_value = str(cell).replace("\ufeff", "").strip()
|
|
if not raw_value:
|
|
continue
|
|
value = raw_value.split("idASIN国家状态价格变体数量", 1)[0].strip() or raw_value
|
|
if value in seen:
|
|
continue
|
|
seen.add(value)
|
|
headers.append(value)
|
|
if value == "缩略图地址8":
|
|
break
|
|
|
|
if not headers:
|
|
return {"success": False, "error": "未读取到有效表头"}
|
|
return {"success": True, "headers": headers}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def get_excel_info(self, file_path: str) -> dict:
|
|
try:
|
|
workbook = load_workbook(file_path, read_only=True, data_only=True)
|
|
sheet = workbook[workbook.sheetnames[0]]
|
|
first_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), None)
|
|
if not first_row:
|
|
workbook.close()
|
|
return {"success": False, "error": "Excel 表头为空"}
|
|
|
|
headers: list[str] = []
|
|
seen: set[str] = set()
|
|
for cell in first_row:
|
|
if cell is None:
|
|
continue
|
|
raw_value = str(cell).replace("\ufeff", "").strip()
|
|
if not raw_value:
|
|
continue
|
|
value = raw_value.split("idASIN国家状态价格变体数量", 1)[0].strip() or raw_value
|
|
if value in seen:
|
|
continue
|
|
seen.add(value)
|
|
headers.append(value)
|
|
if value == "缩略图地址8":
|
|
break
|
|
|
|
total_rows = max(sheet.max_row - 1, 0)
|
|
workbook.close()
|
|
return {"success": True, "headers": headers, "total_rows": total_rows}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def list_txt_templates(self) -> list[dict]:
|
|
try:
|
|
response = self.session.get(
|
|
resolve_url(self.app_base_url, "/api/convert/templates"),
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
if not payload.get("success"):
|
|
return []
|
|
items = payload.get("data") or []
|
|
return [
|
|
{
|
|
"id": str(item.get("templateCode") or item.get("id") or ""),
|
|
"label": item.get("templateName") or "",
|
|
"output_filename": item.get("outputFilename") or "",
|
|
"is_default": bool(item.get("isDefault")),
|
|
"built_in": bool(item.get("builtIn")),
|
|
}
|
|
for item in items
|
|
]
|
|
except Exception:
|
|
templates, _default_template_id = build_effective_templates()
|
|
return [
|
|
{
|
|
"id": template["id"],
|
|
"label": template["label"],
|
|
"output_filename": template["output_filename"],
|
|
"is_default": template.get("is_default", False),
|
|
"built_in": template.get("built_in", False),
|
|
}
|
|
for template in templates.values()
|
|
]
|
|
|
|
def import_txt_template(self, name: str | None = None) -> dict:
|
|
file_path = run_file_dialog(
|
|
lambda _root: filedialog.askopenfilename(
|
|
title="选择 txt 模板文件",
|
|
filetypes=[("TXT files", "*.txt")],
|
|
)
|
|
)
|
|
if not file_path:
|
|
return {"success": False, "error": "用户取消"}
|
|
|
|
try:
|
|
try:
|
|
template_text = Path(file_path).read_text(encoding="utf-8")
|
|
except UnicodeDecodeError:
|
|
template_text = Path(file_path).read_text(encoding="utf-8-sig")
|
|
|
|
response = self.session.post(
|
|
resolve_url(self.app_base_url, "/api/convert/templates/import"),
|
|
json={
|
|
"templateName": (name or "").strip() or Path(file_path).stem,
|
|
"templateContent": template_text,
|
|
},
|
|
timeout=120,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
if not payload.get("success") or not payload.get("data"):
|
|
return {"success": False, "error": payload.get("message") or "导入失败"}
|
|
item = payload.get("data") or {}
|
|
return {
|
|
"success": True,
|
|
"template": {
|
|
"id": str(item.get("templateCode") or item.get("id") or ""),
|
|
"label": item.get("templateName") or "",
|
|
"output_filename": item.get("outputFilename") or "",
|
|
"is_default": bool(item.get("isDefault")),
|
|
"built_in": bool(item.get("builtIn")),
|
|
},
|
|
}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def set_default_txt_template(self, template_id: str) -> dict:
|
|
try:
|
|
response = self.session.post(
|
|
resolve_url(self.app_base_url, f"/api/convert/templates/{template_id}/default"),
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
return {"success": bool(payload.get("success")), "error": payload.get("message") if not payload.get("success") else None}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def delete_txt_template(self, template_id: str) -> dict:
|
|
try:
|
|
response = self.session.delete(
|
|
resolve_url(self.app_base_url, f"/api/convert/templates/{template_id}"),
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
return {"success": bool(payload.get("success")), "error": payload.get("message") if not payload.get("success") else None}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def split_excel_files(self, payload: dict) -> dict:
|
|
paths = payload.get("paths") or []
|
|
selected_columns = payload.get("selected_columns") or []
|
|
split_mode = payload.get("split_mode") or "rows_per_file"
|
|
rows_per_file = payload.get("rows_per_file")
|
|
parts = payload.get("parts")
|
|
|
|
if not isinstance(paths, list) or not paths:
|
|
return {"success": False, "error": "未选择待拆分文件"}
|
|
if not isinstance(selected_columns, list) or not selected_columns:
|
|
return {"success": False, "error": "请至少选择一列"}
|
|
if split_mode not in {"rows_per_file", "parts"}:
|
|
return {"success": False, "error": "拆分模式不合法"}
|
|
if split_mode == "rows_per_file" and (not isinstance(rows_per_file, int) or rows_per_file <= 0):
|
|
return {"success": False, "error": "每份条数不合法"}
|
|
if split_mode == "parts" and (not isinstance(parts, int) or parts <= 0):
|
|
return {"success": False, "error": "拆分份数不合法"}
|
|
|
|
try:
|
|
uploaded_files = []
|
|
for path in paths:
|
|
with open(path, "rb") as file_obj:
|
|
response = self.session.post(
|
|
resolve_url(self.app_base_url, "/api/files/upload"),
|
|
files={"file": (Path(path).name, file_obj)},
|
|
timeout=180,
|
|
)
|
|
response.raise_for_status()
|
|
upload_payload = response.json()
|
|
if not upload_payload.get("success") or not upload_payload.get("data"):
|
|
return {"success": False, "error": upload_payload.get("message") or "上传文件失败"}
|
|
upload_data = upload_payload.get("data") or {}
|
|
uploaded_files.append({
|
|
"fileKey": upload_data.get("fileKey"),
|
|
"originalFilename": upload_data.get("originalFilename") or Path(path).name,
|
|
})
|
|
|
|
run_response = self.session.post(
|
|
resolve_url(self.app_base_url, "/api/split/run"),
|
|
json={
|
|
"files": uploaded_files,
|
|
"selectedColumns": selected_columns,
|
|
"splitMode": split_mode,
|
|
"rowsPerFile": rows_per_file,
|
|
"parts": parts,
|
|
},
|
|
timeout=600,
|
|
)
|
|
run_response.raise_for_status()
|
|
run_payload = run_response.json()
|
|
if not run_payload.get("success") or not run_payload.get("data"):
|
|
return {"success": False, "error": run_payload.get("message") or "数据拆分失败"}
|
|
|
|
data = run_payload.get("data") or {}
|
|
result_items = []
|
|
for item in data.get("items") or []:
|
|
if item.get("success") and item.get("downloadUrl"):
|
|
result_items.append({
|
|
"result_id": item.get("resultId"),
|
|
"source_path": item.get("sourceFilename") or "",
|
|
"output_path": item.get("downloadUrl") or "",
|
|
"success": True,
|
|
"row_count": item.get("rowCount"),
|
|
})
|
|
else:
|
|
result_items.append({
|
|
"source_path": item.get("sourceFilename") or "",
|
|
"success": False,
|
|
"error": item.get("error") or "数据拆分失败",
|
|
})
|
|
|
|
return {
|
|
"success": True,
|
|
"summary": {
|
|
"total": data.get("total") or len(paths),
|
|
"success_count": data.get("successCount") or 0,
|
|
"failed_count": data.get("failedCount") or 0,
|
|
},
|
|
"items": result_items,
|
|
}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def delete_history_item(self, module_type: str, result_id: int) -> dict:
|
|
try:
|
|
response = self.session.delete(
|
|
resolve_url(self.app_base_url, f"/api/{module_type}/history/{result_id}"),
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
return {"success": bool(payload.get("success")), "error": payload.get("message") if not payload.get("success") else None}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def get_convert_history(self) -> dict:
|
|
try:
|
|
response = self.session.get(
|
|
resolve_url(self.app_base_url, "/api/convert/history"),
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
if not payload.get("success") or not payload.get("data"):
|
|
return {"success": False, "error": payload.get("message") or "获取历史失败"}
|
|
|
|
items = payload.get("data", {}).get("items") or []
|
|
return {
|
|
"success": True,
|
|
"items": [
|
|
{
|
|
"result_id": item.get("resultId"),
|
|
"source_path": item.get("sourceFilename") or "",
|
|
"output_path": item.get("downloadUrl") or "",
|
|
"success": bool(item.get("success", True)),
|
|
}
|
|
for item in items
|
|
],
|
|
}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def get_split_history(self) -> dict:
|
|
try:
|
|
response = self.session.get(
|
|
resolve_url(self.app_base_url, "/api/split/history"),
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
if not payload.get("success") or not payload.get("data"):
|
|
return {"success": False, "error": payload.get("message") or "获取历史失败"}
|
|
|
|
items = payload.get("data", {}).get("items") or []
|
|
return {
|
|
"success": True,
|
|
"items": [
|
|
{
|
|
"result_id": item.get("resultId"),
|
|
"source_path": item.get("sourceFilename") or "",
|
|
"output_path": item.get("downloadUrl") or "",
|
|
"success": bool(item.get("success", True)),
|
|
"row_count": item.get("rowCount"),
|
|
}
|
|
for item in items
|
|
],
|
|
}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def get_dedupe_history(self) -> dict:
|
|
try:
|
|
response = self.session.get(
|
|
resolve_url(self.app_base_url, "/api/dedupe/history"),
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
if not payload.get("success") or not payload.get("data"):
|
|
return {"success": False, "error": payload.get("message") or "获取历史失败"}
|
|
|
|
items = payload.get("data", {}).get("items") or []
|
|
return {
|
|
"success": True,
|
|
"items": [
|
|
{
|
|
"result_id": item.get("resultId"),
|
|
"source_path": item.get("sourceFilename") or "",
|
|
"output_path": item.get("downloadUrl") or "",
|
|
"success": bool(item.get("success", True)),
|
|
}
|
|
for item in items
|
|
],
|
|
}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def clean_excel_files(self, payload: dict) -> dict:
|
|
paths = payload.get("paths") or []
|
|
selected_columns = payload.get("selected_columns") or []
|
|
keep_integer_ids = bool(payload.get("keep_integer_ids", True))
|
|
keep_underscore_ids = bool(payload.get("keep_underscore_ids", True))
|
|
|
|
if not isinstance(paths, list) or not paths:
|
|
return {"success": False, "error": "未选择待处理文件"}
|
|
if not isinstance(selected_columns, list) or not selected_columns:
|
|
return {"success": False, "error": "请至少选择一列"}
|
|
if not keep_integer_ids and not keep_underscore_ids:
|
|
return {"success": False, "error": "请至少选择一种 ID 保留规则"}
|
|
|
|
try:
|
|
uploaded_files = []
|
|
for path in paths:
|
|
with open(path, "rb") as file_obj:
|
|
response = self.session.post(
|
|
resolve_url(self.app_base_url, "/api/files/upload"),
|
|
files={"file": (Path(path).name, file_obj)},
|
|
timeout=180,
|
|
)
|
|
response.raise_for_status()
|
|
upload_payload = response.json()
|
|
if not upload_payload.get("success") or not upload_payload.get("data"):
|
|
return {"success": False, "error": upload_payload.get("message") or "上传文件失败"}
|
|
upload_data = upload_payload.get("data") or {}
|
|
uploaded_files.append({
|
|
"fileKey": upload_data.get("fileKey"),
|
|
"originalFilename": upload_data.get("originalFilename") or Path(path).name,
|
|
})
|
|
|
|
run_response = self.session.post(
|
|
resolve_url(self.app_base_url, "/api/dedupe/run"),
|
|
json={
|
|
"files": uploaded_files,
|
|
"selectedColumns": selected_columns,
|
|
"keepIntegerIds": keep_integer_ids,
|
|
"keepUnderscoreIds": keep_underscore_ids,
|
|
},
|
|
timeout=600,
|
|
)
|
|
run_response.raise_for_status()
|
|
run_payload = run_response.json()
|
|
if not run_payload.get("success") or not run_payload.get("data"):
|
|
return {"success": False, "error": run_payload.get("message") or "去重失败"}
|
|
|
|
data = run_payload.get("data") or {}
|
|
result_items = []
|
|
for item in data.get("items") or []:
|
|
if item.get("success") and item.get("downloadUrl"):
|
|
result_items.append({
|
|
"source_path": item.get("sourceFilename") or "",
|
|
"output_path": item.get("downloadUrl") or "",
|
|
"success": True,
|
|
})
|
|
else:
|
|
result_items.append({
|
|
"source_path": item.get("sourceFilename") or "",
|
|
"success": False,
|
|
"error": item.get("error") or "去重失败",
|
|
})
|
|
|
|
return {
|
|
"success": True,
|
|
"summary": {
|
|
"total": data.get("total") or len(paths),
|
|
"success_count": data.get("successCount") or 0,
|
|
"failed_count": data.get("failedCount") or 0,
|
|
},
|
|
"items": result_items,
|
|
}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def convert_excel_to_uk_txt(self, payload: dict) -> dict:
|
|
paths = payload.get("paths") or []
|
|
template_id = payload.get("template_id") or "uk_offer"
|
|
|
|
if not isinstance(paths, list) or not paths:
|
|
return {"success": False, "error": "未选择待转换文件"}
|
|
|
|
try:
|
|
templates = self.list_txt_templates()
|
|
selected_template = next((item for item in templates if item.get("id") == template_id), None)
|
|
if not selected_template:
|
|
return {"success": False, "error": "模板不存在"}
|
|
|
|
uploaded_files = []
|
|
for path in paths:
|
|
with open(path, "rb") as file_obj:
|
|
response = self.session.post(
|
|
resolve_url(self.app_base_url, "/api/files/upload"),
|
|
files={"file": (Path(path).name, file_obj)},
|
|
timeout=180,
|
|
)
|
|
response.raise_for_status()
|
|
upload_payload = response.json()
|
|
if not upload_payload.get("success") or not upload_payload.get("data"):
|
|
return {"success": False, "error": upload_payload.get("message") or "上传文件失败"}
|
|
upload_data = upload_payload.get("data") or {}
|
|
uploaded_files.append({
|
|
"fileKey": upload_data.get("fileKey"),
|
|
"originalFilename": upload_data.get("originalFilename") or Path(path).name,
|
|
})
|
|
|
|
run_response = self.session.post(
|
|
resolve_url(self.app_base_url, "/api/convert/run"),
|
|
json={
|
|
"files": uploaded_files,
|
|
"templateId": str(template_id),
|
|
},
|
|
timeout=600,
|
|
)
|
|
run_response.raise_for_status()
|
|
run_payload = run_response.json()
|
|
if not run_payload.get("success") or not run_payload.get("data"):
|
|
return {"success": False, "error": run_payload.get("message") or "格式转换失败"}
|
|
|
|
data = run_payload.get("data") or {}
|
|
result_items = []
|
|
for item in data.get("items") or []:
|
|
if item.get("success") and item.get("downloadUrl"):
|
|
result_items.append({
|
|
"source_path": item.get("sourceFilename") or "",
|
|
"output_path": item.get("downloadUrl") or "",
|
|
"success": True,
|
|
})
|
|
else:
|
|
result_items.append({
|
|
"source_path": item.get("sourceFilename") or "",
|
|
"success": False,
|
|
"error": item.get("error") or "格式转换失败",
|
|
})
|
|
|
|
return {
|
|
"success": True,
|
|
"summary": {
|
|
"total": data.get("total") or len(paths),
|
|
"success_count": data.get("successCount") or 0,
|
|
"failed_count": data.get("failedCount") or 0,
|
|
},
|
|
"items": result_items,
|
|
}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def open_path(self, path: str) -> dict:
|
|
try:
|
|
if not path:
|
|
return {"success": False, "error": "路径为空"}
|
|
target = Path(path)
|
|
if not target.exists():
|
|
return {"success": False, "error": "路径不存在"}
|
|
os.startfile(str(target))
|
|
return {"success": True}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
def select_folder(self) -> str | None:
|
|
folder = run_file_dialog(
|
|
lambda _root: filedialog.askdirectory(title="选择文件夹")
|
|
)
|
|
return folder or None
|
|
|
|
def save_template_xlsx(self) -> dict:
|
|
return self._download_to_selected_path(
|
|
"/static/品牌文档格式_模板.xlsx",
|
|
"品牌文档格式_模板.xlsx",
|
|
)
|
|
|
|
def save_template_zip(self) -> dict:
|
|
return self._download_to_selected_path(
|
|
"/static/模板2-以文件夹方式上传.zip",
|
|
"模板2-以文件夹方式上传.zip",
|
|
)
|
|
|
|
def save_file_from_url(self, url: str, filename: str) -> dict:
|
|
return self._download_to_selected_path(url, filename)
|
|
|
|
def _download_to_selected_path(self, source_url: str, default_filename: str) -> dict:
|
|
path = run_file_dialog(
|
|
lambda _root: filedialog.asksaveasfilename(
|
|
title="选择保存位置",
|
|
initialfile=default_filename,
|
|
defaultextension=Path(default_filename).suffix or None,
|
|
)
|
|
)
|
|
if not path:
|
|
return {"success": False, "error": "用户取消"}
|
|
|
|
try:
|
|
resolved_url = resolve_url(self.app_base_url, source_url)
|
|
with self.session.get(resolved_url, timeout=120, stream=True) as response:
|
|
response.raise_for_status()
|
|
with open(path, "wb") as file:
|
|
for chunk in response.iter_content(chunk_size=1024 * 128):
|
|
if chunk:
|
|
file.write(chunk)
|
|
return {"success": True, "path": path}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"success": False, "error": str(exc)}
|
|
|
|
|
|
class DistRequestHandler(BaseHTTPRequestHandler):
|
|
backend_base_url = ""
|
|
|
|
def do_GET(self) -> None: # noqa: N802
|
|
self._dispatch()
|
|
|
|
def do_POST(self) -> None: # noqa: N802
|
|
self._dispatch()
|
|
|
|
def do_PUT(self) -> None: # noqa: N802
|
|
self._dispatch()
|
|
|
|
def do_DELETE(self) -> None: # noqa: N802
|
|
self._dispatch()
|
|
|
|
def do_OPTIONS(self) -> None: # noqa: N802
|
|
self._dispatch()
|
|
|
|
def _dispatch(self) -> None:
|
|
if self.path.startswith(PROXY_PREFIXES):
|
|
self._proxy_to_backend()
|
|
return
|
|
self._serve_dist_file()
|
|
|
|
def _proxy_to_backend(self) -> None:
|
|
target_url = resolve_url(self.backend_base_url, self.path)
|
|
body = self._read_request_body()
|
|
headers = {k: v for k, v in self.headers.items() if k.lower() != "host"}
|
|
|
|
try:
|
|
response = requests.request(
|
|
method=self.command,
|
|
url=target_url,
|
|
headers=headers,
|
|
data=body,
|
|
stream=True,
|
|
timeout=120,
|
|
allow_redirects=False,
|
|
)
|
|
except requests.RequestException as exc:
|
|
payload = f"后端代理请求失败：{exc}".encode("utf-8")
|
|
self.send_response(502)
|
|
self.send_header("Content-Type", "text/plain; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
return
|
|
|
|
self.send_response(response.status_code)
|
|
excluded_headers = {"content-encoding", "transfer-encoding", "connection"}
|
|
for key, value in response.headers.items():
|
|
if key.lower() in excluded_headers:
|
|
continue
|
|
self.send_header(key, value)
|
|
self.end_headers()
|
|
for chunk in response.iter_content(chunk_size=1024 * 128):
|
|
if chunk:
|
|
self.wfile.write(chunk)
|
|
|
|
def _serve_dist_file(self) -> None:
|
|
request_path = self.path.split("?", 1)[0].lstrip("/")
|
|
if not request_path:
|
|
file_path = INDEX_FILE
|
|
else:
|
|
file_path = (DIST_DIR / request_path).resolve()
|
|
if not str(file_path).startswith(str(DIST_DIR.resolve())):
|
|
self.send_error(403)
|
|
return
|
|
if not file_path.exists() or file_path.is_dir():
|
|
file_path = INDEX_FILE
|
|
|
|
try:
|
|
content = file_path.read_bytes()
|
|
except FileNotFoundError:
|
|
self.send_error(404)
|
|
return
|
|
|
|
mime_type, _ = mimetypes.guess_type(str(file_path))
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", mime_type or "application/octet-stream")
|
|
self.send_header("Content-Length", str(len(content)))
|
|
self.end_headers()
|
|
self.wfile.write(content)
|
|
|
|
def _read_request_body(self) -> bytes | None:
|
|
content_length = self.headers.get("Content-Length")
|
|
if not content_length:
|
|
return None
|
|
return self.rfile.read(int(content_length))
|
|
|
|
def log_message(self, format: str, *args) -> None: # noqa: A003
|
|
return
|
|
|
|
|
|
class DistServer:
|
|
def __init__(self, host: str, port: int, backend_url: str) -> None:
|
|
handler = type(
|
|
"ConfiguredDistRequestHandler",
|
|
(DistRequestHandler,),
|
|
{"backend_base_url": backend_url.rstrip("/")},
|
|
)
|
|
self.server = ThreadingHTTPServer((host, port), handler)
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.base_url = f"http://{host}:{port}"
|
|
|
|
def start(self) -> None:
|
|
self.thread.start()
|
|
|
|
def stop(self) -> None:
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=2)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Crawler Plugin 桌面端启动器")
|
|
parser.add_argument("--mode", choices=("dev", "dist"), default="dev")
|
|
parser.add_argument("--frontend-url", default="http://127.0.0.1:5173")
|
|
parser.add_argument("--backend-url", default="http://127.0.0.1:8000")
|
|
parser.add_argument("--host", default="127.0.0.1")
|
|
parser.add_argument("--port", type=int, default=0)
|
|
parser.add_argument("--width", type=int, default=1440)
|
|
parser.add_argument("--height", type=int, default=960)
|
|
parser.add_argument("--title", default="数富AI")
|
|
return parser
|
|
|
|
|
|
def parse_args() -> DesktopConfig:
|
|
args = build_parser().parse_args()
|
|
port = args.port or find_free_port()
|
|
return DesktopConfig(
|
|
mode=args.mode,
|
|
frontend_url=args.frontend_url.rstrip("/"),
|
|
backend_url=args.backend_url.rstrip("/"),
|
|
host=args.host,
|
|
port=port,
|
|
width=args.width,
|
|
height=args.height,
|
|
title=args.title,
|
|
)
|
|
|
|
|
|
def ensure_dev_server_ready(frontend_url: str) -> None:
|
|
try:
|
|
response = requests.get(frontend_url, timeout=5)
|
|
response.raise_for_status()
|
|
except requests.RequestException as exc:
|
|
raise SystemExit(
|
|
"Dev server is not reachable.\n"
|
|
f"Expected frontend URL: {frontend_url}\n"
|
|
"Start Vite first with:\n"
|
|
" cd frontend-vue\n"
|
|
" npm run dev"
|
|
) from exc
|
|
|
|
|
|
def main() -> None:
|
|
config = parse_args()
|
|
dist_server = None
|
|
|
|
if config.mode == "dist":
|
|
ensure_dist_exists()
|
|
dist_server = DistServer(config.host, config.port, config.backend_url)
|
|
dist_server.start()
|
|
app_url = dist_server.base_url
|
|
else:
|
|
ensure_dev_server_ready(config.frontend_url)
|
|
app_url = config.frontend_url
|
|
|
|
api = DesktopApi(config.backend_url)
|
|
window = webview.create_window(
|
|
title=config.title,
|
|
url=app_url,
|
|
js_api=api,
|
|
width=config.width,
|
|
height=config.height,
|
|
min_size=(1200, 760),
|
|
)
|
|
|
|
try:
|
|
webview.start()
|
|
finally:
|
|
if dist_server is not None:
|
|
dist_server.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|