feat(runtime): auto-ingest hook after successful download

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
2026-07-24 11:04:04 +08:00
parent 7211846375
commit 62a66467a9

View File

@@ -530,6 +530,35 @@ def _record_business_date(site, kind):
_write("undelivered", state_store.get_offset(site, "expected")) _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): def dispatch_task(ctx, task_spec):
"""执行一条任务。task_spec = {"site", "kind"}。返回 (status, error)。 """执行一条任务。task_spec = {"site", "kind"}。返回 (status, error)。
@@ -554,6 +583,7 @@ def dispatch_task(ctx, task_spec):
if ret is False: if ret is False:
return (state_store.TASK_FAILED, "任务执行失败(重试耗尽)") return (state_store.TASK_FAILED, "任务执行失败(重试耗尽)")
_record_business_date(site, kind) _record_business_date(site, kind)
_persist_to_db(site, kind)
return (state_store.TASK_SUCCESS, None) return (state_store.TASK_SUCCESS, None)
except Exception as e: except Exception as e:
return (state_store.TASK_FAILED, str(e)) return (state_store.TASK_FAILED, str(e))