129 lines
5.1 KiB
Python
129 lines
5.1 KiB
Python
"""RIIN 数据核验:逐天比对远程/本地按状态计数,有差异则拉取修复。
|
|
|
|
对应 riin-verify。mode: recent(近 N 天,不含今天)/ range(指定区间)。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
from prefect import flow, task
|
|
|
|
from api.riin.models import OrderPageRequest
|
|
from flows.riin._pipeline import sync_riin_records
|
|
from flows.shared import (
|
|
SyncDB, create_riin_client, emit_verify_report, fetch_riin_all_pages, format_date, start_of_day,
|
|
)
|
|
|
|
RIIN_COUNT_STATUS_MAP = {
|
|
"assignedSuceessed": "已分配",
|
|
"factoryAuditing": "工厂审核中",
|
|
"inProducing": "生产中",
|
|
"shipped": "已发货",
|
|
"closed": "已关闭",
|
|
"refunding": "退款中",
|
|
"refunded": "已退款",
|
|
"rejecting": "拒审中",
|
|
"rejectedToShop": "已退回店铺",
|
|
}
|
|
|
|
|
|
def _parse_days(mode: str, recent_days: int, date_range: str | None) -> list[datetime]:
|
|
today = start_of_day(datetime.now())
|
|
if mode == "range" and date_range:
|
|
parts = [p.strip() for p in date_range.split("~")]
|
|
from_dt = datetime.fromisoformat(parts[0])
|
|
to_dt = datetime.fromisoformat(parts[1]) if len(parts) > 1 and parts[1] else from_dt
|
|
days = []
|
|
cur = from_dt
|
|
while cur <= to_dt:
|
|
days.append(cur)
|
|
cur += timedelta(days=1)
|
|
return days
|
|
days_n = min(max(recent_days, 1), 90)
|
|
return [today - timedelta(days=i) for i in range(days_n, 0, -1)]
|
|
|
|
|
|
def _count_to_status_map(data: dict) -> dict[str, int]:
|
|
out: dict[str, int] = {}
|
|
for key, count in data.items():
|
|
if key.startswith("dayRange"):
|
|
continue
|
|
label = RIIN_COUNT_STATUS_MAP.get(key)
|
|
if label and count and count > 0:
|
|
out[label] = count
|
|
return out
|
|
|
|
|
|
def _diff(remote: dict[str, int], local: dict[str, int]) -> tuple[dict[str, int], bool]:
|
|
diff: dict[str, int] = {}
|
|
has_diff = False
|
|
for status in set(remote) | set(local):
|
|
r = remote.get(status, 0)
|
|
l = local.get(status, 0)
|
|
if r != l:
|
|
diff[status] = l - r
|
|
has_diff = True
|
|
return diff, has_diff
|
|
|
|
|
|
@flow(name="riin-verify", log_prints=True)
|
|
def riin_verify(mode: str = "recent", recent_days: int = 1, date_range: str | None = None) -> dict:
|
|
days = _parse_days(mode, recent_days, date_range)
|
|
db = SyncDB()
|
|
client = create_riin_client()
|
|
|
|
totals = {k: 0 for k in ("ordersAdded", "ordersUpdated", "ordersRemoved",
|
|
"goodsAdded", "goodsUpdated", "goodsRemoved")}
|
|
days_with_diff = 0
|
|
day_reports = []
|
|
|
|
try:
|
|
for day in days:
|
|
day_str = format_date(day).split(" ")[0]
|
|
day_start = start_of_day(day)
|
|
day_end = day_start + timedelta(days=1)
|
|
|
|
count_resp = client.get_order_count(
|
|
OrderPageRequest(time_type=3, start_time=format_date(day_start), end_time=format_date(day_end))
|
|
)
|
|
remote_map = _count_to_status_map(count_resp.data.model_dump(by_alias=False))
|
|
remote_total = sum(remote_map.values())
|
|
local_counts = db.get_order_count_by_status("RIIN", day_start, day_end)
|
|
local_goods = db.get_local_goods_count("RIIN", day_start, day_end)
|
|
diff, has_diff = _diff(remote_map, local_counts)
|
|
|
|
print(f"[核验 {day_str}] 远程 {remote_total} 本地 {sum(local_counts.values())}"
|
|
+ (f" 差异 {diff}" if has_diff else " 一致,跳过"))
|
|
if not has_diff:
|
|
day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": True})
|
|
continue
|
|
|
|
days_with_diff += 1
|
|
raw_orders = fetch_riin_all_pages(
|
|
client,
|
|
{"time_type": 3, "start_time": format_date(day_start), "end_time": format_date(day_end), "page_size": 200},
|
|
)
|
|
stats = sync_riin_records(raw_orders, client, db)
|
|
for k in ("ordersCreated", "ordersUpdated", "goodsCreated", "goodsUpdated"):
|
|
pass
|
|
totals["ordersAdded"] += stats["ordersCreated"]
|
|
totals["ordersUpdated"] += stats["ordersUpdated"]
|
|
totals["goodsAdded"] += stats["goodsCreated"]
|
|
totals["goodsUpdated"] += stats["goodsUpdated"]
|
|
day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": False,
|
|
"orders": stats["orderCount"], "goods": stats["goodsCount"]})
|
|
finally:
|
|
client.close()
|
|
|
|
print(f"[核验汇总] 共 {len(days)} 天 / 有差异 {days_with_diff} 天 / {totals}")
|
|
emit_verify_report("RIIN 数据核验", day_reports, totals, len(days), days_with_diff)
|
|
return {
|
|
"verifyReport": {"days": day_reports, "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}},
|
|
"syncRange": {"from": format_date(start_of_day(days[0])) if days else "", "to": format_date(start_of_day(days[-1])) if days else ""},
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
riin_verify.serve(name="riin-verify", cron="0 6 * * *")
|