xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: scheduler + CLI + docker-compose + README

Browse Source 
M6 — the scheduling/deployment layer. Makes the system runnable as a
scheduled service + deployable via Docker.

Part A — SyncScheduler (app/sync/scheduler.py):
- APScheduler (AsyncIOScheduler) wraps the orchestrator's sync_* functions.
- Daily full sync at FULL_SYNC_HOUR (default 02:00 local) + incremental
  message sync every MESSAGE_INTERVAL_HOURS (default 4).
- ENTITY_FUNCS registry dispatches run_once("<entity>") to the right runner
  (departments/accounts/friends/groups/voice_calls/phone_calls/sms/messages/
  all_wechat/all).
- Conversations for the message run are discovered from the local customer
  mirror (no extra upstream call).
- Failure policy: every run is wrapped — a blown entity is logged + swallowed
  so the scheduler keeps its next appointment.
- Per-run JSON-line log (level=info/warning/error, totals, failed entities).

Part B — CLI entry point (app/sync/cli.py):
- `python -m app.sync.cli sync [--entity <name>] [--quiet]` — one-shot sync.
- `python -m app.sync.cli scheduler [--full-sync-hour H] [--message-interval-hours N]`
  — long-running scheduler (blocks until SIGINT/SIGTERM).
- Uses _push_settings_into_environ + build_client_from_env (session-cookie
  mode preferred; pcLogin deferred).
- Base.metadata.create_all on every command (fresh-checkout friendly).
- Exit codes: 0 success, 1 generic failure, 2 bad CLI / auth misconfig.

Part C — Docker stack:
- yunke-server/Dockerfile: python:3.12-slim, installs deps + app, single
  image serves both API (uvicorn) + scheduler (CLI).
- yunkes/Dockerfile: multi-stage node:20 build → nginx serves dist/ +
  reverse-proxies /api → backend:8000.
- yunkes/nginx.conf: /api proxy with proxy_buffering off (streaming media),
  SPA fallback, asset caching.
- docker-compose.yml: mysql (8.0, healthcheck, volume) + redis (7-alpine) +
  backend + scheduler + frontend (port 5173 → 80). Env-driven creds + cadence.

Part D — README.md at repo root: dev mode + docker mode quick-start, config
table, architecture overview, current limitations (pcLogin deferred, WA voice/
media deferred, conservative pacing, conversation discovery is local-only).

Verification:
- pytest: 323 passed (was 294; +29 new for scheduler + CLI).
- npm run build: clean.
- python -m app.sync.cli --help: prints usage.
- docker compose config: VALID.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-05 12:11:29 +08:00
parent c8311bb496
commit df29dd027a
11 changed files with 1933 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+177
View File
@@ -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 <http://localhost:5173>.
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 <http://localhost:5173>. 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
```
docker-compose.yml
+163
View File
@@ -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:
yunke-server/.dockerignore
+25
View File
@@ -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
yunke-server/Dockerfile
+35
View File
@@ -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"]
yunke-server/app/sync/cli.py
+310
View File
@@ -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="<command>")
# ``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 <name>, 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())
yunke-server/app/sync/scheduler.py
+501
View File
@@ -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("<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,
"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
yunke-server/tests/test_sync_cli.py
+229
View File
@@ -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(<entity>), 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()
yunke-server/tests/test_sync_scheduler.py
+389
View File
@@ -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("<unknown>")`` 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,
)
yunkes/.dockerignore
+15
View File
@@ -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/
yunkes/Dockerfile
+34
View File
@@ -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.
yunkes/nginx.conf
+55
View File
@@ -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;
}
}