diff --git a/README.md b/README.md new file mode 100644 index 0000000..55d7f99 --- /dev/null +++ b/README.md @@ -0,0 +1,177 @@ +# yunke — 云客 CRM 数据本地化系统 + +A self-hosted mirror of the yunke CRM (`https://crm2.yunkecn.com`): a Python +backend that periodically scrapes WeChat / WhatsApp / phone-call / SMS data +into MySQL + a local object store, plus a Vue SPA for browsing it. The +system is read-only against the upstream CRM — every upstream call is a +list / search endpoint; nothing writes back to yunke. + +## What's here + +``` +. +├── yunke-server/ Python backend (FastAPI + APScheduler + scrapers + repos) +├── yunkes/ Vue 3 + Element Plus frontend (Vite build → nginx) +├── docker-compose.yml Five-service stack (mysql / redis / backend / scheduler / frontend) +└── docs/ Design + plan documents +``` + +## Architecture (1 minute version) + +The backend has three concentric layers: + +1. **Session layer** (`app/session/yunke_client.py`) — a single + `YunkeClient` chokepoint that injects the auth headers, paces requests + with a jittered rate limiter, backs off on risk-control signals (HTTP + 429 or 风控 body markers), and re-logs-in once on 401. +2. **Scraper layer** (`app/scraper/*`) — one paginator + parser per + upstream endpoint, yielding snake_case dicts. Pure-ish: takes a + `YunkeClient`, yields records, never touches the DB. +3. **Sync / persistence layer** (`app/sync/orchestrator.py` + + `app/storage/repos.py`) — the orchestrator runs each entity's + `scrape → upsert → advance watermark` loop, logs every run as a + `SyncRun` row, and never re-raises a per-entity failure (one broken + entity doesn't abort a batch). The repos do dialect-aware idempotent + upserts (SQLite for tests, MySQL for prod) so a re-scrape of the same + row is one refreshed row, not a duplicate. + +Messages have a real incremental window (`[last_timestamp, now]`); every +other entity has no upstream `since` param and relies on idempotent +re-scrape. See the [design doc](docs/superpowers/specs/2026-08-01-yunke-data-mirror-design.md) +for the full data dictionary + risk-control rationale. + +## Quick start + +### Dev mode (no Docker) + +You need: Python 3.11+, Node 20+, and (optionally) a local MySQL. SQLite +works for a quick smoke. + +Backend: + +```bash +cd yunke-server +python -m venv .venv +.venv/Scripts/activate # Windows +# source .venv/bin/activate # *nix +pip install -e ".[dev]" + +# Configure auth + DB (see .env.example). Session-cookie mode preferred: +# YK_USERID=... +# YK_COMPANY=... +# YK_COOKIE="user=...; userToken=..." +cp .env.example .env +# edit .env + +# Run the API: +uvicorn app.api.main:app --reload +``` + +Frontend (separate shell): + +```bash +cd yunkes +npm install +npm run dev # serves on :5173, proxies /api → :8000 +``` + +Open . + +One-shot sync into the configured DB: + +```bash +cd yunke-server +python -m app.sync.cli sync # everything +python -m app.sync.cli sync --entity friends # one entity +python -m app.sync.cli --help # full usage +``` + +### Docker mode (full stack) + +```bash +docker compose up --build +``` + +That brings up: + +| Service | Image / build | Role | Port | +|-----------|---------------------|------------------------------------------------------------|------------| +| `mysql` | `mysql:8.0` | Persistent data store | 3306 | +| `redis` | `redis:7-alpine` | Cache / queue (future use; not yet wired into the app) | — | +| `backend` | `yunke-server/` | FastAPI viewing API (uvicorn) | 8000 | +| `scheduler` | `yunke-server/` | APScheduler loop — daily full sync + every-4h messages | — | +| `frontend`| `yunkes/` | SPA served by nginx, reverse-proxies `/api` to backend:8000 | 5173 → 80 | + +Open . The scheduler emits one JSON-line log per +run; tail with `docker compose logs -f scheduler`. + +## Configuration + +All configuration is via environment / `.env` (the docker-compose stack +passes the host `.env` through to the containers). See +`yunke-server/.env.example` for the full list; the load-bearing ones: + +| Variable | Purpose | +|-----------------------|---------------------------------------------------------------------------| +| `YK_USERID` | Existing-session mode (preferred). Pairs with `YK_COMPANY` (+ `YK_COOKIE`). | +| `YK_COMPANY` | Existing-session mode. | +| `YK_COOKIE` | Optional raw `Cookie:` header for session mode. | +| `YUNKE_USERNAME` | pcLogin mode (deferred — see "Limitations"). Pairs with `YUNKE_PASSWORD`. | +| `YUNKE_PASSWORD` | pcLogin mode. | +| `DB_URL` | SQLAlchemy async URL. Prod: `mysql+asyncmy://...`. Dev: `sqlite+aiosqlite:///./smoke.db`. | +| `REDIS_URL` | Redis URL (currently unused; reserved for future cache / queue). | +| `MEDIA_DIR` | Where downloaded media blobs land (sha256-deduped). | +| `ENCRYPTION_KEY` | Fernet key for encrypting stored credentials. Blank for tests only. | +| `API_TOKEN` | Shared bearer token for the viewing API. Blank = auth disabled (local/intranet trust). | +| `FULL_SYNC_HOUR` | Scheduler: local-time hour for the daily full sync (default `2`). | +| `MESSAGE_INTERVAL_HOURS` | Scheduler: hours between incremental message syncs (default `4`). | + +Auth precedence: session-cookie mode wins when both are set (the existing +live session is always preferred over a fresh pcLogin). + +## Running the scheduler manually + +```bash +cd yunke-server +python -m app.sync.cli scheduler # uses FULL_SYNC_HOUR + MESSAGE_INTERVAL_HOURS +python -m app.sync.cli scheduler --full-sync-hour 3 --message-interval-hours 6 +``` + +Entity names accepted by `sync --entity` and `run_once`: +`departments`, `accounts`, `friends`, `groups`, `voice_calls`, +`phone_calls`, `sms`, `messages`, `all_wechat`, `all` (default). + +## Current limitations + +- **pcLogin deferred.** Programmatic login (`getSecureKey` → `pcLogin`) is + wired and exercises real RSA encryption, but the final `pcLogin` body + field names are inferred and the live CRM has been observed rejecting + them. **Use session-cookie mode** (`YK_USERID` + `YK_COMPANY` + the + captured `YK_COOKIE`) until that's resolved. Capture the cookie/userid + from a logged-in browser tab on `crm2.yunkecn.com`. +- **WhatsApp voice / media deferred.** The WhatsApp scraper covers the + friend list + chat records; voice-call and media-download paths for + WhatsApp are not yet wired. +- **Rate-limited pacing.** Request cadence is deliberately conservative + (default ≈ one request every 1.0–1.8 s with jitter; exponential backoff + on 风控 signals). A full first-time sync is therefore slow (hours for + thousands of conversations). Tunable via `REQUEST_MIN_INTERVAL` / + `REQUEST_JITTER` / `MAX_REQUESTS_PER_RUN` in `app/config.py` — loosen + only after evidence that the CRM tolerates a faster pace. +- **Conversation discovery is local-only.** The incremental message sync + reads the conversation list from the already-mirrored `customer` table, + not the upstream `talkerList` endpoint. A first-time-empty mirror will + therefore not pull any messages until at least one full friend sync has + populated customers. +- **Redis is provisioned but unused.** The compose stack includes a Redis + service for future cache / queue use; nothing in the app reads or writes + it yet. + +## Verification + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest -q # 323 tests +cd yunkes && npm run build # clean production build +cd yunke-server && python -m app.sync.cli --help # CLI usage +docker compose config # validates the stack +``` diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..445d0fa --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,163 @@ +# docker-compose for the yunke data mirror (M6, Part C). +# +# Five services: +# mysql — MySQL 8 data store (volume-backed). +# redis — Redis 7 (future cache/queue use; not yet wired into the app). +# backend — FastAPI viewing API (uvicorn) talking to mysql + redis. +# scheduler — APScheduler loop (python -m app.sync.cli scheduler) that +# periodically scrapes yunke CRM into mysql. +# frontend — yunkes SPA built with Vite, served by nginx, reverse-proxying +# /api to the backend service. +# +# Auth: the scheduler + backend read yunke creds from .env (the host's .env +# is passed through as-is). Session-cookie mode (YK_USERID / YK_COMPANY / +# YK_COOKIE) is preferred; pcLogin mode is wired but deferred (see README). +# +# Usage: +# docker-compose --env-file .env up # build + start everything +# docker-compose up backend # just backend (+ deps) +# docker-compose logs -f scheduler # tail scheduler JSON logs +# +# The frontend's container exposes port 80; mapped to 5173 on the host so +# the same URL (http://localhost:5173) works in dev and prod. + +services: + # ----------------------------------------------------------------------- # + # Data plane + # ----------------------------------------------------------------------- # + mysql: + image: mysql:8.0 + restart: unless-stopped + environment: + MYSQL_DATABASE: ${MYSQL_DATABASE:-yunke} + MYSQL_USER: ${MYSQL_USER:-yunke} + MYSQL_PASSWORD: ${MYSQL_PASSWORD:-yunke} + MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD:-root} + # Default MySQL settings are fine for dev. The container ships its own + # healthcheck (mysqladmin ping); we use it so dependents wait until the + # server is actually accepting connections, not just listening on the port. + healthcheck: + test: ["CMD", "mysqladmin", "ping", "-h", "localhost"] + interval: 10s + timeout: 5s + retries: 10 + volumes: + - mysql_data:/var/lib/mysql + ports: + # Expose for host-side inspection (mysql CLI / DBeaver); omit if you + # don't need cross-container access from the host. + - "3306:3306" + + redis: + image: redis:7-alpine + restart: unless-stopped + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 10s + timeout: 3s + retries: 10 + volumes: + - redis_data:/data + + # ----------------------------------------------------------------------- # + # Application plane — same image, two entrypoints + # ----------------------------------------------------------------------- # + backend: + build: + context: ./yunke-server + dockerfile: Dockerfile + restart: unless-stopped + # uvicorn entrypoint (matches the Dockerfile's default CMD, restated so + # the role of this service is self-documenting at a glance). + command: + - uvicorn + - app.api.main:app + - --host + - 0.0.0.0 + - --port + - "8000" + depends_on: + mysql: + condition: service_healthy + redis: + condition: service_healthy + environment: + # DB_URL points at the mysql *service* (not localhost) — the container + # resolves it via docker-compose DNS. asyncmy is the async driver the + # app expects; the creds match the mysql service block above. + DB_URL: mysql+asyncmy://${MYSQL_USER:-yunke}:${MYSQL_PASSWORD:-yunke}@mysql:3306/${MYSQL_DATABASE:-yunke} + REDIS_URL: redis://redis:6379/0 + MEDIA_DIR: /data/media + # Pass through yunke auth + viewing-API token verbatim from .env so the + # same .env works in dev and docker. + YUNKE_BASE_URL: ${YUNKE_BASE_URL:-https://crm2.yunkecn.com} + YUNKE_USERNAME: ${YUNKE_USERNAME:-} + YUNKE_PASSWORD: ${YUNKE_PASSWORD:-} + YK_USERID: ${YK_USERID:-} + YK_COMPANY: ${YK_COMPANY:-} + YK_COOKIE: ${YK_COOKIE:-} + API_TOKEN: ${API_TOKEN:-} + ENCRYPTION_KEY: ${ENCRYPTION_KEY:-} + volumes: + # Persist downloaded media blobs across container restarts. + - media_data:/data/media + ports: + - "8000:8000" + + scheduler: + build: + context: ./yunke-server + dockerfile: Dockerfile + restart: unless-stopped + # Scheduler entrypoint — same image, different command. Cadence is + # configurable via env (FULL_SYNC_HOUR / MESSAGE_INTERVAL_HOURS) so + # operators can tune without rebuilding. + command: + - python + - -m + - app.sync.cli + - scheduler + depends_on: + mysql: + condition: service_healthy + redis: + condition: service_healthy + environment: + DB_URL: mysql+asyncmy://${MYSQL_USER:-yunke}:${MYSQL_PASSWORD:-yunke}@mysql:3306/${MYSQL_DATABASE:-yunke} + REDIS_URL: redis://redis:6379/0 + MEDIA_DIR: /data/media + YUNKE_BASE_URL: ${YUNKE_BASE_URL:-https://crm2.yunkecn.com} + YUNKE_USERNAME: ${YUNKE_USERNAME:-} + YUNKE_PASSWORD: ${YUNKE_PASSWORD:-} + YK_USERID: ${YK_USERID:-} + YK_COMPANY: ${YK_COMPANY:-} + YK_COOKIE: ${YK_COOKIE:-} + ENCRYPTION_KEY: ${ENCRYPTION_KEY:-} + # Scheduler cadence (env-driven; see app.sync.cli._cmd_scheduler). + FULL_SYNC_HOUR: ${FULL_SYNC_HOUR:-2} + MESSAGE_INTERVAL_HOURS: ${MESSAGE_INTERVAL_HOURS:-4} + volumes: + # Share the media volume with backend so both services see the same + # downloaded blobs (read-only would be ideal but the scheduler is the + # writer; mounting read-write here is the simplest correct config). + - media_data:/data/media + + # ----------------------------------------------------------------------- # + # Frontend — built + served by nginx + # ----------------------------------------------------------------------- # + frontend: + build: + context: ./yunkes + dockerfile: Dockerfile + restart: unless-stopped + depends_on: + - backend + ports: + # Map nginx's :80 to the host's :5173 so the same URL + # (http://localhost:5173) works in dev (vite) and prod (nginx). + - "5173:80" + +volumes: + mysql_data: + redis_data: + media_data: diff --git a/yunke-server/.dockerignore b/yunke-server/.dockerignore new file mode 100644 index 0000000..f1043bc --- /dev/null +++ b/yunke-server/.dockerignore @@ -0,0 +1,25 @@ +# Build artifacts we never want inside the image. +.venv/ +__pycache__/ +*.pyc +*.pyo +*.egg-info/ +.pytest_cache/ +.ruff_cache/ + +# Local DBs / scratch files (the container should use the mysql service). +*.db +*.sqlite +*.sqlite3 +smoke.db + +# Local env — the docker-compose ``environment:`` block injects runtime env. +.env + +# Local media storage (mounted as a volume in docker-compose). +media/ + +# Editor / OS noise. +.vscode/ +.idea/ +.DS_Store diff --git a/yunke-server/Dockerfile b/yunke-server/Dockerfile new file mode 100644 index 0000000..e818578 --- /dev/null +++ b/yunke-server/Dockerfile @@ -0,0 +1,35 @@ +# Backend image: FastAPI + APScheduler + Alembic + scrapers. +# +# Keeps the image small (python:3.12-slim) and the layer cache hot by copying +# only pyproject.toml first and installing deps before the source lands. The +# same image serves both the API (uvicorn) and the scheduler +# (``python -m app.sync.cli scheduler``) — the docker-compose ``command:`` key +# selects which entrypoint runs. +FROM python:3.12-slim + +# Build-time / runtime deps for asyncmy (compiled C extension) + httpx + +# cryptography.``libmariadb`` is what asyncmy links against on Debian. +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + build-essential \ + libmariadb-dev \ + libffi-dev \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +# Install dependencies first (cache-friendly: only re-installs when +# pyproject.toml changes). The project itself is installed in editable-like +# mode by copying the package in afterwards. +COPY pyproject.toml ./ +COPY app/__init__.py app/__init__.py +COPY app/ ./app/ + +# Install the package + dev dependencies (the dev extras include aiosqlite + +# pytest, which are tiny and convenient for in-container smoke checks). Pin +# the install to the local source so the package import path resolves. +RUN pip install --no-cache-dir -e ".[dev]" + +# Default command — overridden per-service in docker-compose. The API service +# runs uvicorn; the scheduler service runs the CLI. +CMD ["uvicorn", "app.api.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/yunke-server/app/sync/cli.py b/yunke-server/app/sync/cli.py new file mode 100644 index 0000000..fca3085 --- /dev/null +++ b/yunke-server/app/sync/cli.py @@ -0,0 +1,310 @@ +"""Command-line entry point for running syncs + the scheduler (M6, Part B). + +Three invocations:: + + # One-shot full sync (every entity) into the configured DB: + python -m app.sync.cli sync + + # One-shot single-entity sync (departments / friends / messages / ...): + python -m app.sync.cli sync --entity friends + + # Long-running scheduler (APScheduler; full sync daily + messages every 4h): + python -m app.sync.cli scheduler + +Auth + DB +--------- +Auth comes from env / ``.env`` via the same helpers the smoke script uses +(:func:`scripts.smoke_wechat._push_settings_into_environ` + +:func:`scripts.smoke_wechat.build_client_from_env`) — session-cookie mode +(``YK_USERID`` + ``YK_COMPANY``) is preferred; pcLogin mode is wired but +deferred. + +The DB is taken from :attr:`Settings.db_url` (MySQL in prod, SQLite in dev / +tests). Tables are created if missing (``Base.metadata.create_all``) on every +command so a fresh checkout works without a separate migration step — the +CLI is intentionally idempotent / bootstrappy that way. Real schema evolution +goes through Alembic; this is a safety net for the operator. + +Output +------ +Progress is printed as JSON lines (one per entity) so the output is both +human-skimmable and machine-parseable. ``--quiet`` suppresses the per-run +JSON lines; ``--help`` lists the entities. + +Exit codes: ``0`` success, ``1`` exception during run, ``2`` bad CLI usage / +missing auth. +""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +import signal +import sys +from typing import Any + +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.sync.scheduler import ENTITY_FUNCS, SyncScheduler + +logger = logging.getLogger("app.sync.cli") + + +# --------------------------------------------------------------------------- +# argparse setup +# --------------------------------------------------------------------------- + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="python -m app.sync.cli", + description=( + "Run a one-shot yunke sync or start the periodic scheduler. " + "Auth + DB come from .env (session-cookie mode preferred)." + ), + ) + sub = parser.add_subparsers(dest="command", required=True, metavar="") + + # ``sync`` ---------------------------------------------------------------- + p_sync = sub.add_parser( + "sync", + help="Run a one-shot sync (all entities by default).", + description=( + "Run a one-shot sync into the configured DB. Without --entity, " + "runs the full set (org tree + messages + phone calls + SMS). " + "With --entity , runs just that one." + ), + ) + p_sync.add_argument( + "--entity", + choices=sorted(ENTITY_FUNCS), + default="all", + help="Entity to sync (default: all).", + ) + p_sync.add_argument( + "--quiet", + action="store_true", + help="Suppress per-run JSON progress lines (errors still logged).", + ) + + # ``scheduler`` ----------------------------------------------------------- + p_sched = sub.add_parser( + "scheduler", + help="Start the APScheduler (long-running).", + description=( + "Start the APScheduler loop: full sync daily at FULL_SYNC_HOUR " + "(default 02:00 local), incremental message sync every " + "MESSAGE_INTERVAL_HOURS (default 4). Runs forever until " + "interrupted (Ctrl-C / SIGTERM)." + ), + ) + p_sched.add_argument( + "--full-sync-hour", + type=int, + default=None, + help=( + "Local-time hour (0-23) for the daily full sync. " + "Defaults to FULL_SYNC_HOUR env var or 2." + ), + ) + p_sched.add_argument( + "--message-interval-hours", + type=int, + default=None, + help=( + "Hours between incremental message syncs. " + "Defaults to MESSAGE_INTERVAL_HOURS env var or 4." + ), + ) + p_sched.add_argument( + "--quiet", + action="store_true", + help="Suppress per-run JSON progress lines (errors still logged).", + ) + + return parser + + +# --------------------------------------------------------------------------- +# Command implementations +# --------------------------------------------------------------------------- + + +async def _cmd_sync(args: argparse.Namespace) -> int: + """Run a one-shot sync and print a per-entity JSON-line summary.""" + settings = get_settings() + + # Create tables if missing (idempotent). Lets a fresh container boot + # without a separate migration step. + engine, session_maker = await _init_db(settings.db_url) + try: + scheduler = SyncScheduler( + engine=engine, session_maker=session_maker + ) + if args.quiet: + # The --quiet flag only governs *our* explicit log echo; the + # orchestrator's own per-entity logger.exception stays on so + # failures are never silent. + logging.getLogger("app.sync").setLevel(logging.WARNING) + results = await scheduler.run_once(args.entity) + _print_sync_summary(args.entity, results) + # Exit non-zero only if every result failed; a partial failure is + # still useful work delivered and should be exit 0 (the JSON line + # surfaces which entities failed). + any_success = any(r.status == "success" for r in results) + return 0 if any_success or not results else 1 + finally: + await engine.dispose() + + +async def _cmd_scheduler(args: argparse.Namespace) -> int: + """Start the long-running scheduler and block until interrupted.""" + # Env-var fallbacks so docker-compose / .env can drive the cadence + # without re-editing the compose command. + full_sync_hour = args.full_sync_hour + if full_sync_hour is None: + full_sync_hour = _env_int("FULL_SYNC_HOUR", default=2) + message_interval_hours = args.message_interval_hours + if message_interval_hours is None: + message_interval_hours = _env_int("MESSAGE_INTERVAL_HOURS", default=4) + + if not (0 <= full_sync_hour <= 23): + print( + f"full-sync-hour must be 0..23, got {full_sync_hour}", + file=sys.stderr, + ) + return 2 + + scheduler = SyncScheduler( + full_sync_hour=full_sync_hour, + message_interval_hours=message_interval_hours, + ) + await scheduler.start() + print( + f"[scheduler] running: full sync daily at hour {full_sync_hour}, " + f"messages every {message_interval_hours}h. Ctrl-C to stop.", + flush=True, + ) + + # Block forever until SIGINT / SIGTERM. AsyncIOScheduler runs jobs on + # its own event loop thread; we just need to keep the process alive. + stop_event = asyncio.Event() + + def _signal_handler(*_: Any) -> None: + print("\n[scheduler] stop signal received, shutting down...", flush=True) + stop_event.set() + + # KeyboardInterrupt is delivered as SIGINT on POSIX, but on Windows + # the convention is a Ctrl-C event; loop.add_signal_handler isn't + # implemented there. Install both: signal.signal for Windows + the + # asyncio version where available. + for sig in (signal.SIGINT, signal.SIGTERM): + try: + asyncio.get_running_loop().add_signal_handler(sig, _signal_handler) + except (NotImplementedError, RuntimeError): + # Windows / non-main thread — fall back to the stdlib handler. + signal.signal(sig, _signal_handler) + + try: + await stop_event.wait() + finally: + await scheduler.stop() + return 0 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +async def _init_db(db_url: str) -> tuple[Any, Any]: + """Create the async engine + session maker and ensure tables exist. + + Returns ``(engine, session_maker)``. The caller owns disposal. Tables + are created with ``Base.metadata.create_all`` — idempotent, so this is + safe to call on a warm DB. + """ + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + engine = create_async_engine(db_url, pool_pre_ping=True, future=True) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + session_maker = async_sessionmaker(engine, expire_on_commit=False) + return engine, session_maker + + +def _env_int(name: str, *, default: int) -> int: + """Read an int env var; fall back to ``default`` on missing / unparseable.""" + import os + + raw = os.environ.get(name, "").strip() + if not raw: + return default + try: + return int(raw) + except ValueError: + print( + f"[cli] {name}={raw!r} is not an int; using default {default}", + file=sys.stderr, + ) + return default + + +def _print_sync_summary(entity: str, results: list[Any]) -> None: + """Print a compact human summary of a finished run (after the JSON lines).""" + if not results: + print(f"[sync] {entity}: no results (run failed before any entity completed)") + return + 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"] + line = ( + f"[sync] {entity}: {len(results)} sub-run(s), " + f"fetched={fetched}, upserted={upserted}" + ) + if failed: + line += f", FAILED={failed}" + print(line, flush=True) + + +def main(argv: list[str] | None = None) -> int: + """CLI entry point. Returns a process exit code.""" + # Configure logging so the scheduler's JSON line + the orchestrator's + # exception tracebacks are visible. Quiet flag is applied per-command + # after parsing. + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + + parser = _build_parser() + args = parser.parse_args(argv) + + try: + if args.command == "sync": + return asyncio.run(_cmd_sync(args)) + if args.command == "scheduler": + return asyncio.run(_cmd_scheduler(args)) + except KeyboardInterrupt: + # So `python -m app.sync.cli scheduler` exits cleanly on Ctrl-C. + return 0 + except ValueError as exc: + # Bad entity name or auth misconfig surfaced as a ValueError — + # print a clear message instead of a traceback. + print(f"[cli] {exc}", file=sys.stderr) + return 2 + except Exception as exc: # noqa: BLE001 + logger.exception("cli command failed") + print(f"[cli] {exc}", file=sys.stderr) + return 1 + + # argparse with required=True subparsers won't get here, but keep the + # lint happy. + parser.print_help(sys.stderr) + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py new file mode 100644 index 0000000..68b25fe --- /dev/null +++ b/yunke-server/app/sync/scheduler.py @@ -0,0 +1,501 @@ +"""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.session.yunke_client import YunkeClient +from app.storage.object_store import ObjectStore +from app.sync.orchestrator import ( + SyncResult, + sync_all_wechat, + sync_departments, + sync_phone_calls, + sync_sms, + sync_wechat_accounts, + sync_wechat_friends, + sync_wechat_groups, + sync_wechat_messages, + sync_wechat_voice_calls, +) + +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 + + +# --------------------------------------------------------------------------- +# Conversation discovery (drives the incremental message sync) +# --------------------------------------------------------------------------- + + +async def _discover_conversations( + session: AsyncSession, + *, + limit: int = DEFAULT_MAX_CONVERSATIONS_PER_RUN, +) -> list[tuple[str, str]]: + """Return distinct ``(owner_account_id, friend_id)`` WeChat pairs to sync. + + Source: the local ``customer`` mirror (where + ``channel='wechat'`` and ``owner_account_id`` is non-empty). 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 == "wechat") + .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_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_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_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) + results.append(await sync_phone_calls(session, client)) + results.append(await sync_sms(session, client)) + 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("")`` 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, + "phone_calls": _run_phone_calls, + "sms": _run_sms, + "messages": _run_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, 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. + """ + 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: + runner = ENTITY_FUNCS[entity] + results = await runner(session, client, ObjectStore()) + 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 diff --git a/yunke-server/tests/test_sync_cli.py b/yunke-server/tests/test_sync_cli.py new file mode 100644 index 0000000..23ab777 --- /dev/null +++ b/yunke-server/tests/test_sync_cli.py @@ -0,0 +1,229 @@ +"""Tests for ``python -m app.sync.cli`` (M6 Part B). + +The CLI's network / DB / scheduler-loop paths are exercised manually (they +need real env + a running DB). These tests verify the testable surface: + +* ``--help`` and the subcommand parsers build without error. +* The ``sync`` command dispatches into :class:`SyncScheduler.run_once` with + the right entity and prints a summary line. +* An unknown subcommand / bad ``--entity`` exits non-zero. +* Env-var fallbacks (``FULL_SYNC_HOUR`` / ``MESSAGE_INTERVAL_HOURS``) work. +""" + +from __future__ import annotations + +import asyncio +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.sync import cli as cli_module +from app.sync.orchestrator import SyncResult + + +# --------------------------------------------------------------------------- +# Argument parser +# --------------------------------------------------------------------------- + + +def test_parser_help_lists_both_subcommands(capsys): + """``--help`` mentions both ``sync`` and ``scheduler``.""" + with pytest.raises(SystemExit) as exc: + cli_module._build_parser().parse_args(["--help"]) + assert exc.value.code == 0 + out = capsys.readouterr().out + assert "sync" in out + assert "scheduler" in out + + +def test_parser_requires_subcommand(): + """No subcommand -> argparse errors (exit code 2).""" + with pytest.raises(SystemExit) as exc: + cli_module._build_parser().parse_args([]) + assert exc.value.code == 2 + + +def test_parser_sync_entity_choices_match_registry(): + """``--entity`` choices mirror the scheduler registry.""" + parser = cli_module._build_parser() + args = parser.parse_args(["sync", "--entity", "friends"]) + assert args.command == "sync" + assert args.entity == "friends" + + +def test_parser_rejects_unknown_entity(): + """An unknown entity name is rejected by argparse (exit code 2).""" + with pytest.raises(SystemExit) as exc: + cli_module._build_parser().parse_args(["sync", "--entity", "nope"]) + assert exc.value.code == 2 + + +def test_parser_scheduler_flags_default_none(): + """``--full-sync-hour`` and ``--message-interval-hours`` default to None (env-fallback path).""" + args = cli_module._build_parser().parse_args(["scheduler"]) + assert args.full_sync_hour is None + assert args.message_interval_hours is None + + +# --------------------------------------------------------------------------- +# main() — exit codes +# --------------------------------------------------------------------------- + + +def test_main_returns_2_on_unknown_subcommand(capsys): + """An unknown subcommand surfaces as exit code 2 (argparse).""" + with pytest.raises(SystemExit) as exc: + cli_module.main(["bogus"]) + assert exc.value.code == 2 + + +def test_main_value_error_becomes_exit_2(monkeypatch, capsys): + """A ValueError surfaced from a command (e.g. unknown entity) is exit 2.""" + async def _boom(_args): + raise ValueError("unknown entity: does_not_exist") + monkeypatch.setattr(cli_module, "_cmd_sync", _boom) + rc = cli_module.main(["sync", "--entity", "friends"]) + assert rc == 2 + err = capsys.readouterr().err + assert "unknown entity" in err + + +def test_main_generic_exception_becomes_exit_1(monkeypatch): + """Any non-ValueError exception is exit 1 (don't print misleading usage).""" + async def _boom(_args): + raise RuntimeError("kaboom") + monkeypatch.setattr(cli_module, "_cmd_sync", _boom) + rc = cli_module.main(["sync"]) + assert rc == 1 + + +# --------------------------------------------------------------------------- +# _cmd_sync — dispatch + summary printing +# --------------------------------------------------------------------------- + + +def test_cmd_sync_dispatches_run_once_and_prints_summary(monkeypatch, capsys): + """``sync`` builds a SyncScheduler, calls run_once(), prints a summary.""" + # ``_cmd_sync`` is async — wrap a coroutine-returning stub via patch. + fake_results = [ + SyncResult(entity="wechat_friend", status="success", fetched=3, upserted=3), + SyncResult(entity="message", status="success", fetched=10, upserted=10, channel="wechat"), + ] + + async def _fake_init_db(_db_url): + engine = MagicMock() + engine.dispose = AsyncMock() + sm = MagicMock() + return engine, sm + + scheduler_mock = MagicMock() + scheduler_mock.run_once = AsyncMock(return_value=list(fake_results)) + + def _ctor(*, engine, session_maker): + # Returned to _cmd_sync which then calls .run_once() on it. + return scheduler_mock + + monkeypatch.setattr(cli_module, "_init_db", _fake_init_db) + monkeypatch.setattr(cli_module, "SyncScheduler", _ctor) + + args = cli_module._build_parser().parse_args(["sync", "--entity", "friends"]) + rc = asyncio.run(cli_module._cmd_sync(args)) + + assert rc == 0 + scheduler_mock.run_once.assert_awaited_once_with("friends") + out = capsys.readouterr().out + assert "[sync] friends" in out + assert "fetched=13" in out + assert "upserted=13" in out + + +def test_cmd_sync_partial_failure_is_exit_0(monkeypatch): + """If at least one sub-result succeeded, exit 0 (partial delivery).""" + fake_results = [ + SyncResult(entity="department", status="success", fetched=2, upserted=2), + SyncResult(entity="wechat_friend", status="failed", fetched=0, upserted=0, error="x"), + ] + async def _fake_init_db(_db_url): + engine = MagicMock() + engine.dispose = AsyncMock() + return engine, MagicMock() + scheduler_mock = MagicMock() + scheduler_mock.run_once = AsyncMock(return_value=list(fake_results)) + monkeypatch.setattr(cli_module, "_init_db", _fake_init_db) + monkeypatch.setattr(cli_module, "SyncScheduler", lambda **kw: scheduler_mock) + + args = cli_module._build_parser().parse_args(["sync"]) + rc = asyncio.run(cli_module._cmd_sync(args)) + assert rc == 0 + + +def test_cmd_sync_all_failed_is_exit_1(monkeypatch): + """Every sub-run failed -> exit 1 (nothing useful delivered).""" + fake_results = [ + SyncResult(entity="department", status="failed", fetched=0, upserted=0, error="x"), + SyncResult(entity="wechat_friend", status="failed", fetched=0, upserted=0, error="y"), + ] + async def _fake_init_db(_db_url): + engine = MagicMock() + engine.dispose = AsyncMock() + return engine, MagicMock() + scheduler_mock = MagicMock() + scheduler_mock.run_once = AsyncMock(return_value=list(fake_results)) + monkeypatch.setattr(cli_module, "_init_db", _fake_init_db) + monkeypatch.setattr(cli_module, "SyncScheduler", lambda **kw: scheduler_mock) + + args = cli_module._build_parser().parse_args(["sync"]) + rc = asyncio.run(cli_module._cmd_sync(args)) + assert rc == 1 + + +# --------------------------------------------------------------------------- +# Env-var fallbacks for the scheduler command +# --------------------------------------------------------------------------- + + +def test_env_int_returns_default_when_missing(monkeypatch): + """Missing env var -> default.""" + monkeypatch.delenv("FULL_SYNC_HOUR", raising=False) + assert cli_module._env_int("FULL_SYNC_HOUR", default=42) == 42 + + +def test_env_int_parses_value(monkeypatch): + """A clean int env var is parsed.""" + monkeypatch.setenv("FULL_SYNC_HOUR", "13") + assert cli_module._env_int("FULL_SYNC_HOUR", default=2) == 13 + + +def test_env_int_falls_back_on_garbage(monkeypatch, capsys): + """Non-int env var falls back to default + prints a warning.""" + monkeypatch.setenv("FULL_SYNC_HOUR", "two") + assert cli_module._env_int("FULL_SYNC_HOUR", default=2) == 2 + err = capsys.readouterr().err + assert "FULL_SYNC_HOUR" in err + + +# --------------------------------------------------------------------------- +# _init_db — create_all is invoked +# --------------------------------------------------------------------------- + + +async def test_init_db_creates_tables(): + """_init_db returns an engine + session maker and runs create_all. + + Uses an in-memory SQLite engine via the real SQLAlchemy path so the + create_all call is exercised end-to-end (not mocked). + """ + from sqlalchemy import text + + engine, sm = await cli_module._init_db("sqlite+aiosqlite:///:memory:") + try: + # Tables created: customer is one of the registered tables. + from app.db.base import Base + + assert "customer" in Base.metadata.tables + # engine + session_maker both usable (no exception = pass). + async with sm() as s: + await s.execute(text("SELECT 1")) + finally: + await engine.dispose() diff --git a/yunke-server/tests/test_sync_scheduler.py b/yunke-server/tests/test_sync_scheduler.py new file mode 100644 index 0000000..065b077 --- /dev/null +++ b/yunke-server/tests/test_sync_scheduler.py @@ -0,0 +1,389 @@ +"""Tests for app.sync.scheduler — periodic-sync wrapper (M6 Part A). + +These tests do NOT hit the network and do NOT start a real APScheduler loop +(that path is exercised manually via ``python -m app.sync.cli scheduler``). +They verify the testable surface of :class:`SyncScheduler`: + +* ``run_once`` dispatches by entity name to the right orchestrator function. +* ``run_once("")`` raises ``ValueError`` with the known-entity list. +* ``_discover_conversations`` reads distinct WeChat pairs from the DB. +* ``_log_run`` emits one JSON line per run with sensible totals + level. +* The daily vs. message-every-N-hours schedule is registered with the right + triggers when :meth:`start` is called (using a fake scheduler fixture). +* Failures inside a runner are swallowed and logged, not re-raised. +""" + +from __future__ import annotations + +import datetime as dt +import json +import logging +from collections.abc import AsyncIterator +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db.base import Base +from app.db.models import Customer +from app.sync import scheduler as scheduler_module +from app.sync.scheduler import ( + ENTITY_FUNCS, + SyncScheduler, + _discover_conversations, + _log_run, +) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +async def _make_engine(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + return engine + + +@pytest.fixture +async def engine(): + eng = await _make_engine() + try: + yield eng + finally: + await eng.dispose() + + +@pytest.fixture +async def session_maker(engine): + return async_sessionmaker(engine, expire_on_commit=False) + + +# --------------------------------------------------------------------------- +# ENTITY_FUNCS registry +# --------------------------------------------------------------------------- + + +def test_entity_funcs_has_expected_names(): + """The registry exposes the documented manual-trigger entity names.""" + expected = { + "departments", + "accounts", + "friends", + "groups", + "voice_calls", + "phone_calls", + "sms", + "messages", + "all_wechat", + "all", + } + assert expected <= set(ENTITY_FUNCS) + + +# --------------------------------------------------------------------------- +# _discover_conversations +# --------------------------------------------------------------------------- + + +async def test_discover_conversations_returns_distinct_wechat_pairs(session_maker): + """Distinct (owner, friend_id) for channel='wechat' only; non-wechat + nulls dropped.""" + async with session_maker() as s: + s.add_all( + [ + Customer( + friend_id="f1", + channel="wechat", + owner_account_id="wx_a", + ), + Customer( + friend_id="f2", + channel="wechat", + owner_account_id="wx_a", + ), + Customer( + friend_id="f3", + channel="wechat", + owner_account_id="wx_b", + ), + # Different channel — should be excluded. + Customer( + friend_id="p1@s.whatsapp.net", + channel="whatsapp", + owner_account_id="wx_a", + ), + # Null owner — should be excluded. + Customer( + friend_id="orphan", + channel="wechat", + owner_account_id=None, + ), + ] + ) + await s.commit() + + pairs = await _discover_conversations(s) + assert sorted(pairs) == sorted([("wx_a", "f1"), ("wx_a", "f2"), ("wx_b", "f3")]) + + +async def test_discover_conversations_respects_limit(session_maker): + """``limit`` caps the work per message-sync run.""" + async with session_maker() as s: + for i in range(10): + s.add( + Customer( + friend_id=f"f{i}", + channel="wechat", + owner_account_id="wx_a", + ) + ) + await s.commit() + + pairs = await _discover_conversations(s, limit=3) + assert len(pairs) == 3 + + +# --------------------------------------------------------------------------- +# _log_run +# --------------------------------------------------------------------------- + + +def test_log_run_emits_one_json_line(capsys, caplog): + """One JSON line, level=info when all sub-runs succeed.""" + results = [ + _mk_result("department", "success", 5, 5), + _mk_result("wechat_friend", "success", 3, 3), + ] + with caplog.at_level(logging.INFO, logger="app.sync.scheduler"): + _log_run("all", results, started_at=dt.datetime.now(dt.UTC)) + out = capsys.readouterr().out.strip().splitlines() + assert len(out) == 1 + payload = json.loads(out[0]) + assert payload["event"] == "sync_run" + assert payload["entity"] == "all" + assert payload["level"] == "info" + assert payload["totals"] == {"fetched": 8, "upserted": 8, "count": 2} + assert payload["failed_entities"] == [] + assert len(payload["results"]) == 2 + + +def test_log_run_level_warning_when_any_failed(capsys): + """A failed sub-result bumps the level to warning + lists the failed entity.""" + results = [ + _mk_result("department", "success", 5, 5), + _mk_result("wechat_friend", "failed", 0, 0, error="boom"), + ] + _log_run("all", results, started_at=dt.datetime.now(dt.UTC)) + payload = json.loads(capsys.readouterr().out.strip()) + assert payload["level"] == "warning" + assert payload["failed_entities"] == ["wechat_friend"] + assert payload["totals"]["fetched"] == 5 + + +def test_log_run_level_error_on_top_level_failure(capsys): + """A top-level error (no results at all) -> level=error + error field.""" + _log_run("all", [], started_at=dt.datetime.now(dt.UTC), error="AuthFailed") + payload = json.loads(capsys.readouterr().out.strip()) + assert payload["level"] == "error" + assert payload["error"] == "AuthFailed" + assert payload["totals"]["count"] == 0 + + +# --------------------------------------------------------------------------- +# SyncScheduler.run_once — dispatch +# --------------------------------------------------------------------------- + + +async def test_run_once_unknown_entity_raises(engine, session_maker): + """An unknown entity name raises ValueError listing the known entities.""" + sched = SyncScheduler(engine=engine, session_maker=session_maker) + with pytest.raises(ValueError, match="unknown entity"): + await sched.run_once("does_not_exist") + + +async def test_run_once_dispatches_to_entity_runner(engine, session_maker, monkeypatch): + """run_once('friends') calls _run_friends (which calls sync_wechat_friends).""" + # Stub _build_client_from_env so no network / login happens. + fake_client = MagicMock() + monkeypatch.setattr( + scheduler_module, + "_build_client_from_env", + AsyncMock(return_value=fake_client), + ) + + captured: dict[str, Any] = {} + + async def fake_runner(session, client, store): + captured["session"] = session + captured["client"] = client + captured["store"] = store + return [_mk_result("wechat_friend", "success", 7, 7)] + + # Patch the registry entry, not the orchestrator function. + monkeypatch.setitem(ENTITY_FUNCS, "friends", fake_runner) + + sched = SyncScheduler(engine=engine, session_maker=session_maker) + results = await sched.run_once("friends") + + assert len(results) == 1 + assert results[0].entity == "wechat_friend" + assert results[0].fetched == 7 + assert captured["client"] is fake_client + assert captured["store"] is not None + + +async def test_run_once_swallows_runner_exception(engine, session_maker, monkeypatch, capsys): + """A blown-up runner is logged + swallowed (returns []; scheduler keeps going).""" + fake_client = MagicMock() + monkeypatch.setattr( + scheduler_module, + "_build_client_from_env", + AsyncMock(return_value=fake_client), + ) + + async def boom(session, client, store): + raise RuntimeError("upstream 500") + + monkeypatch.setitem(ENTITY_FUNCS, "groups", boom) + + sched = SyncScheduler(engine=engine, session_maker=session_maker) + # Must NOT raise. + results = await sched.run_once("groups") + assert results == [] + out = capsys.readouterr().out + assert "sync_run" in out + payload = json.loads(out.strip().splitlines()[-1]) + assert payload["level"] == "error" + assert "upstream 500" in payload["error"] + + +async def test_run_once_swallows_client_build_failure(engine, session_maker, monkeypatch): + """If auth env is missing -> _build_client_from_env raises -> logged + empty list.""" + monkeypatch.setattr( + scheduler_module, + "_build_client_from_env", + AsyncMock(side_effect=RuntimeError("no auth configured")), + ) + sched = SyncScheduler(engine=engine, session_maker=session_maker) + results = await sched.run_once("all") + assert results == [] + + +# --------------------------------------------------------------------------- +# SyncScheduler.start — job registration +# --------------------------------------------------------------------------- + + +async def test_start_registers_two_jobs_with_expected_triggers(engine, session_maker): + """start() wires a daily CronTrigger (full) + an IntervalTrigger (messages).""" + sched = SyncScheduler( + full_sync_hour=3, + message_interval_hours=6, + engine=engine, + session_maker=session_maker, + ) + # Replace the real AsyncIOScheduler with a spy before start() creates it, + # so we can assert on add_job without running the loop. + spy = MagicMock() + spy.add_job = MagicMock() + with patch.object(scheduler_module, "AsyncIOScheduler", return_value=spy): + await sched.start() + + # Two jobs registered. + assert spy.add_job.call_count == 2 + job_ids = {c.kwargs["id"] for c in spy.add_job.call_args_list} + assert job_ids == {"yunke_full_sync", "yunke_message_sync"} + + # Full sync -> CronTrigger(hour=3); messages -> IntervalTrigger(hours=6). + # add_job was called as (func, trigger, kwargs...) so trigger is args[1]. + full_call = next( + c for c in spy.add_job.call_args_list if c.kwargs["id"] == "yunke_full_sync" + ) + full_trigger = full_call.args[1] + # CronTrigger exposes the parsed schedule via str(). + assert str(full_trigger) == "cron[hour='3']" + + msg_call = next( + c for c in spy.add_job.call_args_list if c.kwargs["id"] == "yunke_message_sync" + ) + interval = msg_call.args[1] + # IntervalTrigger exposes its interval as a timedelta on .interval`. + assert interval.interval.total_seconds() == 6 * 3600 + + # The full-sync job passes args=["all"] (so it hits _run_all). + assert full_call.kwargs["args"] == ["all"] + assert msg_call.kwargs["args"] == ["messages"] + + # coalesce=True + max_instances=1 protects against overlap. + for c in spy.add_job.call_args_list: + assert c.kwargs["coalesce"] is True + assert c.kwargs["max_instances"] == 1 + + +async def test_start_is_idempotent(engine, session_maker): + """A second start() is a no-op (does not register a second scheduler).""" + sched = SyncScheduler(engine=engine, session_maker=session_maker) + with patch.object(scheduler_module, "AsyncIOScheduler", return_value=MagicMock()): + await sched.start() + first = sched._scheduler + await sched.start() # second call — ignored. + assert sched._scheduler is first + + +# --------------------------------------------------------------------------- +# Stop / teardown +# --------------------------------------------------------------------------- + + +async def test_stop_disposes_owned_engine(): + """When the scheduler owns its engine, stop() disposes it.""" + # Patch create_async_engine so we don't try to connect to a real DB. + fake_engine = MagicMock() + fake_engine.dispose = AsyncMock() + fake_engine.begin = MagicMock() + # start() will call engine.begin() to create tables; fake it. + class _Ctx: + async def __aenter__(self): + conn = MagicMock() + conn.run_sync = AsyncMock() + return conn + + async def __aexit__(self, *a): + return None + + fake_engine.begin.return_value = _Ctx() + + with patch.object(scheduler_module, "create_async_engine", return_value=fake_engine): + sched = SyncScheduler() + assert sched._owns_engine is True + await sched.stop() + fake_engine.dispose.assert_awaited_once() + + +async def test_stop_does_not_dispose_injected_engine(engine, session_maker): + """An engine injected by the caller is NOT disposed on stop().""" + sched = SyncScheduler(engine=engine, session_maker=session_maker) + assert sched._owns_engine is False + await sched.stop() # must not dispose — caller (test fixture) owns it. + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _mk_result( + entity: str, status: str, fetched: int, upserted: int, *, error: str | None = None): + from app.sync.orchestrator import SyncResult + + return SyncResult( + entity=entity, + status=status, + fetched=fetched, + upserted=upserted, + error=error, + ) diff --git a/yunkes/.dockerignore b/yunkes/.dockerignore new file mode 100644 index 0000000..1a048c4 --- /dev/null +++ b/yunkes/.dockerignore @@ -0,0 +1,15 @@ +# Build artifacts we never want inside the image. +node_modules/ +dist/ +.vite/ + +# Local env — the docker-compose ``environment:`` block injects runtime env. +.env + +# Editor / OS noise. +.vscode/ +.idea/ +.DS_Store + +# Playwright / test scratch. +.playwright-mcp/ diff --git a/yunkes/Dockerfile b/yunkes/Dockerfile new file mode 100644 index 0000000..bb08a45 --- /dev/null +++ b/yunkes/Dockerfile @@ -0,0 +1,34 @@ +# Frontend image: build the SPA with Node, serve the static dist with nginx. +# +# Multi-stage so the final image contains ONLY the compiled static assets + +# nginx — no Node toolchain shipped to production. +FROM node:20-alpine AS build + +WORKDIR /app + +# Install deps first (cache-friendly: only re-installs when package*.json +# changes; node_modules is never copied from the host via .dockerignore). +COPY package.json package-lock.json* ./ +RUN npm ci --no-audit --no-fund + +# Copy sources and build the production bundle into ./dist. +COPY tsconfig*.json vite.config.ts index.html ./ +COPY public/ ./public/ +COPY src/ ./src/ +RUN npm run build + +# --------------------------------------------------------------------------- # +# Runtime stage: nginx serves the static dist and reverse-proxies /api. +# --------------------------------------------------------------------------- # +FROM nginx:1.27-alpine + +# Replace the default nginx config with our /api proxy + SPA fallback. +COPY nginx.conf /etc/nginx/conf.d/default.conf + +# Copy the build output from the build stage into nginx's html root. +COPY --from=build /app/dist /usr/share/nginx/html + +EXPOSE 80 + +# nginx foregrounds by default with the ``nginx -g 'daemon off;'`` directive +# already set on the official image; no CMD override needed. diff --git a/yunkes/nginx.conf b/yunkes/nginx.conf new file mode 100644 index 0000000..a742879 --- /dev/null +++ b/yunkes/nginx.conf @@ -0,0 +1,55 @@ +# nginx config for the yunkes frontend container. +# +# Two responsibilities: +# 1. Serve the SPA's static assets (JS/CSS/images) from /usr/share/nginx/html. +# 2. Reverse-proxy /api to the backend service so the SPA can call the +# FastAPI app on the same origin (no CORS gymnastics in prod). +# +# The SPA fallback (try_files ... /index.html) lets vue-router own deep +# links — a refresh on /customers hits index.html and is resolved client-side. + +server { + listen 80; + server_name _; + + root /usr/share/nginx/html; + index index.html; + + # Gzip text-y responses — meaningful for the (sizeable) JS bundle. + gzip on; + gzip_types text/plain text/css application/javascript application/json image/svg+xml; + gzip_min_length 1024; + + # API reverse proxy. ``backend`` resolves via docker-compose's default + # network; if the name changes, update here. Streaming endpoints + # (/api/media/) work without buffering because we turn proxy_buffering + # off — matches the dev-server behaviour (vite proxies /api/media first). + location /api/ { + proxy_pass http://backend:8000; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_read_timeout 60s; + # Allow websocket-style upgrades (harmless no-op for plain HTTP APIs + # but cheap insurance against future ws use). + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection "upgrade"; + # Disable buffering so /api/media/{hash} streams straight through. + proxy_buffering off; + } + + # Static SPA assets — long cache (immutable content hash in filename). + location ~* \.(?:js|css|woff2?|png|jpe?g|gif|svg|ico)$ { + expires 1y; + add_header Cache-Control "public, immutable"; + try_files $uri =404; + } + + # SPA fallback: any path that isn't a real file resolves to index.html + # so vue-router can pick it up. + location / { + try_files $uri $uri/ /index.html; + } +}