feat(scheduler): 定时同步全面页面化(子进程跑页面脚本,API直调退出调度防风控)
- 每日 full: bash scripts/full_sync_all.sh;每4h messages: scrape_via_page --entity messages - 子进程隔离+asyncio.Lock互斥+非零退出只记JSON日志;自动补DB_URL环境 - ENTITY_FUNCS/run_once/CLI保留为手动调试,不再挂定时触发器 - README同步更新(Docker镜像需Playwright+Chromium);413测试绿 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -1,35 +1,22 @@
|
||||
"""APScheduler-based periodic sync for the yunke data mirror (M6, Part A).
|
||||
|
||||
Wraps the orchestrator's ``sync_*`` functions behind a long-running scheduler
|
||||
so the data mirror stays fresh without manual kicks. Two paces:
|
||||
**Page-based since 2026-08-20** — 定时任务全部改为子进程跑页面抓取器
|
||||
（``scripts/scrape_via_page.py`` / ``scripts/full_sync_all.sh``），不再走
|
||||
API 直调，规避云客风控。Two paces:
|
||||
|
||||
* **Daily full sync** at a configurable hour (default ``02:00`` local time,
|
||||
off-hours) — re-scrapes the org tree, accounts, friends, groups, voice
|
||||
calls, plus phone calls and SMS (all entities that have no upstream
|
||||
``since`` cursor and therefore rely on idempotent re-upsert).
|
||||
* **Incremental message sync** more frequently (default every **4 hours**).
|
||||
Messages DO have a real ``[last_timestamp, now]`` window (see
|
||||
:func:`app.sync.orchestrator.sync_wechat_messages`), so a fast cadence
|
||||
catches new chats promptly without re-pulling history. The conversation
|
||||
list is discovered from the already-mirrored ``customer`` table.
|
||||
* **Daily full sync** at a configurable hour (default ``02:00`` local time)
|
||||
— runs ``bash scripts/full_sync_all.sh`` (accounts/customers/groups/WA 群/
|
||||
voice calls + 录音重链接/voice clips/messages/group messages + 成员派生/
|
||||
WA messages/media,全部浏览器方式;call_recordings 仅走公开 OSS)。
|
||||
* **Incremental message sync** more frequently (default every **4 hours**)
|
||||
— runs ``python scripts/scrape_via_page.py --entity messages
|
||||
--max-conversations N``(对话发现:未扫过的排最前,其余按最近沟通排序)。
|
||||
|
||||
Auth: the scheduler builds a :class:`YunkeClient` per run from env (honoring
|
||||
``.env`` via :func:`scripts.smoke_wechat._push_settings_into_environ`).
|
||||
Session-cookie mode (``YK_USERID`` + ``YK_COMPANY`` [+ ``YK_COOKIE``]) is
|
||||
preferred and is the only mode exercised in production today; pcLogin mode
|
||||
(``YUNKE_USERNAME`` + ``YUNKE_PASSWORD``) is wired but deferred (see README
|
||||
"Current limitations").
|
||||
Subprocess isolation: 浏览器崩溃/断连不会带崩调度进程;子进程内部自带
|
||||
断连重试与浏览器守护循环。同一时刻只允许一个页面子进程(asyncio.Lock)。
|
||||
|
||||
Failure policy: every run is wrapped in a ``try/except`` that logs the error
|
||||
and lets the scheduler keep its next appointment. A single broken entity
|
||||
(scraper 500, network blip) does NOT bring down the loop. Per-entity
|
||||
failures are also captured by the orchestrator as ``status='failed'``
|
||||
``SyncRun`` rows.
|
||||
|
||||
Threading: :class:`SyncScheduler` is built on :class:`AsyncIOScheduler` and
|
||||
is intended to be ``await``-ed inside a long-running asyncio process (the
|
||||
CLI's ``scheduler`` command — see :mod:`app.sync.cli`). ``run_once`` is
|
||||
also awaitable for manual / API-triggered syncs.
|
||||
The API-direct path (``ENTITY_FUNCS`` / ``run_once`` / CLI ``sync --entity``)
|
||||
保留给手动调试,不再挂任何定时触发器。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -38,7 +25,11 @@ import asyncio
|
||||
import datetime as dt
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
from collections.abc import Awaitable, Callable
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import sqlalchemy as sa
|
||||
@@ -79,6 +70,9 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
__all__ = ["SyncScheduler", "ENTITY_FUNCS"]
|
||||
|
||||
# yunke-server 仓库目录（页面脚本/全量脚本的相对路径基准）
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
# Cap on the number of conversations one message-sync run will pull. Bounds
|
||||
# runtime + upstream request volume per run (the orchestrator paces each
|
||||
# request, so 200 conversations × N pages each can still take a while).
|
||||
@@ -487,6 +481,8 @@ class SyncScheduler:
|
||||
)
|
||||
|
||||
self._scheduler: AsyncIOScheduler | None = None
|
||||
# 同一时刻只允许一个页面子进程（full 与 messages 互斥）
|
||||
self._page_lock = asyncio.Lock()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public API
|
||||
@@ -509,17 +505,18 @@ class SyncScheduler:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
sched = AsyncIOScheduler(timezone=self.timezone)
|
||||
# 2026-08-20 起:定时任务全部页面方式(子进程),API 直调退出调度
|
||||
sched.add_job(
|
||||
self._safe_run,
|
||||
self._safe_page_run,
|
||||
CronTrigger(hour=self.full_sync_hour, timezone=self.timezone),
|
||||
args=["all"],
|
||||
args=["full"],
|
||||
id="yunke_full_sync",
|
||||
replace_existing=True,
|
||||
coalesce=True,
|
||||
max_instances=1,
|
||||
)
|
||||
sched.add_job(
|
||||
self._safe_run,
|
||||
self._safe_page_run,
|
||||
IntervalTrigger(hours=self.message_interval_hours),
|
||||
args=["messages"],
|
||||
id="yunke_message_sync",
|
||||
@@ -530,7 +527,7 @@ class SyncScheduler:
|
||||
sched.start()
|
||||
self._scheduler = sched
|
||||
logger.info(
|
||||
"SyncScheduler started: full sync daily at hour=%s, messages every %sh",
|
||||
"SyncScheduler started (page-based): full sync daily at hour=%s, messages every %sh",
|
||||
self.full_sync_hour,
|
||||
self.message_interval_hours,
|
||||
)
|
||||
@@ -565,6 +562,66 @@ class SyncScheduler:
|
||||
# Internals
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def _safe_page_run(self, kind: str) -> None:
|
||||
"""定时入口:子进程跑页面抓取(2026-08-20 起,API 直调退出调度)。
|
||||
|
||||
* ``full`` → ``bash scripts/full_sync_all.sh``(页面方式全量序列)
|
||||
* ``messages`` → ``python scripts/scrape_via_page.py --entity messages
|
||||
--max-conversations N``(增量:未扫对话优先,其余按最近沟通)
|
||||
|
||||
子进程隔离——浏览器崩溃/断连不影响调度进程;非零退出只记 JSON 日志
|
||||
不抛异常(保住下一个周期)。asyncio.Lock 保证 full/messages 互斥。
|
||||
"""
|
||||
async with self._page_lock:
|
||||
started_at = dt.datetime.now(dt.UTC)
|
||||
if kind == "full":
|
||||
bash = shutil.which("bash")
|
||||
if not bash:
|
||||
logger.error("page sync 'full': bash not found on PATH")
|
||||
return
|
||||
args: list[str] = [bash, "scripts/full_sync_all.sh"]
|
||||
elif kind == "messages":
|
||||
args = [
|
||||
sys.executable,
|
||||
"scripts/scrape_via_page.py",
|
||||
"--entity",
|
||||
"messages",
|
||||
"--max-conversations",
|
||||
str(self.max_conversations_per_run),
|
||||
]
|
||||
else:
|
||||
logger.error("page sync: unknown kind %r", kind)
|
||||
return
|
||||
|
||||
rc = await self._page_subprocess(args)
|
||||
finished_at = dt.datetime.now(dt.UTC)
|
||||
line = json.dumps(
|
||||
{
|
||||
"ts": finished_at.isoformat(),
|
||||
"event": "page_sync_run",
|
||||
"kind": kind,
|
||||
"rc": rc,
|
||||
"duration_sec": round((finished_at - started_at).total_seconds(), 3),
|
||||
"level": "info" if rc == 0 else "error",
|
||||
},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
logger.info(line)
|
||||
print(line, flush=True)
|
||||
|
||||
async def _page_subprocess(self, args: list[str]) -> int:
|
||||
"""启动页面子进程并等待结束;继承环境 + 补 DB_URL/PYTHONIOENCODING。"""
|
||||
env = {**os.environ, "PYTHONIOENCODING": "utf-8"}
|
||||
settings = get_settings()
|
||||
if settings.db_url and "DB_URL" not in env:
|
||||
env["DB_URL"] = settings.db_url
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*args,
|
||||
cwd=str(REPO_ROOT),
|
||||
env=env,
|
||||
)
|
||||
return await proc.wait()
|
||||
|
||||
async def _safe_run(
|
||||
self, entity: str, *, log: bool = True
|
||||
) -> list[SyncResult]:
|
||||
|
||||
Reference in New Issue
Block a user