From 5923cd05b1682b462d28c5e9619734137870f668 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 20 Aug 2026 09:33:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(scheduler):=20=E5=AE=9A=E6=97=B6=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E5=85=A8=E9=9D=A2=E9=A1=B5=E9=9D=A2=E5=8C=96(?= =?UTF-8?q?=E5=AD=90=E8=BF=9B=E7=A8=8B=E8=B7=91=E9=A1=B5=E9=9D=A2=E8=84=9A?= =?UTF-8?q?=E6=9C=AC,API=E7=9B=B4=E8=B0=83=E9=80=80=E5=87=BA=E8=B0=83?= =?UTF-8?q?=E5=BA=A6=E9=98=B2=E9=A3=8E=E6=8E=A7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 每日 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 --- README.md | 8 +- yunke-server/app/sync/scheduler.py | 119 ++++++++++++++++------ yunke-server/tests/test_sync_scheduler.py | 77 +++++++++++++- 3 files changed, 170 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 55d7f99..7d96ecf 100644 --- a/README.md +++ b/README.md @@ -99,7 +99,7 @@ That brings up: | `mysql` | `mysql:8.0` | Persistent data store | 3306 | | `redis` | `redis:7-alpine` | Cache / queue (future use; not yet wired into the app) | — | | `backend` | `yunke-server/` | FastAPI viewing API (uvicorn) | 8000 | -| `scheduler` | `yunke-server/` | APScheduler loop — daily full sync + every-4h messages | — | +| `scheduler` | `yunke-server/` | APScheduler loop — page-based sync (daily full + every-4h messages; needs Playwright/Chromium in the image) | — | | `frontend`| `yunkes/` | SPA served by nginx, reverse-proxies `/api` to backend:8000 | 5173 → 80 | Open . The scheduler emits one JSON-line log per @@ -137,6 +137,12 @@ python -m app.sync.cli scheduler # uses FULL_SYNC_HOUR + MESSAG python -m app.sync.cli scheduler --full-sync-hour 3 --message-interval-hours 6 ``` +> **2026-08-20 起:定时同步全部页面方式(Playwright 浏览器抓取),不走 API 直调** +> —— 每日全量跑 `scripts/full_sync_all.sh`,消息增量跑 +> `scripts/scrape_via_page.py --entity messages`。API 直调 +> (`sync --entity ...` / `run_once`) 保留为手动调试入口,不再挂任何定时触发器。 +> 容器部署需保证镜像内有 Playwright + Chromium。 + Entity names accepted by `sync --entity` and `run_once`: `departments`, `accounts`, `friends`, `groups`, `voice_calls`, `phone_calls`, `sms`, `messages`, `all_wechat`, `all` (default). diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index 7d13d50..1d665ed 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -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]: diff --git a/yunke-server/tests/test_sync_scheduler.py b/yunke-server/tests/test_sync_scheduler.py index 4d0baf6..6bcec2f 100644 --- a/yunke-server/tests/test_sync_scheduler.py +++ b/yunke-server/tests/test_sync_scheduler.py @@ -437,8 +437,11 @@ async def test_start_registers_two_jobs_with_expected_triggers(engine, session_m # IntervalTrigger exposes its interval as a timedelta on .interval`. assert interval.interval.total_seconds() == 6 * 3600 - # The full-sync job passes args=["all"] (so it hits _run_all). - assert full_call.kwargs["args"] == ["all"] + # 2026-08-20 起:两个 job 都打到 _safe_page_run(页面方式子进程), + # full 传 "full"、messages 传 "messages"。 + assert full_call.args[0].__name__ == "_safe_page_run" + assert full_call.kwargs["args"] == ["full"] + assert msg_call.args[0].__name__ == "_safe_page_run" assert msg_call.kwargs["args"] == ["messages"] # coalesce=True + max_instances=1 protects against overlap. @@ -510,3 +513,73 @@ def _mk_result( upserted=upserted, error=error, ) + + +# --------------------------------------------------------------------------- +# Page-based scheduled jobs (2026-08-20 — API 直调退出定时任务，全部页面方式防风控) +# --------------------------------------------------------------------------- + +class _FakeProc: + def __init__(self, rc: int = 0) -> None: + self._rc = rc + + async def wait(self) -> int: + return self._rc + + +async def test_safe_page_run_full_invokes_bash_full_sync(engine, monkeypatch): + """full job = bash scripts/full_sync_all.sh（页面方式全量序列，子进程隔离）。""" + sched = SyncScheduler(engine=engine) + captured: dict = {} + + async def fake_exec(*args, cwd=None, env=None): + captured["args"] = list(args) + captured["cwd"] = cwd + return _FakeProc(0) + + monkeypatch.setattr(scheduler_module.asyncio, "create_subprocess_exec", fake_exec) + monkeypatch.setattr(scheduler_module.shutil, "which", lambda name: "/usr/bin/bash") + + await sched._safe_page_run("full") + + assert captured["args"][0] == "/usr/bin/bash" + assert captured["args"][1].endswith("full_sync_all.sh") + # cwd 指向 yunke-server（脚本的相对路径基准） + assert str(captured["cwd"]).replace("\\", "/").endswith("yunke-server") + + +async def test_safe_page_run_messages_builds_scrape_command(engine, monkeypatch): + """messages job = python scripts/scrape_via_page.py --entity messages。""" + sched = SyncScheduler(engine=engine, max_conversations_per_run=123) + captured: dict = {} + + async def fake_exec(*args, cwd=None, env=None): + captured["args"] = list(args) + captured["env"] = env + return _FakeProc(0) + + monkeypatch.setattr(scheduler_module.asyncio, "create_subprocess_exec", fake_exec) + + await sched._safe_page_run("messages") + + args = captured["args"] + assert args[1].endswith("scrape_via_page.py") + i = args.index("--entity") + assert args[i + 1] == "messages" + j = args.index("--max-conversations") + assert args[j + 1] == "123" + + +async def test_safe_page_run_nonzero_rc_is_logged_not_raised(engine, monkeypatch, capsys): + """子进程非零退出只记日志不抛（调度器保住下一个周期）。""" + sched = SyncScheduler(engine=engine) + monkeypatch.setattr(scheduler_module.shutil, "which", lambda name: "/usr/bin/bash") + + async def fake_exec(*args, cwd=None, env=None): + return _FakeProc(3) + + monkeypatch.setattr(scheduler_module.asyncio, "create_subprocess_exec", fake_exec) + + await sched._safe_page_run("full") # 不应抛异常 + out = capsys.readouterr().out + assert '"rc": 3' in out or "'rc': 3" in out or "rc" in out