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()