"""数据核验日报(每日 18:30):近 N 天 RIIN/SDS 订单+商品数据准确性核验并上报企微。

对应 daily-report deployment,cron 30 18 * * *。内容:
1. 复用 riin-verify / sds-verify 逻辑(subflow 调用):远程 vs 本地逐天逐状态计数比对,
   有差异自动拉取修复——与早 6/7 点的每日核验互补,二次确认白天同步质量。
2. 商品数据完整性:窗口内零商品订单 / 商品无名称 / 全库孤儿商品(本地 SQL)。
3. OCR 处理状态与当日失败 flow runs 回顾(事件报警的兜底汇总)。

消息排版遵循 WeCom markdown 子集(无表格,引用块+彩色字体)。
"""

from __future__ import annotations

from datetime import datetime, timedelta

from prefect import flow

from flows.monitor.device_check import check_prefect_failures
from flows.riin.verify import riin_verify
from flows.sds.verify import sds_verify
from flows.shared import SyncDB, emit_sync_summary, send_markdown, start_of_day



def _short_date(date_str: str) -> str:
    return date_str[5:] if len(date_str) == 10 and date_str[:4].isdigit() else date_str


def _affected_units(diff: dict) -> int:
    """状态不符单数 ≈ 本地比远程多的状态计数之和(远程改了状态的单)。"""
    return sum(v for v in (diff or {}).values() if v > 0)


def build_daily_message(
    days: int, window: str, riin: dict, sds: dict,
    goods: dict, ocr: dict, failures: list[dict],
) -> tuple[str, bool]:
    """拼装日报 markdown,返回 (消息, 是否需关注)。结论式极简排版,不暴露内部术语。"""
    lines = [f"📊 **数据核验日报** {datetime.now().strftime('%m-%d %H:%M')}"
             f"（近 {days} 天，{window}）", ""]
    attention = False
    diff_names: list[str] = []

    for name, report in (("RIIN", riin), ("SDS", sds)):
        try:
            days_r = (report.get("verifyReport") or {}).get("days") or []
            summary = (report.get("verifyReport") or {}).get("summary") or {}
            diff_days = [d for d in days_r if not d.get("ok")]
            total = summary.get("totalDays", len(days_r))
            if not diff_days:
                lines.append(f"**{name}** {total}/{total} 天与远程一致 ✅")
            else:
                diff_names.append(name)
                attention = True
                lines.append(f"**{name}** {total - len(diff_days)}/{total} 天与远程一致 ⚠️")
                if len(diff_days) <= 4:
                    per_day = " · ".join(
                        f"{_short_date(d['date'])} {_affected_units(d.get('diff'))}单" for d in diff_days)
                    lines.append(f"> {per_day}状态不符，已自动重新同步")
                else:
                    total_affected = sum(_affected_units(d.get("diff")) for d in diff_days)
                    lines.append(f"> {len(diff_days)} 天共约 {total_affected:,} 单状态不符，已自动重新同步")
        except Exception as err:  # noqa: BLE001
            lines.append(f"**{name}** 核验结果解析失败：{err}")
            attention = True

    zero_total = sum(g["zeroGoodsOrders"] for g in goods.values())
    lines += ["", f"**商品** RIIN {goods['RIIN']['orders']:,} 单/{goods['RIIN']['goods']:,} 行"
              f" · SDS {goods['SDS']['orders']:,} 单/{goods['SDS']['goods']:,} 行"
              + (" ⚠️" if zero_total else " ✅")]
    if zero_total:
        lines.append(f"> 有 {zero_total} 单缺商品数据")
        attention = True
    lines.append(f"**OCR** 已完成 {ocr.get('done', 0):,} · 待处理 {ocr.get('pending', 0):,}"
                 f" · 失败 {ocr.get('error', 0):,}")
    if failures:
        attention = True
        names = "、".join(f"{f['name']}（{f['state']}）" for f in failures[:4])
        lines.append(f"**失败任务(24h)** {len(failures)} 个：{names}")
    else:
        lines.append("**失败任务(24h)** 0 个 ✅")

    lines.append("")
    if attention:
        lines.append("⚠️ **存在需关注项，见上**")
        if diff_names:
            lines.append("> 每天都出现的相同差异属远程统计口径问题，非数据丢失")
    else:
        lines.append("✅ **数据全部一致，无异常**")
    return "\n".join(lines), attention


@flow(name="daily-report", log_prints=True)
def daily_report(days: int = 7) -> dict:
    """近 N 天数据核验 + 商品完整性 + OCR/失败回顾,发企微群日报。"""
    today = start_of_day(datetime.now())
    window_start = today - timedelta(days=days)
    window_end = today  # 与 verify 的 recent 约定一致:近 N 天、不含今天

    riin = riin_verify(mode="recent", recent_days=days)
    sds = sds_verify(mode="recent", recent_days=days)

    db = SyncDB()
    goods = {sys: db.get_goods_integrity(sys, window_start, window_end) for sys in ("RIIN", "SDS")}
    ocr = db.count_ocr_status()
    failures = check_prefect_failures(hours=24)

    window = f"{window_start:%m-%d}~{window_end:%m-%d}"
    content, attention = build_daily_message(days, window, riin, sds, goods, ocr, failures)

    sent = send_markdown(content)
    emit_sync_summary("数据核验日报", [
        ("核验窗口", f"近{days}天"), ("需关注", "是" if attention else "否"),
        ("RIIN 零商品订单", goods["RIIN"]["zeroGoodsOrders"]),
        ("RIIN 商品无名称", goods["RIIN"]["goodsNoName"]),
        ("SDS 零商品订单", goods["SDS"]["zeroGoodsOrders"]),
        ("SDS 商品无名称", goods["SDS"]["goodsNoName"]),
        ("全库孤儿商品", goods["RIIN"]["orphanGoods"]),
        ("OCR 待处理", ocr.get("pending", 0)), ("失败 runs(24h)", len(failures)),
        ("已发企微", "是" if sent else "否"),
    ])
    return {"attention": attention, "sent": sent,
            "goods": goods, "ocr": ocr, "failures24h": len(failures)}


if __name__ == "__main__":
    daily_report.serve(name="daily-report", cron="30 18 * * *")