from __future__ import annotations

import heapq
import math
import sqlite3
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Optional

from ...chat_helpers import (
    _build_avatar_url,
    _load_contact_rows,
    _pick_display_name,
    _should_keep_session,
)
from ...chat_search_index import (
    get_chat_search_index_db_path,
    get_chat_search_index_status,
    start_chat_search_index_build,
)
from ...logging_config import get_logger

logger = get_logger(__name__)


def _year_range_epoch_seconds(year: int) -> tuple[int, int]:
    # Use local time boundaries (same semantics as sqlite "localtime").
    start = int(datetime(year, 1, 1).timestamp())
    end = int(datetime(year + 1, 1, 1).timestamp())
    return start, end


def _mask_name(name: str) -> str:
    s = str(name or "").strip()
    if not s:
        return ""
    if len(s) == 1:
        return "*"
    if len(s) == 2:
        return s[0] + "*"
    return s[0] + ("*" * (len(s) - 2)) + s[-1]


def _format_duration_zh(seconds: int | None) -> str:
    if seconds is None:
        return ""
    try:
        s = int(seconds)
    except Exception:
        s = 0
    if s < 0:
        s = 0

    if s < 60:
        return f"{s}秒"
    m, sec = divmod(s, 60)
    if m < 60:
        return f"{m}分{sec}秒" if sec else f"{m}分钟"
    h, mm = divmod(m, 60)
    if h < 24:
        return f"{h}小时{mm}分钟" if mm else f"{h}小时"
    d, hh = divmod(h, 24)
    return f"{d}天{hh}小时" if hh else f"{d}天"


def _compute_streak_days(doys: list[int]) -> int:
    if not doys:
        return 0
    doys_sorted = sorted({int(x) for x in doys if int(x) > 0})
    if not doys_sorted:
        return 0

    best = 1
    cur = 1
    prev = doys_sorted[0]
    for d in doys_sorted[1:]:
        if d == prev + 1:
            cur += 1
        else:
            cur = 1
        if cur > best:
            best = cur
        prev = d
    return int(best)


def _compute_best_buddy_extras_from_index(*, account_dir: Path, year: int, buddy_username: str) -> dict[str, Any]:
    """Compute a few extra fields for Card07 Bento summary.

    - longestStreakDays: longest consecutive days with any interaction
    - peakHour/peakHourLabel: most active hour of day with this buddy

    Best-effort: returns empty dict on any failure.
    """

    buddy = str(buddy_username or "").strip()
    if not buddy:
        return {}

    index_path = get_chat_search_index_db_path(account_dir)
    try:
        index_status = get_chat_search_index_status(account_dir, source="auto")
        if not bool((index_status.get("index") or {}).get("ready")):
            return {}
    except Exception:
        return {}
    if not index_path.exists():
        return {}

    start_ts, end_ts = _year_range_epoch_seconds(int(year))

    ts_expr = (
        "CASE "
        "WHEN CAST(create_time AS INTEGER) > 1000000000000 "
        "THEN CAST(CAST(create_time AS INTEGER)/1000 AS INTEGER) "
        "ELSE CAST(create_time AS INTEGER) "
        "END"
    )
    where = (
        f"{ts_expr} >= ? AND {ts_expr} < ? "
        "AND db_stem NOT LIKE 'biz_message%' "
        "AND CAST(local_type AS INTEGER) != 10000 "
        "AND username = ? "
        "AND username NOT LIKE '%@chatroom'"
    )

    sql_days = (
        "SELECT DISTINCT "
        "CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) AS doy "
        "FROM ("
        f"  SELECT {ts_expr} AS ts "
        "  FROM message_fts "
        f"  WHERE {where}"
        ") sub "
        "WHERE ts > 0 "
        "ORDER BY doy ASC"
    )

    sql_peak_hour = (
        "SELECT "
        "CAST(strftime('%H', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) AS h, "
        "COUNT(1) AS cnt "
        "FROM ("
        f"  SELECT {ts_expr} AS ts "
        "  FROM message_fts "
        f"  WHERE {where}"
        ") sub "
        "WHERE ts > 0 "
        "GROUP BY h "
        "ORDER BY cnt DESC, h ASC "
        "LIMIT 1"
    )

    conn = sqlite3.connect(str(index_path))
    try:
        has_fts = (
            conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='message_fts' LIMIT 1").fetchone()
            is not None
        )
        if not has_fts:
            return {}

        params = (start_ts, end_ts, buddy)

        doys: list[int] = []
        try:
            rows = conn.execute(sql_days, params).fetchall()
        except Exception:
            rows = []
        for r in rows:
            if not r or r[0] is None:
                continue
            try:
                doys.append(int(r[0]))
            except Exception:
                continue

        longest_streak_days = _compute_streak_days(doys)

        peak_hour: int | None = None
        try:
            row = conn.execute(sql_peak_hour, params).fetchone()
            if row and row[0] is not None:
                peak_hour = int(row[0])
        except Exception:
            peak_hour = None

        out: dict[str, Any] = {"longestStreakDays": int(longest_streak_days)}
        if peak_hour is not None and 0 <= peak_hour <= 23:
            out["peakHour"] = int(peak_hour)
            out["peakHourLabel"] = f"{int(peak_hour):02d}:00"
        return out
    except Exception:
        return {}
    finally:
        try:
            conn.close()
        except Exception:
            pass


@dataclass
class _ConvAgg:
    username: str
    incoming: int
    outgoing: int
    replies: int
    sum_gap: int
    sum_gap_capped: int
    min_gap: int
    max_gap: int

    @property
    def total(self) -> int:
        return int(self.incoming) + int(self.outgoing)

    def avg_gap(self) -> float:
        return float(self.sum_gap) / float(self.replies) if self.replies > 0 else 0.0

    def avg_gap_capped(self) -> float:
        return float(self.sum_gap_capped) / float(self.replies) if self.replies > 0 else 0.0


def _score_conv(*, agg: _ConvAgg, tau_seconds: float) -> float:
    # "聊天频率"：更偏向双向互动（取 min(in, out)）。
    interaction = float(min(int(agg.incoming), int(agg.outgoing)))
    if interaction <= 0.0 or agg.replies <= 0:
        return 0.0

    # "回复频率/速度"：用 capped 平均耗时做一个饱和衰减，避免极端长等待把分数打穿。
    avg_s = float(agg.avg_gap_capped())
    speed_score = 1.0 / (1.0 + (avg_s / float(max(1.0, tau_seconds))))

    volume_score = math.log1p(interaction)
    return float(speed_score * volume_score)


def compute_reply_speed_stats(*, account_dir: Path, year: int) -> dict[str, Any]:
    """
    统计“回复速度”相关指标（全局 + 每个好友），用于 Wrapped 年度总结卡片。

    Notes / 口径说明：
    - 仅统计 1v1（非群聊）会话：username 不以 "@chatroom" 结尾。
    - “一次回复”定义：对方发出消息后，你发送的第一条消息（同一段连续你发的消息只计 1 次）。
    - 默认过滤系统消息（local_type=10000），并排除 biz_message*.db。
    - 优先使用 chat_search_index.db（全量合并所有 shard），没有索引时做 best-effort 降级。
    """

    start_ts, end_ts = _year_range_epoch_seconds(int(year))
    my_username = str(account_dir.name or "").strip()

    # Scoring hyper-params (tuned for "更偏向聊天频率高的" 的直觉)。
    gap_cap_seconds = 6 * 60 * 60  # 6h: scoring 上限（超过当作一样慢）
    tau_seconds = 30 * 60  # 30min: 速度衰减的尺度

    total_replies = 0
    global_fastest: int | None = None
    global_fastest_u: str | None = None
    global_slowest: int | None = None
    global_slowest_u: str | None = None

    reply_gaps: list[int] = []
    reply_stats: dict[str, Any] | None = None

    best_score = -1.0
    best_agg: _ConvAgg | None = None

    # NOTE: Use (score, username, agg) so the heap is always comparable even when scores tie.
    top_heap: list[tuple[float, str, _ConvAgg]] = []
    top_n = 8

    # For "今年你总共给 xxx 人发送过消息" & top-total bar-race.
    sent_to_contacts: set[str] = set()
    # Collect totals for *all* 1v1 sessions so the frontend ranking can naturally grow over time.
    all_totals: dict[str, int] = {}
    # NOTE: Use (total, username, agg) so the heap is always comparable even when totals tie.
    top_total_heap: list[tuple[int, str, _ConvAgg]] = []
    # Keep more than 10 so the bar-race "TOP10" can actually evolve (members can enter/leave over time).
    top_total_n = 100

    # ---- 「谁先开口」：相邻消息间隔 ≥1 小时切分为一次“对话”，首条消息发送方即发起者 ----
    conv_gap_seconds = 3600
    initiative_conversations = 0
    initiated_by_me = 0
    initiated_by_others = 0
    init_by_me_counts: dict[str, int] = {}
    init_to_me_counts: dict[str, int] = {}

    # ---- 「势均力敌」：双方消息量均 ≥50 且 min/max 比值最接近 1 的好友 ----
    mutual_min_each = 50
    mutual_best_agg: _ConvAgg | None = None
    # (ratio, total, username)：比值优先，其次总量更大者，最后用户名字典序保证确定性。
    mutual_best_key: tuple[float, int, str] | None = None

    def consider_conv(agg: _ConvAgg) -> None:
        nonlocal best_score, best_agg
        if not agg.username:
            return
        if agg.replies <= 0:
            return
        if min(agg.incoming, agg.outgoing) <= 0:
            return

        score = _score_conv(agg=agg, tau_seconds=tau_seconds)
        if score > best_score:
            best_score = float(score)
            best_agg = agg

        if score <= 0:
            return
        key = (float(score), str(agg.username), agg)
        if len(top_heap) < top_n:
            heapq.heappush(top_heap, key)
        else:
            heapq.heappushpop(top_heap, key)

    def consider_total(agg: _ConvAgg) -> None:
        nonlocal mutual_best_agg, mutual_best_key
        if not agg.username:
            return
        if agg.total <= 0:
            return
        # Keep the same filtering behavior as other wrapped cards.
        if not _should_keep_session(agg.username, include_official=False):
            return

        if agg.outgoing > 0:
            sent_to_contacts.add(agg.username)

        sent = int(agg.outgoing)
        received = int(agg.incoming)
        if sent >= mutual_min_each and received >= mutual_min_each:
            ratio = float(min(sent, received)) / float(max(sent, received))
            if (
                mutual_best_key is None
                or ratio > mutual_best_key[0]
                or (ratio == mutual_best_key[0] and sent + received > mutual_best_key[1])
                or (
                    ratio == mutual_best_key[0]
                    and sent + received == mutual_best_key[1]
                    and str(agg.username) < mutual_best_key[2]
                )
            ):
                mutual_best_key = (float(ratio), int(sent + received), str(agg.username))
                mutual_best_agg = agg

        total = int(agg.total)
        all_totals[agg.username] = int(total)
        key = (total, str(agg.username), agg)
        if len(top_total_heap) < top_total_n:
            heapq.heappush(top_total_heap, key)
        else:
            heapq.heappushpop(top_total_heap, key)

    used_index = False

    # -------- Preferred path: unified search index --------
    index_path = get_chat_search_index_db_path(account_dir)
    index_status: dict[str, Any] | None = None
    try:
        index_status = get_chat_search_index_status(account_dir, source="auto")
    except Exception:
        index_status = None
    if index_path.exists() and bool(((index_status or {}).get("index") or {}).get("ready")):
        conn = sqlite3.connect(str(index_path))
        try:
            has_fts = (
                conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='message_fts' LIMIT 1").fetchone()
                is not None
            )
            if has_fts and my_username:
                used_index = True
                t0 = time.time()

                ts_expr = (
                    "CASE "
                    "WHEN CAST(create_time AS INTEGER) > 1000000000000 "
                    "THEN CAST(CAST(create_time AS INTEGER)/1000 AS INTEGER) "
                    "ELSE CAST(create_time AS INTEGER) "
                    "END"
                )

                where = (
                    f"{ts_expr} >= ? AND {ts_expr} < ? "
                    "AND db_stem NOT LIKE 'biz_message%' "
                    "AND CAST(local_type AS INTEGER) != 10000 "
                    "AND username NOT LIKE '%@chatroom'"
                )

                # Order by username, then time (ties broken by sort_seq/local_id if possible).
                sql = (
                    "SELECT "
                    "username, sender_username, "
                    f"{ts_expr} AS ts, "
                    "CAST(sort_seq AS INTEGER) AS sort_seq_i, "
                    "CAST(local_id AS INTEGER) AS local_id_i "
                    "FROM message_fts "
                    f"WHERE {where} "
                    "ORDER BY username ASC, ts ASC, sort_seq_i ASC, local_id_i ASC"
                )

                cur = conn.execute(sql, (start_ts, end_ts))

                cur_username: str = ""
                incoming = 0
                outgoing = 0
                replies = 0
                sum_gap = 0
                sum_gap_capped = 0
                min_gap = 0
                max_gap = 0
                prev_other_ts: int | None = None
                prev_any_ts: int | None = None

                def flush() -> None:
                    nonlocal cur_username, incoming, outgoing, replies, sum_gap, sum_gap_capped, min_gap, max_gap
                    if not cur_username:
                        return
                    agg = _ConvAgg(
                        username=cur_username,
                        incoming=int(incoming),
                        outgoing=int(outgoing),
                        replies=int(replies),
                        sum_gap=int(sum_gap),
                        sum_gap_capped=int(sum_gap_capped),
                        min_gap=int(min_gap),
                        max_gap=int(max_gap),
                    )
                    consider_total(agg)
                    consider_conv(agg)

                for row in cur:
                    try:
                        username = str(row[0] or "").strip()
                        sender = str(row[1] or "").strip()
                        ts = int(row[2] or 0)
                    except Exception:
                        continue

                    if ts <= 0 or not username:
                        continue
                    if username != cur_username:
                        # flush old
                        flush()
                        # reset for new conversation
                        cur_username = username
                        incoming = outgoing = replies = 0
                        sum_gap = sum_gap_capped = 0
                        min_gap = max_gap = 0
                        prev_other_ts = None
                        prev_any_ts = None

                    # Drop system/official-ish sessions (best-effort).
                    if not _should_keep_session(username, include_official=False):
                        continue

                    is_me = sender == my_username

                    # 「谁先开口」：会话首条消息、或与上一条消息间隔 ≥ conv_gap_seconds，均视为新对话开始。
                    if prev_any_ts is None or ts - prev_any_ts >= conv_gap_seconds:
                        initiative_conversations += 1
                        if is_me:
                            initiated_by_me += 1
                            init_by_me_counts[username] = init_by_me_counts.get(username, 0) + 1
                        else:
                            initiated_by_others += 1
                            init_to_me_counts[username] = init_to_me_counts.get(username, 0) + 1
                    prev_any_ts = ts

                    if is_me:
                        outgoing += 1
                        if prev_other_ts is not None and ts >= prev_other_ts:
                            gap = int(ts - prev_other_ts)
                            replies += 1
                            total_replies += 1
                            sum_gap += gap
                            sum_gap_capped += min(gap, gap_cap_seconds)
                            reply_gaps.append(int(gap))

                            if replies == 1 or gap < min_gap:
                                min_gap = gap
                            if replies == 1 or gap > max_gap:
                                max_gap = gap

                            if global_fastest is None or gap < global_fastest:
                                global_fastest = gap
                                global_fastest_u = username
                            if global_slowest is None or gap > global_slowest:
                                global_slowest = gap
                                global_slowest_u = username

                            # Only count the first outgoing message as the "reply" to this prompt.
                            prev_other_ts = None
                    else:
                        incoming += 1
                        prev_other_ts = ts

                flush()

                logger.info(
                    "Wrapped card#3 reply_speed computed (search index): account=%s year=%s conversations_top=%s replies=%s db=%s elapsed=%.2fs",
                    str(account_dir.name or "").strip(),
                    int(year),
                    len(top_heap),
                    int(total_replies),
                    str(index_path.name),
                    time.time() - t0,
                )
        finally:
            try:
                conn.close()
            except Exception:
                pass

    if reply_gaps:
        try:
            reply_gaps.sort()
            n = int(len(reply_gaps))
            # Nearest-rank quantiles (deterministic, integer seconds).
            p50_idx = max(0, min(n - 1, int(math.ceil(0.50 * n) - 1)))
            p90_idx = max(0, min(n - 1, int(math.ceil(0.90 * n) - 1)))
            reply_stats = {
                "p50Seconds": int(reply_gaps[p50_idx]),
                "p90Seconds": int(reply_gaps[p90_idx]),
            }
        except Exception:
            reply_stats = None

    # -------- Fallback path: no index --------
    # Best-effort: if the index doesn't exist / isn't ready, auto-start building it (async) so user can
    # retry this page later. We intentionally do NOT block here.
    index_status: dict[str, Any] | None = None
    if not used_index:
        try:
            index_status = get_chat_search_index_status(account_dir, source="auto")
            index = dict(index_status.get("index") or {})
            build = dict(index.get("build") or {})
            index_ready = bool(index.get("ready"))
            build_status = str(build.get("status") or "")
            index_exists = bool(index.get("exists"))

            if (not index_ready) and build_status not in {"building", "error"}:
                start_chat_search_index_build(account_dir, rebuild=bool(index_exists), source="auto")
                index_status = get_chat_search_index_status(account_dir, source="auto")
        except Exception:
            index_status = None

        logger.warning(
            "Wrapped card#3 reply_speed: search index missing/not ready; returning empty stats. account=%s year=%s index=%s",
            str(account_dir.name or "").strip(),
            int(year),
            str(index_path),
        )

    # Sort top buddies by score desc.
    top_buddies: list[tuple[float, _ConvAgg]] = sorted(
        [(score, agg) for score, _, agg in top_heap],
        key=lambda x: (-x[0], x[1].username),
    )
    top_totals: list[tuple[int, _ConvAgg]] = sorted(
        [(total, agg) for total, _, agg in top_total_heap],
        key=lambda x: (-x[0], x[1].username),
    )

    # 「谁先开口」双榜：按次数降序（并列时用户名字典序），各取前 3。
    top_init_by_me = sorted(init_by_me_counts.items(), key=lambda kv: (-int(kv[1]), str(kv[0])))[:3]
    top_init_to_me = sorted(init_to_me_counts.items(), key=lambda kv: (-int(kv[1]), str(kv[0])))[:3]

    # Resolve contact display names/avatars for a small set (bestBuddy + extremes + top list).
    need_usernames: list[str] = []
    if best_agg is not None:
        need_usernames.append(best_agg.username)
    if global_fastest_u:
        need_usernames.append(global_fastest_u)
    if global_slowest_u:
        need_usernames.append(global_slowest_u)
    for _, agg in top_buddies:
        need_usernames.append(agg.username)
    for _, agg in top_totals:
        need_usernames.append(agg.username)
    for u, _ in top_init_by_me:
        need_usernames.append(u)
    for u, _ in top_init_to_me:
        need_usernames.append(u)
    if mutual_best_agg is not None:
        need_usernames.append(mutual_best_agg.username)

    uniq_usernames = []
    seen = set()
    for u in need_usernames:
        if u and u not in seen:
            seen.add(u)
            uniq_usernames.append(u)

    contact_rows = _load_contact_rows(account_dir / "contact.db", uniq_usernames) if uniq_usernames else {}

    def conv_to_obj(score: float | None, agg: _ConvAgg) -> dict[str, Any]:
        row = contact_rows.get(agg.username)
        display = _pick_display_name(row, agg.username)
        avatar = _build_avatar_url(str(account_dir.name or ""), agg.username) if agg.username else ""
        avg_s = agg.avg_gap()
        out: dict[str, Any] = {
            "username": agg.username,
            "displayName": display,
            "maskedName": _mask_name(display),
            "avatarUrl": avatar,
            "incomingMessages": int(agg.incoming),
            "outgoingMessages": int(agg.outgoing),
            "totalMessages": int(agg.total),
            "replyCount": int(agg.replies),
            "avgReplySeconds": float(avg_s),
            "fastestReplySeconds": int(agg.min_gap) if agg.replies > 0 else None,
            "slowestReplySeconds": int(agg.max_gap) if agg.replies > 0 else None,
        }
        if score is not None:
            out["score"] = float(score)
        return out

    best_buddy_obj = None
    if best_agg is not None:
        best_buddy_obj = conv_to_obj(best_score, best_agg)
        if used_index and isinstance(best_buddy_obj, dict) and best_buddy_obj.get("username"):
            extras = _compute_best_buddy_extras_from_index(
                account_dir=account_dir,
                year=int(year),
                buddy_username=str(best_buddy_obj.get("username") or ""),
            )
            if extras:
                best_buddy_obj.update(extras)

    fastest_obj = None
    if global_fastest is not None and global_fastest_u:
        # Use the best agg if it matches; otherwise create a minimal object.
        agg = next((a for _, a in top_buddies if a.username == global_fastest_u), None)
        if agg is None and best_agg is not None and best_agg.username == global_fastest_u:
            agg = best_agg
        if agg is not None:
            fastest_obj = conv_to_obj(None, agg)
            fastest_obj["seconds"] = int(global_fastest)
        else:
            row = contact_rows.get(global_fastest_u)
            display = _pick_display_name(row, global_fastest_u)
            avatar = _build_avatar_url(str(account_dir.name or ""), global_fastest_u) if global_fastest_u else ""
            fastest_obj = {
                "username": global_fastest_u,
                "displayName": display,
                "maskedName": _mask_name(display),
                "avatarUrl": avatar,
                "seconds": int(global_fastest),
            }

    slowest_obj = None
    if global_slowest is not None and global_slowest_u:
        agg = next((a for _, a in top_buddies if a.username == global_slowest_u), None)
        if agg is None and best_agg is not None and best_agg.username == global_slowest_u:
            agg = best_agg
        if agg is not None:
            slowest_obj = conv_to_obj(None, agg)
            slowest_obj["seconds"] = int(global_slowest)
        else:
            row = contact_rows.get(global_slowest_u)
            display = _pick_display_name(row, global_slowest_u)
            avatar = _build_avatar_url(str(account_dir.name or ""), global_slowest_u) if global_slowest_u else ""
            slowest_obj = {
                "username": global_slowest_u,
                "displayName": display,
                "maskedName": _mask_name(display),
                "avatarUrl": avatar,
                "seconds": int(global_slowest),
            }

    top_list = [conv_to_obj(score, agg) for score, agg in top_buddies]

    top_totals_list = [
        {
            **conv_to_obj(None, agg),
            "totalMessages": int(total),
        }
        for total, agg in top_totals
    ]

    def initiative_person(u: str, count: int) -> dict[str, Any]:
        row = contact_rows.get(u)
        display = _pick_display_name(row, u)
        return {
            "username": u,
            "displayName": display,
            "maskedName": _mask_name(display),
            "avatarUrl": _build_avatar_url(str(account_dir.name or ""), u) if u else "",
            "count": int(count),
        }

    mutual_friend_obj: dict[str, Any] | None = None
    if mutual_best_agg is not None:
        sent = int(mutual_best_agg.outgoing)
        received = int(mutual_best_agg.incoming)
        row = contact_rows.get(mutual_best_agg.username)
        display = _pick_display_name(row, mutual_best_agg.username)
        mutual_friend_obj = {
            "username": mutual_best_agg.username,
            "displayName": display,
            "maskedName": _mask_name(display),
            "avatarUrl": _build_avatar_url(str(account_dir.name or ""), mutual_best_agg.username),
            "sentCount": sent,
            "receivedCount": received,
            "ratio": round(float(min(sent, received)) / float(max(sent, received)), 2),
        }

    initiative_obj: dict[str, Any] = {
        "conversationCount": int(initiative_conversations),
        "initiatedByMe": int(initiated_by_me),
        "initiatedByOthers": int(initiated_by_others),
        "initiationRatePct": (
            round(float(initiated_by_me) * 100.0 / float(initiative_conversations), 1)
            if initiative_conversations > 0
            else None
        ),
        "topInitiatedByMe": [initiative_person(u, c) for u, c in top_init_by_me],
        "topInitiatedToMe": [initiative_person(u, c) for u, c in top_init_to_me],
        "mutualFriend": mutual_friend_obj,
    }

    # Prepare "bar race" data: all 1v1 sessions (exclude official/system), cumulative per day.
    race = None
    if used_index and all_totals:
        days_in_year = int((datetime(int(year) + 1, 1, 1) - datetime(int(year), 1, 1)).days)
        u_list = [u for u, _ in sorted(all_totals.items(), key=lambda kv: (-int(kv[1] or 0), str(kv[0] or ""))) if u]
        if days_in_year > 0 and u_list:
            # Convert millisecond timestamps defensively.
            ts_expr = (
                "CASE "
                "WHEN CAST(create_time AS INTEGER) > 1000000000000 "
                "THEN CAST(CAST(create_time AS INTEGER)/1000 AS INTEGER) "
                "ELSE CAST(create_time AS INTEGER) "
                "END"
            )

            base_where = (
                f"{ts_expr} >= ? AND {ts_expr} < ? "
                "AND db_stem NOT LIKE 'biz_message%' "
                "AND CAST(local_type AS INTEGER) != 10000 "
                "AND username NOT LIKE '%@chatroom'"
            )

            sql_daily = (
                "SELECT username, "
                "CAST(strftime('%j', datetime(ts, 'unixepoch', 'localtime')) AS INTEGER) - 1 AS doy, "
                "sender_username, "
                "COUNT(1) AS cnt "
                "FROM ("
                f"  SELECT username, sender_username, {ts_expr} AS ts "
                "  FROM message_fts "
                f"  WHERE {base_where}"
                ") sub "
                "GROUP BY username, doy, sender_username"
            )

            u_set = set(u_list)
            per_user_daily_total: dict[str, list[int]] = {}
            per_user_daily_outgoing: dict[str, list[int]] = {}
            per_user_daily_incoming: dict[str, list[int]] = {}
            try:
                conn2 = sqlite3.connect(str(index_path))
                try:
                    rows = conn2.execute(sql_daily, (start_ts, end_ts)).fetchall()
                finally:
                    conn2.close()
            except Exception:
                rows = []

            for r in rows:
                if not r:
                    continue
                u = str(r[0] or "").strip()
                if not u or u not in u_set:
                    continue
                try:
                    doy = int(r[1] if r[1] is not None else -1)
                    sender = str(r[2] or "").strip()
                    cnt = int(r[3] or 0)
                except Exception:
                    continue
                if cnt <= 0 or doy < 0 or doy >= days_in_year:
                    continue
                daily_total = per_user_daily_total.get(u)
                if daily_total is None:
                    daily_total = [0] * days_in_year
                    per_user_daily_total[u] = daily_total
                daily_total[doy] += cnt

                if sender == my_username:
                    daily_outgoing = per_user_daily_outgoing.get(u)
                    if daily_outgoing is None:
                        daily_outgoing = [0] * days_in_year
                        per_user_daily_outgoing[u] = daily_outgoing
                    daily_outgoing[doy] += cnt
                else:
                    daily_incoming = per_user_daily_incoming.get(u)
                    if daily_incoming is None:
                        daily_incoming = [0] * days_in_year
                        per_user_daily_incoming[u] = daily_incoming
                    daily_incoming[doy] += cnt

            # Ensure we can render display names/avatars for the whole race list.
            extra_usernames = [u for u in u_list if u and u not in contact_rows]
            if extra_usernames:
                try:
                    # sqlite has a default var limit; query in chunks.
                    CHUNK = 900
                    for i in range(0, len(extra_usernames), CHUNK):
                        contact_rows.update(_load_contact_rows(account_dir / "contact.db", extra_usernames[i : i + CHUNK]))
                except Exception:
                    pass

            series: list[dict[str, Any]] = []
            for u in u_list:
                daily_total = per_user_daily_total.get(u)
                if not daily_total:
                    continue
                daily_outgoing = per_user_daily_outgoing.get(u) or [0] * days_in_year
                daily_incoming = per_user_daily_incoming.get(u) or [0] * days_in_year
                cum_total: list[int] = []
                cum_outgoing: list[int] = []
                cum_incoming: list[int] = []
                running_total = 0
                running_outgoing = 0
                running_incoming = 0
                for i in range(days_in_year):
                    running_total += int(daily_total[i] or 0)
                    running_outgoing += int(daily_outgoing[i] or 0)
                    running_incoming += int(daily_incoming[i] or 0)
                    cum_total.append(int(running_total))
                    cum_outgoing.append(int(running_outgoing))
                    cum_incoming.append(int(running_incoming))

                total_messages = int(cum_total[-1]) if cum_total else int(all_totals.get(u) or 0)
                outgoing_messages = int(cum_outgoing[-1]) if cum_outgoing else 0
                incoming_messages = int(cum_incoming[-1]) if cum_incoming else 0

                row = contact_rows.get(u)
                display = _pick_display_name(row, u)
                avatar = _build_avatar_url(str(account_dir.name or ""), u) if u else ""
                series.append(
                    {
                        "username": u,
                        "displayName": display,
                        "maskedName": _mask_name(display),
                        "avatarUrl": avatar,
                        "totalMessages": int(total_messages),
                        "outgoingMessages": int(outgoing_messages),
                        "incomingMessages": int(incoming_messages),
                        "cumulativeCounts": cum_total,
                        "cumulativeOutgoingCounts": cum_outgoing,
                        "cumulativeIncomingCounts": cum_incoming,
                    }
                )

            race = {
                "year": int(year),
                "startDate": f"{int(year)}-01-01",
                "endDate": f"{int(year)}-12-31",
                "days": int(days_in_year),
                "series": series,
            }

    # Load all contacts for lottery animation (up to 50 random contacts)
    all_contacts_list: list[dict[str, Any]] = []
    try:
        contact_db_path = account_dir / "contact.db"
        if contact_db_path.exists():
            conn = sqlite3.connect(str(contact_db_path))
            conn.row_factory = sqlite3.Row
            try:
                # Get contacts that are real users (not chatrooms, not official accounts)
                sql = """
                    SELECT username, remark, nick_name, alias, big_head_url, small_head_url
                    FROM contact
                    WHERE username NOT LIKE '%@chatroom'
                      AND username NOT LIKE 'gh_%'
                      AND username NOT LIKE 'weixin'
                      AND username NOT LIKE 'filehelper'
                      AND username NOT LIKE 'fmessage'
                      AND username NOT IN ('medianote', 'floatbottle', 'shakeapp', 'lbsapp', 'newsapp')
                      AND (nick_name IS NOT NULL AND nick_name != '')
                    ORDER BY RANDOM()
                    LIMIT 50
                """
                rows = conn.execute(sql).fetchall()
                for r in rows:
                    u = str(r["username"] or "").strip()
                    if not u:
                        continue
                    display = _pick_display_name(r, u)
                    avatar = _build_avatar_url(str(account_dir.name or ""), u) if u else ""
                    all_contacts_list.append({
                        "username": u,
                        "displayName": display,
                        "maskedName": _mask_name(display),
                        "avatarUrl": avatar,
                    })
            finally:
                conn.close()
    except Exception:
        pass

    return {
        "year": int(year),
        "sentToContacts": int(len(sent_to_contacts)),
        "replyEvents": int(total_replies),
        "replyStats": reply_stats,
        "fastestReplySeconds": int(global_fastest) if global_fastest is not None else None,
        "longestReplySeconds": int(global_slowest) if global_slowest is not None else None,
        "bestBuddy": best_buddy_obj,
        "fastest": fastest_obj,
        "slowest": slowest_obj,
        "topBuddies": top_list,
        "topTotals": top_totals_list,
        "allContacts": all_contacts_list,
        "initiative": initiative_obj,
        "race": race,
        "settings": {
            "gapCapSeconds": int(gap_cap_seconds),
            "tauSeconds": int(tau_seconds),
            "usedIndex": bool(used_index),
            "indexStatus": index_status,
        },
    }


def build_card_03_reply_speed(*, account_dir: Path, year: int) -> dict[str, Any]:
    stats = compute_reply_speed_stats(account_dir=account_dir, year=year)

    fastest = stats.get("fastestReplySeconds")
    longest = stats.get("longestReplySeconds")
    best = stats.get("bestBuddy") or None
    replies = int(stats.get("replyEvents") or 0)

    if replies <= 0:
        narrative = "今年你还没有可统计的“回复”记录（或尚未构建搜索索引）。"
    else:
        parts: list[str] = []
        if fastest is not None:
            parts.append(f"最快一次，你只用了 {_format_duration_zh(int(fastest))} 就回了消息。")
        if longest is not None:
            parts.append(f"最长一次，你让对方等了 {_format_duration_zh(int(longest))}。")
        if best and isinstance(best, dict) and best.get("displayName"):
            avg_s = best.get("avgReplySeconds")
            try:
                avg_i = int(round(float(avg_s or 0.0)))
            except Exception:
                avg_i = 0
            parts.append(
                f"最像你的聊天搭子是「{_mask_name(str(best.get('displayName') or ''))}」，平均每条回复用时 {_format_duration_zh(avg_i)}。"
            )
        narrative = "".join(parts) if parts else "你的回复速度，藏着你最在意的人。"

    return {
        "id": 3,
        "title": "谁是你「秒回」的置顶关心？",
        "scope": "global",
        "category": "B",
        "status": "ok",
        "kind": "chat/reply_speed",
        "narrative": narrative,
        "data": stats,
    }
