RaptDramaDump/server/services/playlist.py

import json
import re
from pathlib import Path
from urllib.parse import quote, urljoin

from server.services.device import DeviceProfile
from server.services.http import http
from server.services.rapt import API_BASE

CACHE_DIR = Path(__file__).resolve().parents[2] / "cache" / "playlists"


def safe_name(text: str, fallback: str) -> str:
    cleaned = re.sub(r'[<>:"/\\|?*\x00-\x1f]', " ", text or "")
    cleaned = re.sub(r"\s+", " ", cleaned).strip().rstrip(". ")
    return (cleaned[:80] or fallback)


def _items(data):
    if isinstance(data, list):
        return data, len(data)
    items = (data or {}).get("list") or []
    return items, (data or {}).get("count")


def _normalize(item: dict) -> dict | None:
    cid = item.get("cid") or item.get("id")
    if cid is None:
        return None
    idx = item.get("idx") or item.get("number") or 0
    title = str(item.get("msg") or item.get("title") or f"EP.{idx}").strip()
    return {
        "cid": int(cid),
        "idx": int(idx) if str(idx).isdigit() else 0,
        "title": title,
        "src": str(item.get("src") or "").strip(),
    }


def _request_playlist(token: str, device: DeviceProfile, vid: int) -> list[dict]:
    try:
        data = http.request_data("POST", API_BASE + "/app/video/playlist", token, device, {
            "vid": vid,
            "page": 1,
            "before": 1,
        })
        items, total = _items(data)
        episodes = [row for row in (_normalize(item) for item in items if isinstance(item, dict)) if row]
        if episodes and (total is None or len(episodes) >= int(total)):
            return episodes
    except RuntimeError:
        episodes = []
    collected = []
    total = None
    for page in range(1, 41):
        data = http.request_data("POST", API_BASE + "/app/video/playlistv2", token, device, {
            "vid": vid,
            "page": page,
            "before": 1 if page == 1 else 0,
        })
        items, page_total = _items(data)
        if page_total is not None:
            total = page_total
        if not items:
            break
        for item in items:
            if isinstance(item, dict):
                row = _normalize(item)
                if row:
                    collected.append(row)
        if total is not None and len(collected) >= int(total):
            break
    return collected or episodes


def _cache_path(vid: int) -> Path:
    return CACHE_DIR / f"{vid}.json"


def _read_cache(vid: int) -> list[dict]:
    path = _cache_path(vid)
    if not path.is_file():
        return []
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return []
    rows = data.get("episodes") if isinstance(data, dict) else None
    return [row for row in rows if isinstance(row, dict)] if isinstance(rows, list) else []


def _write_cache(vid: int, episodes: list[dict]) -> None:
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    payload = {
        "episodes": [
            {"cid": ep["cid"], "idx": ep["idx"], "title": ep["title"], "src": ep.get("src") or ""}
            for ep in episodes
        ]
    }
    path = _cache_path(vid)
    temporary = path.with_suffix(".json.tmp")
    temporary.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
    temporary.replace(path)


def _filled(episodes: list[dict]) -> int:
    return sum(1 for ep in episodes if ep.get("src"))


def merge_episodes(cached: list[dict], fresh: list[dict]) -> tuple[list[dict], bool]:
    by_cid: dict[int, dict] = {}
    for source in (cached, fresh):
        for ep in source:
            cid = int(ep["cid"])
            current = by_cid.get(cid)
            if current is None:
                by_cid[cid] = {
                    "cid": cid,
                    "idx": int(ep.get("idx") or 0),
                    "title": ep.get("title") or f"EP.{ep.get('idx') or cid}",
                    "src": ep.get("src") or "",
                }
                continue
            if ep.get("title"):
                current["title"] = ep["title"]
            if ep.get("idx"):
                current["idx"] = int(ep["idx"])
            if ep.get("src"):
                current["src"] = ep["src"]
    merged = sorted(by_cid.values(), key=lambda ep: (ep["idx"], ep["cid"]))
    return merged, _filled(fresh) >= _filled(cached) and _filled(fresh) > 0


def load_episodes(token: str, device: DeviceProfile | None, vid: int) -> list[dict]:
    cached = _read_cache(vid)
    fresh = []
    if token and device is not None:
        try:
            fresh = _request_playlist(token, device, vid)
        except RuntimeError:
            fresh = []
    merged, save = merge_episodes(cached, fresh)
    if save:
        _write_cache(vid, merged)
    elif not cached and merged:
        _write_cache(vid, merged)
    return merged


def expand_variant(master_url: str) -> tuple[str, str, list[dict]]:
    master_text = http.read_text(master_url)
    variant_rel = next(
        (line.strip() for line in master_text.splitlines() if line.strip() and not line.startswith("#")),
        "",
    )
    if not variant_rel:
        raise RuntimeError("播放列表没有视频地址")
    if _looks_like_segment(variant_rel):
        variant_url = master_url
        media_text = master_text
    else:
        variant_url = urljoin(master_url, variant_rel)
        media_text = http.read_text(variant_url)
    localized, items = localize_media(media_text, variant_url)
    if not items:
        raise RuntimeError("播放列表没有分片")
    return master_text, localized, items


def localize_media(text: str, base: str) -> tuple[str, list[dict]]:
    urls = _resource_urls(text, base)
    names: list[str] = []
    items: list[dict] = []
    used: set[str] = set()
    known: dict[str, str] = {}
    for index, url in enumerate(urls, start=1):
        name = known.get(url)
        if name is None:
            name = _resource_name(url, index, used)
            known[url] = name
            items.append({"name": name, "url": url})
        names.append(name)
    return _apply_local_names(text, names), items


def _resource_urls(text: str, base: str) -> list[str]:
    urls = []
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        if stripped.startswith("#"):
            urls.extend(urljoin(base, match) for match in re.findall(r'URI="([^"]*)"', stripped))
            continue
        urls.append(urljoin(base, stripped))
    return urls


def _resource_name(url: str, index: int, used: set[str]) -> str:
    raw = Path(url.split("?", 1)[0].rstrip("/")).name
    raw = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", raw).strip(" .") or f"part-{index}.ts"
    candidate = raw
    sequence = index
    while candidate in used:
        suffix = Path(raw).suffix
        stem = Path(raw).stem
        candidate = f"{stem}-{sequence}{suffix}"
        sequence += 1
    used.add(candidate)
    return candidate


def _apply_local_names(text: str, names: list[str]) -> str:
    index = 0

    def take() -> str:
        nonlocal index
        name = names[index] if index < len(names) else ""
        index += 1
        return name

    lines = []
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped:
            lines.append(line)
            continue
        if stripped.startswith("#"):
            lines.append(re.sub(r'URI="([^"]*)"', lambda _match: f'URI="{take()}"', line))
            continue
        lines.append(take() or stripped)
    return "\n".join(lines) + "\n"


def episode_src(token: str, device: DeviceProfile | None, vid: int, cid: int) -> str:
    for ep in load_episodes(token, device, vid):
        if int(ep.get("cid") or 0) == cid and ep.get("src"):
            return str(ep["src"])
    return ""


def playback_playlist(master_url: str) -> str:
    master_text = http.read_text(master_url)
    variant_rel = next(
        (line.strip() for line in master_text.splitlines() if line.strip() and not line.startswith("#")),
        "",
    )
    if not variant_rel:
        raise RuntimeError("播放列表没有视频地址")
    if _looks_like_segment(variant_rel):
        return _rewrite_playlist(master_text, master_url)
    variant_url = urljoin(master_url, variant_rel)
    return _rewrite_playlist(http.read_text(variant_url), variant_url)


def _looks_like_segment(url: str) -> bool:
    path = url.split("?", 1)[0].lower()
    return path.endswith((".ts", ".m4s", ".m4v", ".mp4", ".aac", ".vtt"))


def _proxy_url(url: str) -> str:
    return "/api/play/segment?url=" + quote(url, safe="")


def _rewrite_playlist(text: str, base: str) -> str:
    lines = []
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped:
            lines.append(line)
            continue
        if stripped.startswith("#"):
            lines.append(_rewrite_tag(line, base))
            continue
        lines.append(_proxy_url(urljoin(base, stripped)))
    return "\n".join(lines) + "\n"


def _rewrite_tag(line: str, base: str) -> str:
    def replace(match: re.Match) -> str:
        return 'URI="' + _proxy_url(urljoin(base, match.group(1))) + '"'

    return re.sub(r'URI="([^"]*)"', replace, line)