"""employee_call_daily / employee_metric_daily 维护层(管理域,批次A)。

数据流(v2 §4):
  msg_daily_stats(friend) + first_reply_facts 同源扫描 + customer 生命周期
    → employee_metric_daily (day, channel, user, metric_version)   [Task 6]
  phone_call(按 user_id+start_call_time 日聚合)
    → employee_call_daily (day, user)——通话无渠道维(v2 §1.2 #9)     [本任务]

幽灵员工过滤(v2 §1.3 #11):owner 映射不上或不在 staff 表的不落表。
重算语义:按 day delete+insert,幂等;夜间全量 + (批次F)4h 滚动。
"""
from __future__ import annotations

import logging
import math
from bisect import bisect_right
from collections.abc import Iterable
from itertools import accumulate

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

from app.analytics.lifecycle import classify
from app.db.models import (
    Customer,
    EmployeeCallDaily,
    EmployeeMetricDaily,
    Message,
    MsgDailyStats,
    PhoneCall,
    Staff,
)
from app.storage.first_reply_facts import (
    compute_conversation,
    load_conversation,
    pairs_active_in_days,
)
from app.storage.staff_summary import account_map
from app.storage.summary import day_bounds_ms

logger = logging.getLogger(__name__)

# 通话类型 → 消息渠道(Task0 核实:channel 列可判别,voice_call=微信/
# wa_voice_call=WA;followup 消息查询按此收窄)
_CALL_CHANNEL_TO_MSG: dict[str, str] = {"voice_call": "wechat", "wa_voice_call": "whatsapp"}


async def known_user_ids(session: AsyncSession) -> set[str]:
    """员工主数据 user_id 集合(staff 表;Task0 已核实 41 行全在职)。"""
    rows = (await session.execute(sa.select(Staff.user_id))).scalars().all()
    return {u for u in rows if u}


async def _followup_exists(
    session: AsyncSession,
    msg_channel: str,
    owner_accounts: list[str],
    peer: str,
    call_ts_ms: int,
    window_ms: int = 86_400_000,
) -> bool:
    """通话后 24h 内员工任一账号对该客户有 1:1 消息。

    含客户发来的消息——口径=窗口内存在任一方向 1:1 消息。
    按通话类型收窄消息渠道(Task0:voice_call→wechat/wa_voice_call→whatsapp,
    1:1 通话可匹配率 100%,@chatroom 群通话天然不匹配 1:1 消息)。"""
    if not owner_accounts or not peer:
        return False
    n = (
        await session.execute(
            sa.select(sa.func.count()).select_from(Message).where(
                Message.channel == msg_channel,
                Message.owner_account_id.in_(owner_accounts),
                Message.talker_id == peer,
                sa.or_(Message.room_id.is_(None), Message.room_id == ""),
                Message.timestamp.is_not(None),
                Message.timestamp > call_ts_ms,
                Message.timestamp <= call_ts_ms + window_ms,
            )
        )
    ).scalar_one()
    return bool(n)


async def recompute_call_days(session: AsyncSession, days: Iterable[int]) -> int:
    """重算给定日的通话小表:按 day delete+insert,幂等。返回写入行数。

    口径:接通 = call_duration > 0(Task0:与 call_status=1 差 31 条/1.0%,
    取 dur>0 与现有 roadmap 口径一致);call_seconds 单位=秒(Task0 已核实);
    followup 只对已接通通话判定(未接通的回拨语义留二期)。
    """
    # 员工→该渠道账号(followup 判定用,按消息渠道分组)
    acct = {ch: await account_map(session, ch) for ch in ("wechat", "whatsapp")}
    rev: dict[str, dict[str, list[str]]] = {}  # uid → msg_channel → owners
    for msg_ch, amap in acct.items():
        for owner, uid in amap.items():
            rev.setdefault(uid, {}).setdefault(msg_ch, []).append(owner)

    written = 0
    staff_ids = await known_user_ids(session)  # 幽灵过滤(v2 §1.3 #11)
    for day in sorted(set(days)):
        lo, hi = day_bounds_ms(day)
        calls = (
            await session.execute(
                sa.select(
                    PhoneCall.user_id,
                    PhoneCall.channel,
                    PhoneCall.call_number,
                    PhoneCall.start_call_time,
                    PhoneCall.call_duration,
                ).where(
                    PhoneCall.start_call_time.is_not(None),
                    PhoneCall.start_call_time >= lo,
                    PhoneCall.start_call_time < hi,
                )
            )
        ).all()
        await session.execute(
            sa.delete(EmployeeCallDaily).where(EmployeeCallDaily.day == day)
        )
        # (uid, call_type, number, ts, dur)——msg_channel 按通话类型映射;
        # 未知通话类型(如蜂窝 phone_call)无消息渠道可查,followup 判
        # False 但仍计数
        agg: dict[str, dict] = {}
        for uid, call_ch, number, ts, dur in calls:
            if not uid or uid not in staff_ids:
                continue
            a = agg.setdefault(uid, {"cnt": 0, "sec": 0, "answered": 0, "followup": 0})
            a["cnt"] += 1
            a["sec"] += int(dur or 0)
            if (dur or 0) > 0:
                a["answered"] += 1
                msg_ch = _CALL_CHANNEL_TO_MSG.get(call_ch or "")
                if msg_ch and number:
                    owners = rev.get(uid, {}).get(msg_ch, [])
                    if await _followup_exists(session, msg_ch, owners, str(number), int(ts)):
                        a["followup"] += 1
        for uid, a in sorted(agg.items()):
            session.add(
                EmployeeCallDaily(
                    day=day,
                    user_id=uid,
                    call_count=a["cnt"],
                    call_seconds=a["sec"],
                    answered_call_count=a["answered"],
                    call_followup_count=a["followup"],
                )
            )
            written += 1
    return written


METRIC_VERSION = "v1"


def _percentile(vals: list[int], q: float) -> int | None:
    """最近邻秩分位 ceil(q·n);空列表→None(与 0 严格区分)。"""
    if not vals:
        return None
    s = sorted(vals)
    idx = min(len(s) - 1, max(0, math.ceil(q * len(s)) - 1))
    return s[idx]


async def _lifecycle_daily(
    session: AsyncSession,
    days: list[int],
    channel: str,
    amap: dict[str, str],
    valid_users: set[str],
) -> dict[int, dict[str, dict[str, int]]]:
    """逐日 {day: {user: {state: count}}}——customer 当前归属 + 互动日 sweep。

    数据源:msg_daily_stats friend 行全历史(互动日/条数)。hist 为 ≤as_of
    的历史消息数(时点语义:与 last 同一 bisect 位取按日累积数组 cum[idx],
    as_of 之后的消息不虚增历史日的沉睡门槛)。16k 客户 × 窗口天数
    bisect+classify,夜间全量实测见批次A核实文档;4h tick 只算当日+3 天。
    """
    rows = (
        await session.execute(
            sa.select(
                MsgDailyStats.peer_id, MsgDailyStats.day,
                MsgDailyStats.send_cnt, MsgDailyStats.recv_cnt,
            ).where(
                MsgDailyStats.channel == channel,
                MsgDailyStats.peer_kind == "friend",
            )
        )
    ).all()
    # peer → {day: 消息条数};days_list 与累积数组 cum 对齐:
    # cum[i] = 该客户 ≤days_list[i] 的历史消息数(hist 时点语义)
    day_msgs: dict[str, dict[int, int]] = {}
    for peer, day, snd, rcv in rows:
        dm = day_msgs.setdefault(peer, {})
        dm[day] = dm.get(day, 0) + int(snd or 0) + int(rcv or 0)
    interaction = {p: sorted(dm) for p, dm in day_msgs.items()}

    custs = (
        await session.execute(
            sa.select(Customer.owner_account_id, Customer.friend_id).where(
                Customer.channel == channel,
                sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0),
                Customer.owner_account_id.is_not(None),
            )
        )
    ).all()
    peer_to_user: dict[str, str] = {}
    for owner, peer in custs:
        uid = amap.get(owner or "")
        if uid and uid in valid_users and peer:
            peer_to_user[str(peer)] = uid

    out: dict[int, dict[str, dict[str, int]]] = {d: {} for d in days}
    for peer, days_list in interaction.items():
        uid = peer_to_user.get(peer)
        if uid is None:
            continue
        first = days_list[0]
        cum = list(accumulate(day_msgs[peer][d] for d in days_list))
        for d in days:
            idx = bisect_right(days_list, d) - 1
            if idx < 0:
                continue
            last = days_list[idx]
            prev = days_list[idx - 1] if idx > 0 else None
            hist_asof = cum[idx]
            state = classify(first, last, prev, d, hist_asof)
            if state:
                out[d].setdefault(uid, {}).setdefault(state, 0)
                out[d][uid][state] += 1
    return out


async def recompute_metric_days(
    session: AsyncSession,
    days: Iterable[int],
    channels: tuple[str, ...] = ("wechat", "whatsapp"),
) -> int:
    """重算给定日的宽表:按 (day, channel) delete+insert,幂等。返回写入行数。

    串类列与首响日值:扫窗口内活跃 friend 会话的**全历史**(每会话一次),
    与 first_reply_fact 同一 compute_conversation——分子分母同源(v2 §1.1 #5)。
    """
    days = sorted(set(days))
    valid_users = await known_user_ids(session)
    pairs = await pairs_active_in_days(session, days)
    pairs_by_channel: dict[str, list] = {}
    for p in pairs:
        pairs_by_channel.setdefault(p[0], []).append((p[1], p[2]))

    written = 0
    for channel in channels:
        amap = await account_map(session, channel)
        # ---- 串类+分位:逐会话全历史一次,聚合到 (day, user) ----
        thread_cols: dict[tuple[int, str], dict] = {}
        frs: dict[tuple[int, str], list[int]] = {}
        day_set = set(days)
        for owner, peer in pairs_by_channel.get(channel, []):
            uid = amap.get(owner)
            if uid is None or uid not in valid_users:
                continue
            rows = await load_conversation(session, channel, owner, peer)
            facts, agg = compute_conversation(rows)
            for d, a in agg.items():
                if d not in day_set:
                    continue
                key = (d, uid)
                t = thread_cols.setdefault(
                    key, {"threads": 0, "answered": 0, "fr": 0}
                )
                t["threads"] += a.threads
                t["answered"] += a.answered
                t["fr"] += a.fr_sum_ms
                frs.setdefault(key, []).extend(
                    f.fr_ms for f in facts if f.day == d
                )

        # ---- 生命周期 ----
        life = await _lifecycle_daily(session, days, channel, amap, valid_users)

        # ---- 消息计数+负载:msg_daily_stats(算法无关列) ----
        for day in days:
            mds = (
                await session.execute(
                    sa.select(
                        MsgDailyStats.owner_account_id,
                        MsgDailyStats.peer_kind,
                        MsgDailyStats.peer_id,
                        MsgDailyStats.send_cnt,
                        MsgDailyStats.recv_cnt,
                    ).where(
                        MsgDailyStats.day == day, MsgDailyStats.channel == channel
                    )
                )
            ).all()
            await session.execute(
                sa.delete(EmployeeMetricDaily).where(
                    EmployeeMetricDaily.day == day,
                    EmployeeMetricDaily.channel == channel,
                    EmployeeMetricDaily.metric_version == METRIC_VERSION,
                )
            )
            agg: dict[str, dict] = {}
            for owner, kind, peer, snd, rcv in mds:
                uid = amap.get(owner)
                if uid is None or uid not in valid_users:
                    continue  # 幽灵过滤
                a = agg.setdefault(
                    uid,
                    {"sent": 0, "recv": 0, "friends": set(), "rooms": set()},
                )
                # 消息计数列 = 1:1 口径(v2 §1.1 #2):room 只进 active_rooms
                if kind == "friend":
                    a["sent"] += int(snd or 0)
                    a["recv"] += int(rcv or 0)
                    a["friends"].add(peer)
                elif kind == "room":
                    a["rooms"].add(peer)
            day_life = life.get(day, {})
            # 行并集:消息源 ∪ 生命周期源——纯生命周期日(员工当日无消息
            # 但持有降温/沉睡客户)也要出宽表行,否则客户经营维度漏天
            all_uids = sorted(set(agg) | set(day_life))
            for uid in all_uids:
                a = agg.get(
                    uid, {"sent": 0, "recv": 0, "friends": set(), "rooms": set()}
                )
                key = (day, uid)
                t = thread_cols.get(key, {"threads": 0, "answered": 0, "fr": 0})
                lv = day_life.get(uid, {})
                session.add(
                    EmployeeMetricDaily(
                        day=day,
                        channel=channel,
                        user_id=uid,
                        metric_version=METRIC_VERSION,
                        msg_sent=a["sent"],
                        msg_recv=a["recv"],
                        active_friends=len(a["friends"]),
                        active_rooms=len(a["rooms"]),
                        cust_threads=t["threads"],
                        threads_answered=t["answered"],
                        fr_sum_ms=t["fr"],
                        first_reply_p50_ms=_percentile(frs.get(key, []), 0.5),
                        first_reply_p90_ms=_percentile(frs.get(key, []), 0.9),
                        new_customer_count=lv.get("new", 0),
                        cooling_customer_count=lv.get("cooling", 0),
                        dormant_customer_count=lv.get("dormant", 0),
                        reactivated_customer_count=lv.get("reactivated", 0),
                        data_complete=True,
                    )
                )
                written += 1
            # 防御性 flush:防 autoflush=False 类配置下逐日 add 的 ORM 对象
            # 攒到事务尾(夜间全量约 52k 行;与 Task3 周期性 flush 同教训)
            await session.flush()
    return written