From d74a35844f2d11797de074cc8e6bc10a38602582 Mon Sep 17 00:00:00 2001 From: Misaka Date: Tue, 28 Jul 2026 22:58:47 +0800 Subject: [PATCH] =?UTF-8?q?feat(schedule):=20=E5=91=A8=E6=9C=9F=E6=8A=93?= =?UTF-8?q?=E5=8F=96=E8=B0=83=E5=BA=A6=E5=8F=96=E4=BB=A3=E6=AF=8F=E6=97=A5?= =?UTF-8?q?=E5=AE=9A=E7=82=B9=E4=B8=8B=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 应到/实到数据各自独立周期抓取(启用+激活时段+频率)，抓取与处理解耦，抓完自动落库(复用现有 _persist_to_db 钩子)。 - state_store: 新增 fetch_schedule 表(site,kind)；所有连接加 timeout=3.0 防并发锁；get/set/get_all_fetch_schedules、create_task_if_idle(周期去重)、allowed_kinds；get_all_config 改返回 fetch_schedules；旧 set_schedule 标废弃。修复 get_all_fetch_schedules 列错位(_fetch_spec(r[2:]))。 - server: IntervalTrigger + _in_active_window(含跨午夜) + _enqueue_fetch(就绪门/激活窗口/去重) + _reschedule_fetch + lifespan 按(site,kind)注册 + FetchScheduleSpec + PUT/GET /config 新结构。 - runtime/store/BFF: 零改动。 Co-Authored-By: Claude --- inbound_verify/cli/server.py | 117 +++++++++++++------ inbound_verify/state_store.py | 214 ++++++++++++++++++++++++++++------ 2 files changed, 261 insertions(+), 70 deletions(-) diff --git a/inbound_verify/cli/server.py b/inbound_verify/cli/server.py index 0779828..709978b 100644 --- a/inbound_verify/cli/server.py +++ b/inbound_verify/cli/server.py @@ -17,11 +17,12 @@ import queue import threading import time from contextlib import asynccontextmanager +from datetime import datetime from typing import Dict, Optional import uvicorn from apscheduler.schedulers.background import BackgroundScheduler -from apscheduler.triggers.cron import CronTrigger +from apscheduler.triggers.interval import IntervalTrigger from fastapi import FastAPI, HTTPException from fastapi.responses import FileResponse from pydantic import BaseModel @@ -98,46 +99,76 @@ def _worker_loop(): print(">> [worker] 已退出。") -def _enqueue_undelivered(site): - """定时 job：把该站 undelivered 任务投到队列（worker 串行处理；本线程不碰 Playwright）。""" +def _in_active_window(active_start, active_end): + """当前本地时间是否落在激活时段内（避免半夜空跑）。 + - start/end 任一为空 → 不限时段（24h 活跃） + - start == end（非空）→ 视为全天活跃 + - start < end → 半开区间 [start, end) + - start > end → 跨午夜（如 22:00-06:00）：now >= start 或 now < end + """ + if not active_start or not active_end: + return True + now = datetime.now().strftime("%H:%M") + if active_start == active_end: + return True + if active_start < active_end: + return active_start <= now < active_end + return now >= active_start or now < active_end + + +def _enqueue_fetch(site, kind): + """周期 job 回调：worker 就绪 + 在激活时段内 + 该(site,kind)无未完成任务时，投递一次抓取任务。 + 跑在 APScheduler 线程池线程——绝不碰 Playwright，只经 SQLite + task_queue 通信。""" + if not worker_state["ready"]: + return # worker 未就绪：与 POST /tasks 的 409 同义，下个周期补抓 + cfg = state_store.get_fetch_schedule(site, kind) + if not cfg or not cfg["enabled"]: + return # 已禁用（job 本应已注销，双重保险） + if not _in_active_window(cfg["active_start"], cfg["active_end"]): + return # 不在激活时段，跳过本次 fire try: - tid = state_store.create_task(site, "undelivered") - task_queue.put((tid, {"site": site, "kind": "undelivered"})) - print(f">> [定时] 投递 {site}/undelivered 任务 #{tid}") + tid = state_store.create_task_if_idle(site, kind) + if tid is None: + return # 上一次同类任务还没跑完，跳过避免堆积 + task_queue.put((tid, {"site": site, "kind": kind})) + print(f">> [周期] 投递 {site}/{kind} 任务 #{tid}") except Exception as e: - print(f">> [定时] 投递 {site} 失败: {e}") + print(f">> [周期] 投递 {site}/{kind} 失败: {e}") -def _reschedule_site(site): - """按持久化配置（重新）注册或取消该站的每日定时 job。""" - job_id = f"site_{site}" +def _reschedule_fetch(site, kind): + """按持久化配置（重新）注册或取消该 (site,kind) 的周期抓取 job（IntervalTrigger）。""" + job_id = f"{site}_{kind}" try: scheduler.remove_job(job_id) except Exception: pass - cfg = state_store.get_all_config().get(site) - if not cfg or not cfg.get("schedule_enabled") or not cfg.get("schedule_time"): - return + cfg = state_store.get_fetch_schedule(site, kind) + if not cfg or not cfg["enabled"] or cfg["interval_minutes"] < 1: + return # 未启用或频率非法：不注册（等同取消） try: - hh, mm = cfg["schedule_time"].split(":") scheduler.add_job( - _enqueue_undelivered, - CronTrigger(hour=int(hh), minute=int(mm)), - args=[site], + _enqueue_fetch, + IntervalTrigger(minutes=cfg["interval_minutes"]), + args=[site, kind], id=job_id, replace_existing=True, ) - print(f">> [定时] 已注册 {site} 每日 {cfg['schedule_time']} 下载") + print( + f">> [周期] 已注册 {site}/{kind}：每 {cfg['interval_minutes']} 分钟" + f"（激活 {cfg['active_start'] or '不限'}~{cfg['active_end'] or '不限'}）" + ) except Exception as e: - print(f">> [定时] 注册 {site} 失败: {e}") + print(f">> [周期] 注册 {site}/{kind} 失败: {e}") @asynccontextmanager async def lifespan(_app): """服务启停：起 worker 线程 / 通知 worker 停。""" state_store.init_db() # 先建表/迁移状态库，确保早于 worker 就绪的 /api/status 可用 - for site in ALL_SITES: # 按持久化配置注册各站每日定时 job - _reschedule_site(site) + for site in ALL_SITES: # 按持久化配置注册各站各 kind 的周期抓取 job + for kind in state_store.allowed_kinds(site): + _reschedule_fetch(site, kind) scheduler.start() print(">> [定时] 调度器已启动") t = threading.Thread(target=_worker_loop, daemon=True) @@ -197,20 +228,28 @@ def get_status(): } +class FetchScheduleSpec(BaseModel): + enabled: bool + active_start: str = "" + active_end: str = "" + interval_minutes: int + + class ConfigRequest(BaseModel): expected_offset: Optional[int] = None actual_offset: Optional[int] = None + # DEPRECATED：旧"每日定点"字段，保留一个发布周期兼容旧前端请求（收到即忽略） schedule_enabled: Optional[bool] = None schedule_time: Optional[str] = None settings: Optional[Dict[str, str]] = None + fetch_schedules: Optional[Dict[str, FetchScheduleSpec]] = None def _default_cfg(): return { "expected_offset": 0, "actual_offset": 0, - "schedule_enabled": False, - "schedule_time": "", + "fetch_schedules": {}, } @@ -225,7 +264,6 @@ def get_config(): def set_config(site: str, req: ConfigRequest): if site not in ALL_SITES: raise HTTPException(status_code=400, detail=f"未知站点: {site}") - cur = state_store.get_all_config().get(site, _default_cfg()) # 应到/实到偏移（百世锁定当天） for kind, val in ( ("expected", req.expected_offset), @@ -237,18 +275,25 @@ def set_config(site: str, req: ConfigRequest): status_code=400, detail="百世固定下载当天，不可配置偏移" ) state_store.set_offset(site, kind, val) - # 每日定时（所有站可配，含百世） - if req.schedule_enabled is not None or req.schedule_time is not None: - enabled = ( - req.schedule_enabled - if req.schedule_enabled is not None - else cur["schedule_enabled"] - ) - time_str = ( - req.schedule_time if req.schedule_time is not None else cur["schedule_time"] - ) - state_store.set_schedule(site, enabled, time_str) - _reschedule_site(site) + # 周期抓取调度（应到/实到各自独立；百世只允许 undelivered） + if req.fetch_schedules: + allowed = state_store.allowed_kinds(site) + for kind, spec in req.fetch_schedules.items(): + if kind not in allowed: + raise HTTPException( + status_code=400, + detail=f"站点 {site} 不支持抓取类型 {kind}（允许: {list(allowed)}）", + ) + interval = max(1, min(1440, int(spec.interval_minutes))) + state_store.set_fetch_schedule( + site, + kind, + spec.enabled, + spec.active_start, + spec.active_end, + interval, + ) + _reschedule_fetch(site, kind) # 站点专属配置（密码/账号/路径…） if req.settings: for k, v in req.settings.items(): diff --git a/inbound_verify/state_store.py b/inbound_verify/state_store.py index 7d1363b..874b8d9 100644 --- a/inbound_verify/state_store.py +++ b/inbound_verify/state_store.py @@ -32,7 +32,7 @@ def _now(): def init_db(): """建库建表（幂等）。确保 state 目录存在。""" os.makedirs(os.path.dirname(STATE_DB_PATH), exist_ok=True) - with sqlite3.connect(STATE_DB_PATH) as conn: + with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS site_status ( site TEXT PRIMARY KEY, @@ -121,6 +121,21 @@ def init_db(): 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() @@ -202,14 +217,14 @@ def _upsert(conn, site, **fields): 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: + 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) as conn: + 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="") @@ -223,7 +238,7 @@ def set_data_state(site, kind, ready, generated_at, business_date=None): } if business_date is not None: fields[f"{kind}_business_date"] = business_date or "" - with sqlite3.connect(STATE_DB_PATH) as conn: + with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: _upsert(conn, site, **fields) @@ -231,7 +246,7 @@ def get_all_status(): """返回 {site: {各字段}}；库不存在则返回 {}。""" if not os.path.exists(STATE_DB_PATH): return {} - with sqlite3.connect(STATE_DB_PATH) as conn: + 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, " @@ -260,7 +275,7 @@ def get_all_status(): def set_ingest_state(site, kind, ok, count=0, error=None): """记录一次入库结果（UPSERT）。ok: bool；count: 入库条数；error: 失败原因或 None。""" - with sqlite3.connect(STATE_DB_PATH) as conn: + 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 (?, ?, ?, ?, ?, ?) " @@ -276,7 +291,7 @@ 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) as conn: + 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() @@ -301,7 +316,7 @@ def get_offset(site, kind="expected"): 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) as conn: + with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: row = conn.execute( f"SELECT {col} FROM site_config WHERE site=?", (site,) ).fetchone() @@ -312,7 +327,7 @@ 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) as conn: + 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}, " @@ -324,9 +339,10 @@ def set_offset(site, kind, offset): def set_schedule(site, enabled, time_str): - """设置单站每日定时下载（enabled: bool；time_str: 'HH:MM' 或 ''）。""" + """【DEPRECATED】旧"每日单时点定时"——已被 fetch_schedule 的周期+激活时段模式取代。 + 保留死代码以防外部残留调用；新代码请用 set_fetch_schedule。""" enabled_int = 1 if enabled else 0 - with sqlite3.connect(STATE_DB_PATH) as conn: + 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 (?, ?, ?, ?) " @@ -339,23 +355,128 @@ def set_schedule(site, enabled, time_str): return bool(enabled_int), (time_str or "") -def get_all_config(): - """返回 {site: {expected_offset, actual_offset, schedule_enabled, schedule_time}}。""" - if not os.path.exists(STATE_DB_PATH): - return {} - with sqlite3.connect(STATE_DB_PATH) as conn: - rows = conn.execute( - "SELECT site, expected_offset, actual_offset, schedule_enabled, schedule_time " - "FROM site_config" - ).fetchall() +# ============================ 周期性抓取调度（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 { - r[0]: { - "expected_offset": int(r[1]), - "actual_offset": int(r[2]), - "schedule_enabled": bool(r[3]), - "schedule_time": r[4] or "", + "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 } - for r in rows + 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 } @@ -367,7 +488,7 @@ def get_setting(site, key): """读取单站某个配置值；未设置返回 ''。""" if not os.path.exists(STATE_DB_PATH): return "" - with sqlite3.connect(STATE_DB_PATH) as conn: + 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() @@ -376,7 +497,7 @@ def get_setting(site, key): def set_setting(site, key, value): """设置单站某个配置值（upsert）。""" - with sqlite3.connect(STATE_DB_PATH) as conn: + 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", @@ -389,7 +510,7 @@ def get_site_settings(site): """返回单站全部配置 {key: value}。""" if not os.path.exists(STATE_DB_PATH): return {} - with sqlite3.connect(STATE_DB_PATH) as conn: + with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: rows = conn.execute( "SELECT key, value FROM site_settings WHERE site=?", (site,) ).fetchall() @@ -402,7 +523,32 @@ def get_site_settings(site): def create_task(site, kind): """新建一条 pending 任务，返回其 id。""" now = _now() - with sqlite3.connect(STATE_DB_PATH) as conn: + 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 (?, ?, ?, ?, '', '')", @@ -415,7 +561,7 @@ def create_task(site, kind): 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) as conn: + 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=?", @@ -431,7 +577,7 @@ def update_task(task_id, status, error=None): def get_task(task_id): """返回单条任务 dict，不存在返回 None。""" - with sqlite3.connect(STATE_DB_PATH) as conn: + 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=?", @@ -452,7 +598,7 @@ def get_task(task_id): def list_tasks(limit=20): """返回最近 limit 条任务（按 id 倒序）。""" - with sqlite3.connect(STATE_DB_PATH) as conn: + 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 ?", @@ -475,7 +621,7 @@ def list_tasks(limit=20): def fail_stale_tasks(reason: str = "服务重启，上轮未完成任务，请手动重跑") -> int: """worker 启动时调用：把遗留的 pending/running 任务标记为 failed，实现重启自愈。 返回被清理的任务数量。""" - with sqlite3.connect(STATE_DB_PATH) as conn: + 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 (?, ?)",