import asyncio import logging from datetime import datetime import time from typing import Literal 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, use_checkpoint: bool = True, update_checkpoint: bool = True, sort_direction: Literal["asc", "desc"] = "asc", ) -> dict: """同步明道云开版数据表到暂存表（含可选跨表关联数据）。 - 写入 `api_v1.MDYPlateOrderStaging`： - `raw`：整行原始数据 - `related`：跨表数据（碾平 list[dict]） - 进度记录到 `api_v1.DataSync`（table_name=plate_order） 说明： - max_pages 表示“单次任务最多处理多少页”（不是最大页码） - max_records 表示“单次任务最多处理多少条记录”（0/None 表示不限制） - use_checkpoint 控制本次是否读取 DataSync 游标 - update_checkpoint 控制本次是否写入 DataSync 记录 - sort_direction 为 `asc`（默认）或 `desc`，决定按时间升/降序抓取 """ max_records = max_records or 0 sort_direction = (sort_direction or "asc").lower() if sort_direction not in {"asc", "desc"}: raise ValueError("sort_direction must be 'asc' or 'desc'") is_asc = sort_direction == "asc" last_sync = None if use_checkpoint: 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 and last_sync.page_index 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=is_asc, ) ) 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 use_checkpoint and 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: latest_ctime = record_ctime latest_rowid = rowid else: should_update = ( record_ctime > latest_ctime if is_asc else record_ctime < latest_ctime ) if should_update: 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 if update_checkpoint: 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"{sort_direction} scan; with_related={with_related}; " f"use_checkpoint={use_checkpoint}" ), ) 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, "sort_direction": sort_direction, "use_checkpoint": use_checkpoint, "update_checkpoint": update_checkpoint, } logger.info("明道云开版暂存同步完成: %s", payload) return payload