yeuimu
7c323e9a3f
- 核验日报: 先报订单数量(远程vs本地, verify 修复后重数 verifyReport.post)/商品行数一致性, 再报状态不符单数; SDS 统计口径排除'售后'与本地对称 - OCR: Orders 加 ocrProcessedAt 列(写入自动盖时间戳), 日报报'今日处理 X · 识别失败 Y'(db.count_ocr_today) - 失败任务: check_prefect_failures 补 workflow 名, 日报/设备报列 run 名+状态+时间 - 设备日报: 新增 metrics-agent 侧车(每5分钟采样内存/swap/GPU+每日 du 快照), 报告改 平均/峰值/最低 + 总体/根目录/home 余量 + 增长最快目录 + GPU 温度; 采样缺失降级瞬时值 - Dockerfile 加 tzdata(TZ=Asia/Shanghai); du 用实际占块且容错 rc=1
140 lines
6.4 KiB
Python
140 lines
6.4 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, 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
|
|
|
|
|
|
def _remote_map(remote_stats: list) -> dict[str, int]:
|
|
"""远程状态→数量映射(排除"售后"和无名状态,与本地口径对称)。"""
|
|
out = {getattr(s, "name", None): (getattr(s, "orderNum", 0) or 0) for s in remote_stats}
|
|
out.pop("售后", None)
|
|
out.pop(None, None)
|
|
return out
|
|
|
|
|
|
@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_map = _remote_map(remote_stats)
|
|
remote_total = sum(remote_map.values())
|
|
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"]})
|
|
|
|
# 修复后重数:数量接口 + 本地 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)
|
|
remote_stats = client.get_status(
|
|
OrderQueryParams(pay_date_begin=format_date(day_start), pay_date_end=format_date(day_end))
|
|
)
|
|
remote_map = _remote_map(remote_stats)
|
|
local_counts = db.get_order_count_by_status("SDS", day_start, day_end)
|
|
statuses = set(remote_map) | set(local_counts)
|
|
statuses.discard("售后")
|
|
has_diff = any(remote_map.get(st, 0) != local_counts.get(st, 0) for st in statuses)
|
|
post_diff_days += int(has_diff)
|
|
post_days.append({"date": format_date(day).split(" ")[0],
|
|
"remote": sum(remote_map.values()),
|
|
"local": sum(v for k, v in local_counts.items() if k != "售后"),
|
|
"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("SDS 数据核验", 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__":
|
|
sds_verify.serve(name="sds-verify", cron="0 7 * * *")
|