forked from erp-dev/erp
233 lines
7.1 KiB
Python
233 lines
7.1 KiB
Python
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
|