From 62a66467a9b0e91d089d00489e7f0d4f99c250c6 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Fri, 24 Jul 2026 11:04:04 +0800 Subject: [PATCH] feat(runtime): auto-ingest hook after successful download Co-Authored-By: Claude --- inbound_verify/runtime.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/inbound_verify/runtime.py b/inbound_verify/runtime.py index 2920dbd..70a35b4 100644 --- a/inbound_verify/runtime.py +++ b/inbound_verify/runtime.py @@ -530,6 +530,35 @@ def _record_business_date(site, kind): _write("undelivered", state_store.get_offset(site, "expected")) +def _persist_to_db(site, kind): + """下载成功后把本次数据入库 PostgreSQL（尽力而为，绝不外抛，不影响任务判定）。 + - __compare__ 无源数据，跳过。 + - auto_ingest=false 时跳过（无 PG/cpolar 的开发机）。 + - 懒导入 store 以回避 import 顺序（store↔compare 与 runtime↔compare 共存）。 + - 结果写 state_store.ingest_state，供 /api/status 反映入库健康。 + 所有写库/写状态都包 try/except：失败仅告警，绝不改变 dispatch_task 的 SUCCESS 判定。""" + if site == "__compare__": + return + try: + from inbound_verify import store # 懒导入：冷路径（每下载一次），回避成环 + except Exception as e: + print(f">> [warn] 入库模块不可用: {e}") + return + if not store.ingest_enabled(): + print(">> [入库] 已关闭 (auto_ingest=false)，跳过") + return + try: + count = store.ingest_task(site, kind) + state_store.set_ingest_state(site, kind, ok=True, count=count) + print(f">> [入库] {site}/{kind} 成功，{count} 条") + except Exception as e: + print(f">> [warn] 入库失败 {site}/{kind}: {e}") + try: + state_store.set_ingest_state(site, kind, ok=False, error=str(e)) + except Exception as e2: + print(f">> [warn] 写入库状态也失败: {e2}") + + def dispatch_task(ctx, task_spec): """执行一条任务。task_spec = {"site", "kind"}。返回 (status, error)。 @@ -554,6 +583,7 @@ def dispatch_task(ctx, task_spec): if ret is False: return (state_store.TASK_FAILED, "任务执行失败（重试耗尽）") _record_business_date(site, kind) + _persist_to_db(site, kind) return (state_store.TASK_SUCCESS, None) except Exception as e: return (state_store.TASK_FAILED, str(e))