from __future__ import annotations

"""YouTube 单槽推流协调：与 youtube2 / douyin_youtube_ffplay 语义对齐。"""

import threading
import time
from typing import Any

from web2_youtube_relay_heartbeat import clear_relay_heartbeat, read_relay_heartbeat, write_relay_heartbeat

_slot_lock = threading.Lock()
_slot_owner: str | None = None
_CLAIM_GRACE_SECONDS = 15.0


def _heartbeat_claim_fresh(hb: dict[str, Any]) -> bool:
    if not hb:
        return False
    state = str(hb.get("state") or "").strip().lower()
    if state not in {"claiming", "starting", "streaming", "source_failover"}:
        return False
    updated = float(hb.get("updated_at") or 0)
    return updated > 0 and (time.time() - updated) <= _CLAIM_GRACE_SECONDS + 90.0


def prune_orphan_relay_recording(
    recording: set[str],
    recording_time_list: dict[str, Any],
    *,
    pm2: str,
    base_dir: str,
) -> None:
    """仅清理孤儿占槽：recording 非空且无有效 relay heartbeat。"""
    if not recording:
        return
    hb = read_relay_heartbeat(pm2, base_dir=base_dir)
    if _heartbeat_claim_fresh(hb):
        return
    global _slot_owner
    with _slot_lock:
        recording.clear()
        recording_time_list.clear()
        _slot_owner = None
    clear_relay_heartbeat(pm2, base_dir=base_dir)


def relay_slot_blocks(record_name: str, recording: set[str]) -> bool:
    return any(name for name in recording if name and name != record_name)


def claim_relay_slot(
    record_name: str,
    *,
    pm2: str,
    base_dir: str,
    anchor_name: str = "",
) -> bool:
    global _slot_owner
    with _slot_lock:
        if _slot_owner is not None and _slot_owner != record_name:
            return False
        _slot_owner = record_name
    write_relay_heartbeat(
        pm2,
        {
            "state": "claiming",
            "record_name": record_name,
            "anchor_name": anchor_name,
            "claimed_at": time.time(),
        },
        base_dir=base_dir,
    )
    return True


def release_relay_slot(record_name: str, *, pm2: str = "", base_dir: str = "") -> None:
    global _slot_owner
    with _slot_lock:
        if _slot_owner == record_name:
            _slot_owner = None
    if pm2 and base_dir:
        hb = read_relay_heartbeat(pm2, base_dir=base_dir)
        if str(hb.get("record_name") or "") == record_name:
            clear_relay_heartbeat(pm2, base_dir=base_dir)


def _flv_is_h265(url: str) -> bool:
    lower = (url or "").lower()
    if "codec=h265" in lower or "codec=hevc" in lower:
        return True
    try:
        from src import utils

        codec = utils.get_query_params(url, "codec")
        if codec and str(codec[0]).lower() in {"h265", "hevc"}:
            return True
    except Exception:
        pass
    return False


def youtube_stream_fallback_urls(port_info: dict[str, Any], primary: str) -> list[str]:
    primary_text = (primary or "").strip()
    out: list[str] = []
    seen: set[str] = set()
    for key in ("flv_url", "m3u8_url"):
        url = str(port_info.get(key) or "").strip()
        if not url or url == primary_text or url in seen:
            continue
        if not url.startswith("http"):
            continue
        if key == "flv_url" and _flv_is_h265(url):
            continue
        seen.add(url)
        out.append(url)
    return out