|
|
@@ -0,0 +1,862 @@
|
|
|
+"""快手/视频号 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()
|
Sam Lee