From a5ceb42ac1a4fbedc1a7a762f7d292b589f0293b Mon Sep 17 00:00:00 2001 From: Misaka Date: Thu, 16 Jul 2026 22:44:15 +0800 Subject: [PATCH] =?UTF-8?q?=E9=98=B6=E6=AE=B50=EF=BC=9A=E7=99=BB=E5=BD=95?= =?UTF-8?q?=E6=80=81=E5=BF=83=E8=B7=B3=20+=20=E7=8A=B6=E6=80=81=E6=8C=81?= =?UTF-8?q?=E4=B9=85=E5=8C=96=EF=BC=88=E9=87=8D=E6=9E=84=E5=9C=B0=E5=9F=BA?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 state_store.py：SQLite 状态存储（site_status 表：登录态/数据态/时间戳）， UPSERT 保留字段，重启不丢 - paths.py：加 STATE_DB_PATH（state/state.db） - main_router：提取 probe_site_login / probe_data_file；菜单循环改为 input 后台线程 + 主线程心跳轮询（满足 Playwright sync 线程安全）； 每 30s 探测各站登录态 + 数据文件写库，登录态变化时提示；新增菜单 [12] 站点状态盘 - .gitignore：忽略 state/ 为后续 FastAPI 化与 Web 前端提供状态地基。 Co-Authored-By: Claude --- .gitignore | 1 + main_router.py | 154 ++++++++++++++++++++++++++++++++++++++++++++++++- paths.py | 3 + state_store.py | 136 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 293 insertions(+), 1 deletion(-) create mode 100644 state_store.py diff --git a/.gitignore b/.gitignore index c81798f..723f4ae 100644 --- a/.gitignore +++ b/.gitignore @@ -41,6 +41,7 @@ desktop.ini # Project specific downloads/ output/ +state/ *.xlsx *.xls *.log diff --git a/main_router.py b/main_router.py index 9e54cd1..8cb88db 100644 --- a/main_router.py +++ b/main_router.py @@ -1,15 +1,22 @@ # main_router.py import os +import queue import socket import subprocess +import threading import time import urllib.request +from datetime import datetime + import yaml from playwright.sync_api import sync_playwright from paths import DOWNLOAD_DIR, CONFIG_PATH +# 阶段0：状态持久化（站点登录态 / 数据态，心跳刷新；重启不丢） +import state_store + # 导入抽离出去的各个网点模块 import site_shunxin import site_baishi @@ -41,6 +48,55 @@ READY_SELECTORS = { # 安能是 Electron 桌面应用（不是 Playwright 打开的网页），需要单独启动。 APP_SITES = {"安能"} +# 阶段0：心跳间隔（秒）——各站就绪后，每隔此时长探测一次登录态/数据态 +HEARTBEAT_INTERVAL = 30 + +# 各站最终数据文件名（用于探测"数据是否已跑出来"）；百世为单流程，仅应到未到 +DATA_FILENAMES = { + "顺心": {"expected": "顺心-应到货物数据.xlsx", "actual": "顺心-实到货物数据.xlsx"}, + "中通": {"expected": "中通-应到货物数据.xlsx", "actual": "中通-实到货物数据.xlsx"}, + "韵达": {"expected": "韵达-应到货物数据.xlsx", "actual": "韵达-实到货物数据.xlsx"}, + "安能": {"expected": "安能-应到货物数据.xlsx", "actual": "安能-实到货物数据.xlsx"}, + "百世": {"expected": "百世-应到未到货物数据.xlsx", "actual": ""}, +} + + +def probe_site_login(site_name, pages_map): + """探测单站是否登录（复用就绪轮询判据）。任何异常一律返回 False。 + + 只在主线程调用（Playwright sync 对象绑定主线程）。 + """ + try: + if site_name not in pages_map: + return False + if site_name == "安能": + return site_anneng.anneng_ready() + if site_name == "顺心": + return all( + pg.locator(READY_SELECTORS["顺心"]).is_visible(timeout=500) + for pg in pages_map["顺心"] + ) + return ( + pages_map[site_name] + .locator(READY_SELECTORS[site_name]) + .is_visible(timeout=500) + ) + except Exception: + return False + + +def probe_data_file(site_name, kind): + """探测单站应到/实到数据文件是否存在且为今天。返回 (is_today: bool, mtime_str)。""" + fname = DATA_FILENAMES.get(site_name, {}).get(kind, "") + if not fname: + return (False, "") + path = os.path.join(DOWNLOAD_DIR, fname) + if not os.path.exists(path): + return (False, "") + dt = datetime.fromtimestamp(os.path.getmtime(path)) + is_today = dt.date() == datetime.now().date() + return (is_today, dt.strftime("%Y-%m-%d %H:%M:%S")) + def _find_free_port(): """让操作系统分配一个空闲端口，避免固定端口冲突。""" @@ -420,6 +476,95 @@ def run_multi_site_daemon(): return False return True + # ================================================================== + # 阶段0：后台心跳 + 状态持久化 + # input 放后台线程（只读 stdin，不碰 Playwright），主线程在 _await_command + # 里轮询命令队列并定期跑心跳；所有 page 操作仍在主线程，满足 Playwright + # sync 的线程安全。 + # ================================================================== + state_store.init_db() + sites_to_watch = list(ready_status.keys()) + # 就绪轮询刚通过 → 各站视为已登录写一次初值，后续心跳校正 + for _site in sites_to_watch: + state_store.set_login_state(_site, True) + + last_heartbeat = 0.0 + + def _run_heartbeat(): + """一轮心跳：探测各站登录态 + 数据文件，写状态库；登录态变化时提示。""" + nonlocal last_heartbeat + prev = state_store.get_all_status() + for site_name in sites_to_watch: + logged_in = probe_site_login(site_name, pages_map) + prev_login = prev.get(site_name, {}).get("login_state") + state_store.set_login_state(site_name, logged_in) + now_login = state_store.LOGIN_IN if logged_in else state_store.LOGIN_OUT + if prev_login and prev_login not in ( + now_login, + state_store.LOGIN_UNKNOWN, + ): + print( + f"\n ⚠️【{site_name}】登录态变化: {prev_login} → {now_login}" + ) + for kind in ("expected", "actual"): + ready, gen_at = probe_data_file(site_name, kind) + state_store.set_data_state(site_name, kind, ready, gen_at) + last_heartbeat = time.monotonic() + + def _print_status_board(): + """打印各站登录态 / 数据就绪状态盘（菜单 [12]）。""" + print("\n====================== 站点状态盘 ======================") + status = state_store.get_all_status() + if not status: + print(" （暂无状态记录）") + print("======================================================") + return + login_text = { + state_store.LOGIN_IN: "✅ 已登录", + state_store.LOGIN_OUT: "❌ 未登录", + state_store.LOGIN_UNKNOWN: "❔ 未知", + } + for site_name in sites_to_watch: + s = status.get(site_name) + if not s: + continue + login_mark = login_text.get(s["login_state"], s["login_state"]) + exp = ( + f"应到{'✅' if s['expected_ready'] else '—'} " + f"{s['expected_generated_at'] or '无'}" + ) + act = ( + f"实到{'✅' if s['actual_ready'] else '—'} " + f"{s['actual_generated_at'] or '无'}" + ) + print( + f" 【{site_name}】 {login_mark} | {exp} | {act} " + f"| 探测于 {s['login_checked_at']}" + ) + print("======================================================") + + command_queue = queue.Queue() + + def _input_loop(): + """后台线程：读用户输入塞进队列。不碰任何 Playwright 对象。""" + while True: + try: + command_queue.put(input()) + except EOFError: + return + + threading.Thread(target=_input_loop, daemon=True).start() + + def _await_command(): + """等待一条命令（期间跑心跳）。命令到来即返回。""" + nonlocal last_heartbeat + while True: + try: + return command_queue.get(timeout=0.5) + except queue.Empty: + if time.monotonic() - last_heartbeat >= HEARTBEAT_INTERVAL: + _run_heartbeat() + while True: print("\n====================================================") print(" 物流数据下载主菜单 ") @@ -453,10 +598,14 @@ def run_multi_site_daemon(): " [9] 执行 - 应到未到比对（全站点汇总，输出 output/应到未到数据.xlsx）" ) print("-" * 52) + print(" 站点状态") + print(" [12] 查看 - 各站登录态 / 数据就绪状态") + print("-" * 52) print(" [0] 退出系统") print("====================================================") - choice = input("请输入任务编号并回车: ") + print("请输入任务编号并回车: ", end="", flush=True) + choice = _await_command() try: if choice == "1" and is_site_ready("顺心"): @@ -491,6 +640,8 @@ def run_multi_site_daemon(): run_automation_test(pages_map) elif choice == "9": run_undelivered_compare() + elif choice == "12": + _print_status_board() elif choice == "0": print("\n正在关闭浏览器并退出...") break @@ -507,6 +658,7 @@ def run_multi_site_daemon(): "9", "10", "11", + "12", "0", ]: print("\n⚠️ 无效输入，请查证后回车。") diff --git a/paths.py b/paths.py index bdef9d7..07ba717 100644 --- a/paths.py +++ b/paths.py @@ -13,3 +13,6 @@ DOWNLOAD_DIR = os.path.join(BASE_DIR, "downloads") # 统一的配置文件路径（注意：config.yaml 需与本项目脚本放在同一目录下） CONFIG_PATH = os.path.join(BASE_DIR, "config.yaml") + +# 状态存储（SQLite，阶段0：心跳 / 登录态 / 数据态持久化，重启不丢） +STATE_DB_PATH = os.path.join(BASE_DIR, "state", "state.db") diff --git a/state_store.py b/state_store.py new file mode 100644 index 0000000..5148715 --- /dev/null +++ b/state_store.py @@ -0,0 +1,136 @@ +# state_store.py +# 阶段0：站点状态持久化（SQLite）。记录各站登录态 + 应到/实到数据文件生成时间， +# 供后台心跳刷新、菜单状态盘展示，以及未来 FastAPI / Web 前端读取。 +# 重启不丢——程序重启后状态从本库恢复（登录态会随心跳重新探测校正）。 +# +# 设计：纯 Python（sqlite3 标准库，无新依赖），每次读写开短连接，主线程使用。 + +import os +import sqlite3 +from datetime import datetime + +from paths import STATE_DB_PATH + +# 登录态枚举 +LOGIN_UNKNOWN = "unknown" # 尚未探测过 +LOGIN_IN = "logged_in" +LOGIN_OUT = "logged_out" + + +def _now(): + """本地时间的字符串（到秒），用于时间戳列。""" + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + +def init_db(): + """建库建表（幂等）。确保 state 目录存在。""" + os.makedirs(os.path.dirname(STATE_DB_PATH), exist_ok=True) + with sqlite3.connect(STATE_DB_PATH) as conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS site_status ( + site TEXT PRIMARY KEY, + login_state TEXT, + login_checked_at TEXT, + expected_ready INTEGER, + expected_generated_at TEXT, + actual_ready INTEGER, + actual_generated_at TEXT, + updated_at TEXT + ) + """) + conn.commit() + + +def _upsert(conn, site, **fields): + """更新（或插入）单站：保留未传字段，刷新 updated_at。""" + row = conn.execute( + "SELECT login_state, login_checked_at, expected_ready, expected_generated_at, " + "actual_ready, actual_generated_at FROM site_status WHERE site = ?", + (site,), + ).fetchone() + cur = { + "login_state": LOGIN_UNKNOWN, + "login_checked_at": "", + "expected_ready": 0, + "expected_generated_at": "", + "actual_ready": 0, + "actual_generated_at": "", + } + if row: + ( + cur["login_state"], + cur["login_checked_at"], + cur["expected_ready"], + cur["expected_generated_at"], + cur["actual_ready"], + cur["actual_generated_at"], + ) = row + cur.update(fields) + cur["updated_at"] = _now() + + values = ( + site, + cur["login_state"], + cur["login_checked_at"], + cur["expected_ready"], + cur["expected_generated_at"], + cur["actual_ready"], + cur["actual_generated_at"], + cur["updated_at"], + ) + if row: + conn.execute( + "UPDATE site_status SET login_state=?, login_checked_at=?, expected_ready=?, " + "expected_generated_at=?, actual_ready=?, actual_generated_at=?, updated_at=? " + "WHERE site=?", + values[1:] + (site,), + ) + else: + conn.execute( + "INSERT INTO site_status (site, login_state, login_checked_at, expected_ready, " + "expected_generated_at, actual_ready, actual_generated_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + values, + ) + conn.commit() + + +def set_login_state(site, logged_in): + """更新单站登录态。logged_in: bool。""" + state = LOGIN_IN if logged_in else LOGIN_OUT + with sqlite3.connect(STATE_DB_PATH) as conn: + _upsert(conn, site, login_state=state, login_checked_at=_now()) + + +def set_data_state(site, kind, ready, generated_at): + """更新单站数据态。kind: 'expected'/'actual'；ready: bool；generated_at: str 或 ''。""" + key_ready = "expected_ready" if kind == "expected" else "actual_ready" + key_at = "expected_generated_at" if kind == "expected" else "actual_generated_at" + with sqlite3.connect(STATE_DB_PATH) as conn: + _upsert( + conn, site, **{key_ready: 1 if ready else 0, key_at: generated_at or ""} + ) + + +def get_all_status(): + """返回 {site: {各字段}}；库不存在则返回 {}。""" + if not os.path.exists(STATE_DB_PATH): + return {} + with sqlite3.connect(STATE_DB_PATH) as conn: + rows = conn.execute( + "SELECT site, login_state, login_checked_at, expected_ready, " + "expected_generated_at, actual_ready, actual_generated_at, updated_at " + "FROM site_status" + ).fetchall() + return { + r[0]: { + "login_state": r[1], + "login_checked_at": r[2], + "expected_ready": bool(r[3]), + "expected_generated_at": r[4], + "actual_ready": bool(r[5]), + "actual_generated_at": r[6], + "updated_at": r[7], + } + for r in rows + }