forked from erp-dev/erp
320 lines
11 KiB
Python
320 lines
11 KiB
Python
import asyncio
|
||
import logging
|
||
from datetime import datetime
|
||
import time
|
||
from typing import Literal
|
||
|
||
from django.db import connection
|
||
from django.db.utils import IntegrityError
|
||
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__)
|
||
|
||
_MDY_PLATE_ORDER_STAGING_PK_CONSTRAINT = "api_mdy_plate_order_staging_pkey"
|
||
|
||
|
||
def _reset_mdy_plate_order_staging_id_sequence() -> None:
|
||
"""修复 PostgreSQL 自增序列落后导致的主键重复（duplicate key violates ..._pkey）。
|
||
|
||
常见场景：数据库 restore/copy 后，sequence 未随数据同步到最新值。
|
||
"""
|
||
table_name = api_models.MDYPlateOrderStaging._meta.db_table # 'api_mdy_plate_order_staging'
|
||
pk_column = api_models.MDYPlateOrderStaging._meta.pk.column # 'id'
|
||
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(f"SELECT COALESCE(MAX({pk_column}), 1) FROM {table_name}")
|
||
max_id = cursor.fetchone()[0] or 1
|
||
|
||
cursor.execute("SELECT pg_get_serial_sequence(%s, %s)", [table_name, pk_column])
|
||
seq_name = cursor.fetchone()[0]
|
||
if not seq_name:
|
||
logger.warning("未找到 %s.%s 的 serial sequence，跳过 setval", table_name, pk_column)
|
||
return
|
||
|
||
# is_called=true => nextval 会返回 max_id+1
|
||
cursor.execute("SELECT setval(%s, %s, true)", [seq_name, max_id])
|
||
logger.warning(
|
||
"已重置 sequence=%s 到 max(id)=%s（下次插入将从 %s 开始）",
|
||
seq_name,
|
||
max_id,
|
||
max_id + 1,
|
||
)
|
||
|
||
|
||
def _is_pk_duplicate_error(exc: Exception) -> bool:
|
||
"""判断是否为主键约束冲突（多见于 sequence 未对齐）。"""
|
||
msg = str(exc) or ""
|
||
return _MDY_PLATE_ORDER_STAGING_PK_CONSTRAINT in msg
|
||
|
||
|
||
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
|
||
|
||
defaults = {
|
||
"ctime": _parse_mdy_datetime(row.get("ctime")),
|
||
"utime": _parse_mdy_datetime(row.get("utime")),
|
||
"raw": row,
|
||
"related": related_map.get(rowid, []),
|
||
}
|
||
|
||
try:
|
||
api_models.MDYPlateOrderStaging.objects.update_or_create(
|
||
mdy_rowid=rowid,
|
||
defaults=defaults,
|
||
)
|
||
except IntegrityError as exc:
|
||
# 仅对“主键重复（通常为 sequence 未对齐）”做一次自愈重试。
|
||
if _is_pk_duplicate_error(exc):
|
||
logger.warning(
|
||
"检测到暂存表主键冲突，尝试重置序列后重试一次：rowid=%s err=%s",
|
||
rowid,
|
||
exc,
|
||
)
|
||
_reset_mdy_plate_order_staging_id_sequence()
|
||
api_models.MDYPlateOrderStaging.objects.update_or_create(
|
||
mdy_rowid=rowid,
|
||
defaults=defaults,
|
||
)
|
||
else:
|
||
raise
|
||
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
|