feat(sources): контракт источников, worker и синк ZenMoney + ЦБ

Source.sync(ctx) -> SyncResult пишет только raw_* и возвращает курсор; локи, журнал, ошибки и продвижение курсора берёт на себя worker/runner. ZenMoney читается единственным доступным способом — POST /v8/diff/ по serverTimestamp; токен живёт сутки, поэтому worker ротирует refresh_token через source_credential. Маппер всегда пересобирает core из полных raw_*, так что удаление в ZenMoney исчезает и у нас. ЦБ ходит мимо прокси (trust_env=False) и отдаёт cp1251 с делением на Nominal. Курсы только по рабочим дням — протяжку по календарю делает аналитика. Планировщик — APScheduler в отдельном процессе, на источник advisory-лок sync:<name>, чтобы ручной запуск не пересёкся с плановым.
2026-09-18 13:43:49 +03:00
parent 3fc7a954b9
commit c55fe19e48
31 changed files with 3713 additions and 0 deletions
@@ -0,0 +1,39 @@
+"""Schedule: which source runs when (plan §2.4). Times are in the configured timezone."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from zoneinfo import ZoneInfo
+
+from apscheduler.triggers.base import BaseTrigger
+from apscheduler.triggers.combining import OrTrigger
+from apscheduler.triggers.cron import CronTrigger
+
+
+@dataclass(frozen=True)
+class JobSpec:
+    source: str
+    trigger: BaseTrigger
+    """A source appears at most once: `worker/scheduler.py` keys jobs by `sync:<source>`,
+    so several times of day are one `OrTrigger`, not several specs."""
+
+
+MSK = ZoneInfo("Europe/Moscow")
+"""CBR publishes on Moscow time, so those jobs pin the zone instead of following settings."""
+
+
+def default_schedule() -> list[JobSpec]:
+    # Each entry must name a registered source (see fintracker.sources).
+    return [
+        JobSpec("zenmoney", CronTrigger(minute="0,30")),
+        # official rates appear around 13:30 MSK, the next business day's rate in the evening
+        JobSpec(
+            "cbr",
+            OrTrigger(
+                [
+                    CronTrigger(hour=13, minute=45, timezone=MSK),
+                    CronTrigger(hour=18, minute=0, timezone=MSK),
+                ]
+            ),
+        ),
+    ]
@@ -0,0 +1,36 @@
+"""Postgres advisory locks so a scheduled run and a manual trigger never overlap."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+
+@asynccontextmanager
+async def try_advisory_lock(conn: AsyncConnection, key: str) -> AsyncIterator[bool]:
+    """Yield True if the session-level lock for `key` was acquired; release on exit."""
+    got = (
+        await conn.execute(text("SELECT pg_try_advisory_lock(hashtext(:k))"), {"k": key})
+    ).scalar_one()
+    try:
+        yield bool(got)
+    finally:
+        if got:
+            await conn.execute(text("SELECT pg_advisory_unlock(hashtext(:k))"), {"k": key})
+
+
+@asynccontextmanager
+async def advisory_lock(conn: AsyncConnection, key: str) -> AsyncIterator[None]:
+    """Wait for the session-level lock for `key`, then release it on exit.
+
+    Blocking, unlike `try_advisory_lock`: used where the work must still happen after the
+    current holder is done (a metric refresh queued behind a sync), not be skipped.
+    """
+    await conn.execute(text("SELECT pg_advisory_lock(hashtext(:k))"), {"k": key})
+    try:
+        yield
+    finally:
+        await conn.execute(text("SELECT pg_advisory_unlock(hashtext(:k))"), {"k": key})
@@ -0,0 +1,95 @@
+"""Run one source sync end to end: lock, run log, cursor, error capture."""
+
+from __future__ import annotations
+
+import logging
+import traceback
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+
+from fintracker.config import get_settings
+from fintracker.db import get_engine, get_sessionmaker
+from fintracker.metrics.refresh import refresh_all
+from fintracker.models import RunStatus, SyncRun, SyncState
+from fintracker.sources import registry
+from fintracker.sources.base import SyncContext, SyncResult
+from fintracker.worker.locks import try_advisory_lock
+
+log = logging.getLogger(__name__)
+
+
+class Skipped(Exception):
+    """Another run of the same source holds the lock."""
+
+
+async def run_source(name: str, triggered_by: str) -> SyncRun:
+    source = registry.get(name)
+    settings = get_settings()
+    # the lock lives on its own connection for the whole run
+    async with (
+        get_engine().connect() as lock_conn,
+        try_advisory_lock(lock_conn, f"sync:{name}") as got,
+    ):
+        if not got:
+            raise Skipped(name)
+        async with get_sessionmaker()() as session:
+            state = await session.get(SyncState, name)
+            if state is None:
+                state = SyncState(source=name)
+                session.add(state)
+            run = SyncRun(
+                source=name,
+                status=RunStatus.running,
+                triggered_by=triggered_by,
+                cursor_before=state.cursor,
+            )
+            session.add(run)
+            state.last_run_at = datetime.now(UTC)
+            await session.commit()
+            run_id = run.id
+
+            ctx = SyncContext(
+                session=session,
+                settings=settings,
+                cursor_before=state.cursor,
+                triggered_by=triggered_by,
+            )
+            try:
+                result: SyncResult = await source.sync(ctx)
+            except Exception as exc:
+                await session.rollback()
+                log.exception("sync %s failed", name)
+                async with get_sessionmaker()() as s2:
+                    run2 = await s2.get(SyncRun, run_id)
+                    assert run2 is not None
+                    run2.status = RunStatus.error
+                    run2.finished_at = datetime.now(UTC)
+                    run2.error = f"{type(exc).__name__}: {exc}\n{traceback.format_exc()[-4000:]}"
+                    await s2.commit()
+                    return run2
+
+            run = (await session.execute(select(SyncRun).where(SyncRun.id == run_id))).scalar_one()
+            state = await session.get(SyncState, name)
+            assert state is not None
+            if result.cursor_after is not None:
+                state.cursor = result.cursor_after
+            state.last_success_at = datetime.now(UTC)
+            run.status = RunStatus.ok
+            run.finished_at = datetime.now(UTC)
+            run.cursor_after = state.cursor
+            run.counts = dict(result.counts)
+            run.warnings = list(result.warnings)
+            await session.commit()
+            log.info("sync %s ok: %s", name, result.counts)
+            if result.changed:
+                entry = await refresh_all(session, trigger=f"sync:{name}")
+                if entry.error:
+                    # the download itself succeeded, so the cursor stays advanced (a replay
+                    # would fix nothing); the run is still an error, because the metrics the
+                    # UI reads are now stale. `scheduler._finish_job` picks this up.
+                    run.status = RunStatus.error
+                    run.error = f"metrics refresh failed: {entry.error}"
+                    await session.commit()
+                    log.error("sync %s: metrics refresh failed", name)
+            return run
@@ -0,0 +1,118 @@
+"""Worker process: APScheduler for the timetable + a poller for manual `sync_job` rows."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import signal
+from datetime import UTC, datetime
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from sqlalchemy import select
+
+from fintracker.config import get_settings
+from fintracker.db import get_sessionmaker, reset_engine
+from fintracker.models import JobStatus, SyncJob
+from fintracker.sources import registry
+from fintracker.worker.jobs import default_schedule
+from fintracker.worker.runner import Skipped, run_source
+
+log = logging.getLogger(__name__)
+
+POLL_SECONDS = 5
+
+
+async def _scheduled(source: str) -> None:
+    try:
+        await run_source(source, triggered_by="schedule")
+    except Skipped:
+        log.info("sync %s skipped: already running", source)
+
+
+async def _process_queued_jobs() -> None:
+    async with get_sessionmaker()() as session:
+        jobs = (
+            (
+                await session.execute(
+                    select(SyncJob)
+                    .where(SyncJob.status == JobStatus.queued)
+                    .order_by(SyncJob.requested_at)
+                )
+            )
+            .scalars()
+            .all()
+        )
+    for job in jobs:
+        if job.source not in registry.names():
+            await _finish_job(job.id, JobStatus.error, error=f"unknown source {job.source}")
+            continue
+        await _mark_running(job.id)
+        try:
+            run = await run_source(job.source, triggered_by="manual")
+        except Skipped:
+            await _finish_job(job.id, JobStatus.error, error="already running")
+            continue
+        status = JobStatus.done if run.error is None else JobStatus.error
+        await _finish_job(job.id, status, run_id=run.id, error=run.error)
+
+
+async def _mark_running(job_id) -> None:
+    async with get_sessionmaker()() as session:
+        job = await session.get(SyncJob, job_id)
+        if job is not None:
+            job.status = JobStatus.running
+            job.started_at = datetime.now(UTC)
+            await session.commit()
+
+
+async def _finish_job(job_id, status: JobStatus, *, run_id=None, error: str | None = None) -> None:
+    async with get_sessionmaker()() as session:
+        job = await session.get(SyncJob, job_id)
+        if job is not None:
+            job.status = status
+            job.finished_at = datetime.now(UTC)
+            job.run_id = run_id
+            job.error = error
+            await session.commit()
+
+
+def build_scheduler() -> AsyncIOScheduler:
+    settings = get_settings()
+    scheduler = AsyncIOScheduler(timezone=settings.timezone)
+    for spec in default_schedule():
+        scheduler.add_job(
+            _scheduled,
+            spec.trigger,
+            args=[spec.source],
+            id=f"sync:{spec.source}",
+            max_instances=1,
+            coalesce=True,
+            misfire_grace_time=600,
+        )
+    scheduler.add_job(
+        _process_queued_jobs,
+        "interval",
+        seconds=POLL_SECONDS,
+        id="poll-sync-jobs",
+        max_instances=1,
+        coalesce=True,
+    )
+    return scheduler
+
+
+async def serve_forever() -> None:
+    scheduler = build_scheduler()
+    stop = asyncio.Event()
+    loop = asyncio.get_running_loop()
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        loop.add_signal_handler(sig, stop.set)
+    scheduler.start()
+    log.info(
+        "worker started; sources=%s jobs=%s", registry.names(), [j.id for j in scheduler.get_jobs()]
+    )
+    try:
+        await stop.wait()
+    finally:
+        log.info("worker stopping")
+        scheduler.shutdown(wait=True)
+        await reset_engine()