浅念
9a953553ec
- SearchView.vue: 用 highlightSegments() 返回 {text,hit} 数组替代 v-html+escapeHtml+regex,模板用 <template v-for> + <mark>/<span> 渲染,从结构上消除 XSS 风险(4 处 v-html 全部移除: 消息片段/客户昵称/realName/手机号)
- models.py: Message 新增 ix_message_channel_owner_room_time(channel,owner_account_id,room_id,timestamp),群聊查询过滤 room_id 而非 talker_id,原 talker 索引无法覆盖
- migrations/d8eee96cd017: 新增 room_id 索引的 upgrade/downgrade (offline SQL 已验证)
- base.py + scheduler.py + cli.py: 所有生产 create_async_engine 增加 pool_recycle=3600,在 MySQL wait_timeout(默认 8h)断连前主动回收
Co-Authored-By: Claude <noreply@anthropic.com>
711 lines
28 KiB
Python
711 lines
28 KiB
Python
"""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:
|
||
|
||
* **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.
|
||
|
||
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").
|
||
|
||
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.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import datetime as dt
|
||
import json
|
||
import logging
|
||
from collections.abc import Awaitable, Callable
|
||
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"]
|
||
|
||
# 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
|
||
|
||
# ------------------------------------------------------------------
|
||
# 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)
|
||
sched.add_job(
|
||
self._safe_run,
|
||
CronTrigger(hour=self.full_sync_hour, timezone=self.timezone),
|
||
args=["all"],
|
||
id="yunke_full_sync",
|
||
replace_existing=True,
|
||
coalesce=True,
|
||
max_instances=1,
|
||
)
|
||
sched.add_job(
|
||
self._safe_run,
|
||
IntervalTrigger(hours=self.message_interval_hours),
|
||
args=["messages"],
|
||
id="yunke_message_sync",
|
||
replace_existing=True,
|
||
coalesce=True,
|
||
max_instances=1,
|
||
)
|
||
sched.start()
|
||
self._scheduler = sched
|
||
logger.info(
|
||
"SyncScheduler started: 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_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
|