"""管理域期聚合查询(批次B)——评分/异常/employees API 的单一数据源。

窗口:days=[start_day..end_day] 闭区间 yyyymmdd;上期=等长紧邻前窗;
个人基线=紧邻当前窗之前的 28 天。所有查询走 (day, ...) 索引,不扫消息表
(除 distinct 活跃客户走 msg_daily_stats friend 行)。

口径要点:
- 分位/回复计数:first_reply_fact 窗内行拉回 Python 按员工合并排序取
  最近邻秩 ceil(q·n)(整除算术 (n+1)//2 / (9n+9)//10——SQLite 无原生
  ceil,MySQL `/` 返回浮点,Python 侧取秩零方言风险)。合并按员工而非
  owner:一员工多账号(wx1+wx2→u1,channel=None 时人人如此)的样本必须
  跨账号归并,逐 owner 取分位再映射会互相覆盖。fact 全库全期 ≤6.8万行,
  Python 侧聚合可行。
- 期末存量(new/cooling/dormant):状态快照取「每 (user, channel) 最新
  ≤end 行」——channel=None 时两渠道各取各的最新行再相加(直接取全渠道
  最大 day 的那一行会丢另一渠道的存量)。
- active_rooms_avg = SUM(active_rooms)/全窗天数(含无数据日摊薄),
  非有数据天数——「日均负载」语义。
- days_with_data = COUNT(DISTINCT day):单渠道时与 COUNT(*) 等价
  (PK 含 day+channel,每日至多一行),channel=None 双渠道同日也只计 1 天。
- 通话四列恒全渠道(employee_call_daily 无渠道维),channel 参数不过滤。
"""
from __future__ import annotations

import datetime as dt
import statistics
from collections import namedtuple
from collections.abc import Iterable

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import (
    EmployeeCallDaily,
    EmployeeMetricDaily,
    FirstReplyFact,
    MsgDailyStats,
    Staff,
)
from app.storage.employee_metrics import METRIC_VERSION
from app.storage.staff_summary import account_map

PeriodMetrics = namedtuple("PeriodMetrics", [
    "user_id", "days_with_data", "msg_sent", "msg_recv", "cust_threads",
    "threads_answered", "fr_sum_ms", "fr_p50_ms", "fr_p90_ms",
    "reply_2h_count", "reply_24h_count",
    "active_friends",          # 期内 distinct(直查,非日值求和)
    "active_rooms_avg",        # 日均(全窗天数摊薄)
    "new_customer", "cooling_customer", "dormant_customer",  # 期末单日存量
    "reactivated_customer",    # 期内求和
    "call_count", "call_seconds", "answered_call", "call_followup",
])


def _to_date(k: int) -> dt.date:
    return dt.date(k // 10000, k // 100 % 100, k % 100)


def _to_key(d: dt.date) -> int:
    return d.year * 10000 + d.month * 100 + d.day


def day_range(start: int, end: int) -> list[int]:
    """闭区间日键列表。"""
    out, cur = [], _to_date(start)
    while cur <= _to_date(end):
        out.append(_to_key(cur))
        cur += dt.timedelta(days=1)
    return out


def prev_window(start: int, end: int) -> tuple[int, int]:
    """等长紧邻前窗。"""
    n = len(day_range(start, end))
    pe = _to_date(start) - dt.timedelta(days=1)
    ps = pe - dt.timedelta(days=n - 1)
    return _to_key(ps), _to_key(pe)


def baseline_window(start: int) -> tuple[int, int]:
    """个人基线:紧邻当前窗之前的 28 天。"""
    be = _to_date(start) - dt.timedelta(days=1)
    bs = be - dt.timedelta(days=27)
    return _to_key(bs), _to_key(be)


def _percentile(vals: list[int], q_num: int, q_den: int) -> int | None:
    """最近邻秩分位 ceil(q·n) 的样本值,q=q_num/q_den(1/2、9/10)。

    整数算术 rank=(q_num*n + q_den - 1)//q_den 对正整数 n 恒落在 [1,n]
    (与 math.ceil(q·n) 等价,与批次A _percentile 同口径);空列表→None
    (与 0 严格区分)。
    """
    if not vals:
        return None
    s = sorted(vals)
    n = len(s)
    rank = max(1, min(n, (q_num * n + q_den - 1) // q_den))
    return s[rank - 1]


# 累加器初值:全员(staff)都出结果行,无数据员工为零值而非缺行——
# employees API 全员列表语义。
_ZERO: dict[str, object] = {
    "days_with_data": 0, "msg_sent": 0, "msg_recv": 0, "cust_threads": 0,
    "threads_answered": 0, "fr_sum_ms": 0,
    "fr_p50_ms": None, "fr_p90_ms": None,
    "reply_2h_count": 0, "reply_24h_count": 0,
    "active_friends": 0, "active_rooms_sum": 0,
    "new_customer": 0, "cooling_customer": 0, "dormant_customer": 0,
    "reactivated_customer": 0,
    "call_count": 0, "call_seconds": 0, "answered_call": 0, "call_followup": 0,
}


async def period_metrics(
    session: AsyncSession,
    start_day: int,
    end_day: int,
    channel: str | None,
    user_ids: Iterable[str] | None = None,
) -> dict[str, PeriodMetrics]:
    """窗口 [start_day, end_day] × channel(None=全渠道)的员工期指标。

    user_ids 传入时与 staff 表求交(幽灵过滤);返回含全部目标员工的
    PeriodMetrics(无数据员工为零值行)。channel=None 时两渠道
    account_map 合并(wx_id 与 WA uid 命名空间不撞,无冲突)。
    """
    staff_ids = {
        u for u in (await session.execute(sa.select(Staff.user_id))).scalars() if u
    }
    target = staff_ids if user_ids is None else staff_ids & set(user_ids)
    if not target:
        return {}

    if channel is None:
        amap: dict[str, str] = {}
        for ch in ("wechat", "whatsapp"):
            amap.update(await account_map(session, ch))
    else:
        amap = await account_map(session, channel)

    n_days = len(day_range(start_day, end_day))
    acc: dict[str, dict] = {uid: dict(_ZERO) for uid in target}

    # ---- 宽表 SUM:消息计数/串类/日均负载/复活 ----
    wide_conds = [
        EmployeeMetricDaily.day.between(start_day, end_day),
        EmployeeMetricDaily.metric_version == METRIC_VERSION,
        EmployeeMetricDaily.user_id.in_(target),
    ]
    if channel is not None:
        wide_conds.append(EmployeeMetricDaily.channel == channel)
    wide = (await session.execute(
        sa.select(
            EmployeeMetricDaily.user_id.label("uid"),
            sa.func.count(sa.distinct(EmployeeMetricDaily.day)).label("nd"),
            sa.func.sum(EmployeeMetricDaily.msg_sent).label("sent"),
            sa.func.sum(EmployeeMetricDaily.msg_recv).label("recv"),
            sa.func.sum(EmployeeMetricDaily.cust_threads).label("thr"),
            sa.func.sum(EmployeeMetricDaily.threads_answered).label("ans"),
            sa.func.sum(EmployeeMetricDaily.fr_sum_ms).label("frs"),
            sa.func.sum(EmployeeMetricDaily.active_rooms).label("rooms"),
            sa.func.sum(
                EmployeeMetricDaily.reactivated_customer_count
            ).label("react"),
        ).where(*wide_conds).group_by(EmployeeMetricDaily.user_id)
    )).all()
    for uid, nd, sent, recv, thr, ans, frs, rooms, react in wide:
        a = acc[uid]
        a["days_with_data"] = int(nd or 0)
        a["msg_sent"] = int(sent or 0)
        a["msg_recv"] = int(recv or 0)
        a["cust_threads"] = int(thr or 0)
        a["threads_answered"] = int(ans or 0)
        a["fr_sum_ms"] = int(frs or 0)
        a["active_rooms_sum"] = int(rooms or 0)
        a["reactivated_customer"] = int(react or 0)

    # ---- 期末存量:每 (user, channel) 最新 ≤end 行,跨渠道相加 ----
    stock_conds = [
        EmployeeMetricDaily.day <= end_day,
        EmployeeMetricDaily.metric_version == METRIC_VERSION,
        EmployeeMetricDaily.user_id.in_(target),
    ]
    if channel is not None:
        stock_conds.append(EmployeeMetricDaily.channel == channel)
    latest = (await session.execute(
        sa.select(
            EmployeeMetricDaily.user_id.label("uid"),
            EmployeeMetricDaily.channel.label("ch"),
            sa.func.max(EmployeeMetricDaily.day).label("d"),
        ).where(*stock_conds).group_by(
            EmployeeMetricDaily.user_id, EmployeeMetricDaily.channel
        )
    )).all()
    if latest:
        # (user, channel, day) 含 metric_version 唯一(主键)——OR 点查
        # 精确取回各渠道最新行,再按 user 跨渠道 SUM
        stocks = (await session.execute(
            sa.select(
                EmployeeMetricDaily.user_id.label("uid"),
                sa.func.sum(EmployeeMetricDaily.new_customer_count).label("nw"),
                sa.func.sum(
                    EmployeeMetricDaily.cooling_customer_count
                ).label("cool"),
                sa.func.sum(
                    EmployeeMetricDaily.dormant_customer_count
                ).label("dorm"),
            ).where(sa.or_(*(
                sa.and_(
                    EmployeeMetricDaily.user_id == uid,
                    EmployeeMetricDaily.channel == ch,
                    EmployeeMetricDaily.day == d,
                    EmployeeMetricDaily.metric_version == METRIC_VERSION,
                )
                for uid, ch, d in latest
            ))).group_by(EmployeeMetricDaily.user_id)
        )).all()
        for uid, nw, cool, dorm in stocks:
            a = acc[uid]
            a["new_customer"] = int(nw or 0)
            a["cooling_customer"] = int(cool or 0)
            a["dormant_customer"] = int(dorm or 0)

    # ---- fact 分位 + 2h/24h 回复计数(同一过滤,Python 侧一次聚合) ----
    fr_conds = [FirstReplyFact.day.between(start_day, end_day)]
    if channel is not None:
        fr_conds.append(FirstReplyFact.channel == channel)
    fr_rows = (await session.execute(
        sa.select(
            FirstReplyFact.owner_account_id.label("owner"),
            FirstReplyFact.fr_ms.label("fr"),
        ).where(*fr_conds)
    )).all()
    fr_by_user: dict[str, list[int]] = {}
    for owner, fr in fr_rows:
        uid = amap.get(owner or "")
        if uid is not None and uid in target:
            fr_by_user.setdefault(uid, []).append(int(fr))
    for uid, frs in fr_by_user.items():
        acc[uid]["fr_p50_ms"] = _percentile(frs, 1, 2)
        acc[uid]["fr_p90_ms"] = _percentile(frs, 9, 10)
        # 2h/24h 计数与分位同源同窗——回复率分子分母口径一致
        acc[uid]["reply_2h_count"] = sum(1 for fr in frs if fr <= 7_200_000)
        acc[uid]["reply_24h_count"] = sum(1 for fr in frs if fr <= 86_400_000)

    # ---- distinct 活跃客户:账号归并到员工后按 peer 去重 ----
    af_conds = [
        MsgDailyStats.day.between(start_day, end_day),
        MsgDailyStats.peer_kind == "friend",
    ]
    if channel is not None:
        af_conds.append(MsgDailyStats.channel == channel)
    pairs = (await session.execute(
        sa.select(MsgDailyStats.owner_account_id, MsgDailyStats.peer_id)
        .distinct().where(*af_conds)
    )).all()
    active: dict[str, set[str]] = {}
    for owner, peer in pairs:
        uid = amap.get(owner or "")
        if uid is None or uid not in target:
            continue
        # (uid, peer) 粒度:同一员工多账号(wx1+wx2)的共享客户只计 1
        # ——与宽表日值口径一致(storage 侧账号归并后 add(peer))
        active.setdefault(uid, set()).add(peer or "")
    for uid, pset in active.items():
        acc[uid]["active_friends"] = len(pset)

    # ---- 通话:恒全渠道(表无渠道维,channel 参数不过滤) ----
    calls = (await session.execute(
        sa.select(
            EmployeeCallDaily.user_id.label("uid"),
            sa.func.sum(EmployeeCallDaily.call_count).label("cc"),
            sa.func.sum(EmployeeCallDaily.call_seconds).label("cs"),
            sa.func.sum(EmployeeCallDaily.answered_call_count).label("ca"),
            sa.func.sum(EmployeeCallDaily.call_followup_count).label("cf"),
        ).where(
            EmployeeCallDaily.day.between(start_day, end_day),
            EmployeeCallDaily.user_id.in_(target),
        ).group_by(EmployeeCallDaily.user_id)
    )).all()
    for uid, cc, cs, ca, cf in calls:
        a = acc[uid]
        a["call_count"] = int(cc or 0)
        a["call_seconds"] = int(cs or 0)
        a["answered_call"] = int(ca or 0)
        a["call_followup"] = int(cf or 0)

    return {
        uid: PeriodMetrics(
            user_id=uid,
            days_with_data=a["days_with_data"],
            msg_sent=a["msg_sent"],
            msg_recv=a["msg_recv"],
            cust_threads=a["cust_threads"],
            threads_answered=a["threads_answered"],
            fr_sum_ms=a["fr_sum_ms"],
            fr_p50_ms=a["fr_p50_ms"],
            fr_p90_ms=a["fr_p90_ms"],
            reply_2h_count=a["reply_2h_count"],
            reply_24h_count=a["reply_24h_count"],
            active_friends=a["active_friends"],
            active_rooms_avg=(a["active_rooms_sum"] / n_days) if n_days else 0.0,
            new_customer=a["new_customer"],
            cooling_customer=a["cooling_customer"],
            dormant_customer=a["dormant_customer"],
            reactivated_customer=a["reactivated_customer"],
            call_count=a["call_count"],
            call_seconds=a["call_seconds"],
            answered_call=a["answered_call"],
            call_followup=a["call_followup"],
        )
        for uid, a in acc.items()
    }


def team_medians(metrics: dict[str, PeriodMetrics]) -> dict[str, float | None]:
    """每数值字段的全员中位数(纯函数,评分归一用)。

    None(无样本的分位等)剔除后取中位——statistics.median:奇数取中位
    样本、偶数取中间两数均值;全员 None → None(调用方按无基线处理)。
    """
    out: dict[str, float | None] = {}
    for field in PeriodMetrics._fields:
        if field == "user_id":
            continue
        vals = [
            v for v in (getattr(m, field) for m in metrics.values())
            if v is not None
        ]
        out[field] = statistics.median(vals) if vals else None
    return out