"""比价记录 CRUD:上报 upsert(按 user_id+trace_id 幂等) + 派生字段 + 明细分页。

派生逻辑:best_* / saved_amount_cents / is_source_best 全部从 comparison_results 算出
(协议保证已按 price 升序、rank=1 最便宜),客户端不用自己算、也不可信它算。
"""
from __future__ import annotations

from datetime import datetime, timedelta

from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session, defer

from app.core.rewards import CN_TZ
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.comparison import ComparisonRecord
from app.models.savings import SavingsRecord
from app.models.user import User
from app.schemas.compare_record import ComparisonRecordIn

DAILY_COMPARE_START_LIMIT = 100


class DailyCompareStartLimitExceeded(Exception):
    """The authenticated user has consumed today's comparison-start quota."""


class ComparisonTraceOwnershipError(Exception):
    """A trace id already belongs to a different authenticated user."""


def _yuan_to_cents(yuan: float | None) -> int | None:
    """元(float)→ 分(int)。None 透传。"""
    if yuan is None:
        return None
    return round(yuan * 100)


def _product_names_from_items(items: list | None) -> str | None:
    """下单商品 items([{name, qty, specs?}])→ 顿号分隔的商品名串(去重保序),
    供 admin「商品」列展示 + 商品名 LIKE 搜索。空 / 无名 → None;超列宽(512)截断留余量。"""
    if not items:
        return None
    names: list[str] = []
    for it in items:
        name = it.get("name") if isinstance(it, dict) else None
        if not name:
            continue
        s = str(name).strip()
        if s and s not in names:
            names.append(s)
    joined = "、".join(names)
    return joined[:500] or None


# ---- 失败记录的展示文案(记录页失败卡「原因」行)------------------------------
# information 具体就直出;笼统(_GENERIC_INFO)则从 platform_results 捞一条用户可读的业务
# 原因;捞不到 → None(端侧显示品牌兜底「网络开小差…」)。pricebot 把 store_closed /
# no_delivery 漏成了 status=failed,这里按 reason 关键字补判;打烊类 reason 常带一坨脏店名
# (店名+月售+起送+配送…),统一成简短模板。自动化黑话(搜索失败/读价失败/购物车残留/裸
# FAILED…)不给用户看 → 归入品牌兜底。

# pricebot 组不出具体原因时的笼统 information(线上统计的大头),一律走品牌兜底。
_GENERIC_INFO = {
    "比价过程出错，请稍后重试",
    "比价出错",
    "比价未完成",
    "done 参数缺少可验证的目标平台结果",
}
# 干净业务结局 status(直接可信),按展示优先级(越靠前越先选)。
_BIZ_STATUS_PRIORITY = (
    "below_minimum",
    "no_delivery",
    "store_closed",
    "items_not_found",
    "store_not_found",
)


def _store_closed_text(reason: str | None) -> str:
    """打烊/暂停营业/休息类 reason 常带脏店名元数据 → 只留结论,套简短模板。"""
    r = reason or ""
    if "暂停营业" in r:
        state = "暂停营业"
    elif "休息" in r:
        state = "休息中"
    else:
        state = "已打烊"
    return f"门店{state}，无法比价"


def _target_display_reason(platform_results: dict | None) -> str | None:
    """从逐平台结果里挑一条"可展示给用户"的失败原因;挑不到返回 None。
    ① status 命中干净业务结局集 → 直接采信(打烊套模板,其余用 reason);
    ② 补判 pricebot 漏成 status=failed 的两类:打烊(套模板)、单点不配送(reason 本身干净);
    自动化黑话(搜索失败/读价失败/购物车残留/裸 FAILED…)一律不展示 → None。"""
    pr = platform_results or {}
    targets = [
        v for v in pr.values() if isinstance(v, dict) and not v.get("is_source")
    ]
    for want in _BIZ_STATUS_PRIORITY:  # ① 干净 status 优先
        for v in targets:
            if v.get("status") == want:
                if want == "store_closed":
                    return _store_closed_text(v.get("reason"))
                if v.get("reason"):
                    return v["reason"]
    for v in targets:  # ② 漏成 failed 的业务结局补判
        if v.get("status") != "failed":
            continue
        reason = (v.get("reason") or "").strip()
        if any(k in reason for k in ("打烊", "暂停营业", "休息")):
            return _store_closed_text(reason)
        if "单点不配送" in reason:
            return reason
    return None


def _derive_fail_display(
    information: str | None, platform_results: dict | None
) -> str | None:
    """失败记录展示文案:information 具体则直出;笼统则从 platform_results 捞/补判;
    都拿不到 → None(端侧品牌兜底)。仅在 status=failed 时调用。"""
    info = (information or "").strip()
    text = info if (info and info not in _GENERIC_INFO) else _target_display_reason(
        platform_results
    )
    return text[:256] if text else None


def _derive(payload: ComparisonRecordIn) -> dict:
    """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
    results = payload.comparison_results
    _pr = payload.platform_results or {}

    def _is_short(r) -> bool:
        # 缺菜(漏菜)店: 少买了菜总价虚低, 不参与最优评选。逐平台 skipped 在 platform_results, 行里没有。
        # platform_results 内层结构宽松(pricebot/老客户端透传, 可伪造), 值非 dict 时按"不缺菜"处理, 不崩。
        info = _pr.get(r.platform_id) if r.platform_id else None
        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0

    # 最优 = 非缺菜里 rank 最小(=最便宜)的一条;协议已升序,但不信顺序,显式按 rank/price 取。
    # 源平台永远全菜, 故全目标缺菜时回落到源(is_source_best、saved=0), 不把虚低价当最低。
    priced = [r for r in results if r.price is not None]
    clean = [r for r in priced if not _is_short(r)]
    best = None
    if clean:
        best = min(
            clean,
            key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
        )

    source_price_cents = _yuan_to_cents(payload.source_price)
    if source_price_cents is None:
        # 源价没单独给,从 comparison_results 里的 is_source 行兜底
        src_row = next((r for r in results if r.is_source and r.price is not None), None)
        if src_row is not None:
            source_price_cents = _yuan_to_cents(src_row.price)

    best_price_cents = _yuan_to_cents(best.price) if best else None

    saved_amount_cents = None
    if source_price_cents is not None and best_price_cents is not None:
        saved_amount_cents = source_price_cents - best_price_cents

    is_source_best = best.is_source if best is not None else None

    # status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
    # → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
    status = payload.record_status or payload.status
    if status is None:
        has_valid_target = any(
            (not r.is_source) and r.price is not None for r in results
        )
        status = "success" if has_valid_target else "failed"

    return {
        "source_price_cents": source_price_cents,
        "best_platform_id": best.platform_id if best else None,
        "best_platform_name": best.platform_name if best else None,
        "best_price_cents": best_price_cents,
        "saved_amount_cents": saved_amount_cents,
        "is_source_best": is_source_best,
        "status": status,
        "fail_reason": (
            _derive_fail_display(payload.information, _pr)
            if status == "failed"
            else None
        ),
    }


def upsert_record(
    db: Session, *, user_id: int, payload: ComparisonRecordIn
) -> ComparisonRecord:
    """按 **trace_id** 幂等写入(唯一键已从 user_id+trace_id 改为 trace_id):已存在则合并
    (回填 null user_id + 覆盖字段,但**不降级 success**),否则新建。

    灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
    新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
    """
    # 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
    # _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
    if payload.platforms:
        derived = _derive_from_platforms(payload.platforms, payload.record_status)
        # 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
        derived["fail_reason"] = (
            _derive_fail_display(payload.information, payload.platform_results or {})
            if derived["status"] == "failed"
            else None
        )
        # 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。
        # 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived)
        # 与 _derive_from_platforms 同名键撞键 TypeError)。
        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
            if not derived.get(_k):
                derived[_k] = getattr(payload, _k)
    else:
        derived = _derive(payload)
        # _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补,
        # 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。
        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
            derived[_k] = getattr(payload, _k)
    items = [it.model_dump(exclude_none=True) for it in payload.items]
    fields = dict(
        device_id=payload.device_id,
        business_type=payload.business_type,
        product_names=_product_names_from_items(items),
        # store_name / source_platform_id / source_platform_name / source_package 统一由
        # derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
        information=payload.information,
        best_deeplink=payload.best_deeplink,
        trace_url=payload.trace_url,
        total_dish_count=payload.total_dish_count,
        skipped_dish_count=payload.skipped_dish_count,
        items=items,
        comparison_results=[r.model_dump() for r in payload.comparison_results],
        platforms=list(payload.platforms or []),
        skipped_dish_names=list(payload.skipped_dish_names),
        # 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
        device_model=payload.device_model,
        device_manufacturer=payload.device_manufacturer,
        rom_vendor=payload.rom_vendor,
        rom_name=payload.rom_name,
        rom_version=payload.rom_version,
        android_version=payload.android_version,
        android_sdk=payload.android_sdk,
        app_version=payload.app_version,
        app_version_code=payload.app_version_code,
        source_app_version=payload.source_app_version,
        longitude=payload.longitude,
        latitude=payload.latitude,
        total_ms=payload.total_ms,
        step_count=payload.step_count,
        raw_payload=payload.model_dump(),
        **derived,
    )

    # 按 trace_id 定位(唯一键已改 trace_id)：后端 harvest 可能已建行,这里的客户端上报
    # (灰度期老客户端 / 新客户端不再走这条)与之 reconcile。
    existing = db.execute(
        select(ComparisonRecord).where(
            ComparisonRecord.trace_id == payload.trace_id,
        )
    ).scalar_one_or_none()

    if existing is not None:
        # 不降级:harvest 或更早上报已落成 success,收尾期 fromFailure 的 cancelled/failed
        # 不许把它盖回去(老 comparisonReported bug 的服务端兜底)——只补 user_id / trace_url。
        if existing.status == "success" and fields.get("status") != "success":
            if existing.user_id is None and user_id is not None:
                existing.user_id = user_id
            if fields.get("trace_url"):
                existing.trace_url = fields["trace_url"]
            db.commit()
            db.refresh(existing)
            return existing
        # 只**填**空缺 user_id、不 reassign:trace_id 全局唯一,一条 trace 只属一个用户;
        # 绝不把已有归属的记录改判给另一个上报者(仅 harvest 建的 null-user 行在此绑上)。
        if existing.user_id is None:
            existing.user_id = user_id
        for k, v in fields.items():
            setattr(existing, k, v)
        db.commit()
        db.refresh(existing)
        return existing

    # created_at 显式存 naive 北京 wall-clock(同 savings_record):客户端「比价记录」页
    # formatRecordTime 原样切片 created_at 字符串、不转时区,而默认 server_default=func.now()
    # 在 SQLite 下返回 UTC → 直接慢 8h。覆盖分支不动 created_at(保留首次创建时间)。
    rec = ComparisonRecord(
        user_id=user_id,
        trace_id=payload.trace_id,
        created_at=datetime.now(CN_TZ).replace(tzinfo=None),
        **fields,
    )
    db.add(rec)
    db.commit()
    db.refresh(rec)
    return rec


# ============================================================
# 后端 harvest：app-server 从 pricebot 透传响应里直接落库(不靠客户端上报)。
# 帧0 建行(running) → 最终 done 更新(success/failed) → finalize 更新(aborted)。
# 全按 trace_id upsert；success 行永不被后到的 failed/aborted 降级。
# ============================================================


def _derive_from_results(
    results: list[dict], platform_results: dict | None = None,
    record_status: str | None = None,
) -> dict:
    """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。

    platform_results(done.params.platform_results): 逐平台 skipped_dish_count 在这里(行里没有)。
    传入则派生 best 时排除缺菜(漏菜)店 —— 少买了菜总价虚低, 不能当记录级"最低价"/算虚假省额;
    源平台永远全菜, 故全目标缺菜时 best 回落到源(is_source_best、不虚报省)。不传→纯 rank/price, 行为不变。"""
    _pr = platform_results or {}

    def _is_short(r: dict) -> bool:
        # platform_results 内层结构宽松(pricebot/客户端透传), 值非 dict 时按"不缺菜"处理, 不崩。
        pid = r.get("platform_id")
        info = _pr.get(pid) if pid else None
        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0

    priced = [r for r in results if r.get("price") is not None]
    clean = [r for r in priced if not _is_short(r)]  # 缺菜店排除出最优评选
    best = None
    if clean:
        best = min(
            clean,
            key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]),
        )
    src_row = next((r for r in results if r.get("is_source")), None)

    source_price_cents = None
    if src_row is not None and src_row.get("price") is not None:
        source_price_cents = _yuan_to_cents(src_row["price"])
    best_price_cents = _yuan_to_cents(best["price"]) if best else None
    saved_amount_cents = None
    if source_price_cents is not None and best_price_cents is not None:
        saved_amount_cents = source_price_cents - best_price_cents

    has_valid_target = any(
        (not r.get("is_source")) and r.get("price") is not None for r in results
    )
    return {
        "source_platform_id": (src_row or {}).get("platform_id"),
        "source_platform_name": (src_row or {}).get("platform_name"),
        "source_package": (src_row or {}).get("package"),
        "source_price_cents": source_price_cents,
        "best_platform_id": best.get("platform_id") if best else None,
        "best_platform_name": best.get("platform_name") if best else None,
        "best_price_cents": best_price_cents,
        "saved_amount_cents": saved_amount_cents,
        "is_source_best": best.get("is_source") if best else None,
        "store_name": (src_row or {}).get("store_name") or None,
        # 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
        # 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
        # 回退老的 success/failed 二态派生, 向后兼容。
        "status": record_status or ("success" if has_valid_target else "failed"),
    }


def _derive_from_platforms(
    platforms: list, record_status: str | None = None,
) -> dict:
    """从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。

    best_* 直接取 platforms 里 is_best 的那一行、source_* 取 role=source 行,与前端读的
    platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优,
    消除"标量列 vs platforms"双源不一致)。platforms 非空时优先走这里;老 pricebot 无
    platforms 时调用方回退 _derive_from_results(向后兼容)。"""
    rows = [p for p in (platforms or []) if isinstance(p, dict)]
    src = next((p for p in rows if p.get("role") == "source"), None)
    best = next((p for p in rows if p.get("is_best")), None)
    source_price_cents = _yuan_to_cents(src.get("price")) if src else None
    best_price_cents = _yuan_to_cents(best.get("price")) if best else None
    saved_amount_cents = None
    if source_price_cents is not None and best_price_cents is not None:
        saved_amount_cents = source_price_cents - best_price_cents
    has_valid_target = any(
        p.get("role") != "source" and p.get("price") is not None for p in rows
    )
    # store_name: 优先源行; recompare 场景源平台自己当目标、源行被目标覆盖(pricebot
    # _build_platform_rows 有意去重, platforms 无 role=source 行)→ 回退 best 行 → 首个有店名
    # 的行(显示现场实际比到的店), 免得记录页店名空掉兜底显示成"比价"。正常比价有源行不走回退。
    store_name = (
        (src or {}).get("store_name")
        or (best or {}).get("store_name")
        or next((p.get("store_name") for p in rows if p.get("store_name")), None)
    )
    return {
        "source_platform_id": (src or {}).get("platform_id"),
        "source_platform_name": (src or {}).get("platform_name"),
        "source_package": (src or {}).get("package"),
        "source_price_cents": source_price_cents,
        "best_platform_id": (best or {}).get("platform_id"),
        "best_platform_name": (best or {}).get("platform_name"),
        "best_price_cents": best_price_cents,
        "saved_amount_cents": saved_amount_cents,
        "is_source_best": (best.get("role") == "source") if best else None,
        "store_name": store_name or None,
        "status": record_status or ("success" if has_valid_target else "failed"),
    }


def _device_cols_from_info(device_info: dict | None) -> dict:
    """step 帧 device_info({locale,brand,model,android_version,rom_version})→ 表列。
    step 帧只带这几项;rom_name / android_sdk / app_version / 耗时步数 harvest 拿不到 → 留 None
    (灰度期老客户端 fromComparison 会补齐;要新客户端也全带需扩 collectDeviceInfo,后续)。"""
    d = device_info or {}
    rv = str(d.get("rom_version") or "")
    return {
        "device_model": d.get("model") or None,
        "device_manufacturer": d.get("brand") or None,
        "android_version": d.get("android_version") or None,
        "rom_version": int(rv) if rv.isdigit() else None,
    }


def _get_by_trace(db: Session, trace_id: str) -> ComparisonRecord | None:
    return db.execute(
        select(ComparisonRecord).where(ComparisonRecord.trace_id == trace_id)
    ).scalar_one_or_none()


def reserve_daily_start(
    db: Session,
    *,
    user_id: int,
    trace_id: str,
    business_type: str = "food",
    device_id: str | None = None,
    now: datetime | None = None,
) -> tuple[ComparisonRecord, int]:
    """Atomically reserve one of a user's 100 Beijing-day comparison starts.

    ``trace_id`` makes client retries idempotent. Locking the user row serializes
    concurrent starts for one account, so parallel requests cannot both consume
    the final available slot. The reservation is the existing ``running``
    comparison row; later result reporting updates that same row.
    """
    db.execute(select(User.id).where(User.id == user_id).with_for_update()).scalar_one()

    existing = _get_by_trace(db, trace_id)
    if existing is not None:
        if existing.user_id not in (None, user_id):
            raise ComparisonTraceOwnershipError
        if existing.user_id is None:
            existing.user_id = user_id
            if existing.device_id is None and device_id:
                existing.device_id = device_id
            db.commit()
            db.refresh(existing)

        existing_at = existing.created_at
        if existing_at.tzinfo is not None:
            existing_at = existing_at.astimezone(CN_TZ).replace(tzinfo=None)
        day_start = existing_at.replace(hour=0, minute=0, second=0, microsecond=0)
        day_end = day_start + timedelta(days=1)
        used = db.scalar(
            select(func.count(ComparisonRecord.id)).where(
                ComparisonRecord.user_id == user_id,
                ComparisonRecord.created_at >= day_start,
                ComparisonRecord.created_at < day_end,
            )
        ) or 0
        return existing, int(used)

    current = now or datetime.now(CN_TZ)
    if current.tzinfo is not None:
        current = current.astimezone(CN_TZ).replace(tzinfo=None)
    day_start = current.replace(hour=0, minute=0, second=0, microsecond=0)
    day_end = day_start + timedelta(days=1)
    used = db.scalar(
        select(func.count(ComparisonRecord.id)).where(
            ComparisonRecord.user_id == user_id,
            ComparisonRecord.created_at >= day_start,
            ComparisonRecord.created_at < day_end,
        )
    ) or 0
    if used >= DAILY_COMPARE_START_LIMIT:
        raise DailyCompareStartLimitExceeded

    rec = ComparisonRecord(
        trace_id=trace_id,
        user_id=user_id,
        business_type=business_type or "food",
        device_id=device_id,
        status="running",
        created_at=current,
    )
    db.add(rec)
    db.commit()
    db.refresh(rec)
    return rec, int(used) + 1


def harvest_running(
    db: Session,
    *,
    trace_id: str,
    user_id: int | None,
    business_type: str = "food",
    device_id: str | None = None,
    device_info: dict | None = None,
    trace_url: str | None = None,
) -> ComparisonRecord:
    """帧0(或任一尚未建行的帧)建/补 running 行。幂等:已存在只补空缺(user_id/trace_url/
    device_id/机型),绝不动已落定的 status / 结果。"""
    rec = _get_by_trace(db, trace_id)
    if rec is None:
        rec = ComparisonRecord(
            trace_id=trace_id,
            user_id=user_id,
            business_type=business_type or "food",
            device_id=device_id,
            status="running",
            trace_url=trace_url,
            created_at=datetime.now(CN_TZ).replace(tzinfo=None),
            **_device_cols_from_info(device_info),
        )
        db.add(rec)
        db.commit()
        db.refresh(rec)
        return rec
    changed = False
    if rec.user_id is None and user_id is not None:
        rec.user_id = user_id
        changed = True
    if not rec.trace_url and trace_url:
        rec.trace_url = trace_url
        changed = True
    if rec.device_id is None and device_id:
        rec.device_id = device_id
        changed = True
    for k, v in _device_cols_from_info(device_info).items():
        if getattr(rec, k) is None and v is not None:
            setattr(rec, k, v)
            changed = True
    if changed:
        db.commit()
        db.refresh(rec)
    return rec


def harvest_done(
    db: Session,
    *,
    trace_id: str,
    user_id: int | None,
    done_params: dict,
    business_type: str = "food",
    device_id: str | None = None,
    device_info: dict | None = None,
    trace_url: str | None = None,
) -> tuple[ComparisonRecord, bool]:
    """最终 done 帧:running 行 → 终态(success/failed)+结果+trace_url。
    返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
    行不存在(理论上帧0已建;防御)则新建。"""
    results = done_params.get("comparison_results") or []
    # 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
    # record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
    platforms = done_params.get("platforms") or []
    record_status = done_params.get("record_status")
    # 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
    # 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无
    # platforms 时回退从 comparison_results 派生(向后兼容)。
    if platforms:
        derived = _derive_from_platforms(platforms, record_status)
        # 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退)
        _item_row = (
            next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None)
            or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None)
            or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None)
        )
        items = (_item_row or {}).get("items") or []
    else:
        derived = _derive_from_results(
            results, done_params.get("platform_results"), record_status
        )
        # pricebot 已把源单菜品塞进 comparison_results[源行].items
        items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
    # 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算
    fail_reason = (
        _derive_fail_display(
            done_params.get("information"), done_params.get("platform_results")
        )
        if derived["status"] == "failed"
        else None
    )
    fields = dict(
        business_type=business_type or "food",
        information=done_params.get("information") or None,
        fail_reason=fail_reason,
        # best_deeplink 来自客户端剪贴板采集,harvest 拿不到 → 留空(灰度期 fromComparison 会补;
        # 纯 harvest 行「再次比价」退化为按 package 拉起 App。要精确深链需客户端另传,后续)。
        trace_url=trace_url or done_params.get("trace_url"),
        total_dish_count=done_params.get("total_dish_count"),
        skipped_dish_count=done_params.get("skipped_dish_count"),
        skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
        comparison_results=results,
        platforms=platforms,
        items=items,
        product_names=_product_names_from_items(items),
        raw_payload=done_params,
        **derived,
        **_device_cols_from_info(device_info),
    )

    rec = _get_by_trace(db, trace_id)
    was_success = rec is not None and rec.status == "success"
    if rec is None:
        rec = ComparisonRecord(
            trace_id=trace_id,
            user_id=user_id,
            created_at=datetime.now(CN_TZ).replace(tzinfo=None),
            device_id=device_id,
            **fields,
        )
        db.add(rec)
    else:
        if user_id is not None and rec.user_id is None:
            rec.user_id = user_id
        if device_id and rec.device_id is None:
            rec.device_id = device_id
        for k, v in fields.items():
            setattr(rec, k, v)
    db.commit()
    db.refresh(rec)
    newly_success = (rec.status == "success") and not was_success
    return rec, newly_success


def harvest_abort(
    db: Session,
    *,
    trace_id: str,
    status: str,
    reason: str | None,
    trace_url: str | None = None,
) -> ComparisonRecord | None:
    """finalize(用户终止/超时/异常,无 done 帧):running 行 → aborted/failed + trace_url。
    **不降级 success**:行已 success(收尾取消那种 finalize 后到)只 refresh trace_url。
    行不存在(极少:帧0没建成)→ 返回 None,不凭空造。"""
    rec = _get_by_trace(db, trace_id)
    if rec is None:
        return None
    if trace_url and not rec.trace_url:
        rec.trace_url = trace_url
    if rec.status != "success":
        rec.status = status or "cancelled"
        if reason:
            rec.information = reason
    db.commit()
    db.refresh(rec)
    return rec


def _ordered_shop_name_select(user_id: int):
    """该用户「真实下单」(source='compare')覆盖到的店名 select,给「已下单」筛选当子查询。

    口径与 [_ordered_shop_names] 完全一致,只是时机不同:那边是**拿到本页之后**按 candidates
    反查打标;这边是**分页之前**就要过滤,拿不到 candidates,只能整段下推成子查询。
    没有先捞成集合再展开 IN (...) 字面量 —— 重度用户下单过的店名可能上千,展开会撞 SQLite
    的绑定变量上限,而且又变回了那个「随下单量线性变慢」的老写法。
    """
    return select(SavingsRecord.shop_name).where(
        SavingsRecord.user_id == user_id,
        SavingsRecord.source == "compare",
        SavingsRecord.shop_name.is_not(None),
    )


def _like_escape(kw: str) -> str:
    """转义 LIKE 通配符(百分号 / 下划线 / 反斜杠),让用户输入只按字面量匹配(配合 escape 参数)。

    不转义的话搜一个「%」就等于把整表拉回来。
    """
    return kw.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def _ordered_shop_names(db: Session, user_id: int, candidates: set[str]) -> set[str]:
    """[candidates] 里哪些店名被该用户「真实下单」(source='compare')覆盖过,用来打「已下单」。

    只认 compare(归因命中后真实上报),demo 演示数据不算。下单上报不带 trace_id,
    只能按店名对齐——两边店名同源(都来自比价意图识别阶段的门店名 query),精确相等即视为同店。
    语义=店级:同一家店比价过多次,这些记录会一并标「已下单」。

    ⚠️ 只查**本页出现过的店名**(candidates ≤ limit 条),不再把该用户全部下单店名捞回内存:
    老写法随下单量线性增长,重度用户几千行全读一遍只为跟 50 条记录取交集。空集合直接返回
    (避免 IN () 非法)。
    """
    if not candidates:
        return set()
    rows = db.execute(
        select(SavingsRecord.shop_name).where(
            SavingsRecord.user_id == user_id,
            SavingsRecord.source == "compare",
            SavingsRecord.shop_name.in_(candidates),
        ).distinct()
    ).scalars().all()
    return {s for s in rows if s}


def _ad_coins_by_trace(db: Session, user_id: int, trace_ids: list[str]) -> dict[str, int]:
    """各 trace_id「看广告赚的金币」:按 trace_id 聚合该用户已发放(granted)的信息流广告金币。

    广告金币来自比价等候期信息流(biz_type=feed_ad_reward);客户端结算时把本场 trace_id 带上,
    落到 ad_feed_reward_record.trace_id。此处按 trace_id 求和 = 这次比价看广告实发的金币。
    trace_id 仅比价场景客户端带(领券/福利/旧客户端为 NULL),故按 trace_id 过滤天然只算比价广告。
    空集合直接返回(避免 IN () 非法)。
    """
    if not trace_ids:
        return {}
    rows = db.execute(
        select(
            AdFeedRewardRecord.trace_id,
            func.coalesce(func.sum(AdFeedRewardRecord.coin), 0),
        ).where(
            AdFeedRewardRecord.user_id == user_id,
            AdFeedRewardRecord.status == "granted",
            AdFeedRewardRecord.trace_id.in_(trace_ids),
        ).group_by(AdFeedRewardRecord.trace_id)
    ).all()
    return {tid: int(coin) for tid, coin in rows if tid}


# 列表出参(ComparisonRecordOut)根本不读、但 select(ORM) 默认会一并捞回来的重型 JSON 列:
#   - raw_payload:done.params 上报体全量,**每条记录都有**(harvest 与 POST 两条写路径都落)。
#     单条几 KB~几十 KB,一页 50 条就是稳定几百 KB~几 MB 的白读 + 白反序列化。
#   - llm_calls:每次 LLM 调用的 input_messages + output 全文。只有走老客户端 POST /compare/record
#     的记录才有(_backfill_llm_calls 回填;harvest 路径不落),但有的时候单条就能到 MB 级 —— 一页里
#     混进几条这种记录,整个请求就被它们拖住。
#   - llm_price_snapshot:逐模型单价快照,同样只在回填时落。
# 三列全部读出来再被 pydantic 丢掉,是「比价记录/全部记录」页慢的主要来源。
# ⚠️ defer 的列一旦在别处被读到会触发**逐行**懒加载(N+1);列表这条链路(ComparisonRecordOut
# 不声明这三个字段 → 不会 getattr 到)是安全的。详情接口 get_record 不 defer,raw_payload 照常返回。
_LIST_DEFERRED = (
    ComparisonRecord.raw_payload,
    ComparisonRecord.llm_calls,
    ComparisonRecord.llm_price_snapshot,
)


def list_records(
    db: Session,
    user_id: int,
    *,
    limit: int = 20,
    cursor: int | None = None,
    ordered: bool | None = None,
    keyword: str | None = None,
) -> tuple[list[ComparisonRecord], int | None]:
    """比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记 + 「看广告赚的金币」(瞬态,不写库)。"""
    stmt = (
        select(ComparisonRecord)
        .where(ComparisonRecord.user_id == user_id)
        .options(*(defer(col) for col in _LIST_DEFERRED))
    )
    if cursor is not None:
        stmt = stmt.where(ComparisonRecord.id < cursor)
    # 「已下单」tab 与搜索框的过滤都下推到这里,不能留给客户端对整页结果 filter ——
    # 分页之后一页里可能一条都不命中,列表看着就是空的/卡住的,得翻很多页才蹦出一条。
    if ordered:
        stmt = stmt.where(
            ComparisonRecord.store_name.in_(_ordered_shop_name_select(user_id))
        )
    kw = (keyword or "").strip()
    if kw:
        # product_names 是写路径从 items[].name 派生的普通文本列(items 本身是 JSON,SQLite 下
        # 中文被 ensure_ascii 转义,没法直接 LIKE)—— 搜「菜名」靠的就是它。
        # ilike:PG 原生 ILIKE,SQLite 渲染成 lower() LIKE lower(),两边都忽略大小写。
        pattern = f"%{_like_escape(kw)}%"
        stmt = stmt.where(
            or_(
                ComparisonRecord.store_name.ilike(pattern, escape="\\"),
                ComparisonRecord.product_names.ilike(pattern, escape="\\"),
            )
        )
    # 排序与 ix_comparison_user_created(user_id, created_at, id)对齐 —— DESC/DESC 正好是该索引的
    # 反向扫,PG 免排序直接取前 limit 条。改排序方向前先想清楚索引还吃不吃得上。
    stmt = stmt.order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc()).limit(limit)

    items = list(db.execute(stmt).scalars().all())
    next_cursor = items[-1].id if len(items) == limit else None

    # 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。
    # ordered / ad_coins_earned 均非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。
    page_shops = {it.store_name for it in items if it.store_name}
    # ordered=True 时上面已按同一口径(_ordered_shop_name_select)筛过,本页必然全是已下单,
    # 省掉这次反查;其余情况照旧按本页店名反查 savings。
    ordered_shops = page_shops if ordered else _ordered_shop_names(db, user_id, page_shops)
    # 「本次比价看广告赚的金币」:按本页 trace_id 一次性聚合(同 ordered 范式)。
    ad_coins = _ad_coins_by_trace(db, user_id, [it.trace_id for it in items])
    for it in items:
        it.ordered = bool(it.store_name and it.store_name in ordered_shops)
        it.ad_coins_earned = ad_coins.get(it.trace_id, 0)

    return items, next_cursor


def count_success(db: Session, user_id: int) -> int:
    """该用户成功比价(status='success')的条数。比价战绩里程碑的解锁进度源。"""
    return db.execute(
        select(func.count(ComparisonRecord.id)).where(
            ComparisonRecord.user_id == user_id,
            ComparisonRecord.status == "success",
        )
    ).scalar_one()


def get_stats(db: Session, user_id: int) -> tuple[int, int]:
    """「我的」页省钱战绩卡的**比价口径**聚合:(成功比价数, 累计发现可省额/分)。

    - 完成比价 = status='success' 的比价记录数(同 count_success)。
    - 累计发现可省 = 各成功比价 saved_amount_cents(源价−最低价)之和;null(没省/未识别)不计。
    与「下单成交记账」savings_record 解耦:比过价就计,不要求下单(见 savings.py)。
    """
    row = db.execute(
        select(
            func.count(ComparisonRecord.id),
            func.coalesce(func.sum(ComparisonRecord.saved_amount_cents), 0),
        ).where(
            ComparisonRecord.user_id == user_id,
            ComparisonRecord.status == "success",
        )
    ).one()
    return int(row[0]), int(row[1])


def get_record(db: Session, user_id: int, record_id: int) -> ComparisonRecord | None:
    """取单条(限本人,避免越权读他人记录)。附「看广告赚的金币」瞬态属性(同 list_records)。"""
    rec = db.execute(
        select(ComparisonRecord).where(
            ComparisonRecord.id == record_id,
            ComparisonRecord.user_id == user_id,
        )
    ).scalar_one_or_none()
    if rec is not None:
        rec.ad_coins_earned = _ad_coins_by_trace(db, user_id, [rec.trace_id]).get(rec.trace_id, 0)
    return rec
