LogisticsHubIPA/InboundVerify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(schedule): 周期抓取调度取代每日定点下载

Browse Source 
应到/实到数据各自独立周期抓取(启用+激活时段+频率)，抓取与处理解耦，抓完自动落库(复用现有 _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 <noreply@anthropic.com>
This commit is contained in:
Misaka
2026-07-28 22:58:47 +08:00
parent 82ab6d8be3
commit d74a35844f
2 changed files with 261 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
inbound_verify/state_store.py
View File

@@ -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 (?, ?)",