"""中国邮政揽收系统 - Excel 自动填单工具 主程序入口，提供 Tkinter GUI 界面。 """ import asyncio import json import os import sys import threading from datetime import datetime from tkinter import ( BOTH, BOTTOM, DISABLED, END, HORIZONTAL, LEFT, NORMAL, RIGHT, VERTICAL, X, Y, BooleanVar, Button, Checkbutton, Entry, Frame, Label, LabelFrame, Listbox, Scrollbar, StringVar, Text, Tk, Toplevel, filedialog, messagebox, ttk, ) from excel_reader import read_input_excel, validate_required_columns from field_mapping import get_defaults, get_field_mapping, get_fixed_values, get_timing_config, load_config from log import user as _log_user # ─── 全局引擎引用 ────────────────────────────────────────── _engine = None _loop = None def _get_writable_path(name: str) -> str: """获取可写文件路径（config.json, session.json）。 开发模式：脚本所在目录。 打包模式：exe 所在目录（_MEIPASS 是只读的）。 """ if getattr(sys, 'frozen', False): return os.path.join(os.path.dirname(sys.executable), name) else: return os.path.join(os.path.dirname(os.path.abspath(__file__)), name) def _get_config_path(): return _get_writable_path("config.json") def _get_session_path(): return _get_writable_path("session.json") def _resource_path(relative_path): """获取只读资源文件路径（pickup.ico 等）。 开发模式：脚本所在目录。 打包模式：_MEIPASS（PyInstaller 解压临时目录）。 """ if getattr(sys, 'frozen', False): return os.path.join(sys._MEIPASS, relative_path) else: return os.path.join(os.path.dirname(os.path.abspath(__file__)), relative_path) # ─── 异步引擎后台线程 ────────────────────────────────────── def _run_async(coro): """在新线程中运行异步协程。""" global _loop _loop = asyncio.new_event_loop() asyncio.set_event_loop(_loop) _loop.run_until_complete(coro) def _run_in_thread(coro): """在后台线程启动异步任务。""" t = threading.Thread(target=_run_async, args=(coro,), daemon=True) t.start() return t # ─── 主窗口 ──────────────────────────────────────────────── class App: def __init__(self): self.root = Tk() self.root.title("自动填单工具 v2.0") self.root.geometry("900x650") self.root.minsize(800, 550) if sys.platform == "win32": self.root.iconbitmap(_resource_path("pickup.ico")) else: # 在 Linux 上什么都不做，或者用 .xbm 文件 # self.root.iconbitmap(_resource_path("pickup.xbm")) pass self.config = {} self.excel_path = StringVar() self.delay_var = StringVar(value="500") self.status_var = StringVar(value="就绪") self.engine_status_var = StringVar(value="未启动") self.running = False self.paused = False self.results = [] # 固定值编辑框变量 self.fv_product = StringVar() self.fv_customer = StringVar() self.fv_destination = StringVar() self.fv_force_destination = BooleanVar() self.fv_district_code = StringVar() self.fv_force_district = BooleanVar() self.fv_sender_name = StringVar() self.fv_force_sender_name = BooleanVar() self.fv_sender_addr = StringVar() self.fv_force_sender_addr = BooleanVar() # 搜索基础设施（懒加载） self._search_bridge = None self._search_sess = None self._search_api = None self._loading_config = False self._search_debounce = None # 产品/客户预缓存（batch 期间复用） self._product_cached_choice = None self._product_cached_detail = None self._customer_cached_choice = None self._customer_cached_detail = None self._sender_realname_cached = None self._build_ui() # 监听固定值输入框变化，触发产品/客户搜索 self.fv_product.trace_add("write", self._on_product_change) self.fv_customer.trace_add("write", self._on_customer_change) # 关闭窗口时清理 self.root.protocol("WM_DELETE_WINDOW", self._on_close) # 日志统一输出到 stderr（终端 / 控制台窗口可见） self._load_config() def _build_ui(self): """构建界面。""" # ── 配置区 ── config_frame = LabelFrame(self.root, text="配置", padx=10, pady=5) config_frame.pack(fill=X, padx=10, pady=5) # 引擎状态 row_status = Frame(config_frame) row_status.pack(fill=X, pady=2) Label(row_status, text="引擎状态:", width=10, anchor="e").pack(side="left") self.status_label = Label(row_status, textvariable=self.engine_status_var, fg="gray") self.status_label.pack(side="left", padx=5) # Excel 文件 row2 = Frame(config_frame) row2.pack(fill=X, pady=2) Label(row2, text="Excel文件:", width=10, anchor="e").pack(side="left") self.file_label = Label(row2, text="未选择", fg="gray", anchor="w") self.file_label.pack(side="left", fill=X, expand=True, padx=5) Button(row2, text="选择文件", command=self._pick_file, width=10).pack( side="right" ) # 延时设置 row3 = Frame(config_frame) row3.pack(fill=X, pady=2) Label(row3, text="间隔(毫秒):", width=10, anchor="e").pack(side="left") Entry(row3, textvariable=self.delay_var, width=8).pack(side="left", padx=5) Label(row3, text="(每单提交之间的等待时间)", fg="gray").pack(side="left") # 按钮 btn_frame = Frame(config_frame) btn_frame.pack(fill=X, pady=5) self.login_btn = Button( btn_frame, text="🔑 登录", command=self._do_login, width=8 ) self.login_btn.pack(side="left", padx=2) self.start_btn = Button( btn_frame, text="▶ 开始处理", command=self._start, bg="#4CAF50", fg="white", width=12, height=1 ) self.start_btn.pack(side="left", padx=2) self.pause_btn = Button( btn_frame, text="⏸ 暂停", command=self._pause, width=10, state=DISABLED ) self.pause_btn.pack(side="left", padx=2) self.stop_btn = Button( btn_frame, text="⏹ 停止", command=self._stop, width=10, state=DISABLED ) self.stop_btn.pack(side="left", padx=2) # 状态 Label( btn_frame, textvariable=self.status_var, fg="blue" ).pack(side="right", padx=10) # ── 固定值设置 ── fixed_frame = LabelFrame(self.root, text="固定值设置", padx=10, pady=5) fixed_frame.pack(fill=X, padx=10, pady=5) # Row 0: 产品 | 客户 r0 = Frame(fixed_frame) r0.pack(fill=X, pady=2) Label(r0, text="产品:", width=10, anchor="e").pack(side="left") self.fv_product_entry = Entry(r0, textvariable=self.fv_product, width=20) self.fv_product_entry.pack(side="left", padx=5) Label(r0, text="客户:", width=10, anchor="e").pack(side="left", padx=(30, 0)) self.fv_customer_entry = Entry(r0, textvariable=self.fv_customer, width=20) self.fv_customer_entry.pack(side="left", padx=5) # Row 1: 寄达地 | 无影响 r1 = Frame(fixed_frame) r1.pack(fill=X, pady=2) Label(r1, text="寄达地:", width=10, anchor="e").pack(side="left") Entry(r1, textvariable=self.fv_destination, width=20).pack(side="left", padx=5) Checkbutton(r1, text="强制覆盖", variable=self.fv_force_destination).pack(side="left", padx=5) Label(r1, text="无影响", fg="gray", font=("", 8)).pack(side="left") # Row 2: 寄达地区划代码 | 强制覆盖 | 直接决定寄达地 r2 = Frame(fixed_frame) r2.pack(fill=X, pady=2) Label(r2, text="寄达地区划代码:", width=12, anchor="e").pack(side="left") Entry(r2, textvariable=self.fv_district_code, width=20).pack(side="left", padx=5) Checkbutton(r2, text="强制覆盖", variable=self.fv_force_district).pack(side="left", padx=5) Label(r2, text="（直接决定寄达地）", fg="gray", font=("", 8)).pack(side="left") # Row 3: 寄件人姓名 | 强制覆盖 r3 = Frame(fixed_frame) r3.pack(fill=X, pady=2) Label(r3, text="寄件人姓名:", width=12, anchor="e").pack(side="left") Entry(r3, textvariable=self.fv_sender_name, width=20).pack(side="left", padx=5) Checkbutton(r3, text="强制覆盖", variable=self.fv_force_sender_name).pack(side="left", padx=5) # Row 4: 寄件地址 | 强制覆盖 r4 = Frame(fixed_frame) r4.pack(fill=X, pady=2) Label(r4, text="寄件地址:", width=12, anchor="e").pack(side="left") Entry(r4, textvariable=self.fv_sender_addr, width=20).pack(side="left", padx=5) Checkbutton(r4, text="强制覆盖", variable=self.fv_force_sender_addr).pack(side="left", padx=5) # ── 进度条 ── progress_frame = Frame(self.root) progress_frame.pack(fill=X, padx=10, pady=2) self.progress = ttk.Progressbar( progress_frame, mode="determinate", length=100 ) self.progress.pack(fill=X) self.progress_label = Label(progress_frame, text="", fg="gray") self.progress_label.pack() # ── 提交结果表格 ── result_frame = LabelFrame(self.root, text="提交结果", padx=10, pady=5) result_frame.pack(fill="both", expand=True, padx=10, pady=5) columns = ("单号", "处理状态", "重量", "产品", "客户", "寄达地区划代码", "寄件人姓名", "寄件地址") self.result_tree = ttk.Treeview(result_frame, columns=columns, show="headings", height=8) col_widths = [140, 80, 60, 110, 110, 110, 80, 160] for col, width in zip(columns, col_widths): self.result_tree.heading(col, text=col) self.result_tree.column(col, width=width, anchor="center") self.result_tree.pack(side="left", fill="both", expand=True) scroll_y = Scrollbar(result_frame, orient=VERTICAL, command=self.result_tree.yview) scroll_y.pack(side=RIGHT, fill=Y) self.result_tree.configure(yscrollcommand=scroll_y.set) # ─── 配置管理 ─────────────────────────────────────────── def _load_config(self): """加载配置文件。首次运行时从 _MEIPASS 复制到 exe 目录。""" try: self._loading_config = True config_path = _get_config_path() # 打包模式首次运行：从 _MEIPASS 复制 config.json 到可写目录 if getattr(sys, 'frozen', False) and not os.path.exists(config_path): bundled = os.path.join(sys._MEIPASS, "config.json") if os.path.exists(bundled): import shutil shutil.copy2(bundled, config_path) self.config = load_config(config_path) timing = get_timing_config(self.config) delay_ms = timing.get("delay_between_rows_ms", 500) self.delay_var.set(str(delay_ms)) # 加载固定值到编辑框 fv = self.config.get("fixed_values", {}) self.fv_product.set(fv.get("产品名称", {}).get("value", "")) self.fv_customer.set(fv.get("客户名称", {}).get("value", "")) self.fv_destination.set(fv.get("寄达地", {}).get("value", "")) self.fv_force_destination.set(fv.get("寄达地", {}).get("force", False)) self.fv_district_code.set(fv.get("寄达地区划", {}).get("value", "")) self.fv_force_district.set(fv.get("寄达地区划", {}).get("force", False)) self.fv_sender_name.set(fv.get("寄件人", {}).get("value", "")) self.fv_force_sender_name.set(fv.get("寄件人", {}).get("force", False)) self.fv_sender_addr.set(fv.get("寄件地址", {}).get("value", "")) self.fv_force_sender_addr.set(fv.get("寄件地址", {}).get("force", False)) self._loading_config = False # 检查已保存的登录会话 self._check_session_status() except Exception as e: self._log(f"[error] 配置加载失败: {e}") messagebox.showwarning("配置错误", f"配置文件加载失败:\n{e}") def _check_session_status(self): """检查已保存的登录会话状态并更新 UI。""" try: from login_helper import load_session_info info = load_session_info() if info and len(info.get("cookies", [])) > 0: self.engine_status_var.set(f"已登录 ({len(info['cookies'])} cookies)") self.status_label.configure(fg="green") else: self.engine_status_var.set("未登录") self.status_label.configure(fg="orange") except Exception: self.engine_status_var.set("未登录") self.status_label.configure(fg="orange") # ─── 固定值保存 ──────────────────────────────────────── def _save_fixed_from_entries(self): """从编辑框保存固定值到 config.json。""" new_fixed = { "产品名称": {"value": self.fv_product.get().strip(), "force": False}, "客户名称": {"value": self.fv_customer.get().strip(), "force": False}, "寄达地": {"value": self.fv_destination.get().strip(), "force": self.fv_force_destination.get()}, "寄达地区划": {"value": self.fv_district_code.get().strip(), "force": self.fv_force_district.get()}, "寄件人": {"value": self.fv_sender_name.get().strip(), "force": self.fv_force_sender_name.get()}, "寄件地址": {"value": self.fv_sender_addr.get().strip(), "force": self.fv_force_sender_addr.get()}, } self.config["fixed_values"] = new_fixed config_path = _get_config_path() with open(config_path, "r", encoding="utf-8") as f: cfg = json.load(f) cfg["fixed_values"] = new_fixed with open(config_path, "w", encoding="utf-8") as f: json.dump(cfg, f, ensure_ascii=False, indent=4) def _add_result_row(self, result): """添加一行到结果表格。""" values = ( result.get("单号", ""), result.get("status", ""), result.get("重量", ""), result.get("产品", ""), result.get("客户", ""), result.get("寄达地区划代码", ""), result.get("寄件人姓名", ""), result.get("寄件地址", ""), ) self.result_tree.insert("", "end", values=values) # ─── 产品/客户搜索 ────────────────────────────────────── def _init_search_infra(self): """在登录后（或重新登录后）初始化搜索基础设施（token bridge + session + api）。""" # 销毁旧的基础设施（可能是过期 cookies 的 session） if self._search_bridge: self._search_bridge.stop() self._search_bridge = None self._search_sess = None self._search_api = None try: from pickup_api import TokenBridge, PickupSession, PickupAPI config = load_config(_get_config_path()) url = config.get("system", {}).get("url", "https://10.4.188.1/pickup-web/a/postwaybill/main") self._search_bridge = TokenBridge() self._search_bridge.start() self._search_sess = PickupSession(url) session_path = _get_session_path() if not self._search_sess.load_session(session_path): self._search_bridge.stop() self._search_bridge = None self._search_sess = None self._log("[error] 搜索初始化失败：未找到登录会话") return resources = self._search_sess.fetch_page_resources() minimal_html = ( f"" ) self._search_bridge.initialize(minimal_html, resources["scripts"], resources["base_url"]) self._search_api = PickupAPI(self._search_sess, self._search_bridge) try: self._search_api.query_ecommerce("B") self._search_api.query_deliver_date() self._search_api.check_sheet_warning() except Exception: pass except Exception as e: self._log(f"[error] 搜索初始化失败: {e}") if self._search_bridge: self._search_bridge.stop() self._search_bridge = None self._search_sess = None def _on_product_change(self, *args): if self._loading_config: return if self._search_debounce: self.root.after_cancel(self._search_debounce) keyword = self.fv_product.get().strip() if not keyword or len(keyword) < 1: return self._search_debounce = self.root.after(300, lambda: self._do_product_search(keyword)) def _on_customer_change(self, *args): if self._loading_config: return if self._search_debounce: self.root.after_cancel(self._search_debounce) keyword = self.fv_customer.get().strip() if not keyword or len(keyword) < 1: return self._search_debounce = self.root.after(300, lambda: self._do_customer_search(keyword)) def _do_product_search(self, keyword: str): if self._search_api is None: return try: candidates = self._search_api.query_product(keyword) if not candidates: return names = [c.get("productName", "") for c in candidates] self._show_search_dropdown( self.fv_product_entry, names, lambda idx: self._on_product_selected(candidates[idx]), ) except Exception as e: self._log(f"[error] 产品搜索失败: {e}") def _do_customer_search(self, keyword: str): if self._search_api is None: return try: candidates = self._search_api.query_customer(keyword) if not candidates: return names = [ c.get("customerFullName", c.get("customerName", "")) for c in candidates ] self._show_search_dropdown( self.fv_customer_entry, names, lambda idx: self._on_customer_selected(candidates[idx]), ) except Exception as e: self._log(f"[error] 客户搜索失败: {e}") def _show_search_dropdown(self, entry: Entry, options: list[str], on_select): """在输入框正下方弹出下拉候选列表。""" popup = Toplevel(self.root) popup.overrideredirect(True) popup.attributes("-topmost", True) # 定位在输入框正下方 x = entry.winfo_rootx() y = entry.winfo_rooty() + entry.winfo_height() w = entry.winfo_width() popup.geometry(f"{w}x180+{x}+{y}") frame = Frame(popup, bd=1, relief="solid") frame.pack(fill="both", expand=True) listbox = Listbox(frame, font=("", 10), selectmode="browse", activestyle="none") scroll = Scrollbar(frame, orient=VERTICAL, command=listbox.yview) listbox.configure(yscrollcommand=scroll.set) scroll.pack(side=RIGHT, fill=Y) listbox.pack(side=LEFT, fill=BOTH, expand=True) for opt in options: listbox.insert(END, opt) def select(e=None): sel = listbox.curselection() if sel: popup.destroy() on_select(sel[0]) def cancel(e=None): popup.destroy() listbox.bind("", select) listbox.bind("", select) listbox.bind("", cancel) listbox.focus_set() if listbox.size() > 0: listbox.selection_set(0) # 点击外部关闭 def on_focus_out(e): popup.after(150, lambda: popup.destroy() if not popup.focus_get() else None) popup.bind("", on_focus_out) popup.wait_window() def _on_product_selected(self, choice: dict): """产品选中：缓存选择并获取详情。""" self._product_cached_choice = choice self._loading_config = True # 阻止 trace 再次触发搜索 self.fv_product.set(choice.get("productName", "")) self._loading_config = False # 获取产品详情 try: biz_id = choice.get("productId", "") biz_no = choice.get("bizProductNo", "") if biz_id and biz_no and self._search_api: self._product_cached_detail = self._search_api.select_product(biz_id, biz_no) except Exception as e: self._log(f"[error] 获取产品详情失败: {e}") self._product_cached_detail = {} def _on_customer_selected(self, choice: dict): """客户选中：缓存选择并获取详情和寄件人实名。""" self._customer_cached_choice = choice self._loading_config = True # 阻止 trace 再次触发搜索 self.fv_customer.set(choice.get("customerFullName", choice.get("customerName", ""))) self._loading_config = False # 获取客户详情 try: cust_id = choice.get("customerId", "") if cust_id and self._search_api: detail = self._search_api.select_customer(str(cust_id)) if isinstance(detail, dict) and "_text" not in detail: self._customer_cached_detail = detail base = detail.get("baseCustome", {}) if isinstance(detail.get("baseCustome"), dict) else {} sender_no = base.get("businessCode") or base.get("customerSubCode") or "" if sender_no: self._sender_realname_cached = self._search_api.find_sender_history(str(sender_no)) except Exception as e: self._log(f"[error] 获取客户详情失败: {e}") self._customer_cached_detail = {} # ─── 文件选择 ─────────────────────────────────────────── def _pick_file(self): path = filedialog.askopenfilename( title="选择 Excel 文件", filetypes=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")], ) if path: self.excel_path.set(path) self.file_label.configure(text=path, fg="black") def _on_close(self): """关闭窗口；atexit 会强制清理 Node.js 子进程。""" global _engine if _engine: _engine.request_stop() if self._search_bridge: self._search_bridge.stop() self.root.destroy() # ─── 控制按钮 ─────────────────────────────────────────── def _do_login(self): """弹窗让用户粘贴一键导出的 cookies JSON。""" from login_helper import parse_json_and_save, load_session_info win = Toplevel(self.root) win.title("登录揽收系统") win.geometry("620x360") win.transient(self.root) win.grab_set() Label(win, text="登录揽收系统", font=("", 12, "bold"), pady=8).pack() Label(win, text="使用 Cookies 导出工具，将 10.4.188.1 的所有 cookies 导出为 JSON，粘贴到下方：").pack(anchor="w", padx=10) json_text = Text(win, height=12, wrap="word", font=("Consolas", 9)) json_text.pack(fill=BOTH, expand=True, padx=10, pady=5) def do_save(): try: n = parse_json_and_save(json_text.get("1.0", "end-1c")) self.engine_status_var.set("已登录") self.status_label.configure(fg="green") self._check_session_status() self._init_search_infra() win.destroy() except Exception as e: messagebox.showerror("解析失败", f"JSON 解析失败:\n{e}", parent=win) btn_f = Frame(win) btn_f.pack(pady=10) Button(btn_f, text="保存并关闭", command=do_save, bg="#4CAF50", fg="white", width=15, height=1).pack() info = load_session_info() if info: Label(win, text=f"当前已保存: {len(info.get('cookies', []))} 个 cookies", fg="gray").pack() def _start(self): if self.running: return if not self.excel_path.get(): messagebox.showwarning("提示", "请先选择 Excel 文件") return try: rows, headers = read_input_excel(self.excel_path.get()) except Exception as e: messagebox.showerror("读取错误", f"读取 Excel 失败:\n{e}") return if not rows: messagebox.showwarning("提示", "Excel 中没有数据") return # 验证必要字段是否有数据来源（固定值 > Excel > 默认值） fixed_values = get_fixed_values(self.config) defaults = get_defaults(self.config) missing = validate_required_columns(headers, fixed_values, defaults) if missing: msg = ( f"以下字段在所有来源中均未找到值:\n" + "\n".join(f" - {c}（请填入Excel / 固定值 / 默认值任一）" for c in missing) ) messagebox.showwarning("缺少必要字段", msg) return self.running = True self.paused = False self._set_controls_state("running") # 保存固定值到配置文件 self._save_fixed_from_entries() # 清空结果表格 for item in self.result_tree.get_children(): self.result_tree.delete(item) # 创建结果工作簿 self.results = [] # 更新配置中的延时 try: delay_ms = int(self.delay_var.get()) self.config.setdefault("timing", {})["delay_between_rows_ms"] = delay_ms except ValueError: pass self._log(f"开始处理 {len(rows)} 条记录...") self.progress["maximum"] = len(rows) self.progress["value"] = 0 self.progress_label.configure(text=f"0 / {len(rows)}") # 在后台线程启动异步引擎 _run_in_thread(self._run_engine(rows, headers)) async def _run_engine(self, rows, headers): """后台运行引擎（在 async 线程中执行）。""" global _engine try: from engine import PickupEngine field_mapping = get_field_mapping(self.config) defaults = get_defaults(self.config) timing = get_timing_config(self.config) fixed_values = get_fixed_values(self.config) _engine = PickupEngine( url="https://10.4.188.1/pickup-web/a/postwaybill/main", field_mapping=field_mapping, defaults=defaults, timing=timing, log_callback=self._log_async, fixed_values=fixed_values, choose_callback=self._choose_callback, confirm_callback=self._confirm_callback, ) # 注入已缓存的搜索基础设施和产品/客户选择 if self._search_api is not None: _engine.inject_infrastructure(self._search_sess, self._search_bridge, self._search_api) if self._product_cached_choice is not None: _engine._product_choice = self._product_cached_choice if self._product_cached_detail is not None: _engine._product_detail = self._product_cached_detail if self._customer_cached_choice is not None: _engine._customer_choice = self._customer_cached_choice if self._customer_cached_detail is not None: _engine._customer_detail = self._customer_cached_detail if self._sender_realname_cached is not None: _engine._sender_realname = self._sender_realname_cached self._update_status("启动中...") try: await _engine.start() except Exception as e: self._log_async(f"[error] 启动失败: {e}") self._update_status("启动失败") await _engine.stop() return # 验证已加载了 session，且包含关键 cookie if not _engine._sess or not _engine._sess._http.cookies: self._log_async("[提示] 未找到登录会话，请点击'登录'按钮完成登录后再开始处理") self._update_status("未登录") await _engine.stop() return cookie_names = {c.name for c in _engine._sess._http.cookies} missing = [n for n in ("JSESSIONID", "jdpt.session.id") if n not in cookie_names] if missing: self._log_async(f"[提示] 缺少关键 cookie: {', '.join(missing)}，请重新登录并手动填写") self._update_status("未登录") await _engine.stop() return # 初始化页面资源和 token bridge try: await _engine.begin_processing() self._update_status("后台运行中") self._log_async("HTTP API 引擎已就绪") except Exception as e: self._log_async(f"[error] 初始化失败: {e}") self._update_status("启动失败") await _engine.stop() return for i, row in enumerate(rows): if _engine._stop_requested: break result = await _engine.process_single_row(row) self.results.append(result) # 更新结果表格 self.root.after(0, lambda r=result: self._add_result_row(r)) # 更新进度 self.root.after(0, lambda cur=i+1, total=len(rows): self._update_progress(cur, total)) # 行间延时 delay = timing.get("delay_between_rows_ms", 3000) / 1000 if i < len(rows) - 1: await asyncio.sleep(delay) self._log_async(f"\n处理完成！共处理 {len(self.results)} 条记录") except Exception as e: self._log_async(f"[error] 引擎异常: {e}") import traceback traceback.print_exc() finally: if _engine: await _engine.stop() self.root.after(0, self._on_finished) def _update_progress(self, current, total): self.progress["value"] = current self.progress_label.configure(text=f"{current} / {total}") self.status_var.set(f"处理中: {current}/{total}") def _on_finished(self): global _engine self.running = False self.paused = False _engine = None self._set_controls_state("stopped") self.status_var.set("就绪") # 清空产品/客户输入框（batch 结束后重置） self.fv_product.set("") self.fv_customer.set("") # 清空缓存的选择 self._product_cached_choice = None self._product_cached_detail = None self._customer_cached_choice = None self._customer_cached_detail = None self._sender_realname_cached = None def _pause(self): global _engine if not self.running: return if self.paused: self.paused = False if _engine: _engine.resume() self.pause_btn.configure(text="⏸ 暂停") self.status_var.set("处理中...") self._log("已恢复") else: self.paused = True if _engine: _engine.request_pause() self.pause_btn.configure(text="▶ 继续") self.status_var.set("已暂停") self._log("已暂停") def _stop(self): global _engine self.running = False if _engine: _engine.request_stop() self._set_controls_state("stopped") self.status_var.set("正在停止...") self._log("用户停止") def _set_controls_state(self, state): if state == "running": self.start_btn.configure(state=DISABLED) self.pause_btn.configure(state=NORMAL) self.stop_btn.configure(state=NORMAL) else: self.start_btn.configure(state=NORMAL) self.pause_btn.configure(state=DISABLED, text="⏸ 暂停") self.stop_btn.configure(state=DISABLED) # ─── 日志 ─────────────────────────────────────────────── def _log(self, msg): """线程安全的日志输出。""" self.root.after(0, self._log_sync, msg) def _log_async(self, msg): """从异步线程调用（直接入队到 GUI 线程）。""" self.root.after(0, self._log_sync, msg) def _update_status(self, status: str): """更新引擎状态显示。""" color_map = {"后台运行中": "green", "启动失败": "red", "未登录": "orange"} self.root.after(0, lambda: self.engine_status_var.set(status)) self.root.after(0, lambda s=status: self.status_label.configure( fg=color_map.get(s, "gray") )) async def _choose_callback(self, title: str, options: list[str]) -> int | None: """弹出选择框让用户从 autocomplete 结果中选择。返回索引或 None。""" loop = asyncio.get_event_loop() future = loop.create_future() result = {"index": None} def show(): win = Toplevel(self.root) win.title(title) win.geometry("400x300") win.transient(self.root) win.grab_set() Label(win, text=f"请选择{title}，选择后批量复用：", pady=10).pack() frame = Frame(win) frame.pack(fill="both", expand=True, padx=10) listbox = Listbox(frame) scroll = Scrollbar(frame, orient=VERTICAL, command=listbox.yview) listbox.configure(yscrollcommand=scroll.set) scroll.pack(side=RIGHT, fill=Y) listbox.pack(side=LEFT, fill=BOTH, expand=True) for i, opt in enumerate(options): listbox.insert(END, f"[{i+1}] {opt}") def on_select(): sel = listbox.curselection() if sel: result["index"] = sel[0] win.destroy() listbox.bind("", lambda e: on_select()) # 关闭窗口 → result["index"] 为 None → engine 停止 batch def on_cancel(): result["index"] = None win.destroy() btn_frame = Frame(win) btn_frame.pack(pady=10) Button(btn_frame, text="确认选择", command=on_select, bg="#4CAF50", fg="white", width=12).pack(side=LEFT, padx=5) Button(btn_frame, text="取消", command=on_cancel, bg="red", fg="white", width=8).pack(side=LEFT, padx=5) win.protocol("WM_DELETE_WINDOW", on_cancel) listbox.focus_set() # 等窗口关闭后通知 async 线程 def wait_then_resolve(): win.wait_window() loop.call_soon_threadsafe(lambda: future.set_result(result["index"])) threading.Thread(target=wait_then_resolve, daemon=True).start() self.root.after(0, show) return await future async def _confirm_callback(self, fields: dict[str, str]) -> bool: """弹出确认框显示表单数据，用户确认后返回 True。""" loop = asyncio.get_event_loop() future = loop.create_future() result = {"confirmed": False} def show(): win = Toplevel(self.root) win.title("确认提交信息 - 确认后自动批量处理") win.geometry("450x400") win.transient(self.root) win.grab_set() Label(win, text="请确认以下信息无误后点击确认：", pady=10, font=("", 10, "bold")).pack() frame = Frame(win) frame.pack(fill=BOTH, expand=True, padx=20, pady=5) text = Text(frame, wrap="word", font=("Consolas", 10), height=16) text.pack(side=LEFT, fill=BOTH, expand=True) scroll = Scrollbar(frame, command=text.yview) scroll.pack(side=RIGHT, fill=Y) text.configure(yscrollcommand=scroll.set) # 系统标记（不展示） from engine import _SYSTEM_FLAGS sys_keys = set(_SYSTEM_FLAGS.keys()) # 关键字段：始终展示，值也翻译为中文 from field_mapping import reverse_translate key_fields = [ ("产品名称", "bizProductName", ""), ("客户名称", "senders", ""), ("客户代码", "senderNo", ""), ("邮件号", "waybillNo", ""), ("重量(g)", "realWeight", ""), ("寄达地", "receiverArriveOrgName", ""), ("收件人", "receiverLinker", ""), ("收件人手机", "receiverMobile", ""), ("收件人地址", "receiverAddr", ""), ("寄件人", "senderLinker", ""), ("寄件人手机", "senderMobile", ""), ("寄件人地址", "senderAddr", ""), ("运输类型", "transferType", "运输类型"), ("内件属性", "contentsAttribute", "内件属性"), ("电商平台", "ecommerceNo", "电商平台"), ("付款方式", "paymentMode", "付款方式"), ("内件品名", "reserved5", ""), ("保价金额", "insuranceAmount", ""), ("代收货款", "codAmount", ""), ("备注", "senderNotes", ""), ] if not isinstance(fields, dict): text.insert(END, f"错误: 表单数据异常 ({type(fields).__name__})") text.configure(state=DISABLED) return shown = set() for label, key, col_name in key_fields: val = fields.get(key, "") or "空" if col_name and val != "空": val = reverse_translate(col_name, val) text.insert(END, f"{label}: {val}\n") shown.add(key) # 展示其余非系统标记的有值字段 for k, v in fields.items(): if v and k not in shown and k not in sys_keys: text.insert(END, f"{k}: {v}\n") text.configure(state=DISABLED) btn_frame = Frame(win) btn_frame.pack(pady=10) Button(btn_frame, text="确认提交", command=lambda: (result.__setitem__("confirmed", True), win.destroy()), bg="#4CAF50", fg="white", width=12, height=1).pack(side=LEFT, padx=10) Button(btn_frame, text="取消", command=win.destroy, bg="red", fg="white", width=10).pack(side=LEFT, padx=10) def wait_then_resolve(): win.wait_window() loop.call_soon_threadsafe(lambda: future.set_result(result["confirmed"])) threading.Thread(target=wait_then_resolve, daemon=True).start() self.root.after(0, show) return await future def _log_sync(self, msg): _log_user(msg) # ─── 入口 ────────────────────────────────────────────────── def main(): app = App() app.root.mainloop() if __name__ == "__main__": main()