feat: batch update for plate order

api_v1/mdy_plate_order_sync.py

@@ -0,0 +1,232 @@
+import asyncio
+import logging
+from datetime import datetime
+import time
+
+from django.utils import timezone
+
+from api_v1 import models as api_models
+from flower.utils.mingdaoyun.fetch import (
+    fetch_plate_orders_from_mingdaoyun,
+    fetch_row_by_rowid_from_mingdaoyun,
+)
+from flower.utils.mingdaoyun.mappings import plate_order_related_worksheet_map
+from flower.utils.mingdaoyun.relations import (
+    extract_plate_order_related_rowids,
+    flatten_plate_order_related_row,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _parse_mdy_datetime(value: str | None):
+    if not value:
+        return None
+    try:
+        dt = datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
+    except ValueError:
+        return None
+    if timezone.is_naive(dt):
+        dt = timezone.make_aware(dt, timezone.get_current_timezone())
+    return dt
+
+
+async def _fetch_related_flattened(
+    plate_order_row: dict,
+    *,
+    max_related_per_type: int = 5,
+    request_interval_seconds: float = 0.0,
+) -> list[dict]:
+    """根据开版主表 row 的 Relation 字段，拉取并碾平关联表数据。"""
+
+    related_rowids_map = extract_plate_order_related_rowids(plate_order_row)
+    if not related_rowids_map:
+        return []
+
+    flattened: list[dict] = []
+    for related_key, rowids in related_rowids_map.items():
+        wsid = plate_order_related_worksheet_map.get(related_key)
+        if not wsid:
+            continue
+
+        for rid in rowids[: max_related_per_type or 0]:
+            related_row = await fetch_row_by_rowid_from_mingdaoyun(
+                worksheet_id=wsid,
+                rowid=rid,
+            )
+            if not related_row:
+                continue
+
+            flattened.append(flatten_plate_order_related_row(related_key, related_row))
+            if request_interval_seconds:
+                await asyncio.sleep(request_interval_seconds)
+
+    return flattened
+
+
+async def _fetch_related_for_rows(
+    rows: list[dict],
+    *,
+    max_related_per_type: int,
+    request_interval_seconds: float = 0.0,
+) -> dict[str, list[dict]]:
+    """批量拉取并碾平关联表数据（仅网络请求，不涉及 ORM）。"""
+
+    result: dict[str, list[dict]] = {}
+    for row in rows:
+        rowid = row.get("rowid")
+        if not rowid:
+            continue
+        result[rowid] = await _fetch_related_flattened(
+            row,
+            max_related_per_type=max_related_per_type,
+            request_interval_seconds=request_interval_seconds,
+        )
+    return result
+
+
+def sync_mdy_plate_orders_to_staging(
+    *,
+    page_size: int = 100,
+    max_pages: int | None = None,
+    max_records: int | None = None,
+    with_related: bool = True,
+    max_related_per_type: int = 5,
+    request_interval_seconds: float = 0.02,
+) -> dict:
+    """同步明道云开版数据表到暂存表（含可选跨表关联数据）。
+
+    - 写入 `api_v1.MDYPlateOrderStaging`：
+      - `raw`：整行原始数据
+      - `related`：跨表数据（碾平 list[dict]）
+    - 进度记录到 `api_v1.DataSync`（table_name=plate_order）
+
+    说明：
+    - max_pages 表示“单次任务最多处理多少页”（不是最大页码）
+    - max_records 表示“单次任务最多处理多少条记录”（0/None 表示不限制）
+    """
+
+    max_records = max_records or 0
+
+    last_sync = api_models.DataSync.objects.filter(
+        table_name=api_models.DataSync.TableName.PLATE_ORDER
+    ).order_by("-created_at").first()
+    last_ctime = last_sync.last_ctime if last_sync else None
+    last_rowid = last_sync.last_rowid if last_sync else ""
+    start_page_index = last_sync.page_index if last_sync else 1
+
+    synced_rows = 0
+    page_index = max(1, start_page_index)
+    pages_processed = 0
+    total_count = 0
+    latest_ctime = last_ctime
+    latest_rowid = last_rowid
+
+    while True:
+        if max_pages is not None and pages_processed >= max_pages:
+            break
+        if max_records and synced_rows >= max_records:
+            break
+
+        rows, total = asyncio.run(
+            fetch_plate_orders_from_mingdaoyun(
+                page=page_index,
+                page_size=page_size,
+                sort_id="ctime",
+                is_asc=True,
+            )
+        )
+        if request_interval_seconds:
+            time.sleep(request_interval_seconds)
+        total_count = total
+        if not rows:
+            break
+
+        hit_max_records = False
+        to_process: list[dict] = []
+        for row in rows:
+            if max_records and synced_rows + len(to_process) >= max_records:
+                hit_max_records = True
+                break
+
+            rowid = row.get("rowid")
+            if not rowid:
+                continue
+
+            record_ctime = _parse_mdy_datetime(row.get("ctime"))
+
+            # 跳过已同步到的游标
+            if last_ctime and record_ctime:
+                if record_ctime < last_ctime:
+                    continue
+                if record_ctime == last_ctime and last_rowid and rowid == last_rowid:
+                    continue
+
+            to_process.append(row)
+
+        related_map: dict[str, list[dict]] = {}
+        if with_related and to_process:
+            related_map = asyncio.run(
+                _fetch_related_for_rows(
+                    to_process,
+                    max_related_per_type=max_related_per_type,
+                    request_interval_seconds=request_interval_seconds,
+                )
+            )
+
+        for row in to_process:
+            rowid = row.get("rowid")
+            if not rowid:
+                continue
+
+            api_models.MDYPlateOrderStaging.objects.update_or_create(
+                mdy_rowid=rowid,
+                defaults={
+                    "ctime": _parse_mdy_datetime(row.get("ctime")),
+                    "utime": _parse_mdy_datetime(row.get("utime")),
+                    "raw": row,
+                    "related": related_map.get(rowid, []),
+                },
+            )
+            synced_rows += 1
+
+            record_ctime = _parse_mdy_datetime(row.get("ctime"))
+            if record_ctime:
+                if latest_ctime is None or record_ctime > latest_ctime:
+                    latest_ctime = record_ctime
+                    latest_rowid = rowid
+                elif record_ctime == latest_ctime:
+                    latest_rowid = rowid
+
+        pages_processed += 1
+        if hit_max_records:
+            break
+
+        if len(rows) < page_size:
+            break
+        page_index += 1
+
+    record_last_ctime = latest_ctime or last_ctime
+    record_last_rowid = latest_rowid or last_rowid
+
+    api_models.DataSync.objects.create(
+        table_name=api_models.DataSync.TableName.PLATE_ORDER,
+        page_index=page_index,
+        page_size=page_size,
+        synced_rows=synced_rows,
+        total_count=total_count,
+        last_ctime=record_last_ctime,
+        last_rowid=record_last_rowid,
+        note=f"asc scan; with_related={with_related}",
+    )
+
+    payload = {
+        "synced_rows": synced_rows,
+        "page_index": page_index,
+        "page_size": page_size,
+        "total_count": total_count,
+        "last_ctime": record_last_ctime.isoformat() if record_last_ctime else None,
+        "with_related": with_related,
+    }
+    logger.info("明道云开版暂存同步完成: %s", payload)
+    return payload