"""SDS 数据核验:逐天比对远程/本地按状态计数(排除"售后"),有差异则拉取修复。 对应 sds-verify。mode: recent(近 N 天,不含今天)/ range(指定区间)。 """ from __future__ import annotations from datetime import datetime, timedelta from prefect import flow, task from api.sds.models import OrderQueryParams from flows.sds._pipeline import sync_sds_records from flows.shared import ( SyncDB, create_sds_client, emit_verify_report, fetch_sds_all_pages, format_date, get_channel_type_map, refresh_channel_type_map, start_of_day, ) 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 n = min(max(recent_days, 1), 90) return [today - timedelta(days=i) for i in range(n, 0, -1)] def _diff(remote_stats: list, local: dict[str, int]) -> tuple[dict[str, int], bool]: diff: dict[str, int] = {} has_diff = False statuses = {getattr(s, "name", None) for s in remote_stats} | set(local) statuses.discard("售后") statuses.discard(None) for status in statuses: remote = next((getattr(s, "orderNum", 0) for s in remote_stats if getattr(s, "name", None) == status), 0) local_n = local.get(status, 0) if remote != local_n: diff[status] = local_n - remote has_diff = True return diff, has_diff @flow(name="sds-verify", log_prints=True) def sds_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_sds_client() channel_map = get_channel_type_map(client) totals = {k: 0 for k in ("ordersAdded", "ordersUpdated", "goodsAdded", "goodsUpdated")} 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) remote_stats = client.get_status( OrderQueryParams(pay_date_begin=format_date(day_start), pay_date_end=format_date(day_end)) ) remote_total = sum(getattr(s, "orderNum", 0) or 0 for s in remote_stats) local_counts = db.get_order_count_by_status("SDS", day_start, day_end) diff, has_diff = _diff(remote_stats, 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 = fetch_sds_all_pages( client, {"pay_date_begin": format_date(day_start), "pay_date_end": format_date(day_end), "size": 500}, ) stats = sync_sds_records(raw, channel_map, db, channel_refresh=lambda: refresh_channel_type_map(client)) 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("SDS 数据核验", 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__": sds_verify.serve(name="sds-verify", cron="0 7 * * *")