"""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) 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"]}) # 修复后重数:数量 API + 本地 SQL 很便宜,给出修复后的数量一致性结论(口径与 _diff 相同) post = None if days_with_diff: post_days, post_diff_days = [], 0 for day in days: day_start, day_end = start_of_day(day), start_of_day(day) + 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)) local_counts = db.get_order_count_by_status("RIIN", day_start, day_end) diff, has_diff = _diff(remote_map, local_counts) post_diff_days += int(has_diff) post_days.append({"date": format_date(day).split(" ")[0], "remote": sum(remote_map.values()), "local": sum(local_counts.values()), "ok": not has_diff}) post = {"days": post_days, "daysWithDiff": post_diff_days, "remote": sum(d["remote"] for d in post_days), "local": sum(d["local"] for d in post_days)} print(f"[核验重数] 修复后 远程 {post['remote']} 本地 {post['local']} / 有差异 {post_diff_days} 天") 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, "post": post, "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 * * *")