"""YouTube relay quality scoring (from sh2 youtube_quality_guard.py, client-simplified)."""

from __future__ import annotations

import re
import time
from typing import Any

from web2_douyin_source_display import extract_douyin_source_meta

_FFMPEG_KV_RE = re.compile(
    r"(?P<key>frame|fps|bitrate|speed|drop|dup)\s*=\s*(?P<value>[^\s]+)", re.IGNORECASE
)
_TIME_RE = re.compile(r"time\s*=\s*(?P<value>\d{2}:\d{2}:\d{2}(?:\.\d+)?)", re.IGNORECASE)
_URL_RE = re.compile(r"https?://[^\s\"'<>]+", re.IGNORECASE)
_ROOM_RE = re.compile(r"live\.douyin\.com/([a-zA-Z0-9_-]+)", re.IGNORECASE)


def _as_float(value: Any) -> float | None:
    text = str(value or "").strip().lower().replace("x", "").replace("kbits/s", "").replace("kb/s", "")
    text = text.replace(",", "")
    match = re.search(r"-?\d+(?:\.\d+)?", text)
    if not match:
        return None
    try:
        return float(match.group(0))
    except ValueError:
        return None


def _as_int(value: Any) -> int | None:
    parsed = _as_float(value)
    return None if parsed is None else max(0, int(parsed))


_IDLE_MARKERS = (
    "等待直播",
    "waiting",
    "获取失败",
    "douyin web api failed",
    "无录制任务",
    "网址内容获取失败",
    "没有正在监测的直播",
)


def _line_implies_relay_idle(line: str) -> bool:
    lower = line.lower()
    if "frame=" in lower and "speed=" in lower:
        return False
    return any(marker in lower for marker in _IDLE_MARKERS) or "error" in lower or "failed" in lower


def _latest_ffmpeg_stats_line(text: str, *, lookback: int = 12) -> str:
    lines = [line.strip() for line in str(text or "").splitlines() if line.strip()]
    if not lines:
        return ""
    tail = lines[-lookback:]
    stats_idx = -1
    stats_line = ""
    for idx in range(len(tail) - 1, -1, -1):
        line = tail[idx]
        lower = line.lower()
        if "frame=" in lower and "speed=" in lower:
            stats_idx = idx
            stats_line = line
            break
    if stats_idx < 0:
        return ""
    if any(_line_implies_relay_idle(line) for line in tail[stats_idx + 1 :]):
        return ""
    return stats_line


def parse_ffmpeg_metrics(*texts: str) -> dict[str, Any]:
    raw = "\n".join(str(item or "") for item in texts)
    tail = raw[-16000:]
    metrics: dict[str, Any] = {}
    stats_line = _latest_ffmpeg_stats_line(tail)
    if stats_line:
        for match in _FFMPEG_KV_RE.finditer(stats_line):
            key = match.group("key").lower()
            value = match.group("value")
            if key in {"frame", "drop", "dup"}:
                metrics[key] = _as_int(value)
            elif key in {"fps", "speed", "bitrate"}:
                metrics[key] = _as_float(value)
        time_match = _TIME_RE.search(stats_line)
        if time_match:
            metrics["encoded_time"] = time_match.group("value")
    lower = tail.lower()
    metrics["error_count"] = sum(
        lower.count(token)
        for token in (
            "error",
            "failed",
            "connection refused",
            "timed out",
            "broken pipe",
            "invalid data",
        )
    )
    metrics["reconnect_count"] = sum(lower.count(token) for token in ("reconnect", "retry", "connection reset"))
    return metrics


def _count_urls(text: str) -> int:
    return len(_URL_RE.findall(str(text or "")))


def _ini_has_key(text: str) -> bool:
    return bool(re.search(r"^\s*key\s*=\s*\S+", str(text or ""), re.IGNORECASE | re.MULTILINE))


def _ini_has_rtmps(text: str) -> bool:
    lower = str(text or "").lower()
    return any(token in lower for token in ("rtmps = 是", "rtmps=true", "rtmps = true", "rtmps=1"))


def _source_hint(url_lines: str, recent_log: str) -> dict[str, str]:
    combined = f"{url_lines}\n{recent_log}"
    urls = _URL_RE.findall(combined)
    source_url = urls[0] if urls else ""
    room = _room_from_text(source_url) if source_url else ""
    return {"source_url": source_url, "source_room_id": room}


def _room_from_text(text: str) -> str:
    m = _ROOM_RE.search(text or "")
    return m.group(1) if m else ""


def _lane_score(row: dict[str, Any], metrics: dict[str, Any], cfg: dict[str, str]) -> tuple[int, list[dict[str, str]]]:
    score = 100
    issues: list[dict[str, str]] = []

    if not row.get("ffmpeg_active") and not row.get("is_pushing"):
        score -= 35
        issues.append({"level": "critical", "text": "FFmpeg 未检测到推流进程"})
    if not row.get("is_pushing"):
        score -= 25
        issues.append({"level": "critical", "text": "当前没有稳定推流输出"})

    speed = metrics.get("speed") or row.get("ffmpeg_speed")
    if isinstance(speed, (int, float)) and speed < 0.98:
        score -= 18
        issues.append({"level": "warning", "text": f"编码速度 {speed:.2f}x，低于实时要求"})
    if row.get("ffmpeg_speed_low"):
        score -= 10
        issues.append({"level": "warning", "text": "近期推流速度持续偏低"})
    if row.get("ffmpeg_progress_stalled"):
        score -= 15
        issues.append({"level": "warning", "text": "FFmpeg 进度停滞"})

    biz = str(row.get("business_status") or "")
    if biz == "youtube_key_rejected":
        score -= 40
        issues.append({"level": "critical", "text": "YouTube 串流密钥被拒绝"})
    elif biz == "misconfigured":
        score -= 30
        issues.append({"level": "critical", "text": "串流密钥或配置不完整"})

    if int(metrics.get("error_count") or 0) > 0:
        score -= min(30, int(metrics.get("error_count") or 0) * 6)
        issues.append({"level": "warning", "text": "日志出现 FFmpeg/网络错误"})

    url_count = _count_urls(cfg.get("url_lines", ""))
    has_key = _ini_has_key(cfg.get("youtube_ini_text", ""))
    if not has_key:
        score = max(0, score - 30)
        issues.append({"level": "critical", "text": "YouTube 推流 key 未配置"})
    if url_count == 0:
        score = max(0, score - 15)
        issues.append({"level": "warning", "text": "没有配置转播源 URL"})
    if has_key and not _ini_has_rtmps(cfg.get("youtube_ini_text", "")):
        score = max(0, score - 8)
        issues.append({"level": "warning", "text": "建议使用 RTMPS 推流"})

    watch = str(cfg.get("watch_url") or "").strip()
    if not watch:
        score = max(0, score - 6)
        issues.append({"level": "info", "text": "未配置 YouTube 目标监听地址（弹幕/观众数）"})

    if not issues and row.get("is_pushing"):
        issues.append({"level": "ok", "text": "推流链路正常"})
    return max(0, min(100, score)), issues[:6]


def build_youtube_quality_guard_report(
    rows: list[dict[str, Any]],
    *,
    live_summary: dict[str, Any] | None = None,
    config_by_process: dict[str, dict[str, str]] | None = None,
) -> dict[str, Any]:
    live_summary = live_summary or {}
    config_by_process = config_by_process or {}
    lanes: list[dict[str, Any]] = []
    recommendations: list[dict[str, str]] = []

    for row in rows:
        process = str(row.get("process") or row.get("pm2") or "")
        pushing = bool(row.get("is_pushing"))
        metrics = (
            parse_ffmpeg_metrics(str(row.get("recent_log") or ""), str(row.get("recent_error") or ""))
            if pushing
            else {}
        )
        cfg = config_by_process.get(process) or {}
        source_meta = extract_douyin_source_meta(cfg.get("url_lines", ""), str(row.get("recent_log") or ""))
        source_title = str(source_meta.get("source_title") or row.get("source_title") or "").strip()
        source_room_id = str(source_meta.get("source_room_id") or row.get("source_room_id") or "").strip()
        source_url = str(source_meta.get("source_url") or row.get("source_url") or "").strip()
        source_display = str(
            source_meta.get("source_display")
            or row.get("source_display")
            or source_title
            or (f"抖音直播间 {source_room_id}" if source_room_id else "")
        ).strip()
        score, issues = _lane_score(row, metrics, cfg)
        rtmps = _ini_has_rtmps(cfg.get("youtube_ini_text", ""))
        lanes.append(
            {
                "process": process,
                "label": row.get("label") or process,
                "source_title": source_title,
                "source_room_id": source_room_id,
                "source_url": source_url,
                "source_display": source_display,
                "status": row.get("business_status") or row.get("process_status") or "",
                "is_pushing": pushing,
                "ffmpeg_active": bool(row.get("ffmpeg_active") or row.get("ffmpeg_pids")),
                "score": score,
                "metrics": metrics,
                "rtmps": rtmps,
                "has_stream_key": _ini_has_key(cfg.get("youtube_ini_text", "")),
                "issues": issues,
            }
        )

    pushing_count = sum(1 for lane in lanes if lane["is_pushing"])
    active_events = int(
        ((live_summary.get("stats") or {}).get("active_events") or len(live_summary.get("active") or [])) or 0
    )
    sample_count = int(((live_summary.get("stats") or {}).get("sample_count") or 0) or 0)
    viewer_peak = int(((live_summary.get("stats") or {}).get("viewer_peak") or 0) or 0)
    score = int(sum(lane["score"] for lane in lanes) / len(lanes)) if lanes else 0

    if pushing_count == 0:
        recommendations.append({"level": "critical", "text": "没有正在推流的线路"})
    if active_events == 0 and pushing_count > 0:
        recommendations.append({"level": "warning", "text": "推流已存在但事件中心未形成活动事件"})
    if sample_count == 0 and pushing_count > 0:
        recommendations.append({"level": "info", "text": "尚无 Studio/浏览器采样，可安装 ingest 扩展增强观测"})
    elif all(lane["score"] >= 80 for lane in lanes if lane["is_pushing"]):
        recommendations.append({"level": "ok", "text": "推流质量良好，保持监控即可"})
    if any(lane["status"] == "relay_slow" for lane in lanes):
        recommendations.append({"level": "warning", "text": "存在推流速度偏低的线路，检查 CPU/GPU 与网络"})

    return {
        "generated_at": time.time(),
        "score": max(0, min(100, score)),
        "summary": {
            "lanes": len(lanes),
            "pushing_lanes": pushing_count,
            "active_events": active_events,
            "sample_count": sample_count,
            "viewer_peak": viewer_peak,
        },
        "lanes": lanes,
        "recommendations": recommendations[:8],
    }