浅念
933 lines
38 KiB
Python
933 lines
38 KiB
Python
"""APScheduler-based periodic sync for the yunke data mirror (M6, Part A).
|
||
|
||
**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)
|
||
— 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``(对话发现:未扫过的排最前,其余按最近沟通排序)。
|
||
|
||
Subprocess isolation: 浏览器崩溃/断连不会带崩调度进程;子进程内部自带
|
||
断连重试与浏览器守护循环。同一时刻只允许一个页面子进程(asyncio.Lock)。
|
||
|
||
The API-direct path (``ENTITY_FUNCS`` / ``run_once`` / CLI ``sync --entity``)
|
||
保留给手动调试,不再挂任何定时触发器。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import datetime as dt
|
||
import json
|
||
import logging
|
||
import os
|
||
import shutil
|
||
import sys
|
||
import time
|
||
from collections.abc import Awaitable, Callable
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import sqlalchemy as sa
|
||
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
||
from apscheduler.triggers.cron import CronTrigger
|
||
from apscheduler.triggers.interval import IntervalTrigger
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||
|
||
from app.config import get_settings
|
||
from app.db import models # noqa: F401 (register tables on Base.metadata)
|
||
from app.db.base import Base
|
||
from app.db.models import SyncRun
|
||
from app.session.yunke_client import YunkeClient
|
||
from app.storage.object_store import ObjectStore
|
||
from app.sync.orchestrator import (
|
||
SyncResult,
|
||
sync_all_wechat,
|
||
sync_call_recordings,
|
||
sync_departments,
|
||
sync_message_images,
|
||
sync_phone_calls,
|
||
sync_sms,
|
||
sync_voice_clip_audio,
|
||
sync_voice_clips,
|
||
sync_wa_voice_calls,
|
||
sync_wechat_accounts,
|
||
sync_wechat_friends,
|
||
sync_wechat_groups,
|
||
sync_wechat_messages,
|
||
sync_wechat_voice_calls,
|
||
sync_whatsapp_accounts,
|
||
sync_whatsapp_friends,
|
||
sync_whatsapp_groups,
|
||
sync_whatsapp_messages,
|
||
)
|
||
|
||
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).
|
||
DEFAULT_MAX_CONVERSATIONS_PER_RUN = 200
|
||
|
||
# Advisory-lock TTL: a ``running`` SyncRun row older than this is considered
|
||
# stale (the process probably crashed mid-run) and a new invocation may
|
||
# proceed despite it. 60 min is generous — any entity fits inside — but
|
||
# short enough that a crashed run auto-recovers within an hour. Tuned to
|
||
# protect against concurrent CLI + scheduler invocations hammering 云控
|
||
# with the same scope twice.
|
||
ADVISORY_LOCK_TTL_MINUTES = 60
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Conversation discovery (drives the incremental message sync)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _discover_conversations(
|
||
session: AsyncSession,
|
||
*,
|
||
channel: str = "wechat",
|
||
limit: int = DEFAULT_MAX_CONVERSATIONS_PER_RUN,
|
||
) -> list[tuple[str, str]]:
|
||
"""Return distinct ``(owner_account_id, friend_id)`` pairs to sync.
|
||
|
||
Source: the local ``customer`` mirror, filtered by ``channel``
|
||
(default ``'wechat'``; pass ``'whatsapp'`` for WA conversations). Reading
|
||
the already-mirrored table — instead of the upstream ``talkerList``
|
||
endpoint — keeps the discovery call off the risk-control budget and makes
|
||
the incremental run purely a function of the customers we already know
|
||
about. Order is unspecified; ``limit`` caps work per run.
|
||
"""
|
||
from app.db.models import Customer
|
||
|
||
stmt = (
|
||
sa.select(Customer.owner_account_id, Customer.friend_id)
|
||
.where(Customer.channel == channel)
|
||
.where(Customer.owner_account_id.is_not(None))
|
||
.where(Customer.owner_account_id != "")
|
||
.distinct()
|
||
.limit(limit)
|
||
)
|
||
rows = (await session.execute(stmt)).all()
|
||
return [
|
||
(owner, friend_id)
|
||
for owner, friend_id in rows
|
||
if owner and friend_id
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Entity registry — name -> async callable(session, client, store) -> list[SyncResult]
|
||
# ---------------------------------------------------------------------------
|
||
#
|
||
# One entry per logical "thing the scheduler can run". Each callable returns a
|
||
# list of SyncResult (one for the entity, or one per conversation for the
|
||
# message entry). Keeping this as a registry means run_once("friends") and the
|
||
# scheduler's daily job can share the same dispatch.
|
||
|
||
|
||
async def _run_departments(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_departments(session, client)]
|
||
|
||
|
||
async def _run_accounts(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_wechat_accounts(session, client)]
|
||
|
||
|
||
async def _run_friends(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_wechat_friends(session, client)]
|
||
|
||
|
||
async def _run_groups(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_wechat_groups(session, client)]
|
||
|
||
|
||
async def _run_voice_calls(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_wechat_voice_calls(session, client)]
|
||
|
||
|
||
async def _run_wa_voice_calls(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_wa_voice_calls(session, client)]
|
||
|
||
|
||
async def _run_wa_accounts(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_whatsapp_accounts(session, client)]
|
||
|
||
|
||
async def _run_wa_friends(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_whatsapp_friends(session, client)]
|
||
|
||
|
||
async def _run_wa_groups(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_whatsapp_groups(session, client)]
|
||
|
||
|
||
async def _run_phone_calls(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_phone_calls(session, client)]
|
||
|
||
|
||
async def _run_sms(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
return [await sync_sms(session, client)]
|
||
|
||
|
||
async def _run_voice_clips(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
"""Per-employee scrape of 长语音 (voice_clip) across all staff."""
|
||
return [await sync_voice_clips(session, client)]
|
||
|
||
|
||
async def _run_call_recordings(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
"""Recording-localization pass for already-mirrored ``phone_call`` rows."""
|
||
return [await sync_call_recordings(session, client, store)]
|
||
|
||
|
||
async def _run_voice_clip_audio(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
"""Audio-localization pass for already-mirrored ``voice_clip`` rows."""
|
||
return [await sync_voice_clip_audio(session, client, store)]
|
||
|
||
|
||
async def _run_message_images(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
"""Image / media localization pass for already-mirrored ``message`` rows."""
|
||
return [await sync_message_images(session, client, store)]
|
||
|
||
|
||
async def _run_messages(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
"""Incremental message sync across all locally-known WeChat conversations."""
|
||
conversations = await _discover_conversations(session)
|
||
results: list[SyncResult] = []
|
||
for wechat_id, friend_id in conversations:
|
||
results.append(
|
||
await sync_wechat_messages(
|
||
session, client, store, wechat_id, friend_id
|
||
)
|
||
)
|
||
return results
|
||
|
||
|
||
async def _run_wa_messages(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
"""Incremental message sync across all locally-known WA conversations.
|
||
|
||
Mirrors :func:`_run_messages`: discovers ``(uid, friend_uid)`` pairs from
|
||
the local ``customer`` mirror (``channel='whatsapp'``) and calls
|
||
:func:`sync_whatsapp_messages` per conversation. ``store`` is accepted
|
||
for signature parity with the other runners but is NOT used — WA media
|
||
download is deferred to Phase 2.
|
||
"""
|
||
conversations = await _discover_conversations(session, channel="whatsapp")
|
||
results: list[SyncResult] = []
|
||
for uid, friend_uid in conversations:
|
||
results.append(
|
||
await sync_whatsapp_messages(session, client, uid, friend_uid)
|
||
)
|
||
return results
|
||
|
||
|
||
async def _run_all_wechat(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
"""Full WeChat sequence: org/entities first, then per-conversation messages.
|
||
|
||
Conversations are discovered from the local ``customer`` mirror so the
|
||
daily job is self-contained (no separate message kick needed).
|
||
"""
|
||
conversations = await _discover_conversations(session)
|
||
return await sync_all_wechat(
|
||
session, client, store, conversations=conversations
|
||
)
|
||
|
||
|
||
async def _run_all(
|
||
session: AsyncSession, client: YunkeClient, store: ObjectStore
|
||
) -> list[SyncResult]:
|
||
"""Everything: full WeChat sequence + phone calls + SMS.
|
||
|
||
This is the entity the scheduler's daily job runs.
|
||
"""
|
||
results = await _run_all_wechat(session, client, store)
|
||
# WhatsApp entities (accounts + friends). WA voice calls are already
|
||
# included via ``_run_all_wechat``; accounts + friends need their own
|
||
# kick here. WA friends depend on the department tree (root dept id),
|
||
# which ``_run_all_wechat`` seeds via sync_departments — so order matters.
|
||
results.append(await sync_whatsapp_accounts(session, client))
|
||
results.append(await sync_whatsapp_friends(session, client))
|
||
# WA chat-groups: share the ChatGroup table with wechat groups (channel
|
||
# distinguishes). Same department_id dependency as wa_friends. UNVERIFIED
|
||
# group-type switch — see scrape_whatsapp_groups.
|
||
results.append(await sync_whatsapp_groups(session, client))
|
||
# WA messages: per-conversation incremental sync, discovered from the
|
||
# already-mirrored ``customer`` table (channel='whatsapp'). Must run
|
||
# AFTER wa_friends (which seeds the customer rows the discovery reads).
|
||
results.extend(await _run_wa_messages(session, client, store))
|
||
results.append(await sync_phone_calls(session, client))
|
||
results.append(await sync_sms(session, client))
|
||
# 长语音 (voice_clip): per-employee scrape across both staff-account
|
||
# tables. Must run AFTER wechat_accounts + wa_accounts (which seed the
|
||
# user_id -> channel mapping the voice-clip sync reads).
|
||
results.append(await sync_voice_clips(session, client))
|
||
# Voice-clip audio localization: download OSS audio for already-mirrored
|
||
# voice_clip rows. Runs AFTER voice_clips so freshly-scraped rows are
|
||
# eligible. Commits internally (mirrors call_recordings) — placing it last
|
||
# in the daily job means no later in-batch entity can be affected by the
|
||
# mid-batch commit boundary.
|
||
results.append(await sync_voice_clip_audio(session, client, store))
|
||
# Message image / media localization: download OSS images / videos / files
|
||
# for already-mirrored ``message`` rows. Runs AFTER messages (synced
|
||
# earlier in this same daily job via _run_all_wechat + _run_wa_messages)
|
||
# so freshly-scraped rows are eligible. Commits internally (mirrors
|
||
# call_recordings / voice_clip_audio) — also placed at the tail of the
|
||
# daily job for the same mid-batch commit-boundary reason.
|
||
results.append(await sync_message_images(session, client, store))
|
||
return results
|
||
|
||
|
||
# Logical name -> runner. ``messages`` is the fast-cadence entry; ``all`` is
|
||
# the daily full-sync entry. The single-entity names are exposed for manual
|
||
# ``run_once("<entity>")`` triggers from the CLI / API.
|
||
ENTITY_FUNCS: dict[
|
||
str, Callable[[AsyncSession, YunkeClient, ObjectStore], Awaitable[list[SyncResult]]]
|
||
] = {
|
||
"departments": _run_departments,
|
||
"accounts": _run_accounts,
|
||
"friends": _run_friends,
|
||
"groups": _run_groups,
|
||
"voice_calls": _run_voice_calls,
|
||
"wa_voice_calls": _run_wa_voice_calls,
|
||
"wa_accounts": _run_wa_accounts,
|
||
"wa_friends": _run_wa_friends,
|
||
"wa_groups": _run_wa_groups,
|
||
"phone_calls": _run_phone_calls,
|
||
"sms": _run_sms,
|
||
"voice_clips": _run_voice_clips,
|
||
"call_recordings": _run_call_recordings,
|
||
"voice_clip_audio": _run_voice_clip_audio,
|
||
"message_images": _run_message_images,
|
||
"messages": _run_messages,
|
||
"wa_messages": _run_wa_messages,
|
||
"all_wechat": _run_all_wechat,
|
||
"all": _run_all,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# JSON-line logger — one line per run, easy to ship to any aggregator
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _log_run(
|
||
entity: str,
|
||
results: list[SyncResult],
|
||
*,
|
||
started_at: dt.datetime,
|
||
error: str | None = None,
|
||
) -> None:
|
||
"""Emit one JSON-line log entry summarising a finished run.
|
||
|
||
Always emits exactly one line, even on failure, so a log shipper can
|
||
parse the stream uniformly. ``level`` is ``info`` on full success,
|
||
``warning`` if any sub-result failed, ``error`` if the run itself blew
|
||
up before producing results.
|
||
"""
|
||
finished_at = dt.datetime.now(dt.UTC)
|
||
fetched = sum(r.fetched for r in results)
|
||
upserted = sum(r.upserted for r in results)
|
||
failed = [r.entity for r in results if r.status != "success"]
|
||
|
||
if error is not None:
|
||
level = "error"
|
||
elif failed:
|
||
level = "warning"
|
||
else:
|
||
level = "info"
|
||
|
||
payload: dict[str, Any] = {
|
||
"ts": finished_at.isoformat(),
|
||
"event": "sync_run",
|
||
"entity": entity,
|
||
"started_at": started_at.isoformat(),
|
||
"duration_sec": round((finished_at - started_at).total_seconds(), 3),
|
||
"results": [
|
||
{
|
||
"entity": r.entity,
|
||
"status": r.status,
|
||
"fetched": r.fetched,
|
||
"upserted": r.upserted,
|
||
"channel": r.channel,
|
||
"scope_id": r.scope_id,
|
||
"error": r.error,
|
||
}
|
||
for r in results
|
||
],
|
||
"totals": {"fetched": fetched, "upserted": upserted, "count": len(results)},
|
||
"failed_entities": failed,
|
||
"level": level,
|
||
}
|
||
if error is not None:
|
||
payload["error"] = error
|
||
|
||
line = json.dumps(payload, ensure_ascii=False, default=str)
|
||
# Logger so operators can route with structured-log filters; the JSON body
|
||
# is the whole message (no %-format) so a downstream parser sees one
|
||
# object per line.
|
||
logger.info(line)
|
||
# Also echo to stdout for human/`docker logs` consumers; flush so a
|
||
# piped reader sees it immediately.
|
||
print(line, flush=True)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SyncScheduler
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class SyncScheduler:
|
||
"""Periodic-sync driver wrapping an :class:`AsyncIOScheduler`.
|
||
|
||
Construction is cheap (no I/O); :meth:`start` opens the engines and
|
||
registers jobs, :meth:`stop` tears them down. :meth:`run_once` is a
|
||
standalone manual trigger that does NOT require :meth:`start` — useful
|
||
from the CLI / API.
|
||
|
||
Parameters
|
||
----------
|
||
full_sync_hour:
|
||
Local-time hour at which the daily full sync fires (0..23). Default 2.
|
||
message_interval_hours:
|
||
Hours between incremental message syncs. Default 4.
|
||
session_maker / engine:
|
||
Optional injected async engine + session maker (tests inject an
|
||
in-memory SQLite). When omitted, a production engine + maker are
|
||
built from :attr:`Settings.db_url`.
|
||
timezone:
|
||
APScheduler timezone string for the daily cron. Defaults to the
|
||
local machine timezone (matches operator expectation).
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
full_sync_hour: int = 2,
|
||
message_interval_hours: int = 4,
|
||
engine: Any | None = None,
|
||
session_maker: async_sessionmaker[AsyncSession] | None = None,
|
||
timezone: str | None = None,
|
||
max_conversations_per_run: int = DEFAULT_MAX_CONVERSATIONS_PER_RUN,
|
||
) -> None:
|
||
self.full_sync_hour = full_sync_hour
|
||
self.message_interval_hours = message_interval_hours
|
||
self.timezone = timezone
|
||
self.max_conversations_per_run = max_conversations_per_run
|
||
|
||
# Engine ownership flag — if the caller injected an engine, they own
|
||
# its lifecycle (a test will dispose it). Otherwise we own + dispose
|
||
# on :meth:`stop`.
|
||
self._owns_engine = engine is None
|
||
if engine is not None:
|
||
self._engine = engine
|
||
self._session_maker = session_maker or async_sessionmaker(
|
||
engine, expire_on_commit=False
|
||
)
|
||
else:
|
||
settings = get_settings()
|
||
self._engine = create_async_engine(
|
||
settings.db_url,
|
||
pool_pre_ping=True,
|
||
pool_recycle=3600, # recycle before MySQL wait_timeout closes us
|
||
future=True,
|
||
)
|
||
self._session_maker = async_sessionmaker(
|
||
self._engine, expire_on_commit=False
|
||
)
|
||
|
||
self._scheduler: AsyncIOScheduler | None = None
|
||
# 同一时刻只允许一个页面子进程（full 与 messages 互斥）
|
||
self._page_lock = asyncio.Lock()
|
||
|
||
# ------------------------------------------------------------------
|
||
# Public API
|
||
# ------------------------------------------------------------------
|
||
|
||
async def start(self) -> None:
|
||
"""Open the engine, create tables if missing, register jobs, start.
|
||
|
||
Idempotent: calling :meth:`start` twice without an intervening
|
||
:meth:`stop` is a no-op (the second call logs and returns).
|
||
"""
|
||
if self._scheduler is not None:
|
||
logger.warning("SyncScheduler.start() called twice; ignoring.")
|
||
return
|
||
|
||
# Create tables if missing (mirrors scripts.smoke_wechat). Cheap on a
|
||
# warm DB (a no-op CREATE that already-exists). Lets the scheduler
|
||
# boot on a fresh container without a separate migration step.
|
||
async with self._engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
|
||
sched = AsyncIOScheduler(timezone=self.timezone)
|
||
# 2026-08-20 起:定时任务全部页面方式(子进程),API 直调退出调度
|
||
sched.add_job(
|
||
self._safe_page_run,
|
||
CronTrigger(hour=self.full_sync_hour, timezone=self.timezone),
|
||
args=["full"],
|
||
id="yunke_full_sync",
|
||
replace_existing=True,
|
||
coalesce=True,
|
||
max_instances=1,
|
||
)
|
||
sched.add_job(
|
||
self._safe_page_run,
|
||
IntervalTrigger(hours=self.message_interval_hours),
|
||
args=["messages"],
|
||
id="yunke_message_sync",
|
||
replace_existing=True,
|
||
coalesce=True,
|
||
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,
|
||
)
|
||
# 4h 分析链:stat_daily_staff 滚动3天刷新 + 敏感词增量扫描(纯 DB,
|
||
# 无浏览器子进程;与消息增量同节奏,读滞后语义与「数据截至」一致)
|
||
sched.add_job(
|
||
self._analytics_tick,
|
||
IntervalTrigger(hours=self.message_interval_hours),
|
||
id="yunke_analytics_tick",
|
||
replace_existing=True,
|
||
coalesce=True,
|
||
max_instances=1,
|
||
)
|
||
sched.start()
|
||
self._scheduler = sched
|
||
logger.info(
|
||
"SyncScheduler started (page-based): full sync daily at hour=%s, messages every %sh",
|
||
self.full_sync_hour,
|
||
self.message_interval_hours,
|
||
)
|
||
|
||
async def stop(self, *, wait: bool = True) -> None:
|
||
"""Stop the scheduler and (if owned) dispose the engine."""
|
||
if self._scheduler is not None:
|
||
self._scheduler.shutdown(wait=wait)
|
||
self._scheduler = None
|
||
if self._owns_engine:
|
||
await self._engine.dispose()
|
||
|
||
async def run_once(self, entity: str = "all") -> list[SyncResult]:
|
||
"""Run one sync for ``entity`` (e.g. ``"friends"``, ``"all"``).
|
||
|
||
Does NOT require :meth:`start` — opens its own client + session for
|
||
the run and tears them down before returning. Returns the
|
||
orchestrator's :class:`SyncResult` list (one per entity / per
|
||
conversation). Raises ``ValueError`` on an unknown entity name so
|
||
the CLI can surface usage cleanly.
|
||
"""
|
||
if entity not in ENTITY_FUNCS:
|
||
raise ValueError(
|
||
f"unknown entity: {entity!r}. "
|
||
f"Known: {sorted(ENTITY_FUNCS)}"
|
||
)
|
||
# _safe_run handles the client build + try/except + logging; surface
|
||
# the results it produced (possibly an empty list on hard failure).
|
||
return await self._safe_run(entity, log=True)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 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``(增量:未扫对话优先,其余按最近沟通),末尾
|
||
追加 ``--entity moments --max-pages 3``(朋友圈浅扫,每员工最多 3 页)
|
||
|
||
子进程隔离——浏览器崩溃/断连不影响调度进程;非零退出只记 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
|
||
commands = [([bash, "scripts/full_sync_all.sh"], 22 * 3600)] # 22h 上限:防极端情况拖进次日相位(全量常态 10-20h)
|
||
elif kind == "messages":
|
||
# 2026-09-10 调度根治(用户报「员工列表时间差 2 天」):
|
||
# ① 序列头部刷员工账号表(accounts ~1min)——员工列表「数据
|
||
# 更新时间/最新消息时间」从此最多滞后 4h(此前只在每日
|
||
# full 刷,长任务挤掉 full 就停更);
|
||
# ② 消息实体只扫**近 24h 活跃**对话(--active-window-hours;
|
||
# 从未扫过的不受限)——全对话遍历(WA 9,357 个×~3.5s≈9h)
|
||
# 曾把序列拖到 15h+,超过 4h 间隔→与 02:00 full 背靠背
|
||
# 互相挤死;陈旧对话无新消息,每日 full 兜底;
|
||
# ③ 每命令超时保险丝(超时 kill 下轮补,增量幂等)。
|
||
# friends 仍先刷(最近沟通时间是排序键+活跃窗口的判定源)。
|
||
msg_entities = [
|
||
("accounts", None, 900),
|
||
("wa-accounts", None, 900),
|
||
("friends", None, 2400),
|
||
("wa-friends", None, 2400),
|
||
("messages", 24, 4800),
|
||
("wa-messages", 24, 4800),
|
||
# chatroom-stats 全窗(~1min):刷 chat_group.last_chat_time——
|
||
# 群消息活跃窗的判定源。不加则窗口用的是昨晚 full 的快照,
|
||
# 「昨天安静、今天突然活跃」的群会被窗口漏到明晚(2026-09-10)
|
||
("chatroom-stats", None, 1800),
|
||
("group-messages", 24, 4800),
|
||
("wa-group-messages", 24, 2400),
|
||
("moments", None, 2400),
|
||
]
|
||
commands = [
|
||
(
|
||
[
|
||
sys.executable,
|
||
"scripts/scrape_via_page.py",
|
||
"--entity",
|
||
entity,
|
||
]
|
||
+ (
|
||
[
|
||
"--max-conversations",
|
||
"99999",
|
||
"--active-window-hours",
|
||
str(window),
|
||
]
|
||
if window
|
||
else (
|
||
["--max-pages", "3"]
|
||
if entity == "moments"
|
||
else []
|
||
)
|
||
),
|
||
timeout,
|
||
)
|
||
for entity, window, timeout in msg_entities
|
||
]
|
||
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",
|
||
],
|
||
6 * 3600, # 单实体 6h 上限(2026-09-10:防周日的全历史
|
||
# 对齐拖进周一,把 02:00 full 又挤掉——同类事故根治)
|
||
)
|
||
for e in entities
|
||
]
|
||
else:
|
||
logger.error("page sync: unknown kind %r", kind)
|
||
return
|
||
|
||
rc = 0
|
||
for args, timeout in commands:
|
||
rc = await self._page_subprocess(args, timeout=timeout) or rc
|
||
if rc != 0:
|
||
logger.warning("page sync %s: command failed rc=%s, skipping rest", kind, rc)
|
||
break
|
||
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 _analytics_tick(self) -> None:
|
||
"""4h 分析链(staff 滚动3天 + 敏感词增量 + 管理域滚动)——异常只记日志不断链。
|
||
|
||
两段独立事务:段1 staff+敏感词(既有);段2 管理域(fact 近3天活跃会话
|
||
+ 宽表/通话 滚动3天,批次F 2026-09-10)。段2 失败只回滚自身,不牵连
|
||
段1 已提交结果(v2 Task12:指标失败不中断主分析链)。纯 DB 无浏览器
|
||
子进程,不抢 ``_page_lock``;``max_instances=1`` 防自身重叠即可。与
|
||
夜间 full_sync 的汇总/管理步重叠时,两侧 DELETE+INSERT 在重叠 PK
|
||
范围可能锁等待——非致命:tick 吞异常下轮重试,夜间脚本幂等自愈。
|
||
|
||
时区:``day_key`` 内部钉死 Asia/Shanghai,喂 UTC epoch
|
||
(``time.time()*1000``)即无歧义。
|
||
"""
|
||
from app.storage.summary import day_key
|
||
|
||
today = day_key(int(time.time() * 1000))
|
||
# ── 段1:staff 滚动 + 敏感词增量(既有,语义原样) ──
|
||
try:
|
||
async with self._session_maker() as session:
|
||
from app.analytics.sensitive import scanner
|
||
from app.storage.staff_summary import recompute_days, rolling_days
|
||
|
||
n = await recompute_days(session, rolling_days(today, 3))
|
||
scan = await scanner.run_incremental(session)
|
||
await session.commit()
|
||
logger.info(
|
||
"[analytics-tick] staff_days=%s scanned=%s hits=%s",
|
||
n, scan.get("scanned"), scan.get("hits"),
|
||
)
|
||
except Exception: # noqa: BLE001 — scheduler MUST NOT crash
|
||
logger.exception("analytics tick failed")
|
||
|
||
# ── 段2:管理域滚动(fact 活跃会话 + 宽表/通话,当日+滚动3天) ──
|
||
try:
|
||
async with self._session_maker() as session:
|
||
from app.storage.employee_metrics import (
|
||
recompute_call_days,
|
||
recompute_metric_days,
|
||
)
|
||
from app.storage.first_reply_facts import (
|
||
pairs_active_in_days,
|
||
recompute_pairs,
|
||
)
|
||
from app.storage.staff_summary import rolling_days
|
||
|
||
rd = rolling_days(today, 3)
|
||
pairs = await pairs_active_in_days(session, rd)
|
||
f = await recompute_pairs(session, pairs)
|
||
w = await recompute_metric_days(session, rd)
|
||
c = await recompute_call_days(session, rd)
|
||
await session.commit()
|
||
logger.info(
|
||
"[analytics-tick] mgmt fact_pairs=%s wide_rows=%s call_rows=%s",
|
||
f, w, c,
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
logger.exception("management metrics tick failed")
|
||
|
||
async def _page_subprocess(self, args: list[str], *, timeout: float | None = None) -> int:
|
||
"""启动页面子进程并等待结束;继承环境 + 补 DB_URL/PYTHONIOENCODING。
|
||
|
||
``timeout``(秒,2026-09-10 调度根治保险丝):超时 kill 子进程返回
|
||
非零——上游 break 跳过本序列剩余命令,增量幂等下轮自动补,防止
|
||
单命令拖死整个调度相位(4h 序列曾因此 15h+ 与 full 互相挤死)。
|
||
"""
|
||
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,
|
||
)
|
||
if timeout is None:
|
||
return await proc.wait()
|
||
try:
|
||
return await asyncio.wait_for(proc.wait(), timeout)
|
||
except asyncio.TimeoutError:
|
||
logger.error(
|
||
"page sync subprocess timeout after %.0fs (killed): %s",
|
||
timeout, " ".join(args[-6:]),
|
||
)
|
||
proc.kill()
|
||
await proc.wait()
|
||
return 124 # 超时语义(同 shell timeout 惯例)
|
||
|
||
async def _safe_run(
|
||
self, entity: str, *, log: bool = True
|
||
) -> list[SyncResult]:
|
||
"""Run one entity, swallowing + logging any unexpected exception.
|
||
|
||
This is the failure-policy wrapper that protects the scheduler loop:
|
||
a raised exception becomes a logged error and an empty result list,
|
||
and the scheduler keeps its next appointment.
|
||
|
||
Concurrency: an advisory lock refuses to run if another ``_safe_run``
|
||
for the same entity is still active. The lock is implemented by
|
||
staking a ``running`` SyncRun row keyed on the scheduler entity name
|
||
(e.g. ``"all"`` / ``"messages"`` / ``"friends"``) at the start and
|
||
finalizing it on completion. A concurrent invocation sees the row and
|
||
skips. The 60-min TTL (:data:`ADVISORY_LOCK_TTL_MINUTES`) auto-recovers
|
||
from a crashed process: a stale ``running`` row eventually falls
|
||
outside the window and a new run may proceed.
|
||
"""
|
||
started_at = dt.datetime.now(dt.UTC)
|
||
client: YunkeClient | None = None
|
||
results: list[SyncResult] = []
|
||
try:
|
||
client = await _build_client_from_env()
|
||
async with self._session_maker() as session:
|
||
# --- Advisory lock check ------------------------------
|
||
# ``SyncRun.started_at`` is a non-tz-aware DateTime; reading
|
||
# a row back yields a naive datetime, so the cutoff is built
|
||
# naive-UTC to keep the comparison type-consistent.
|
||
cutoff = (
|
||
dt.datetime.now(dt.UTC).replace(tzinfo=None)
|
||
- dt.timedelta(minutes=ADVISORY_LOCK_TTL_MINUTES)
|
||
)
|
||
existing = (
|
||
await session.execute(
|
||
sa.select(SyncRun)
|
||
.where(SyncRun.entity_type == entity)
|
||
.where(SyncRun.status == "running")
|
||
.where(SyncRun.started_at > cutoff)
|
||
.limit(1)
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
logger.warning(
|
||
"sync %s: skipped — another run is active"
|
||
" (started %s)",
|
||
entity,
|
||
existing.started_at.isoformat(),
|
||
)
|
||
skipped = SyncResult(
|
||
entity=entity,
|
||
status="skipped",
|
||
fetched=0,
|
||
upserted=0,
|
||
error=f"another sync is running for {entity}",
|
||
)
|
||
if log:
|
||
_log_run(entity, [skipped], started_at=started_at)
|
||
return [skipped]
|
||
|
||
# --- Stake our own 'running' row ----------------------
|
||
# A wrapping SyncRun at the scheduler-entity level (distinct
|
||
# from any sub-entity rows the orchestrator creates internally).
|
||
# On a caught exception the session rolls back and this row
|
||
# disappears with everything else — no phantom to recover
|
||
# from. On the success path the row is finalized + committed
|
||
# below. If the process hard-crashes between flush and commit,
|
||
# the row stays 'running' and blocks retries for up to TTL.
|
||
run = SyncRun(
|
||
entity_type=entity,
|
||
status="running",
|
||
started_at=dt.datetime.now(dt.UTC),
|
||
)
|
||
session.add(run)
|
||
await session.flush()
|
||
|
||
runner = ENTITY_FUNCS[entity]
|
||
results = await runner(session, client, ObjectStore())
|
||
|
||
# --- Finalize the wrapping SyncRun --------------------
|
||
run.status = (
|
||
"success"
|
||
if all(r.status == "success" for r in results)
|
||
else "failed"
|
||
)
|
||
run.records_fetched = sum(r.fetched for r in results)
|
||
run.records_upserted = sum(r.upserted for r in results)
|
||
run.finished_at = dt.datetime.now(dt.UTC)
|
||
if run.status == "failed":
|
||
joined = "; ".join(
|
||
filter(None, (r.error for r in results))
|
||
)
|
||
run.error = joined or None
|
||
await session.commit()
|
||
except Exception as exc: # noqa: BLE001 — scheduler MUST NOT crash
|
||
logger.exception("sync %s failed", entity)
|
||
if log:
|
||
_log_run(entity, results, started_at=started_at, error=repr(exc))
|
||
return results
|
||
|
||
if log:
|
||
_log_run(entity, results, started_at=started_at)
|
||
return results
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Client factory (env-driven; session-cookie mode preferred)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _build_client_from_env() -> YunkeClient:
|
||
"""Construct + authenticate a :class:`YunkeClient` from env / ``.env``.
|
||
|
||
Lifts non-empty auth values from :func:`app.config.get_settings` (which
|
||
loads ``.env``) into ``os.environ`` first, then delegates to
|
||
:func:`scripts.smoke_wechat.build_client_from_env` so the precedence
|
||
(session-cookie mode wins; pcLogin fallback) stays in one place.
|
||
|
||
In login mode, performs the actual ``await client.login()`` (real RSA +
|
||
``pcLogin``); in session-cookie mode the client is already authenticated
|
||
and no network call is made here.
|
||
|
||
Raises ``RuntimeError`` if no auth env is configured at all.
|
||
"""
|
||
# Imported lazily so a misconfigured ``scripts`` module never breaks
|
||
# import of :mod:`app.sync.scheduler` (e.g. in unit tests that inject
|
||
# their own engine / client).
|
||
from scripts.smoke_wechat import (
|
||
_push_settings_into_environ,
|
||
build_client_from_env,
|
||
)
|
||
|
||
_push_settings_into_environ()
|
||
client = build_client_from_env()
|
||
if client is None:
|
||
raise RuntimeError(
|
||
"no yunke auth configured: set YK_USERID+YK_COMPANY (preferred) "
|
||
"or YUNKE_USERNAME+YUNKE_PASSWORD in .env / environment."
|
||
)
|
||
if not client.logged_in:
|
||
# pcLogin mode — actually log in now. May raise; the caller's
|
||
# _safe_run wrapper will catch + log.
|
||
await client.login()
|
||
return client
|