1007 lines
36 KiB
Python
1007 lines
36 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, app_base_url: str) -> None:
|
||
self.app_base_url = app_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]:
|
||
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:
|
||
template_text = Path(file_path).read_text(encoding="utf-8")
|
||
except UnicodeDecodeError:
|
||
template_text = Path(file_path).read_text(encoding="utf-8-sig")
|
||
except Exception as exc: # noqa: BLE001
|
||
return {"success": False, "error": str(exc)}
|
||
|
||
template = recognize_txt_template(template_text, (name or "").strip() or None)
|
||
if not template:
|
||
return {"success": False, "error": "无法识别模板"}
|
||
|
||
user_templates, default_template_id = load_user_templates()
|
||
user_templates[template["id"]] = template
|
||
save_user_templates(user_templates, default_template_id)
|
||
|
||
return {
|
||
"success": True,
|
||
"template": {
|
||
"id": template["id"],
|
||
"label": template["label"],
|
||
"output_filename": template["output_filename"],
|
||
"is_default": False,
|
||
"built_in": False,
|
||
},
|
||
}
|
||
|
||
def set_default_txt_template(self, template_id: str) -> dict:
|
||
templates, _default_template_id = build_effective_templates()
|
||
if template_id not in templates:
|
||
return {"success": False, "error": "模板不存在"}
|
||
|
||
user_templates, _ = load_user_templates()
|
||
save_user_templates(user_templates, template_id)
|
||
return {"success": True}
|
||
|
||
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")
|
||
output_dir = payload.get("output_dir") or ""
|
||
|
||
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": "拆分份数不合法"}
|
||
if not output_dir:
|
||
return {"success": False, "error": "未选择输出目录"}
|
||
|
||
output_dir_path = Path(output_dir)
|
||
if not output_dir_path.exists() or not output_dir_path.is_dir():
|
||
return {"success": False, "error": "输出目录不存在"}
|
||
|
||
items: list[dict] = []
|
||
success_count = 0
|
||
failed_count = 0
|
||
|
||
for path in paths:
|
||
try:
|
||
workbook = load_workbook(path)
|
||
sheet = workbook[workbook.sheetnames[0]]
|
||
header_cells = next(sheet.iter_rows(min_row=1, max_row=1))
|
||
header_map: dict[str, int] = {}
|
||
for index, cell in enumerate(header_cells):
|
||
value = normalize_cell_text(cell.value)
|
||
if not value:
|
||
continue
|
||
value = value.split("idASIN国家状态价格变体数量", 1)[0].strip() or value
|
||
if value in header_map:
|
||
continue
|
||
header_map[value] = index
|
||
if value == "缩略图地址8":
|
||
break
|
||
|
||
missing = [column for column in selected_columns if column not in header_map]
|
||
if missing:
|
||
workbook.close()
|
||
items.append({
|
||
"source_path": path,
|
||
"success": False,
|
||
"error": f"缺少列：{'、'.join(missing)}",
|
||
})
|
||
failed_count += 1
|
||
continue
|
||
|
||
rows_data: list[list[str]] = []
|
||
for row in sheet.iter_rows(min_row=2, values_only=True):
|
||
projected = [
|
||
normalize_cell_text(row[header_map[column]]) if header_map[column] < len(row) else ""
|
||
for column in selected_columns
|
||
]
|
||
rows_data.append(projected)
|
||
|
||
total_rows = len(rows_data)
|
||
if total_rows == 0:
|
||
workbook.close()
|
||
items.append({
|
||
"source_path": path,
|
||
"success": False,
|
||
"error": "没有可拆分的数据行",
|
||
})
|
||
failed_count += 1
|
||
continue
|
||
|
||
if split_mode == "rows_per_file":
|
||
chunk_size = rows_per_file
|
||
chunks = [rows_data[i : i + chunk_size] for i in range(0, total_rows, chunk_size)]
|
||
else:
|
||
actual_parts = min(parts, total_rows)
|
||
base = total_rows // actual_parts
|
||
remainder = total_rows % actual_parts
|
||
chunks = []
|
||
start = 0
|
||
for idx in range(actual_parts):
|
||
size = base + (1 if idx < remainder else 0)
|
||
end = start + size
|
||
chunks.append(rows_data[start:end])
|
||
start = end
|
||
|
||
from openpyxl import Workbook
|
||
|
||
source = Path(path)
|
||
for index, chunk in enumerate(chunks, start=1):
|
||
output_workbook = Workbook()
|
||
output_sheet = output_workbook.active
|
||
output_sheet.title = sheet.title
|
||
output_sheet.append(selected_columns)
|
||
for row_values in chunk:
|
||
output_sheet.append(row_values)
|
||
|
||
output_path = build_named_output_path(output_dir, f"{source.stem}_split_{index}.xlsx")
|
||
output_workbook.save(output_path)
|
||
output_workbook.close()
|
||
|
||
items.append({
|
||
"source_path": path,
|
||
"output_path": output_path,
|
||
"success": True,
|
||
"row_count": len(chunk),
|
||
})
|
||
|
||
workbook.close()
|
||
success_count += 1
|
||
except Exception as exc: # noqa: BLE001
|
||
items.append({
|
||
"source_path": path,
|
||
"success": False,
|
||
"error": str(exc),
|
||
})
|
||
failed_count += 1
|
||
|
||
return {
|
||
"success": True,
|
||
"summary": {
|
||
"total": len(paths),
|
||
"success_count": success_count,
|
||
"failed_count": failed_count,
|
||
"output_dir": str(output_dir_path),
|
||
},
|
||
"items": items,
|
||
}
|
||
|
||
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))
|
||
output_dir = payload.get("output_dir") or ""
|
||
|
||
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 保留规则"}
|
||
if not output_dir:
|
||
return {"success": False, "error": "未选择输出目录"}
|
||
|
||
output_dir_path = Path(output_dir)
|
||
if not output_dir_path.exists() or not output_dir_path.is_dir():
|
||
return {"success": False, "error": "输出目录不存在"}
|
||
|
||
items: list[dict] = []
|
||
success_count = 0
|
||
failed_count = 0
|
||
|
||
for path in paths:
|
||
try:
|
||
workbook = load_workbook(path)
|
||
sheet = workbook[workbook.sheetnames[0]]
|
||
header_cells = next(sheet.iter_rows(min_row=1, max_row=1))
|
||
headers: list[str] = []
|
||
header_map: dict[str, int] = {}
|
||
for index, cell in enumerate(header_cells):
|
||
value = normalize_cell_text(cell.value)
|
||
if not value:
|
||
continue
|
||
value = value.split("idASIN国家状态价格变体数量", 1)[0].strip() or value
|
||
if value in header_map:
|
||
continue
|
||
header_map[value] = index
|
||
headers.append(value)
|
||
if value == "缩略图地址8":
|
||
break
|
||
|
||
missing = [column for column in selected_columns if column not in header_map]
|
||
if missing:
|
||
workbook.close()
|
||
items.append({
|
||
"source_path": path,
|
||
"success": False,
|
||
"error": f"缺少列：{'、'.join(missing)}",
|
||
})
|
||
failed_count += 1
|
||
continue
|
||
|
||
from openpyxl import Workbook
|
||
|
||
output_workbook = Workbook()
|
||
output_sheet = output_workbook.active
|
||
output_sheet.title = sheet.title
|
||
output_sheet.append(selected_columns)
|
||
|
||
id_column_index = header_map.get("id")
|
||
|
||
for row in sheet.iter_rows(min_row=2, values_only=True):
|
||
original_values = list(row)
|
||
if id_column_index is not None:
|
||
id_value = original_values[id_column_index] if id_column_index < len(original_values) else None
|
||
if not should_keep_id(id_value, keep_integer_ids, keep_underscore_ids):
|
||
continue
|
||
|
||
cleaned_values = [
|
||
normalize_cell_text(original_values[header_map[column]])
|
||
if header_map[column] < len(original_values)
|
||
else ""
|
||
for column in selected_columns
|
||
]
|
||
output_sheet.append(cleaned_values)
|
||
|
||
output_path = build_output_path(output_dir, path)
|
||
output_workbook.save(output_path)
|
||
output_workbook.close()
|
||
workbook.close()
|
||
|
||
items.append({
|
||
"source_path": path,
|
||
"output_path": output_path,
|
||
"success": True,
|
||
})
|
||
success_count += 1
|
||
except Exception as exc: # noqa: BLE001
|
||
items.append({
|
||
"source_path": path,
|
||
"success": False,
|
||
"error": str(exc),
|
||
})
|
||
failed_count += 1
|
||
|
||
return {
|
||
"success": True,
|
||
"summary": {
|
||
"total": len(paths),
|
||
"success_count": success_count,
|
||
"failed_count": failed_count,
|
||
"output_dir": str(output_dir_path),
|
||
},
|
||
"items": items,
|
||
}
|
||
|
||
def convert_excel_to_uk_txt(self, payload: dict) -> dict:
|
||
paths = payload.get("paths") or []
|
||
output_dir = payload.get("output_dir") or ""
|
||
template_id = payload.get("template_id") or "uk_offer"
|
||
|
||
if not isinstance(paths, list) or not paths:
|
||
return {"success": False, "error": "未选择待转换文件"}
|
||
if not output_dir:
|
||
return {"success": False, "error": "未选择输出目录"}
|
||
|
||
templates, _default_template_id = build_effective_templates()
|
||
template = templates.get(template_id)
|
||
if not template:
|
||
return {"success": False, "error": "模板不存在"}
|
||
|
||
output_dir_path = Path(output_dir)
|
||
if not output_dir_path.exists() or not output_dir_path.is_dir():
|
||
return {"success": False, "error": "输出目录不存在"}
|
||
|
||
items: list[dict] = []
|
||
success_count = 0
|
||
failed_count = 0
|
||
|
||
for path in paths:
|
||
try:
|
||
workbook = load_workbook(path, read_only=True, data_only=True)
|
||
sheet = workbook[workbook.sheetnames[0]]
|
||
header_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), None)
|
||
if not header_row:
|
||
workbook.close()
|
||
items.append({"source_path": path, "success": False, "error": 'Excel 表头为空'})
|
||
failed_count += 1
|
||
continue
|
||
|
||
header_map: dict[str, int] = {}
|
||
for index, cell in enumerate(header_row):
|
||
value = normalize_cell_text(cell)
|
||
if not value or value in header_map:
|
||
continue
|
||
header_map[value] = index
|
||
|
||
required = template["required_source_columns"]
|
||
missing = [column for column in required if column not in header_map]
|
||
if missing:
|
||
workbook.close()
|
||
items.append({
|
||
"source_path": path,
|
||
"success": False,
|
||
"error": f"缺少列：{'、'.join(missing)}",
|
||
})
|
||
failed_count += 1
|
||
continue
|
||
|
||
rows: list[str] = []
|
||
rows.extend(template["preamble_lines"])
|
||
rows.append('\t'.join(template["header_columns"]))
|
||
for _ in range(template.get("blank_header_rows_after_schema", 0)):
|
||
rows.append('\t' * (len(template["header_columns"]) - 1))
|
||
|
||
batch_id = generate_snowflake_like_id()
|
||
row_index = 1
|
||
for row in sheet.iter_rows(min_row=2, values_only=True):
|
||
line_values: list[str] = []
|
||
asin = normalize_cell_text(row[header_map['ASIN']] if header_map['ASIN'] < len(row) else '')
|
||
if not asin:
|
||
continue
|
||
|
||
for column in template["header_columns"]:
|
||
if column == 'sku':
|
||
line_values.append(f'{batch_id}-{row_index}')
|
||
continue
|
||
if column in template["field_mapping"]:
|
||
source_column = template["field_mapping"][column]
|
||
value = normalize_cell_text(row[header_map[source_column]] if header_map[source_column] < len(row) else '')
|
||
line_values.append(value)
|
||
continue
|
||
if column in template["defaults"]:
|
||
line_values.append(str(template["defaults"][column]))
|
||
continue
|
||
line_values.append('')
|
||
|
||
rows.append('\t'.join(line_values))
|
||
row_index += 1
|
||
|
||
output_path = build_named_output_path(output_dir, template["output_filename"])
|
||
Path(output_path).write_text('\n'.join(rows), encoding='utf-8')
|
||
workbook.close()
|
||
|
||
items.append({
|
||
"source_path": path,
|
||
"output_path": output_path,
|
||
"success": True,
|
||
})
|
||
success_count += 1
|
||
except Exception as exc: # noqa: BLE001
|
||
items.append({
|
||
"source_path": path,
|
||
"success": False,
|
||
"error": str(exc),
|
||
})
|
||
failed_count += 1
|
||
|
||
return {
|
||
"success": True,
|
||
"summary": {
|
||
"total": len(paths),
|
||
"success_count": success_count,
|
||
"failed_count": failed_count,
|
||
"output_dir": str(output_dir_path),
|
||
},
|
||
"items": items,
|
||
}
|
||
|
||
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(app_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()
|