inkreach-prefect/flows/riin/verify.py

"""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 * * *")