# -*- coding: utf-8 -*-
"""看板指标计算。

这里的函数不访问数据库、网络或时钟，便于离线单元测试。SQL 查询与数据整形
放在 ``reaction_board.web.queries``，前端只负责展示。
"""

from __future__ import annotations

import math
from dataclasses import dataclass
from statistics import median as _stat_median
from statistics import pstdev
from typing import Iterable, Mapping, Sequence

METRICS: tuple[str, ...] = (
    "view",
    "like",
    "coin",
    "favorite",
    "share",
    "reply",
    "danmaku",
)

DEFAULT_HEAT_WEIGHTS: dict[str, float] = {
    "view": 0.30,
    "like": 0.20,
    "coin": 0.20,
    "favorite": 0.15,
    "share": 0.10,
    "reply": 0.03,
    "danmaku": 0.02,
}

MIN_VIEW_FOR_INTERACTION = 100
MIN_VIEW_FOR_GROWTH_RATE = 200
MIN_EFFICIENCY_DURATION = 600
DEFAULT_MATURITY_DAYS = 2
DEFAULT_SHRINK_MINUTES = 120.0
_IQR_TO_SIGMA = 1.349


def percentile(values: Sequence[float], p: float) -> float:
    """线性插值分位数；``p`` 取 0~1。空序列返回 0。"""
    if not values:
        return 0.0
    ordered = sorted(float(value) for value in values)
    if len(ordered) == 1:
        return ordered[0]
    p = min(max(float(p), 0.0), 1.0)
    position = (len(ordered) - 1) * p
    lower = math.floor(position)
    upper = math.ceil(position)
    if lower == upper:
        return ordered[lower]
    weight = position - lower
    return ordered[lower] * (1 - weight) + ordered[upper] * weight


def robust_center_scale(values: Sequence[float]) -> tuple[float, float]:
    """中位数与稳健标准差（IQR / 1.349）。

    当 IQR 为 0（指标全相同或样本过少）时退化为总体标准差；仍为 0 时返回
    ``(center, 0)``，调用方应把该维度的 z 值视为 0。
    """
    if not values:
        return 0.0, 0.0
    numbers = [float(value) for value in values]
    center = float(_stat_median(numbers))
    scale = (percentile(numbers, 0.75) - percentile(numbers, 0.25)) / _IQR_TO_SIGMA
    if scale <= 1e-12:
        scale = float(pstdev(numbers)) if len(numbers) > 1 else 0.0
    if scale <= 1e-12:
        scale = 0.0
    return center, scale


def _metric_value(row: Mapping[str, object], metric: str) -> float:
    value = row.get(metric)
    if value is None or isinstance(value, bool):
        return 0.0
    try:
        return max(0.0, float(value))
    except (TypeError, ValueError):
        return 0.0


def content_hours(duration_seconds: object) -> float | None:
    """把视频总时长（秒）换算为内容小时；缺失或非正数返回 ``None``。

    多 P 视频的 ``duration`` 已是各 P 之和，因此这里不需要再乘 P 数。
    """
    if duration_seconds is None or isinstance(duration_seconds, bool):
        return None
    try:
        seconds = float(duration_seconds)
    except (TypeError, ValueError):
        return None
    if seconds <= 0:
        return None
    return seconds / 3600.0


def views_per_hour(view: object, duration_seconds: object) -> float | None:
    """播放/内容小时；时长缺失或非正数时返回 ``None``，不做除零。"""
    hours = content_hours(duration_seconds)
    if hours is None:
        return None
    return _metric_value({"view": view}, "view") / hours


def interactions_per_hour(
    metrics: Mapping[str, object] | None,
    duration_seconds: object,
    *,
    interaction_metrics: Sequence[str] = ("like", "coin", "favorite"),
) -> float | None:
    """互动/内容小时；统计缺失或时长无效时返回 ``None``。"""
    hours = content_hours(duration_seconds)
    if hours is None or metrics is None:
        return None
    interactions = sum(_metric_value(metrics, metric) for metric in interaction_metrics)
    return interactions / hours


def age_days(pubdate: object, reference: int | float) -> int | None:
    """发布到参考时点（通常为最新数据日）的整天数；无效发布时间返回 ``None``。"""
    if pubdate is None or isinstance(pubdate, bool):
        return None
    try:
        published = int(pubdate)
        now = int(reference)
    except (TypeError, ValueError):
        return None
    if published <= 0:
        return None
    return max(0, (now - published) // 86400)


def is_mature(
    pubdate: object,
    reference: int | float,
    maturity_days: int = DEFAULT_MATURITY_DAYS,
) -> bool:
    """是否已达到成熟窗口；发布时间缺失视为未成熟。"""
    days = age_days(pubdate, reference)
    return days is not None and days >= max(0, int(maturity_days))


def median_of(values: Iterable[object]) -> float | None:
    """忽略无效值的中位数；没有有效值返回 ``None``。"""
    numbers: list[float] = []
    for value in values:
        if value is None or isinstance(value, bool):
            continue
        try:
            numbers.append(float(value))
        except (TypeError, ValueError):
            continue
    if not numbers:
        return None
    return float(_stat_median(numbers))


def effective_shrink_minutes(
    per_up_minutes: Iterable[object],
    fallback: float = DEFAULT_SHRINK_MINUTES,
) -> float:
    """自适应收缩强度：各 UP 总内容分钟的中位数，缺失时回退固定值。"""
    positives: list[float] = []
    for minutes in per_up_minutes:
        if minutes is None or isinstance(minutes, bool):
            continue
        try:
            value = float(minutes)
        except (TypeError, ValueError):
            continue
        if value > 0:
            positives.append(value)
    value = median_of(positives)
    if value is None or value <= 0:
        return float(fallback)
    return value


def eb_rate(
    total: object,
    exposure: object,
    prior_rate: float,
    prior_strength: float,
) -> float | None:
    """Gamma–Poisson 经验贝叶斯收缩后的速率。

    ``exposure`` 与 ``prior_strength`` 必须使用同一单位（本项目中为分钟），
    返回单位与 ``prior_rate`` 一致（每分钟）。
    """
    if prior_strength < 0:
        raise ValueError("prior_strength 不能为负")
    try:
        total_value = max(0.0, float(total))
        exposure_value = max(0.0, float(exposure))
        prior = max(0.0, float(prior_rate))
    except (TypeError, ValueError):
        return None
    if prior_strength <= 0:
        return None if exposure_value <= 0 else total_value / exposure_value
    return (total_value + prior_strength * prior) / (exposure_value + prior_strength)


def normalized_weights(weights: Mapping[str, float] | None = None) -> dict[str, float]:
    source = weights or DEFAULT_HEAT_WEIGHTS
    result: dict[str, float] = {}
    for metric in METRICS:
        value = source.get(metric, DEFAULT_HEAT_WEIGHTS[metric])
        try:
            result[metric] = max(0.0, float(value))
        except (TypeError, ValueError):
            result[metric] = 0.0
    total = sum(result.values())
    if total <= 0:
        return {metric: 0.0 for metric in METRICS}
    return {metric: value / total for metric, value in result.items()}


def heat_scores(
    rows: Iterable[Mapping[str, object]],
    weights: Mapping[str, float] | None = None,
) -> dict[int, float]:
    """对每行最新指标计算综合热度。

    每个指标先 ``log1p``，再按中位数/IQR 做稳健标准化，最后加权求和。结果只
    用于同一批数据内的排序，不跨日期比较。
    """
    items = list(rows)
    if not items:
        return {}
    normalized = normalized_weights(weights)
    log_values: dict[str, list[float]] = {}
    centers: dict[str, float] = {}
    scales: dict[str, float] = {}
    for metric in METRICS:
        values = [math.log1p(_metric_value(row, metric)) for row in items]
        log_values[metric] = values
        centers[metric], scales[metric] = robust_center_scale(values)

    scores: dict[int, float] = {}
    for index, row in enumerate(items):
        if "aid" not in row:
            raise ValueError("heat_scores 需要每行包含 aid")
        score = 0.0
        for metric in METRICS:
            weight = normalized[metric]
            if weight == 0:
                continue
            scale = scales[metric]
            if scale <= 0:
                z = 0.0
            else:
                z = (log_values[metric][index] - centers[metric]) / scale
            score += weight * z
        scores[int(row["aid"])] = score
    return scores


def score_percentiles(scores: Mapping[int, float]) -> dict[int, float]:
    """把分数映射为 0~100 的百分位，最高者 100，最低者 0。"""
    if not scores:
        return {}
    ordered = sorted(scores.values())
    total = len(ordered)
    if total == 1:
        return {aid: 100.0 for aid in scores}
    result: dict[int, float] = {}
    for aid, score in scores.items():
        below = sum(1 for value in ordered if value < score)
        result[aid] = round(below / (total - 1) * 100, 2)
    return result


def interaction_rate(
    metrics: Mapping[str, object],
    *,
    min_view: int = MIN_VIEW_FOR_INTERACTION,
) -> float | None:
    """互动率 = (点赞 + 投币 + 收藏) / 播放；播放不足门槛返回 None。"""
    view = _metric_value(metrics, "view")
    if view < min_view or view <= 0:
        return None
    interactions = sum(_metric_value(metrics, metric) for metric in ("like", "coin", "favorite"))
    return interactions / view


@dataclass(frozen=True)
class Growth:
    available: bool
    delta: int | None = None
    per_day: float | None = None
    rate: float | None = None
    gap_days: int | None = None

    def as_dict(self) -> dict[str, object]:
        return {
            "available": self.available,
            "delta": self.delta,
            "per_day": self.per_day,
            "rate": self.rate,
            "gap_days": self.gap_days,
        }


def compute_growth(
    current: int | float | None,
    previous: int | float | None,
    gap_days: int | float | None,
    *,
    allow_zero_gap: bool = False,
) -> Growth:
    """计算累计差值、日均差值和相对增速。

    缺失基线或间隔为负时返回不可用。默认间隔不足 1 天也不可用，不把 0 当成真实增长。
    ``allow_zero_gap=True`` 时允许同一天的两次采样（或相对 0）计算增量，``per_day`` 为 None。
    """
    if current is None or previous is None or gap_days is None:
        return Growth(available=False)
    gap = int(gap_days)
    if gap < 0:
        return Growth(available=False)
    if gap == 0 and not allow_zero_gap:
        return Growth(available=False)
    current_value = max(0, int(current))
    previous_value = max(0, int(previous))
    delta = current_value - previous_value
    return Growth(
        available=True,
        delta=delta,
        per_day=None if gap == 0 else delta / gap,
        rate=delta / max(previous_value, 1),
        gap_days=gap,
    )


def normalize_cover(url: object) -> str | None:
    """把 B 站封面统一为 https；空值返回 None。"""
    if not isinstance(url, str):
        return None
    value = url.strip()
    if not value:
        return None
    if value.startswith("//"):
        return "https:" + value
    if value.startswith("http://"):
        return "https://" + value[len("http://"):]
    return value


def is_loopback(host: str) -> bool:
    value = (host or "").strip().lower()
    return value in {"127.0.0.1", "localhost", "::1"}