105 lines
4.2 KiB
Python
105 lines
4.2 KiB
Python
"""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, 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, "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, "diff": diff, "ok": False,
|
|
"orders": stats["orderCount"], "goods": stats["goodsCount"]})
|
|
finally:
|
|
client.close()
|
|
|
|
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 * * *")
|