content-find-agent-new/content_agent/integrations/platform_video_url.py

"""Platform video URL candidate selection.

Keeps full temporary URLs out of runtime diagnostics while preserving enough
field-path/probe evidence to debug no-play-url cases.
"""

from __future__ import annotations

import re
from collections import Counter
from dataclasses import dataclass
from typing import Any, Callable
from urllib.parse import urlparse

ProbeFn = Callable[[str, str], dict[str, Any]]
MAX_VIDEO_URL_CANDIDATES = 3

VIDEO_HINT_RE = re.compile(r"(video|play|mp4|m3u8|media|download|src)", re.I)
IMAGE_HINT_RE = re.compile(r"(image|cover|poster|thumbnail|thumb|heif|jpg|jpeg|png|kvif)", re.I)
AVATAR_HINT_RE = re.compile(r"(avatar|qlogo|uhead)", re.I)
AUDIO_HINT_RE = re.compile(r"(audio|bgm|music|m4a|mp3)", re.I)
PAGE_HINT_RE = re.compile(r"(content_link|share_url|page_url|account_link|link)$", re.I)
AD_HINT_RE = re.compile(
    r"(^|[._/\-\[\]&?=#])(?:ad|ads|advert|commercial|promotion|material|marketing|营销|广告|推广)($|[._/\-\[\]&?=#])",
    re.I,
)
VIDEO_EXT_RE = re.compile(r"\.(mp4|m3u8)(?:$|[?&#])", re.I)


@dataclass
class VideoUrlCandidate:
    path: str
    url: str
    host: str
    kind: str
    reject_reason: str = ""
    http_status: int | None = None
    content_type: str = ""
    content_range: str = ""
    range_bytes: int = 0
    looks_like_video: bool = False
    probe_status: str = "not_probed"


def select_video_url(
    platform: str,
    sources: list[tuple[str, dict[str, Any]]],
    *,
    probe_fn: ProbeFn | None = None,
) -> dict[str, Any]:
    candidates = _unique_candidates(
        candidate
        for source_name, payload in sources
        for candidate in find_url_candidates(payload, prefix=f"$.{source_name}")
    )
    for candidate in candidates:
        if candidate.kind != "video":
            continue
        if probe_fn is None:
            continue
        try:
            probe = probe_fn(candidate.url, platform)
        except Exception as exc:  # noqa: BLE001 - diagnostics should survive upstream quirks.
            probe = {"probe_status": "failed", "reject_reason": f"probe_failed:{type(exc).__name__}"}
        _apply_probe(candidate, probe)

    selected = _select_best(candidates, platform)
    result = {
        "play_url": selected.url if selected else None,
        "media_failure_reason": None if selected else "no_valid_play_url",
        "platform_raw_payload": build_diagnostic_payload(candidates, selected),
        "video_url_candidates": serialize_video_url_candidates(
            candidates,
            selected,
            platform,
            limit=MAX_VIDEO_URL_CANDIDATES,
        ),
    }
    if result["media_failure_reason"]:
        result["platform_raw_payload"]["no_valid_play_url"] = True
    return result


def serialize_video_url_candidates(
    candidates: list[VideoUrlCandidate],
    selected: VideoUrlCandidate | None,
    platform: str,
    *,
    limit: int = MAX_VIDEO_URL_CANDIDATES,
) -> list[dict[str, Any]]:
    ordered = _ordered_video_candidates(candidates, selected, platform)[: max(0, limit)]
    result: list[dict[str, Any]] = []
    for index, candidate in enumerate(ordered):
        result.append(
            _compact_candidate(
                {
                    "url": candidate.url,
                    "host": candidate.host,
                    "path": candidate.path,
                    "source": _source_from_path(candidate.path),
                    "probe_status": candidate.probe_status,
                    "content_type": candidate.content_type,
                    "content_range": candidate.content_range,
                    "selected": bool(selected and candidate.url == selected.url),
                    "candidate_index": index,
                }
            )
        )
    return result


def find_url_candidates(value: Any, *, prefix: str = "$") -> list[VideoUrlCandidate]:
    candidates: list[VideoUrlCandidate] = []
    if isinstance(value, dict):
        for key, item in value.items():
            candidates.extend(find_url_candidates(item, prefix=f"{prefix}.{key}"))
    elif isinstance(value, list):
        for index, item in enumerate(value):
            candidates.extend(find_url_candidates(item, prefix=f"{prefix}[{index}]"))
    elif isinstance(value, str) and value.startswith(("http://", "https://")):
        kind, reject = classify_url_candidate(prefix, value)
        candidates.append(
            VideoUrlCandidate(
                path=prefix,
                url=value,
                host=urlparse(value).netloc,
                kind=kind,
                reject_reason=reject,
            )
        )
    return candidates


def classify_url_candidate(path: str, url: str) -> tuple[str, str]:
    haystack = f"{path} {url}".lower()
    if AD_HINT_RE.search(haystack):
        return "ad_or_material", "ad_or_material_path"
    if AUDIO_HINT_RE.search(haystack):
        return "audio", "audio_or_bgm_path"
    if AVATAR_HINT_RE.search(haystack):
        return "avatar", "avatar_path"
    if IMAGE_HINT_RE.search(haystack):
        return "image", "image_path"
    if PAGE_HINT_RE.search(path):
        return "page", "page_link_path"
    if VIDEO_EXT_RE.search(url) or VIDEO_HINT_RE.search(path):
        return "video", ""
    return "unknown", "weak_video_signal"


def probe_url_with_httpx(
    url: str,
    platform: str,
    *,
    http_client: Any,
    bytes_to_probe: int = 1024 * 1024,
    timeout_seconds: float = 30.0,
) -> dict[str, Any]:
    get = getattr(http_client, "get", None)
    if not callable(get):
        return {"probe_status": "failed", "reject_reason": "probe_client_missing_get"}
    headers = _download_headers(platform)
    headers["Range"] = f"bytes=0-{max(bytes_to_probe - 1, 0)}"
    response = get(url, headers=headers, follow_redirects=True, timeout=timeout_seconds)
    content = response.content or b""
    content_type = response.headers.get("content-type", "")
    return {
        "http_status": response.status_code,
        "content_type": content_type,
        "content_range": response.headers.get("content-range", ""),
        "range_bytes": len(content),
        "looks_like_video": _looks_like_video(content, content_type),
        "probe_status": "verified"
        if response.status_code in {200, 206} and _looks_like_video(content, content_type)
        else "failed",
    }


def build_diagnostic_payload(
    candidates: list[VideoUrlCandidate],
    selected: VideoUrlCandidate | None,
) -> dict[str, Any]:
    reject_reasons = Counter(
        candidate.reject_reason or candidate.probe_status
        for candidate in candidates
        if candidate.kind != "video" or candidate.probe_status == "failed" or candidate.reject_reason
    )
    payload: dict[str, Any] = {
        "selected_video_url_path": selected.path if selected else None,
        "selected_video_url_host": selected.host if selected else None,
        "selected_video_url_probe_status": selected.probe_status if selected else None,
        "selected_video_url_content_type": selected.content_type if selected else None,
        "selected_video_url_content_range": selected.content_range if selected else None,
        "video_url_candidate_counts": dict(Counter(candidate.kind for candidate in candidates)),
        "video_url_reject_reasons": dict(reject_reasons),
    }
    return {key: value for key, value in payload.items() if value not in (None, "", {}, [])}


def _select_best(candidates: list[VideoUrlCandidate], platform: str) -> VideoUrlCandidate | None:
    video_candidates = [candidate for candidate in candidates if candidate.kind == "video"]
    verified = [
        candidate
        for candidate in video_candidates
        if candidate.probe_status == "verified"
        and candidate.http_status in {200, 206}
        and (candidate.looks_like_video or "video" in candidate.content_type.lower())
    ]
    if verified:
        return sorted(verified, key=lambda candidate: _selection_score(candidate, platform, fallback=False))[0]
    if not video_candidates:
        return None
    selected = sorted(video_candidates, key=lambda candidate: _selection_score(candidate, platform, fallback=True))[0]
    selected.probe_status = "failed_fallback" if selected.probe_status != "not_probed" else "not_probed_fallback"
    return selected


def _ordered_video_candidates(
    candidates: list[VideoUrlCandidate],
    selected: VideoUrlCandidate | None,
    platform: str,
) -> list[VideoUrlCandidate]:
    video_candidates = [candidate for candidate in candidates if candidate.kind == "video"]
    ordered: list[VideoUrlCandidate] = []
    seen: set[str] = set()
    if selected:
        ordered.append(selected)
        seen.add(selected.url)
    for candidate in sorted(
        video_candidates,
        key=lambda item: _selection_score(
            item,
            platform,
            fallback=item.probe_status not in {"verified"},
        ),
    ):
        if candidate.url in seen:
            continue
        ordered.append(candidate)
        seen.add(candidate.url)
    return ordered


def _selection_score(candidate: VideoUrlCandidate, platform: str, *, fallback: bool) -> tuple[Any, ...]:
    detail_bonus = 0 if platform == "kuaishou" and ".detail." in candidate.path else 1
    preferred_host = 0
    if platform == "shipinhao":
        preferred_host = 0 if "findermp.video.qq.com" in candidate.host else 1
    elif platform == "kuaishou":
        preferred_host = 0 if ("kwai" in candidate.host or "kwimgs" in candidate.host) else 1
    elif platform == "douyin":
        host = candidate.host.lower()
        if "v11-weba.douyinvod.com" in host:
            preferred_host = 3
        elif "douyinvod.com" in host:
            preferred_host = 0
        else:
            preferred_host = 1
    return (
        detail_bonus,
        preferred_host,
        fallback and candidate.probe_status == "failed",
        candidate.http_status not in {200, 206},
        not candidate.looks_like_video,
        candidate.path.count("."),
        candidate.path,
    )


def _apply_probe(candidate: VideoUrlCandidate, probe: dict[str, Any]) -> None:
    candidate.http_status = _int_or_none(probe.get("http_status"))
    candidate.content_type = str(probe.get("content_type") or "")
    candidate.content_range = str(probe.get("content_range") or "")
    candidate.range_bytes = int(probe.get("range_bytes") or 0)
    candidate.looks_like_video = bool(probe.get("looks_like_video"))
    candidate.probe_status = str(probe.get("probe_status") or "failed")


def _unique_candidates(candidates: Any) -> list[VideoUrlCandidate]:
    seen: set[str] = set()
    result: list[VideoUrlCandidate] = []
    for candidate in candidates:
        if candidate.url in seen:
            continue
        seen.add(candidate.url)
        result.append(candidate)
    return result


def _download_headers(platform: str) -> dict[str, str]:
    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/126 Safari/537.36",
        "Accept": "*/*",
    }
    if platform == "shipinhao":
        headers["Referer"] = "https://channels.weixin.qq.com/"
    elif platform == "kuaishou":
        headers["Referer"] = "https://www.kuaishou.com/"
    elif platform == "douyin":
        headers["Referer"] = "https://www.douyin.com/"
    return headers


def _looks_like_video(content: bytes, content_type: str) -> bool:
    head = content[:512]
    return "video" in content_type.lower() or b"ftyp" in head or b"moov" in head or b"mdat" in head


def _int_or_none(value: Any) -> int | None:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _source_from_path(path: str) -> str:
    if not path.startswith("$."):
        return ""
    remainder = path[2:]
    for separator in (".", "["):
        if separator in remainder:
            return remainder.split(separator, 1)[0]
    return remainder


def _compact_candidate(candidate: dict[str, Any]) -> dict[str, Any]:
    return {key: value for key, value in candidate.items() if value not in (None, "", {}, [])}