浅念
9cbd796975
Task 2.7 — the coordination layer tying scrapers + repos together with
resumable, incremental sync.
app/sync/watermarks.py — SyncState CRUD keyed on (entity_type, channel,
scope_id): get_watermark / get_last_timestamp / get_last_version /
set_watermark (upsert). A partial advance (timestamp-only) does not clobber
the other cursor with NULL.
app/sync/orchestrator.py — the sync brain:
* sync_wechat_messages: time-window incremental. Reads last_timestamp for
(message, wechat, "<wechat_id>:<friend_id>"), queries [last_ts, now] via
scrape_wechat_messages, upserts each, best-effort downloads media, advances
the watermark to max(timestamp). A re-run with no new messages queries an
empty window and fetches nothing — the watermark advanced past.
* sync_{departments,wechat_accounts,wechat_friends,wechat_groups,
wechat_voice_calls}: full re-scrape + idempotent upsert (these list APIs
have no since-param; 'incremental' = re-scrape all, rely on PK idempotency).
Watermark records last_synced_at only.
* sync_all_wechat: coordinator running departments -> accounts -> friends ->
groups -> voice_calls, then per-conversation messages for an explicit
(wechat_id, friend_id) list. Auto-discovery deferred.
Each run logs a SyncRun row (started -> success/failed, counts, error).
Failure policy: a scraper that raises is caught, recorded as failed, and a
failed SyncResult is returned — the orchestrator does not re-raise, so one
broken entity does not abort the batch. Media-download failures are
best-effort. The orchestrator never commits; callers own the transaction.
Tests (TDD red -> green; in-memory SQLite; scrapers/repos monkeypatched):
- watermarks: set/get, upsert (same key twice -> 1 row), partial-advance
preserves the other cursor, defaults when absent.
- orchestrator messages: first run upserts 2 + advances watermark to T2;
second run scrape called with start==T2 and upserts 0 (incremental proof);
media failure does not abort; SyncRun success recorded.
- orchestrator friends: 3 records upserted, SyncRun success, watermark
last_synced_at set.
- failure policy: scraper raises -> SyncRun failed + failed result, no raise
(both friends and message paths).
- sync_all_wechat: 5 entities + 2 message runs in order, all success.
Suite: 149 -> 163 passing. ruff clean.
Co-Authored-By: Claude <noreply@anthropic.com>
49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
"""Incremental-sync coordination layer (Task 2.7).
|
|
|
|
This package ties the scrapers (``app.scraper.*``) and the persistence layer
|
|
(``app.storage.repos``) together with **resumable, incremental** sync via the
|
|
``SyncState`` watermark table and the ``SyncRun`` audit log.
|
|
|
|
Public surface:
|
|
|
|
* :mod:`app.sync.watermarks` — SyncState CRUD (per-entity / per-channel /
|
|
per-scope cursor get/set + convenience accessors).
|
|
* :mod:`app.sync.orchestrator` — the sync brain. One entrypoint per entity
|
|
(``sync_wechat_messages`` / ``sync_wechat_friends`` / ...) plus a coordinator
|
|
(:func:`app.sync.orchestrator.sync_all_wechat`). Each run is logged as a
|
|
``SyncRun`` row (started → success/failed) so the dashboard / ops view can
|
|
report progress and errors.
|
|
"""
|
|
|
|
from app.sync.orchestrator import (
|
|
SyncResult,
|
|
sync_all_wechat,
|
|
sync_departments,
|
|
sync_wechat_accounts,
|
|
sync_wechat_friends,
|
|
sync_wechat_groups,
|
|
sync_wechat_messages,
|
|
sync_wechat_voice_calls,
|
|
)
|
|
from app.sync.watermarks import (
|
|
get_last_timestamp,
|
|
get_last_version,
|
|
get_watermark,
|
|
set_watermark,
|
|
)
|
|
|
|
__all__ = [
|
|
"SyncResult",
|
|
"get_last_timestamp",
|
|
"get_last_version",
|
|
"get_watermark",
|
|
"set_watermark",
|
|
"sync_all_wechat",
|
|
"sync_departments",
|
|
"sync_wechat_accounts",
|
|
"sync_wechat_friends",
|
|
"sync_wechat_groups",
|
|
"sync_wechat_messages",
|
|
"sync_wechat_voice_calls",
|
|
]
|