feat(scheduler): 每周日05:00全量对齐job(四消息实体--full-history,修撤回/编辑+消除滞后窗口)
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -524,6 +524,17 @@ class SyncScheduler:
|
|||||||
coalesce=True,
|
coalesce=True,
|
||||||
max_instances=1,
|
max_instances=1,
|
||||||
)
|
)
|
||||||
|
# 每周全量对齐(周日 05:00):--full-history 强制翻全历史,修增量模式
|
||||||
|
# 抓不到的撤回/编辑 + 消除 lastChatTime 滞后窗口
|
||||||
|
sched.add_job(
|
||||||
|
self._safe_page_run,
|
||||||
|
CronTrigger(day_of_week="sun", hour=5, timezone=self.timezone),
|
||||||
|
args=["full-history"],
|
||||||
|
id="yunke_weekly_full_history",
|
||||||
|
replace_existing=True,
|
||||||
|
coalesce=True,
|
||||||
|
max_instances=1,
|
||||||
|
)
|
||||||
sched.start()
|
sched.start()
|
||||||
self._scheduler = sched
|
self._scheduler = sched
|
||||||
logger.info(
|
logger.info(
|
||||||
@@ -595,6 +606,22 @@ class SyncScheduler:
|
|||||||
str(self.max_conversations_per_run),
|
str(self.max_conversations_per_run),
|
||||||
],
|
],
|
||||||
]
|
]
|
||||||
|
elif kind == "full-history":
|
||||||
|
# 每周全量对齐(周日 05:00):强制全量翻历史,对齐增量模式抓不到的
|
||||||
|
# 撤回/编辑消息 + 彻底消除 lastChatTime 滞后窗口(2026-08-21)
|
||||||
|
entities = ["messages", "group-messages", "wa-messages", "wa-group-messages"]
|
||||||
|
commands = [
|
||||||
|
[
|
||||||
|
sys.executable,
|
||||||
|
"scripts/scrape_via_page.py",
|
||||||
|
"--entity",
|
||||||
|
e,
|
||||||
|
"--max-conversations",
|
||||||
|
"99999",
|
||||||
|
"--full-history",
|
||||||
|
]
|
||||||
|
for e in entities
|
||||||
|
]
|
||||||
else:
|
else:
|
||||||
logger.error("page sync: unknown kind %r", kind)
|
logger.error("page sync: unknown kind %r", kind)
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -416,10 +416,10 @@ async def test_start_registers_two_jobs_with_expected_triggers(engine, session_m
|
|||||||
with patch.object(scheduler_module, "AsyncIOScheduler", return_value=spy):
|
with patch.object(scheduler_module, "AsyncIOScheduler", return_value=spy):
|
||||||
await sched.start()
|
await sched.start()
|
||||||
|
|
||||||
# Two jobs registered.
|
# Three jobs registered (2026-08-21 加每周全量对齐).
|
||||||
assert spy.add_job.call_count == 2
|
assert spy.add_job.call_count == 3
|
||||||
job_ids = {c.kwargs["id"] for c in spy.add_job.call_args_list}
|
job_ids = {c.kwargs["id"] for c in spy.add_job.call_args_list}
|
||||||
assert job_ids == {"yunke_full_sync", "yunke_message_sync"}
|
assert job_ids == {"yunke_full_sync", "yunke_message_sync", "yunke_weekly_full_history"}
|
||||||
|
|
||||||
# Full sync -> CronTrigger(hour=3); messages -> IntervalTrigger(hours=6).
|
# Full sync -> CronTrigger(hour=3); messages -> IntervalTrigger(hours=6).
|
||||||
# add_job was called as (func, trigger, kwargs...) so trigger is args[1].
|
# add_job was called as (func, trigger, kwargs...) so trigger is args[1].
|
||||||
@@ -444,6 +444,16 @@ async def test_start_registers_two_jobs_with_expected_triggers(engine, session_m
|
|||||||
assert msg_call.args[0].__name__ == "_safe_page_run"
|
assert msg_call.args[0].__name__ == "_safe_page_run"
|
||||||
assert msg_call.kwargs["args"] == ["messages"]
|
assert msg_call.kwargs["args"] == ["messages"]
|
||||||
|
|
||||||
|
# Weekly full-history: Sunday 05:00 CronTrigger.
|
||||||
|
weekly_call = next(
|
||||||
|
c for c in spy.add_job.call_args_list if c.kwargs["id"] == "yunke_weekly_full_history"
|
||||||
|
)
|
||||||
|
weekly_trigger = weekly_call.args[1]
|
||||||
|
assert "day_of_week='sun'" in str(weekly_trigger)
|
||||||
|
assert "hour='5'" in str(weekly_trigger) or "hour='5'" in repr(weekly_trigger)
|
||||||
|
assert weekly_call.args[0].__name__ == "_safe_page_run"
|
||||||
|
assert weekly_call.kwargs["args"] == ["full-history"]
|
||||||
|
|
||||||
# coalesce=True + max_instances=1 protects against overlap.
|
# coalesce=True + max_instances=1 protects against overlap.
|
||||||
for c in spy.add_job.call_args_list:
|
for c in spy.add_job.call_args_list:
|
||||||
assert c.kwargs["coalesce"] is True
|
assert c.kwargs["coalesce"] is True
|
||||||
|
|||||||
Reference in New Issue
Block a user