refactor: unify data preparations and research input snapshots

backend/app/catalog/sync.py

@@ -4,6 +4,7 @@ import asyncio
 import math
 import re
 from urllib.parse import parse_qs, urlparse
+from uuid import uuid4
 
 from sqlalchemy import select
 
@@ -64,14 +65,15 @@ def normalize(raw, dataset_id):
     )
 
 
-async def sync_catalog(runner, job_id, payload):
+async def sync_catalog(runner, job_id, payload, *, batch_id=None, full=False):
     scope = Scope.model_validate(payload["scope"])
     dataset_id = payload.get("dataset_id")
+    batch_id = batch_id or job_id
     async with runner.sessions() as db:
-        checkpoint = (await db.get(Job, job_id)).checkpoint
-    if checkpoint.get("done"):
-        return
-    offset = checkpoint.get("offset", 0)
+        batch = await db.get(CatalogBatch, batch_id)
+        if batch.complete:
+            return
+        offset = batch.offset
     while True:
         await runner.checkpoint(job_id, {"next_retry_at": None})
         raw = await runner.client.catalog_page(scope.model_dump(), dataset_id, offset)
@@ -97,12 +99,12 @@ async def sync_catalog(runner, job_id, payload):
             job = await db.get(Job, job_id)
             if job.cancel_requested:
                 raise asyncio.CancelledError()
-            batch = await db.get(CatalogBatch, job_id)
+            batch = await db.get(CatalogBatch, batch_id)
             added = 0
             for entry in entries:
-                if await db.get(CatalogEntry, (job_id, entry["id"])):
+                if await db.get(CatalogEntry, (batch_id, entry["id"])):
                     continue
-                db.add(CatalogEntry(batch_id=job_id, **entry))
+                db.add(CatalogEntry(batch_id=batch_id, **entry))
                 await db.flush()
                 added += 1
                 owner = dataset_id or entry["id"]
@@ -112,25 +114,28 @@ async def sync_catalog(runner, job_id, payload):
             if rows and not added:
                 raise WqError("平台分页重复且未前进，已保留进度", "invalid_response")
             batch.count += added
-            job.processed = batch.count
+            if not full:
+                job.processed = batch.count
             offset += len(rows)
-            job.checkpoint = dict(offset=offset, done=not more)
+            batch.offset = offset
+            job.checkpoint = {**job.checkpoint, "offset": offset, "done": not more, "current_field_count": batch.count}
             job.updated_at = now()
             if not more:
                 batch.complete, batch.completed_at = True, now()
-                job.total = batch.count
+                if not full:
+                    job.total = batch.count
                 if dataset_id:
                     dataset = await db.scalar(
                         select(CatalogDataset)
                         .where(CatalogDataset.scope_key == scope.key(), CatalogDataset.id == dataset_id)
                         .with_for_update()
                     )
-                    dataset.field_version = job_id
+                    dataset.field_version = batch_id
                 else:
                     scope_row = await db.get(CatalogScope, scope.key())
-                    scope_row.catalog_version, scope_row.synced_at = job_id, now()
+                    scope_row.catalog_version, scope_row.synced_at = batch_id, now()
                     ids = (
-                        await db.scalars(select(CatalogEntry.id).where(CatalogEntry.batch_id == job_id))
+                        await db.scalars(select(CatalogEntry.id).where(CatalogEntry.batch_id == batch_id))
                     ).all()
                     for item_id in ids:
                         if not await db.get(CatalogDataset, (scope.key(), item_id)):
@@ -138,3 +143,66 @@ async def sync_catalog(runner, job_id, payload):
             await db.commit()
         if not more:
             return
+
+
+async def sync_full_catalog(runner, job_id, payload):
+    """Resume each dataset batch independently; only publish complete enumerations."""
+    scope = Scope.model_validate(payload["scope"])
+    options = await runner.client.get_platform_setting_options()
+    if not any(r["instrument_type"] == scope.instrument_type and r["region"] == scope.region
+               and r["delay"] == scope.delay and scope.universe in r["universes"]
+               for r in options["instrument_options"]):
+        async with runner.sessions.begin() as db:
+            job = await db.get(Job, job_id)
+            job.checkpoint = {**job.checkpoint, "error_code": "invalid_scope"}
+        raise WqError("平台不支持该研究范围", "invalid_scope")
+    async with runner.sessions.begin() as db:
+        job = await db.get(Job, job_id)
+        job.checkpoint = {**{k: v for k, v in job.checkpoint.items() if k != "error_code"}, "phase": "catalog"}
+    await sync_catalog(runner, job_id, {"scope": payload["scope"]}, full=True)
+    async with runner.sessions() as db:
+        ids = list(await db.scalars(select(CatalogEntry.id).where(CatalogEntry.batch_id == job_id)
+                                   .order_by(CatalogEntry.id)))
+        job = await db.get(Job, job_id)
+        failures = dict(job.checkpoint.get("failures", {}))
+    completed = 0
+    await runner.checkpoint(job_id, {"total": len(ids)})
+    for dataset_id in ids:
+        async with runner.sessions.begin() as db:
+            batch = await db.scalar(select(CatalogBatch).where(CatalogBatch.job_id == job_id,
+                                                               CatalogBatch.dataset_id == dataset_id))
+            if not batch:
+                batch = CatalogBatch(id=str(uuid4()), job_id=job_id, scope_key=scope.key(), dataset_id=dataset_id)
+                db.add(batch)
+                await db.flush()
+            batch_id, complete = batch.id, batch.complete
+            job = await db.get(Job, job_id)
+            if job.cancel_requested:
+                raise asyncio.CancelledError()
+            job.checkpoint = {**job.checkpoint, "phase": "fields", "dataset_id": dataset_id,
+                              "datasets_completed": completed, "datasets_total": len(ids),
+                              "offset": batch.offset, "current_field_count": batch.count}
+        if not complete:
+            try:
+                await sync_catalog(runner, job_id, {"scope": payload["scope"], "dataset_id": dataset_id},
+                                   batch_id=batch_id, full=True)
+            except WqError as exc:
+                if exc.code in ("disconnected", "authentication_failed", "identity_mismatch", "verification_required", "network_error"):
+                    raise
+                failures[dataset_id] = str(exc)
+        if complete or (await _batch_complete(runner, batch_id)):
+            failures.pop(dataset_id, None)
+            completed += 1
+        async with runner.sessions.begin() as db:
+            job = await db.get(Job, job_id)
+            job.processed, job.failed = completed, len(failures)
+            job.checkpoint = {**job.checkpoint, "datasets_completed": completed, "failures": failures}
+    async with runner.sessions.begin() as db:
+        job = await db.get(Job, job_id)
+        job.error = f"{len(failures)} 个数据集同步失败" if failures else None
+        job.checkpoint = {**job.checkpoint, "phase": "finished"}
+
+
+async def _batch_complete(runner, batch_id):
+    async with runner.sessions() as db:
+        return (await db.get(CatalogBatch, batch_id)).complete