yunke/yunke-server/app/sync/scheduler.py

"""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
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,
        )
        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: list[list[str]] = [[bash, "scripts/full_sync_all.sh"]]
            elif kind == "messages":
                # 先刷客户列表(最近沟通时间是云客列表的实时排序键,friends 快照
                # 滞后会让本地排序与云客第一页完全不同,2026-08-20),再抓全部四类
                # 消息增量(2026-08-21 全接入:水位增量后单对话 1 页,可承受);
                # 末尾追加员工朋友圈浅扫(2026-08-25,max-pages 3 即每员工最多
                # 3 页,新帖通常落第一页;全量翻页由每日 full 走 full_sync_all.sh)
                commands = [
                    [sys.executable, "scripts/scrape_via_page.py", "--entity", "friends"],
                    [sys.executable, "scripts/scrape_via_page.py", "--entity", "wa-friends"],
                ] + [
                    [
                        sys.executable,
                        "scripts/scrape_via_page.py",
                        "--entity",
                        e,
                        "--max-conversations",
                        "99999",
                    ]
                    for e in ("messages", "wa-messages", "group-messages", "wa-group-messages")
                ] + [
                    [
                        sys.executable,
                        "scripts/scrape_via_page.py",
                        "--entity",
                        "moments",
                        "--max-pages",
                        "3",
                    ]
                ]
            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:
                logger.error("page sync: unknown kind %r", kind)
                return

            rc = 0
            for args in commands:
                rc = await self._page_subprocess(args) 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 _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]:
        """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