content-find-agent-new/tests/test_platform_video_url_candidates.py

"""快手/视频号 URL 解包诊断工具。

pytest 默认只跑离线单测。真实探测手动执行:

    .venv/bin/python tests/test_platform_video_url_candidates.py --real-probe --probe-oss

真实响应和完整临时 URL 只写入 data/platform_url_probe/，该目录不提交。
提交到文档的内容应使用脚本生成的 *_sanitized_report.md。
"""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import time
from collections import Counter, defaultdict
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
from urllib.parse import urljoin, urlparse

import httpx


KUAISHOU_RUN_ID = "v1_run_94448252f5cd"
SHIPINHAO_RUN_ID = "v1_run_35f4f7ccd1cc"
DEFAULT_QUERIES = [
    "怀旧经典歌曲",
    "歌词唱词",
    "歌词同步滚动呈现",
    "怀旧金曲",
    "歌词",
    "经典老歌",
    "世界杯主题曲",
    "音乐盘点",
    "热门歌曲",
    "视频剪辑歌词",
]
VIDEO_HINT_RE = re.compile(r"(video|play|mp4|m3u8|media|download|src)", re.I)
IMAGE_HINT_RE = re.compile(r"(image|cover|avatar|poster|thumbnail|thumb|qlogo|heif|jpg|jpeg|png|kvif)", 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|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 UrlCandidate:
    path: str
    url: str
    host: str
    url_kind: str
    reject_hint: str = ""
    http_status: int | None = None
    content_type: str = ""
    content_length: int | None = None
    content_range: str = ""
    range_bytes: int = 0
    looks_like_video: bool = False
    ffprobe_ok: bool = False
    duration_seconds: float | None = None
    width: int | None = None
    height: int | None = None
    selected: bool = False
    oss_status: str = ""
    oss_failure_type: str = ""
    oss_url_present: bool = False
    oss_elapsed_seconds: float | None = None
    oss_response_shape: str = ""


def find_url_candidates(value: Any, *, prefix: str = "$") -> list[UrlCandidate]:
    candidates: list[UrlCandidate] = []
    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(
            UrlCandidate(
                path=prefix,
                url=value,
                host=urlparse(value).netloc,
                url_kind=kind,
                reject_hint=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 IMAGE_HINT_RE.search(haystack):
        return "image", "image_or_avatar_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_candidate", ""
    return "unknown", "weak_video_signal"


def probe_candidate(
    candidate: UrlCandidate,
    *,
    platform: str,
    client: httpx.Client,
    bytes_to_probe: int,
    timeout_seconds: float,
) -> None:
    headers = download_headers(platform)
    headers["Range"] = f"bytes=0-{max(bytes_to_probe - 1, 0)}"
    try:
        response = client.get(candidate.url, headers=headers, follow_redirects=True, timeout=timeout_seconds)
        candidate.http_status = response.status_code
        candidate.content_type = response.headers.get("content-type", "")
        candidate.content_length = _int_or_none(response.headers.get("content-length"))
        candidate.content_range = response.headers.get("content-range", "")
        candidate.range_bytes = len(response.content or b"")
        candidate.looks_like_video = looks_like_video(response.content, candidate.content_type)
        probe_ffprobe(candidate, response.content)
    except Exception as exc:  # noqa: BLE001 - diagnostic output should retain exception type.
        candidate.reject_hint = candidate.reject_hint or f"http_probe_failed:{type(exc).__name__}"


def select_video_url(candidates: list[UrlCandidate], platform: str) -> UrlCandidate | None:
    viable = [
        item
        for item in candidates
        if item.url_kind == "video_candidate"
        and not item.reject_hint
        and item.http_status in {200, 206}
        and (item.looks_like_video or "video" in item.content_type.lower() or item.ffprobe_ok)
    ]
    if not viable:
        viable = [
            item
            for item in candidates
            if item.url_kind == "video_candidate" and not item.reject_hint
        ]
    if not viable:
        return None
    return sorted(viable, key=lambda item: selection_score(item, platform))[0]


def selection_score(candidate: UrlCandidate, platform: str) -> tuple[Any, ...]:
    path = candidate.path
    detail_bonus = 0 if "detail" in path else 1
    preferred_host = 0
    if platform == "shipinhao":
        preferred_host = 0 if "findermp.video.qq.com" in candidate.host else 1
    if platform == "kuaishou":
        preferred_host = 0 if "kwaicdn" in candidate.host or "kwai" in candidate.host else 1
    return (
        detail_bonus,
        preferred_host,
        candidate.http_status not in {200, 206},
        not candidate.looks_like_video,
        not candidate.ffprobe_ok,
        path.count("."),
        path,
    )


def run_historical_audit(runtime_root: Path) -> dict[str, Any]:
    return {
        "kuaishou": audit_run(runtime_root / KUAISHOU_RUN_ID),
        "shipinhao": audit_run(runtime_root / SHIPINHAO_RUN_ID),
    }


def audit_run(run_dir: Path) -> dict[str, Any]:
    content = _read_jsonl(run_dir / "discovered_content_items.jsonl")
    media = _read_jsonl(run_dir / "content_media_records.jsonl")
    evidence = _read_jsonl(run_dir / "pattern_recall_evidence.jsonl")
    decisions = _read_jsonl(run_dir / "rule_decisions.jsonl")
    by_content_id = {row.get("content_discovery_id"): row for row in content}
    by_platform_id = {row.get("platform_content_id"): row for row in content}
    media_by_platform_id = {row.get("platform_content_id"): row for row in media}
    evidence_by_content_id = {row.get("content_discovery_id"): row for row in evidence}
    rows = []
    for decision in decisions:
        target = decision.get("decision_target_id")
        item = by_content_id.get(target) or by_platform_id.get(target) or {}
        platform_id = item.get("platform_content_id") or target
        media_row = media_by_platform_id.get(platform_id, {})
        evidence_row = evidence_by_content_id.get(item.get("content_discovery_id"), {})
        evidence_summary = evidence_row.get("evidence_summary") or {}
        evidence_raw = evidence_row.get("raw_payload") or {}
        timing = evidence_summary.get("timing_metrics") or evidence_raw.get("timing_metrics") or {}
        video_fetch = timing.get("video_fetch") or {}
        gemini_request = timing.get("gemini_request") or {}
        media_raw = media_row.get("raw_payload") or {}
        platform_raw = item.get("platform_raw_payload") or {}
        rows.append(
            {
                "platform_content_id": platform_id,
                "search_query_id": item.get("search_query_id"),
                "decision_action": decision.get("decision_action"),
                "play_url_host": urlparse(str(media_row.get("play_url") or "")).netloc,
                "play_url_path_found_in_platform_raw": path_for_url(platform_raw, media_row.get("play_url")),
                "platform_raw_key_count": len(platform_raw),
                "has_full_platform_raw": has_full_raw_item(platform_raw),
                "content_media_status": media_row.get("content_media_status"),
                "oss_url_present": bool(media_row.get("oss_url")),
                "oss_archive_last_error": media_raw.get("oss_archive_last_error") or media_raw.get("failure_reason"),
                "gemini_video_source": video_fetch.get("gemini_video_source"),
                "download_seconds": _ms_to_seconds(video_fetch.get("download_duration_ms")),
                "ffmpeg_seconds": _ms_to_seconds(video_fetch.get("ffmpeg_duration_ms")),
                "gemini_seconds": _ms_to_seconds(gemini_request.get("total_duration_ms")),
                "failure_type": evidence_summary.get("failure_type") or evidence_raw.get("failure_type"),
            }
        )
    return {
        "run_id": run_dir.name,
        "rows": rows,
        "counts": {
            "decision_action": dict(Counter(row["decision_action"] for row in rows)),
            "media_status": dict(Counter(row["content_media_status"] for row in rows)),
            "oss_error": dict(Counter(row["oss_archive_last_error"] for row in rows if row["oss_archive_last_error"])),
            "play_url_host": dict(Counter(row["play_url_host"] for row in rows if row["play_url_host"])),
        },
    }


def inspect_real_platforms(
    *,
    queries: list[str],
    output_dir: Path,
    items_per_query: int,
    bytes_to_probe: int,
    kuaishou_detail_limit_per_query: int,
    probe_oss_enabled: bool,
    oss_limit_per_platform: int,
    oss_timeout_seconds: float,
    oss_send_referer: bool,
) -> dict[str, Any]:
    output_dir.mkdir(parents=True, exist_ok=True)
    result = {
        "created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
        "queries": queries,
        "platforms": {},
    }
    for platform in ["kuaishou", "shipinhao"]:
        print(f"[probe] platform={platform} start", flush=True)
        result["platforms"][platform] = inspect_platform(
            platform,
            queries,
            output_dir=output_dir,
            items_per_query=items_per_query,
            bytes_to_probe=bytes_to_probe,
            kuaishou_detail_limit_per_query=kuaishou_detail_limit_per_query,
            probe_oss_enabled=probe_oss_enabled,
            oss_limit=oss_limit_per_platform,
            oss_timeout_seconds=oss_timeout_seconds,
            oss_send_referer=oss_send_referer,
        )
        print(f"[probe] platform={platform} done", flush=True)
        time.sleep(15)
    return result


def inspect_platform(
    platform: str,
    queries: list[str],
    *,
    output_dir: Path,
    items_per_query: int,
    bytes_to_probe: int,
    kuaishou_detail_limit_per_query: int,
    probe_oss_enabled: bool,
    oss_limit: int,
    oss_timeout_seconds: float,
    oss_send_referer: bool,
) -> dict[str, Any]:
    run = {
        "platform": platform,
        "searches": [],
        "items": [],
        "selected": [],
        "summary": {},
    }
    seen_content_ids: set[str] = set()
    with httpx.Client() as client:
        for query_index, query in enumerate(queries, start=1):
            print(f"[probe] {platform} query={query_index}/{len(queries)} page=1 keyword={query}", flush=True)
            search_result = fetch_search(platform, query, cursor="")
            raw_search_path = output_dir / f"{platform}_q{query_index:02d}_p1_raw_search.json"
            raw_search_path.write_text(json.dumps(search_result, ensure_ascii=False, indent=2), encoding="utf-8")
            run["searches"].append(search_summary(query, 1, search_result))
            items = _items_from_search(search_result)
            print(f"[probe] {platform} query={query_index} page=1 items={len(items)}", flush=True)
            inspect_items(
                platform,
                query,
                1,
                items[:items_per_query],
                seen_content_ids,
                output_dir,
                run,
                client,
                bytes_to_probe,
                kuaishou_detail_limit_per_query,
            )
            if platform == "shipinhao":
                cursor = _next_cursor(search_result)
                if cursor:
                    time.sleep(15)
                    print(f"[probe] {platform} query={query_index}/{len(queries)} page=2 keyword={query}", flush=True)
                    page2 = fetch_search(platform, query, cursor=cursor)
                    (output_dir / f"{platform}_q{query_index:02d}_p2_raw_search.json").write_text(
                        json.dumps(page2, ensure_ascii=False, indent=2),
                        encoding="utf-8",
                    )
                    run["searches"].append(search_summary(query, 2, page2))
                    print(f"[probe] {platform} query={query_index} page=2 items={len(_items_from_search(page2))}", flush=True)
                    inspect_items(
                        platform,
                        query,
                        2,
                        _items_from_search(page2)[: max(1, items_per_query // 2)],
                        seen_content_ids,
                        output_dir,
                        run,
                        client,
                        bytes_to_probe,
                        kuaishou_detail_limit_per_query,
                    )
            if query_index < len(queries):
                time.sleep(15)
        selected = [item for item in run["items"] if item.get("selected")]
        for oss_index, item in enumerate(selected[:oss_limit], start=1):
            if probe_oss_enabled:
                print(f"[probe] {platform} oss={oss_index}/{min(len(selected), oss_limit)} content_id={item.get('content_id')}", flush=True)
                probe_selected_oss(item, platform, client, oss_timeout_seconds, send_referer=oss_send_referer)
            run["selected"].append(item)
    run["summary"] = summarize_platform(run)
    return run


def inspect_items(
    platform: str,
    query: str,
    page: int,
    items: list[dict[str, Any]],
    seen_content_ids: set[str],
    output_dir: Path,
    run: dict[str, Any],
    client: httpx.Client,
    bytes_to_probe: int,
    kuaishou_detail_limit_per_query: int,
) -> None:
    for index, item in enumerate(items, start=1):
        content_id = str(item.get("channel_content_id") or f"{query}_{page}_{index}")
        if content_id in seen_content_ids:
            continue
        seen_content_ids.add(content_id)
        raw_sources = [{"source": "search", "payload": item}]
        if platform == "kuaishou" and content_id and index <= kuaishou_detail_limit_per_query:
            time.sleep(15)
            detail = fetch_kuaishou_detail(content_id)
            (output_dir / f"{platform}_{content_id}_detail.json").write_text(
                json.dumps(detail, ensure_ascii=False, indent=2),
                encoding="utf-8",
            )
            raw_sources.append({"source": "detail", "payload": _detail_item(detail)})
        candidates = []
        for source in raw_sources:
            for candidate in find_url_candidates(source["payload"], prefix=f"$.{source['source']}"):
                if candidate.url_kind == "video_candidate":
                    probe_candidate(
                        candidate,
                        platform=platform,
                        client=client,
                        bytes_to_probe=bytes_to_probe,
                        timeout_seconds=30.0,
                    )
                candidates.append(candidate)
        candidates = unique_candidates(candidates)
        selected = select_video_url(candidates, platform)
        if selected:
            selected.selected = True
        run["items"].append(
            {
                "platform": platform,
                "query": query,
                "page": page,
                "content_id": content_id,
                "title": item.get("title") or item.get("body_text") or "",
                "content_type": item.get("content_type"),
                "raw_candidate_count": len(candidates),
                "video_candidate_count": sum(1 for candidate in candidates if candidate.url_kind == "video_candidate"),
                "selected": asdict(selected) if selected else None,
                "candidates": [asdict(candidate) for candidate in candidates],
            }
        )


def probe_selected_oss(
    item: dict[str, Any],
    platform: str,
    client: httpx.Client,
    timeout_seconds: float,
    *,
    send_referer: bool,
) -> None:
    selected = item.get("selected")
    if not selected:
        return
    payload = {
        "src_url": selected["url"],
        "src_type": "video",
        "use_proxy": True,
    }
    if send_referer:
        payload["referer"] = download_headers(platform)
    selected["oss_payload_mode"] = "with_referer_dict" if send_referer else "no_referer"
    endpoint = os.environ.get("CONTENT_AGENT_OSS_UPLOAD_URL") or _load_project_env().get("CONTENT_AGENT_OSS_UPLOAD_URL") or "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream"
    started = time.monotonic()
    try:
        response = client.post(endpoint, json=payload, timeout=timeout_seconds)
        elapsed = round(time.monotonic() - started, 3)
        body = response.json()
        oss_object = body.get("oss_object") if isinstance(body, dict) else None
        selected["oss_status"] = f"http_{response.status_code}"
        selected["oss_elapsed_seconds"] = elapsed
        selected["oss_response_shape"] = response_shape(body)
        selected["oss_url_present"] = bool(isinstance(oss_object, dict) and oss_object.get("cdn_url"))
        if not selected["oss_url_present"]:
            selected["oss_failure_type"] = "oss_upload_response_invalid"
    except Exception as exc:  # noqa: BLE001
        selected["oss_status"] = "exception"
        selected["oss_failure_type"] = type(exc).__name__
        selected["oss_elapsed_seconds"] = round(time.monotonic() - started, 3)


def sanitize_report_payload(historical: dict[str, Any], real_probe: dict[str, Any]) -> dict[str, Any]:
    return {
        "historical": historical,
        "real_probe": {
            "created_at": real_probe.get("created_at"),
            "queries": real_probe.get("queries"),
            "platforms": {
                platform: {
                    "summary": data.get("summary", {}),
                    "searches": data.get("searches", []),
                    "selected": [sanitize_item(item) for item in data.get("selected", [])],
                }
                for platform, data in (real_probe.get("platforms") or {}).items()
            },
        },
    }


def sanitize_item(item: dict[str, Any]) -> dict[str, Any]:
    selected = item.get("selected") or {}
    candidates = item.get("candidates") or []
    return {
        "platform": item.get("platform"),
        "query": item.get("query"),
        "page": item.get("page"),
        "content_id": item.get("content_id"),
        "title_sample": _short(item.get("title") or "", 80),
        "content_type": item.get("content_type"),
        "raw_candidate_count": item.get("raw_candidate_count"),
        "video_candidate_count": item.get("video_candidate_count"),
        "candidate_kind_counts": dict(Counter(candidate.get("url_kind") for candidate in candidates)),
        "candidate_hosts": dict(Counter(candidate.get("host") for candidate in candidates if candidate.get("host"))),
        "selected_path": selected.get("path"),
        "selected_host": selected.get("host"),
        "selected_http_status": selected.get("http_status"),
        "selected_content_type": selected.get("content_type"),
        "selected_content_range": selected.get("content_range"),
        "selected_looks_like_video": selected.get("looks_like_video"),
        "selected_ffprobe_ok": selected.get("ffprobe_ok"),
        "selected_duration_seconds": selected.get("duration_seconds"),
        "oss_status": selected.get("oss_status"),
        "oss_failure_type": selected.get("oss_failure_type"),
        "oss_url_present": selected.get("oss_url_present"),
        "oss_elapsed_seconds": selected.get("oss_elapsed_seconds"),
        "oss_response_shape": selected.get("oss_response_shape"),
        "oss_payload_mode": selected.get("oss_payload_mode"),
    }


def render_markdown_report(payload: dict[str, Any]) -> str:
    lines = [
        "# 快手/视频号 URL 解包与 OSS 批量验证报告",
        "",
        f"生成时间：{time.strftime('%Y-%m-%d %H:%M:%S')}",
        "",
        "## 1. 历史 run 对照",
        "",
    ]
    for platform, audit in payload["historical"].items():
        lines.extend([
            f"### {platform}",
            f"- run_id: `{audit['run_id']}`",
            f"- decision_action: `{audit['counts']['decision_action']}`",
            f"- media_status: `{audit['counts']['media_status']}`",
            f"- oss_error: `{audit['counts']['oss_error']}`",
            f"- play_url_host: `{audit['counts']['play_url_host']}`",
            "- 字段路径反查：当前历史 runtime 的 `platform_raw_payload` 只保留 content/account id，不能完整反查 URL 原始字段路径。",
            "",
        ])
    lines.extend(["## 2. 真实接口探测与 selected URL OSS 验证", ""])
    for platform, data in payload["real_probe"]["platforms"].items():
        summary = data["summary"]
        lines.extend([
            f"### {platform}",
            f"- search_count: `{summary.get('search_count')}`",
            f"- item_count: `{summary.get('item_count')}`",
            f"- selected_count: `{summary.get('selected_count')}`",
            f"- url_kind_counts: `{summary.get('url_kind_counts')}`",
            f"- selected_hosts: `{summary.get('selected_hosts')}`",
            f"- oss_counts: `{summary.get('oss_counts')}`",
            f"- oss_failure_counts: `{summary.get('oss_failure_counts')}`",
            "",
            "| content_id | query | selected_path | host | http | content_type | range | oss | failure |",
            "|---|---|---|---|---:|---|---|---|---|",
        ])
        for item in data.get("selected", [])[:30]:
            lines.append(
                "| {content_id} | {query} | `{path}` | `{host}` | {http} | `{ctype}` | `{range_}` | `{oss}` | `{failure}` |".format(
                    content_id=item.get("content_id") or "",
                    query=_short(item.get("query") or "", 20),
                    path=item.get("selected_path") or "",
                    host=item.get("selected_host") or "",
                    http=item.get("selected_http_status") or "",
                    ctype=item.get("selected_content_type") or "",
                    range_=item.get("selected_content_range") or "",
                    oss=item.get("oss_status") or "",
                    failure=item.get("oss_failure_type") or "",
                )
            )
        lines.append("")
    lines.extend([
        "## 3. 结论口径",
        "",
        "- 快手/视频号 URL 解包不能再把 `video_url_list[0]` 当作唯一事实，应先递归分类 URL 候选。",
        "- 广告/素材、图片/封面、头像、BGM/音频、页面链接必须从正片候选中排除；视频号也按同样规则检查广告 MP4。",
        "- 如果 selected URL 轻量验证通过但 OSS invalid，应先归因为 OSS 转存兼容/响应结构问题，而不是直接判定 URL 选错。",
        "- 历史 run 里的 raw payload 不足以完整追溯字段路径，后续生产运行应保存 URL 候选摘要到 raw_payload。",
        "",
    ])
    return "\n".join(lines)


def summarize_platform(run: dict[str, Any]) -> dict[str, Any]:
    items = run["items"]
    candidates = [candidate for item in items for candidate in item.get("candidates", [])]
    selected = [item.get("selected") for item in items if item.get("selected")]
    return {
        "search_count": len(run["searches"]),
        "item_count": len(items),
        "selected_count": len(selected),
        "url_kind_counts": dict(Counter(candidate.get("url_kind") for candidate in candidates)),
        "selected_hosts": dict(Counter(item.get("host") for item in selected if item.get("host"))),
        "selected_paths": dict(Counter(item.get("path") for item in selected if item.get("path"))),
        "oss_counts": dict(Counter(item.get("oss_status") for item in selected if item.get("oss_status"))),
        "oss_failure_counts": dict(Counter(item.get("oss_failure_type") for item in selected if item.get("oss_failure_type"))),
    }


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


def fetch_search(platform: str, query: str, *, cursor: str = "") -> dict[str, Any]:
    base_url = _required_env("CONTENTFIND_API_CRAWAPI_BASE_URL")
    path = "/crawler/kuai_shou/keyword_v2" if platform == "kuaishou" else "/crawler/shi_pin_hao/keyword"
    payload = {"keyword": query}
    if platform == "shipinhao":
        payload["cursor"] = cursor
    retry_events = []
    attempts = 3 if platform == "shipinhao" else 1
    with httpx.Client() as client:
        for attempt in range(1, attempts + 1):
            response = client.post(urljoin(base_url.rstrip("/") + "/", path.lstrip("/")), json=payload, timeout=60.0)
            response.raise_for_status()
            body = response.json()
            if platform == "shipinhao" and body.get("code") == 25011 and attempt < attempts:
                retry_events.append({"attempt": attempt, "code": body.get("code"), "msg": body.get("msg")})
                time.sleep(float(attempt))
                continue
            if retry_events:
                body["_retry_events"] = retry_events
            return body
    raise RuntimeError("unreachable fetch_search retry state")


def fetch_kuaishou_detail(content_id: str) -> dict[str, Any]:
    base_url = _required_env("CONTENTFIND_API_CRAWAPI_BASE_URL")
    with httpx.Client() as client:
        response = client.post(urljoin(base_url.rstrip("/") + "/", "crawler/kuai_shou/detail"), json={"content_id": content_id}, timeout=60.0)
        response.raise_for_status()
        return response.json()


def search_summary(query: str, page: int, response: dict[str, Any]) -> dict[str, Any]:
    block = response.get("data") if isinstance(response.get("data"), dict) else {}
    return {
        "query": query,
        "page": page,
        "code": response.get("code"),
        "msg": response.get("msg"),
        "item_count": len(block.get("data") if isinstance(block.get("data"), list) else []),
        "has_more": block.get("has_more"),
        "next_cursor_present": bool(block.get("next_cursor")),
        "retry_events": response.get("_retry_events") or [],
    }


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/"
    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 probe_ffprobe(candidate: UrlCandidate, content: bytes) -> None:
    if not content:
        return
    tmp = Path("data/platform_url_probe/.tmp_probe.mp4")
    tmp.parent.mkdir(parents=True, exist_ok=True)
    try:
        tmp.write_bytes(content)
        completed = subprocess.run(
            ["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,duration", "-of", "json", str(tmp)],
            check=False,
            capture_output=True,
            text=True,
            timeout=10,
        )
        if completed.returncode != 0:
            return
        data = json.loads(completed.stdout or "{}")
        stream = (data.get("streams") or [{}])[0]
        candidate.ffprobe_ok = True
        candidate.width = _int_or_none(stream.get("width"))
        candidate.height = _int_or_none(stream.get("height"))
        try:
            candidate.duration_seconds = round(float(stream.get("duration")), 3)
        except (TypeError, ValueError):
            candidate.duration_seconds = None
    except Exception:
        return
    finally:
        try:
            tmp.unlink()
        except OSError:
            pass


def path_for_url(payload: Any, url: Any, prefix: str = "$") -> str:
    if not url:
        return ""
    if isinstance(payload, dict):
        for key, value in payload.items():
            found = path_for_url(value, url, f"{prefix}.{key}")
            if found:
                return found
    if isinstance(payload, list):
        for index, value in enumerate(payload):
            found = path_for_url(value, url, f"{prefix}[{index}]")
            if found:
                return found
    if payload == url:
        return prefix
    return ""


def has_full_raw_item(payload: dict[str, Any]) -> bool:
    return any(key in payload for key in ["video_url_list", "image_url_list", "content_link", "bgm_data"])


def response_shape(value: Any) -> str:
    if not isinstance(value, dict):
        return type(value).__name__
    pieces = []
    for key in sorted(value):
        item = value[key]
        pieces.append(f"{key}({','.join(sorted(item))})" if isinstance(item, dict) else str(key))
    return " / ".join(pieces)


def _items_from_search(response: dict[str, Any]) -> list[dict[str, Any]]:
    block = response.get("data") if isinstance(response.get("data"), dict) else {}
    items = block.get("data")
    return items if isinstance(items, list) else []


def _next_cursor(response: dict[str, Any]) -> str:
    block = response.get("data") if isinstance(response.get("data"), dict) else {}
    return str(block.get("next_cursor") or "")


def _detail_item(response: dict[str, Any]) -> dict[str, Any]:
    block = response.get("data") if isinstance(response.get("data"), dict) else {}
    item = block.get("data", block) if isinstance(block, dict) else {}
    return item if isinstance(item, dict) else {}


def _read_jsonl(path: Path) -> list[dict[str, Any]]:
    if not path.exists():
        return []
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def _load_project_env(env_file: str | Path = ".env") -> dict[str, str]:
    path = Path(env_file)
    if not path.exists():
        return {}
    env: dict[str, str] = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith("#") or "=" not in stripped:
            continue
        key, value = stripped.split("=", 1)
        env[key.strip()] = value.strip().strip('"').strip("'")
    return env


def _required_env(key: str) -> str:
    value = os.environ.get(key) or _load_project_env().get(key)
    if not value:
        raise RuntimeError(f"missing required env: {key}")
    return value


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


def _ms_to_seconds(value: Any) -> float | None:
    try:
        return round(float(value) / 1000.0, 3)
    except (TypeError, ValueError):
        return None


def _short(value: str, limit: int) -> str:
    return value if len(value) <= limit else value[: limit - 1] + "…"


def test_find_url_candidates_classifies_video_image_audio_page_and_ad():
    payload = {
        "video_url_list": [{"video_url": "https://v.example/a.mp4?x=1"}],
        "image_url_list": [{"image_url": "https://img.example/a.jpg"}],
        "bgm_data": {"play_url": "https://audio.example/a.m4a"},
        "content_link": "https://page.example/item",
        "ads": {"video_url": "https://ad.example/ad.mp4"},
    }

    by_url = {item.url: item for item in find_url_candidates(payload)}

    assert by_url["https://v.example/a.mp4?x=1"].url_kind == "video_candidate"
    assert by_url["https://img.example/a.jpg"].url_kind == "image"
    assert by_url["https://audio.example/a.m4a"].url_kind == "audio"
    assert by_url["https://page.example/item"].url_kind == "page"
    assert by_url["https://ad.example/ad.mp4"].url_kind == "ad_or_material"


def test_select_video_url_ignores_ad_mp4_and_prefers_verified_video():
    ad = UrlCandidate("$.ads.video_url", "https://ad.example/ad.mp4", "ad.example", "ad_or_material", "ad_or_material_path", http_status=200, content_type="video/mp4", looks_like_video=True)
    video = UrlCandidate("$.search.video_url_list[0].video_url", "https://v.example/a.mp4", "v.example", "video_candidate", "", http_status=206, content_type="video/mp4", looks_like_video=True)

    assert select_video_url([ad, video], "kuaishou") is video


def test_select_video_url_prefers_kuaishou_detail_over_search():
    search = UrlCandidate("$.search.video_url_list[0].video_url", "https://v.kwaicdn.test/search.mp4", "v.kwaicdn.test", "video_candidate", "", http_status=206, content_type="video/mp4", looks_like_video=True)
    detail = UrlCandidate("$.detail.video_url_list[0].video_url", "https://v.kwaicdn.test/detail.mp4", "v.kwaicdn.test", "video_candidate", "", http_status=206, content_type="video/mp4", looks_like_video=True)

    assert select_video_url([search, detail], "kuaishou") is detail


def test_ad_classifier_does_not_match_download_or_head_substrings():
    kind, reject = classify_url_candidate(
        "$.search.video_url_list[0].video_url",
        "https://findermp.video.qq.com/251/20304/stodownload?head=1&token=abc",
    )

    assert kind == "video_candidate"
    assert reject == ""


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--real-probe", action="store_true")
    parser.add_argument("--probe-oss", action="store_true")
    parser.add_argument("--queries", default=",".join(DEFAULT_QUERIES))
    parser.add_argument("--items-per-query", type=int, default=4)
    parser.add_argument("--kuaishou-detail-limit-per-query", type=int, default=3)
    parser.add_argument("--oss-limit-per-platform", type=int, default=20)
    parser.add_argument("--oss-timeout-seconds", type=float, default=180.0)
    parser.add_argument("--oss-send-referer", action="store_true")
    parser.add_argument("--bytes", type=int, default=1024 * 1024)
    parser.add_argument("--output-dir", default="")
    args = parser.parse_args()

    timestamp = time.strftime("%Y%m%d_%H%M%S")
    output_dir = Path(args.output_dir or f"data/platform_url_probe/{timestamp}")
    historical = run_historical_audit(Path("runtime/v1"))
    if args.real_probe:
        queries = [item.strip() for item in args.queries.split(",") if item.strip()]
        real_probe = inspect_real_platforms(
            queries=queries,
            output_dir=output_dir,
            items_per_query=args.items_per_query,
            bytes_to_probe=args.bytes,
            kuaishou_detail_limit_per_query=args.kuaishou_detail_limit_per_query,
            probe_oss_enabled=args.probe_oss,
            oss_limit_per_platform=args.oss_limit_per_platform,
            oss_timeout_seconds=args.oss_timeout_seconds,
            oss_send_referer=args.oss_send_referer,
        )
    else:
        real_probe = {"created_at": time.strftime("%Y-%m-%d %H:%M:%S"), "queries": [], "platforms": {}}
    payload = sanitize_report_payload(historical, real_probe)
    output_dir.mkdir(parents=True, exist_ok=True)
    (output_dir / "historical_audit.json").write_text(json.dumps(historical, ensure_ascii=False, indent=2), encoding="utf-8")
    (output_dir / "sanitized_report.json").write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    report = render_markdown_report(payload)
    (output_dir / "sanitized_report.md").write_text(report, encoding="utf-8")
    print(report)


if __name__ == "__main__":
    main()