"""埋点健康度聚合(埋点成功率 / 上报成功率)。

只存原始累计快照,查询时在 Python 侧差分聚合(admin 低频、量级小,跨 PG/SQLite 无方言坑;
与 cps.py / coupon_data.py 同款约定)。差分按 (device_id, epoch_id, event) 分区、created_at
升序,相邻做差、负值夹 0;每增量按其快照 created_at 归入北京天桶。
"""
from __future__ import annotations

from collections import defaultdict
from datetime import UTC, datetime

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

from app.core import rewards
from app.models.analytics_selfstat import AnalyticsSelfStat as H
from app.models.analytics_selfstat import AnalyticsSelfStatEvent as E

_COUNTS = ("attempted", "drop_capture", "delivered", "drop_undelivered")


def diff_snapshots(rows: list[dict]) -> list[dict]:
    """累计快照行 → 每快照增量行(纯逻辑)。

    rows 每行含 device_id/epoch_id/event/created_at/app_ver/oem/os + 四个累计计数。
    返回每行含 dims + created_at + 四个增量 d_*(分区首行增量=累计值;负值夹 0)。
    """
    parts: dict[tuple, list[dict]] = defaultdict(list)
    for r in rows:
        parts[(r["device_id"], r["epoch_id"], r["event"])].append(r)

    out: list[dict] = []
    for group in parts.values():
        group.sort(key=lambda r: (r["created_at"], r.get("id", 0)))
        prev = {k: 0 for k in _COUNTS}
        for r in group:
            deltas = {f"d_{k}": max(0, int(r[k]) - prev[k]) for k in _COUNTS}
            out.append({
                "device_id": r["device_id"], "epoch_id": r["epoch_id"], "event": r["event"],
                "created_at": r["created_at"], "app_ver": r["app_ver"],
                "oem": r["oem"], "os": r["os"], **deltas,
            })
            prev = {k: int(r[k]) for k in _COUNTS}
    return out


def _cn_day(dt: datetime) -> str:
    """created_at(UTC 口径)→ 北京日期字符串 YYYY-MM-DD。naive 当 UTC,tz-aware 直接换算。"""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=UTC)
    return dt.astimezone(rewards.CN_TZ).date().isoformat()


def _rates(sums: dict) -> dict:
    """由四个增量和派生两段率(分母 0 → None)。"""
    persisted_denom = sums["attempted"]
    report_denom = sums["delivered"] + sums["drop_undelivered"]
    return {
        **sums,
        "track_success_rate": (
            (sums["attempted"] - sums["drop_capture"]) / persisted_denom
            if persisted_denom else None
        ),
        "report_success_rate": (
            sums["delivered"] / report_denom if report_denom else None
        ),
    }


def _sum_deltas(deltas: list[dict]) -> dict:
    return {k: sum(d[f"d_{k}"] for d in deltas) for k in _COUNTS}


def _fetch_rows(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
    """取 [from, to) 区间行 + 每分区在 from 左侧的最后一条基线行(供第一条区间增量做差)。"""
    cols = (
        H.id, H.device_id, H.epoch_id, E.event, H.created_at,
        H.app_ver, H.oem, H.os,
        E.attempted, E.drop_capture, E.delivered, E.drop_undelivered,
    )
    in_range = db.execute(
        select(*cols).join(E, E.snapshot_id == H.id)
        .where(H.created_at >= date_from, H.created_at < date_to)
    ).mappings().all()

    # 注:基线子查询无下界扫 from 左侧全量(spec §7 已接受的取舍;量级变大再上物化 rollup)。
    # 用 max(id) 而非 max(created_at) 选"最新一条":id 严格单调,避免 SQLite 秒级时间戳撞车时选歧义。
    sub = (
        select(H.device_id, H.epoch_id, E.event, func.max(H.id).label("max_id"))
        .join(E, E.snapshot_id == H.id)
        .where(H.created_at < date_from)
        .group_by(H.device_id, H.epoch_id, E.event)
        .subquery()
    )
    baseline = db.execute(
        select(*cols).join(E, E.snapshot_id == H.id).join(
            sub, sub.c.max_id == H.id
        )
    ).mappings().all()

    return [dict(r) for r in list(baseline) + list(in_range)]


def _in_range_deltas(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
    """差分后只保留 created_at ∈ [from, to) 的增量(基线行被差分用后丢弃)。

    Python 侧过滤需对齐 tz 口径:SQLite 返回 naive UTC,PG 返回 aware UTC。
    统一转成 naive UTC 再比较,兼容两种后端。
    """
    def _to_naive_utc(dt: datetime) -> datetime:
        if dt.tzinfo is not None:
            return dt.astimezone(UTC).replace(tzinfo=None)
        return dt

    from_naive = _to_naive_utc(date_from)
    to_naive = _to_naive_utc(date_to)
    deltas = diff_snapshots(_fetch_rows(db, date_from, date_to))
    return [d for d in deltas if from_naive <= _to_naive_utc(d["created_at"]) < to_naive]


def overview(db: Session, date_from: datetime, date_to: datetime) -> dict:
    deltas = _in_range_deltas(db, date_from, date_to)
    return _rates(_sum_deltas(deltas))


def trend(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
    deltas = _in_range_deltas(db, date_from, date_to)
    by_day: dict[str, list[dict]] = defaultdict(list)
    for d in deltas:
        by_day[_cn_day(d["created_at"])].append(d)
    return [
        {"day": day, **_rates(_sum_deltas(items))}
        for day, items in sorted(by_day.items())
    ]


def breakdown(db: Session, date_from: datetime, date_to: datetime, dim: str) -> list[dict]:
    if dim not in ("event", "app_ver", "oem"):
        raise ValueError(f"invalid dim: {dim!r}")
    deltas = _in_range_deltas(db, date_from, date_to)
    by_key: dict[str, list[dict]] = defaultdict(list)
    for d in deltas:
        by_key[d.get(dim) or "(unknown)"].append(d)
    rows = [{"key": key, **_rates(_sum_deltas(items))} for key, items in by_key.items()]
    rows.sort(key=lambda r: (r["report_success_rate"] is None, r["report_success_rate"] or 0.0))
    return rows
