feat: implement scoped dataset catalog and template input drafts

2026-09-08 09:22:53 +08:00
parent 404a4d8a04
commit 01d169d118
30 changed files with 3024 additions and 37 deletions
@@ -0,0 +1,139 @@
+"""Publish complete enumerations only; retain staging checkpoints and old versions."""
+
+import asyncio
+import math
+import re
+from urllib.parse import parse_qs, urlparse
+
+from sqlalchemy import select
+
+from ..models import CatalogBatch, CatalogDataset, CatalogEntry, CatalogNote, CatalogScope, Job, now
+from ..worldquant import WqError
+from .contracts import Scope
+
+
+def identifier(value):
+    if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_.-]{1,200}", value):
+        raise WqError("平台目录包含无法识别的 ID，已保留进度", "invalid_response")
+    return value
+
+
+def label(value):
+    if isinstance(value, dict):
+        value = value.get("name") or value.get("id")
+    return value if isinstance(value, str) and value else None
+
+
+def number(value, integer=False):
+    if (
+        isinstance(value, bool)
+        or not isinstance(value, (int, float))
+        or not math.isfinite(value)
+        or value < 0
+    ):
+        return None
+    return int(value) if integer and value == int(value) else None if integer else value
+
+
+def normalize(raw, dataset_id):
+    if not isinstance(raw, dict):
+        raise WqError("平台目录记录格式无法识别", "invalid_response")
+    item_id = identifier(raw.get("id"))
+    owner = raw.get("dataset")
+    owner = owner.get("id") if isinstance(owner, dict) else owner
+    if dataset_id and owner != dataset_id:
+        raise WqError("平台返回了其他数据集的字段", "invalid_response")
+    coverage = number(raw.get("coverage"))
+    # BRAIN coverage is a fraction. Never guess that a value >1 means percent.
+    # Real-account schema/units still require read-only integration verification.
+    if coverage is not None and coverage > 1:
+        raise WqError("平台覆盖率单位无法确认，应为 0–1", "invalid_response")
+    return dict(
+        id=item_id,
+        name=label(raw.get("name")) or item_id,
+        category=label(raw.get("category")),
+        subcategory=label(raw.get("subcategory")),
+        field_type=label(raw.get("type")) if dataset_id else None,
+        coverage=coverage,
+        user_count=number(raw.get("userCount"), True),
+        alpha_count=number(raw.get("alphaCount"), True),
+        field_count=number(raw.get("fieldCount"), True),
+        description=label(raw.get("description")),
+        unit=label(raw.get("unit")),
+    )
+
+
+async def sync_catalog(runner, job_id, payload):
+    scope = Scope.model_validate(payload["scope"])
+    dataset_id = payload.get("dataset_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)
+    while True:
+        await runner.checkpoint(job_id, {"next_retry_at": None})
+        raw = await runner.client.catalog_page(scope.model_dump(), dataset_id, offset)
+        rows = raw.get("results")
+        if not isinstance(rows, list):
+            raise WqError("平台目录缺少 results，已保留进度", "invalid_response")
+        entries = [normalize(r, dataset_id) for r in rows]
+        # Always probe to exhaustion if next is absent; count alone cannot prove completeness.
+        next_page = raw.get("next")
+        if "next" in raw and next_page is not None:
+            if not isinstance(next_page, str) or not next_page:
+                raise WqError("平台 next 分页格式无法识别", "invalid_response")
+            parsed = urlparse(next_page)
+            expected_path = "/data-fields" if dataset_id else "/data-sets"
+            offsets = parse_qs(parsed.query).get("offset", [])
+            if parsed.path.rstrip("/") != expected_path or offsets != [str(offset + len(rows))]:
+                raise WqError("平台 next 分页未按预期前进", "invalid_response")
+        more = next_page is not None if "next" in raw else bool(rows)
+        count = number(raw.get("count"), True)
+        if (more and not rows) or (not more and count is not None and offset + len(rows) < count):
+            raise WqError("平台分页提前结束，未发布不完整集合", "invalid_response")
+        async with runner.sessions() as db:
+            job = await db.get(Job, job_id)
+            if job.cancel_requested:
+                raise asyncio.CancelledError()
+            batch = await db.get(CatalogBatch, job_id)
+            added = 0
+            for entry in entries:
+                if await db.get(CatalogEntry, (job_id, entry["id"])):
+                    continue
+                db.add(CatalogEntry(batch_id=job_id, **entry))
+                await db.flush()
+                added += 1
+                owner = dataset_id or entry["id"]
+                field_id = entry["id"] if dataset_id else ""
+                if not await db.get(CatalogNote, (scope.key(), owner, field_id)):
+                    db.add(CatalogNote(scope_key=scope.key(), dataset_id=owner, field_id=field_id))
+            if rows and not added:
+                raise WqError("平台分页重复且未前进，已保留进度", "invalid_response")
+            batch.count += added
+            job.processed = batch.count
+            offset += len(rows)
+            job.checkpoint = dict(offset=offset, done=not more)
+            job.updated_at = now()
+            if not more:
+                batch.complete, batch.completed_at = True, now()
+                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
+                else:
+                    scope_row = await db.get(CatalogScope, scope.key())
+                    scope_row.catalog_version, scope_row.synced_at = job_id, now()
+                    ids = (
+                        await db.scalars(select(CatalogEntry.id).where(CatalogEntry.batch_id == job_id))
+                    ).all()
+                    for item_id in ids:
+                        if not await db.get(CatalogDataset, (scope.key(), item_id)):
+                            db.add(CatalogDataset(scope_key=scope.key(), id=item_id))
+            await db.commit()
+        if not more:
+            return