# state_store.py # 阶段0：站点状态持久化（SQLite）。记录各站登录态 + 应到/实到数据文件生成时间， # 供后台心跳刷新、菜单状态盘展示，以及未来 FastAPI / Web 前端读取。 # 重启不丢——程序重启后状态从本库恢复（登录态会随心跳重新探测校正）。 # # 设计：纯 Python（sqlite3 标准库，无新依赖），每次读写开短连接，主线程使用。 import os import sqlite3 from datetime import datetime from inbound_verify.paths import STATE_DB_PATH # 登录态枚举 LOGIN_UNKNOWN = "unknown" # 尚未探测过 LOGIN_IN = "logged_in" LOGIN_OUT = "logged_out" # 任务状态枚举（task_history.status） TASK_PENDING = "pending" # 已入队，待执行 TASK_RUNNING = "running" # 正在执行 TASK_SUCCESS = "success" # 成功（有数据） TASK_NO_DATA = "no_data" # 成功但本站本次无数据 TASK_FAILED = "failed" # 失败（重试耗尽 / 未登录 / 异常） 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, timeout=3.0) 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, undelivered_ready INTEGER, undelivered_generated_at TEXT, expected_business_date TEXT, actual_business_date TEXT, undelivered_business_date TEXT, updated_at TEXT ) """) # 旧库迁移：补 undelivered 两列（新库已含；重复添加抛 OperationalError，忽略） for _col, _typedef in [ ("undelivered_ready", "INTEGER NOT NULL DEFAULT 0"), ("undelivered_generated_at", "TEXT NOT NULL DEFAULT ''"), ("expected_business_date", "TEXT NOT NULL DEFAULT ''"), ("actual_business_date", "TEXT NOT NULL DEFAULT ''"), ("undelivered_business_date", "TEXT NOT NULL DEFAULT ''"), ]: try: conn.execute(f"ALTER TABLE site_status ADD COLUMN {_col} {_typedef}") except sqlite3.OperationalError: pass conn.execute(""" CREATE TABLE IF NOT EXISTS task_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, site TEXT, kind TEXT, status TEXT, started_at TEXT, finished_at TEXT, error TEXT ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS site_config ( site TEXT PRIMARY KEY, expected_offset INTEGER NOT NULL DEFAULT 0, actual_offset INTEGER NOT NULL DEFAULT 0, schedule_enabled INTEGER NOT NULL DEFAULT 0, schedule_time TEXT NOT NULL DEFAULT '', updated_at TEXT ) """) # 旧库迁移：补 schedule 两列 + expected/actual 偏移（新库已含；重复添加抛错忽略） for _col, _typedef in [ ("schedule_enabled", "INTEGER NOT NULL DEFAULT 0"), ("schedule_time", "TEXT NOT NULL DEFAULT ''"), ("expected_offset", "INTEGER NOT NULL DEFAULT 0"), ("actual_offset", "INTEGER NOT NULL DEFAULT 0"), ]: try: conn.execute(f"ALTER TABLE site_config ADD COLUMN {_col} {_typedef}") except sqlite3.OperationalError: pass # 旧库若有 date_offset 列，把值搬到 expected/actual（一次性；新库无此列则跳过） try: conn.execute( "UPDATE site_config SET expected_offset=date_offset, actual_offset=date_offset " "WHERE expected_offset=0 AND date_offset IS NOT NULL AND date_offset>0" ) except sqlite3.OperationalError: pass conn.execute(""" CREATE TABLE IF NOT EXISTS site_settings ( site TEXT, key TEXT, value TEXT, PRIMARY KEY (site, key) ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS ingest_state ( site TEXT, kind TEXT, ok INTEGER, ingested_at TEXT, count INTEGER, error TEXT, PRIMARY KEY (site, kind) ) """) # 周期性抓取调度（每 site×kind 一行）：取代旧 site_config.schedule_* 的每日单时点。 # enabled=总开关；active_start/end=激活时段"HH:MM"（空=不限时段，避免半夜空跑）； # interval_minutes=激活时段内的抓取间隔。百世 kind 固定为 undelivered（无应到/实到二分）。 conn.execute(""" CREATE TABLE IF NOT EXISTS fetch_schedule ( site TEXT NOT NULL, kind TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 0, active_start TEXT NOT NULL DEFAULT '', active_end TEXT NOT NULL DEFAULT '', interval_minutes INTEGER NOT NULL DEFAULT 30, updated_at TEXT, PRIMARY KEY (site, kind) ) """) conn.commit() def _upsert(conn, site, **fields): """更新（或插入）单站：保留未传字段，刷新 updated_at。""" row = conn.execute( "SELECT login_state, login_checked_at, expected_ready, expected_generated_at, " "expected_business_date, actual_ready, actual_generated_at, actual_business_date, " "undelivered_ready, undelivered_generated_at, undelivered_business_date " "FROM site_status WHERE site = ?", (site,), ).fetchone() cur = { "login_state": LOGIN_UNKNOWN, "login_checked_at": "", "expected_ready": 0, "expected_generated_at": "", "expected_business_date": "", "actual_ready": 0, "actual_generated_at": "", "actual_business_date": "", "undelivered_ready": 0, "undelivered_generated_at": "", "undelivered_business_date": "", } if row: ( cur["login_state"], cur["login_checked_at"], cur["expected_ready"], cur["expected_generated_at"], cur["expected_business_date"], cur["actual_ready"], cur["actual_generated_at"], cur["actual_business_date"], cur["undelivered_ready"], cur["undelivered_generated_at"], cur["undelivered_business_date"], ) = 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["expected_business_date"], cur["actual_ready"], cur["actual_generated_at"], cur["actual_business_date"], cur["undelivered_ready"], cur["undelivered_generated_at"], cur["undelivered_business_date"], cur["updated_at"], ) if row: conn.execute( "UPDATE site_status SET login_state=?, login_checked_at=?, expected_ready=?, " "expected_generated_at=?, expected_business_date=?, actual_ready=?, " "actual_generated_at=?, actual_business_date=?, undelivered_ready=?, " "undelivered_generated_at=?, undelivered_business_date=?, 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, expected_business_date, actual_ready, actual_generated_at, " "actual_business_date, undelivered_ready, undelivered_generated_at, " "undelivered_business_date, 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, timeout=3.0) as conn: _upsert(conn, site, login_state=state, login_checked_at=_now()) def reset_login_states(sites): """启动时把给定站点的登录态重置为 unknown（避免显示上一会话的陈旧登录态）。 登录态是会话级的；数据态（文件就绪）会话无关、保留不动，心跳就绪后会重新探测。""" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: for site in sites: _upsert(conn, site, login_state=LOGIN_UNKNOWN, login_checked_at="") def set_data_state(site, kind, ready, generated_at, business_date=None): """更新单站数据态。kind: 'expected'/'actual'/'undelivered'；ready: bool； generated_at: str。business_date: str 或 None——None 时保留原值（心跳不覆盖下载快照）。""" fields = { f"{kind}_ready": 1 if ready else 0, f"{kind}_generated_at": generated_at or "", } if business_date is not None: fields[f"{kind}_business_date"] = business_date or "" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: _upsert(conn, site, **fields) def get_all_status(): """返回 {site: {各字段}}；库不存在则返回 {}。""" if not os.path.exists(STATE_DB_PATH): return {} with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: rows = conn.execute( "SELECT site, login_state, login_checked_at, expected_ready, " "expected_generated_at, expected_business_date, actual_ready, " "actual_generated_at, actual_business_date, undelivered_ready, " "undelivered_generated_at, undelivered_business_date, 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], "expected_business_date": r[5], "actual_ready": bool(r[6]), "actual_generated_at": r[7], "actual_business_date": r[8], "undelivered_ready": bool(r[9]), "undelivered_generated_at": r[10], "undelivered_business_date": r[11], "updated_at": r[12], } for r in rows } def set_ingest_state(site, kind, ok, count=0, error=None): """记录一次入库结果（UPSERT）。ok: bool；count: 入库条数；error: 失败原因或 None。""" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: conn.execute( "INSERT INTO ingest_state (site, kind, ok, ingested_at, count, error) " "VALUES (?, ?, ?, ?, ?, ?) " "ON CONFLICT(site, kind) DO UPDATE SET " "ok=excluded.ok, ingested_at=excluded.ingested_at, " "count=excluded.count, error=excluded.error", (site, kind, 1 if ok else 0, _now(), int(count or 0), error or ""), ) conn.commit() def get_all_ingest_state(): """返回 {site: {kind: {ok, ingested_at, count, error}}}；库不存在返回 {}。""" if not os.path.exists(STATE_DB_PATH): return {} with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: rows = conn.execute( "SELECT site, kind, ok, ingested_at, count, error FROM ingest_state" ).fetchall() out = {} for site, kind, ok, ingested_at, count, error in rows: out.setdefault(site, {})[kind] = { "ok": bool(ok), "ingested_at": ingested_at or "", "count": int(count or 0), "error": error or "", } return out # ============================ 下载日期偏移（site_config）============================ MAX_DATE_OFFSET = 30 # 0=今天，最大回溯 30 天 def get_offset(site, kind="expected"): """读取单站下载日期偏移（kind: 'expected'/'actual'；0=今天，1=昨天…）；未配置返回 0。""" col = "expected_offset" if kind == "expected" else "actual_offset" if not os.path.exists(STATE_DB_PATH): return 0 with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: row = conn.execute( f"SELECT {col} FROM site_config WHERE site=?", (site,) ).fetchone() return int(row[0]) if row else 0 def set_offset(site, kind, offset): """设置单站下载日期偏移（kind: 'expected'/'actual'），钳制到 [0, MAX_DATE_OFFSET]。""" col = "expected_offset" if kind == "expected" else "actual_offset" offset = max(0, min(MAX_DATE_OFFSET, int(offset))) with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: conn.execute( f"INSERT INTO site_config (site, {col}, updated_at) VALUES (?, ?, ?) " f"ON CONFLICT(site) DO UPDATE SET {col}=excluded.{col}, " f"updated_at=excluded.updated_at", (site, offset, _now()), ) conn.commit() return offset def set_schedule(site, enabled, time_str): """【DEPRECATED】旧"每日单时点定时"——已被 fetch_schedule 的周期+激活时段模式取代。 保留死代码以防外部残留调用；新代码请用 set_fetch_schedule。""" enabled_int = 1 if enabled else 0 with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: conn.execute( "INSERT INTO site_config (site, schedule_enabled, schedule_time, updated_at) " "VALUES (?, ?, ?, ?) " "ON CONFLICT(site) DO UPDATE SET " "schedule_enabled=excluded.schedule_enabled, " "schedule_time=excluded.schedule_time, updated_at=excluded.updated_at", (site, enabled_int, time_str or "", _now()), ) conn.commit() return bool(enabled_int), (time_str or "") # ============================ 周期性抓取调度（fetch_schedule）============================ # 取代旧 site_config.schedule_* 的"每日单时点"：每 site×kind 一行， # 在激活时段 [active_start, active_end) 内按 interval_minutes 周期抓取。 # 应到/实到各自独立配置；百世只有 undelivered（站点直供未到明细，无应到/实到二分）。 # 各站允许的周期抓取 kind（单一来源，server.py 复用） SITE_FETCH_KINDS = { "顺心": ("expected", "actual"), "中通": ("expected", "actual"), "韵达": ("expected", "actual"), "安能": ("expected", "actual"), "百世": ("undelivered",), } DEFAULT_FETCH_SCHEDULE = { "enabled": False, "active_start": "", "active_end": "", "interval_minutes": 30, } def allowed_kinds(site): """该站允许的周期抓取 kind 元组；未知站点返回空元组。""" return SITE_FETCH_KINDS.get(site, ()) def _fetch_spec(row): """把 fetch_schedule 行转成 spec dict。""" return { "enabled": bool(row[0]), "active_start": row[1] or "", "active_end": row[2] or "", "interval_minutes": int(row[3]), } def get_fetch_schedule(site, kind): """读单 (site,kind) 周期抓取配置；未配置返回 None。""" if not os.path.exists(STATE_DB_PATH): return None with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: row = conn.execute( "SELECT enabled, active_start, active_end, interval_minutes " "FROM fetch_schedule WHERE site=? AND kind=?", (site, kind), ).fetchone() return _fetch_spec(row) if row else None def set_fetch_schedule(site, kind, enabled, active_start, active_end, interval_minutes): """UPSERT 单 (site,kind) 周期抓取配置；返回写入后的 spec dict。""" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: conn.execute( "INSERT INTO fetch_schedule " "(site, kind, enabled, active_start, active_end, interval_minutes, updated_at) " "VALUES (?, ?, ?, ?, ?, ?, ?) " "ON CONFLICT(site, kind) DO UPDATE SET " "enabled=excluded.enabled, active_start=excluded.active_start, " "active_end=excluded.active_end, interval_minutes=excluded.interval_minutes, " "updated_at=excluded.updated_at", ( site, kind, 1 if enabled else 0, (active_start or ""), (active_end or ""), int(interval_minutes), _now(), ), ) conn.commit() return { "enabled": bool(enabled), "active_start": active_start or "", "active_end": active_end or "", "interval_minutes": int(interval_minutes), } def get_all_fetch_schedules(): """返回 {site: {kind: spec}}；对每个站点的每个 allowed kind 都补齐（缺失用默认值）。 保证前端永远拿到完整 kind 集，不必前端补默认 spec。""" out = {} if not os.path.exists(STATE_DB_PATH): for site, kinds in SITE_FETCH_KINDS.items(): out[site] = {k: dict(DEFAULT_FETCH_SCHEDULE) for k in kinds} return out with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: rows = conn.execute( "SELECT site, kind, enabled, active_start, active_end, interval_minutes " "FROM fetch_schedule" ).fetchall() by_key = {(r[0], r[1]): _fetch_spec(r[2:]) for r in rows} for site, kinds in SITE_FETCH_KINDS.items(): out[site] = { k: by_key.get((site, k), dict(DEFAULT_FETCH_SCHEDULE)) for k in kinds } return out def get_all_config(): """返回 {site: {expected_offset, actual_offset, fetch_schedules}}。 站点集以 fetch_schedule 的 allowed kinds 为准（覆盖全业务站点）；offsets 缺失默认 0。""" schedules = get_all_fetch_schedules() offsets = {} if os.path.exists(STATE_DB_PATH): with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: rows = conn.execute( "SELECT site, expected_offset, actual_offset FROM site_config" ).fetchall() offsets = { r[0]: {"expected_offset": int(r[1]), "actual_offset": int(r[2])} for r in rows } return { site: { "expected_offset": offsets.get(site, {}).get("expected_offset", 0), "actual_offset": offsets.get(site, {}).get("actual_offset", 0), "fetch_schedules": schedules.get(site, {}), } for site in schedules } # ============================ 站点键值配置（site_settings）============================ # 各站专属配置（百世密码、韵达账密、安能 exe 路径…），由前端配置弹窗设置。 def get_setting(site, key): """读取单站某个配置值；未设置返回 ''。""" if not os.path.exists(STATE_DB_PATH): return "" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: row = conn.execute( "SELECT value FROM site_settings WHERE site=? AND key=?", (site, key) ).fetchone() return row[0] if row else "" def set_setting(site, key, value): """设置单站某个配置值（upsert）。""" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: conn.execute( "INSERT INTO site_settings (site, key, value) VALUES (?, ?, ?) " "ON CONFLICT(site, key) DO UPDATE SET value=excluded.value", (site, key, value if value is not None else ""), ) conn.commit() def get_site_settings(site): """返回单站全部配置 {key: value}。""" if not os.path.exists(STATE_DB_PATH): return {} with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: rows = conn.execute( "SELECT key, value FROM site_settings WHERE site=?", (site,) ).fetchall() return {r[0]: r[1] for r in rows} # ============================ 任务历史 ============================ def create_task(site, kind): """新建一条 pending 任务，返回其 id。""" now = _now() with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: cur = conn.execute( "INSERT INTO task_history (site, kind, status, started_at, finished_at, error) " "VALUES (?, ?, ?, ?, '', '')", (site, kind, TASK_PENDING, now), ) conn.commit() return cur.lastrowid def create_task_if_idle(site, kind): """周期调度专用：若该 (site,kind) 已有 pending/running 任务则返回 None（跳过本次周期）， 否则建一条 pending 任务返回其 id。单连接内 check-then-insert，靠 SQLite 写锁把竞态压到忽略不计。 与 create_task 的区别：手动触发(POST /tasks)用 create_task（用户点的必建）；周期 job 用本函数 ——上一次还没跑完时跳过，避免同 (site,kind) 任务堆积。手动建的任务会让紧随其后的周期 fire 判到 inflight 而跳过，天然互斥。""" now = _now() with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: row = conn.execute( "SELECT 1 FROM task_history WHERE site=? AND kind=? " "AND status IN (?, ?) LIMIT 1", (site, kind, TASK_PENDING, TASK_RUNNING), ).fetchone() if row: return None cur = conn.execute( "INSERT INTO task_history (site, kind, status, started_at, finished_at, error) " "VALUES (?, ?, ?, ?, '', '')", (site, kind, TASK_PENDING, now), ) conn.commit() return cur.lastrowid def update_task(task_id, status, error=None): """更新任务状态。终态(success/no_data/failed)写入 finished_at。""" finished = _now() if status in (TASK_SUCCESS, TASK_NO_DATA, TASK_FAILED) else "" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: if finished: conn.execute( "UPDATE task_history SET status=?, finished_at=?, error=? WHERE id=?", (status, finished, error or "", task_id), ) else: conn.execute( "UPDATE task_history SET status=?, error=? WHERE id=?", (status, error or "", task_id), ) conn.commit() def get_task(task_id): """返回单条任务 dict，不存在返回 None。""" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: row = conn.execute( "SELECT id, site, kind, status, started_at, finished_at, error " "FROM task_history WHERE id=?", (task_id,), ).fetchone() if not row: return None return { "id": row[0], "site": row[1], "kind": row[2], "status": row[3], "started_at": row[4], "finished_at": row[5], "error": row[6], } def list_tasks(limit=20): """返回最近 limit 条任务（按 id 倒序）。""" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: rows = conn.execute( "SELECT id, site, kind, status, started_at, finished_at, error " "FROM task_history ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() return [ { "id": r[0], "site": r[1], "kind": r[2], "status": r[3], "started_at": r[4], "finished_at": r[5], "error": r[6], } for r in rows ] def fail_stale_tasks(reason: str = "服务重启，上轮未完成任务，请手动重跑") -> int: """worker 启动时调用：把遗留的 pending/running 任务标记为 failed，实现重启自愈。 返回被清理的任务数量。""" with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: cur = conn.execute( "UPDATE task_history SET status=?, finished_at=?, error=? " "WHERE status IN (?, ?)", (TASK_FAILED, _now(), reason, TASK_PENDING, TASK_RUNNING), ) conn.commit() return cur.rowcount