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

"""视频号(shipinhao)接入 client (V3-M1C).

复用 crawapi_http 共享基座(HTTP/限流/env)。search 对暂时性故障(25011/网络/
超时)按 platform_profiles/shipinhao.json 的口径重试(3 次、退避 1-2-4s),试满
抛 ContentAgentError 走既有失败通道。归一化输出与抖音同构(canonical 键集合一致)。
blogger/account_info 上游 blocked,fetch_author_works 返回 [] 不请求、不抛。
"""

from __future__ import annotations

import re
import time
from pathlib import Path
from typing import Any, Callable

from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations.crawapi_http import (
    CrawapiTransientError,
    RateLimiter,
    _env,
    _load_env_file,
    _optional_positive_int,
    content_format,
    post_crawapi_json,
    score_from_statistics,
    search_previous_discovery_step,
)
from content_agent.integrations import platform_video_url

SEARCH_RATE_LIMIT_BUCKET = "shipinhao_search"
TRANSIENT_BUSINESS_CODES = {"25011"}
SHIPINHAO_SEARCH_MIN_INTERVAL_SECONDS = 10.0
SHIPINHAO_SEARCH_MAX_INTERVAL_SECONDS = 12.0
_TAG_RE = re.compile(r"#([^\s#@（(]+)")


def _retry_transient(
    fn: Callable[[], Any],
    *,
    attempts: int,
    backoff_seconds: tuple[int, ...],
    sleep_fn: Callable[[float], None],
) -> Any:
    for attempt in range(attempts):
        try:
            return fn()
        except CrawapiTransientError:
            if attempt == attempts - 1:
                raise
            sleep_fn(backoff_seconds[min(attempt, len(backoff_seconds) - 1)])


def _normalize_shipinhao_item(
    query: dict[str, Any],
    item: dict[str, Any],
    index: int,
    has_more: bool,
    next_cursor: str,
    video_url_selection: dict[str, Any] | None = None,
) -> dict[str, Any]:
    video_url_selection = video_url_selection or platform_video_url.select_video_url(
        "shipinhao",
        [("search", item)],
        probe_fn=None,
    )
    title = item.get("title") or ""
    statistics = {
        "digg_count": int(item.get("like_count") or 0),
        "comment_count": int(item.get("comment_count") or 0),
        "share_count": int(item.get("share_count") or 0),
        "collect_count": int(item.get("collect_count") or 0),
        "play_count": int(item.get("play_count") or 0),
    }
    topic_list = item.get("topic_list") or []
    tags = [t if str(t).startswith("#") else f"#{t}" for t in topic_list if t]
    if not tags:
        tags = [f"#{m}" for m in _TAG_RE.findall(title)]
    platform_content_id = str(item.get("channel_content_id") or "")
    platform_author_id = str(item.get("channel_account_id") or "")
    publish_ms = item.get("publish_timestamp")
    platform_raw_payload = {
        "channel_content_id": platform_content_id,
        "channel_account_id": platform_author_id,
        **dict(video_url_selection.get("platform_raw_payload") or {}),
    }
    result = {
        "content_discovery_id": f"{query['search_query_id']}_content_{index:03d}",
        "search_query_id": query["search_query_id"],
        "platform": "shipinhao",
        "platform_content_id": platform_content_id,
        "platform_content_format": content_format(item.get("content_type") or "video"),
        "play_url": video_url_selection.get("play_url"),
        "description": title,
        "platform_author_id": platform_author_id,
        "author_display_name": item.get("channel_account_name") or "",
        "statistics": statistics,
        "tags": list(dict.fromkeys(tags)),
        "text_extra": [],
        "create_time": int(publish_ms) // 1000 if publish_ms else None,
        "has_more": has_more,
        "next_cursor": next_cursor,
        "score": score_from_statistics(statistics),
        "risk_level": "unknown",
        "discovery_relation": "derived_from_pattern_demand",
        "discovery_start_source": query["discovery_start_source"],
        "previous_discovery_step": search_previous_discovery_step(query),
        "content_metadata_source": "shipinhao_keyword_search",
        "platform_auth_mode": "no_bearer",
        "platform_raw_payload": platform_raw_payload,
    }
    if video_url_selection.get("media_failure_reason"):
        result["media_failure_reason"] = video_url_selection["media_failure_reason"]
    if video_url_selection.get("video_url_candidates"):
        result["video_url_candidates"] = video_url_selection["video_url_candidates"]
    return result


def _research_keyword(item: dict[str, Any], query: dict[str, Any]) -> str:
    title = " ".join(str(item.get("title") or "").split())
    if title:
        return title[:80]
    return str(query.get("search_query") or "")[:80]


def _strip_transient_fields(result: dict[str, Any]) -> dict[str, Any]:
    return {key: value for key, value in result.items() if not str(key).startswith("_platform_")}


class CrawapiShipinhaoClient:
    requires_progressive_search_rate_limit = True

    def __init__(
        self,
        base_url: str,
        keyword_path: str = "/crawler/shi_pin_hao/keyword",
        timeout_seconds: float = 60.0,
        max_results_per_query: int | None = 5,
        max_attempts: int = 3,
        backoff_seconds: tuple[int, ...] = (1, 2, 4),
        http_client: Any | None = None,
        rate_limiter: RateLimiter | None = None,
        sleep_fn: Callable[[float], None] = time.sleep,
        video_url_probe_fn: platform_video_url.ProbeFn | None = None,
    ) -> None:
        import httpx

        self.base_url = base_url.rstrip("/") + "/"
        self.keyword_path = keyword_path.lstrip("/")
        self.timeout_seconds = timeout_seconds
        self.max_results_per_query = max_results_per_query
        self.max_attempts = max_attempts
        self.backoff_seconds = backoff_seconds
        self.http_client = http_client or httpx.Client(timeout=timeout_seconds)
        self.rate_limiter = rate_limiter
        self.sleep_fn = sleep_fn
        self.video_url_probe_fn = video_url_probe_fn

    @classmethod
    def from_env(cls, env_path: str | Path = ".env") -> "CrawapiShipinhaoClient":
        env = _load_env_file(env_path)
        return cls(
            base_url=_env("CONTENTFIND_API_CRAWAPI_BASE_URL", env, required=True),
            timeout_seconds=float(
                _env("CONTENTFIND_API_CRAWAPI_TIMEOUT_SECONDS", env, default="180")
            ),
            max_results_per_query=_optional_positive_int(
                _env("CONTENTFIND_SHIPINHAO_MAX_RESULTS_PER_QUERY", env, default="5")
            ),
            rate_limiter=RateLimiter(
                min_interval_seconds=SHIPINHAO_SEARCH_MIN_INTERVAL_SECONDS,
                max_interval_seconds=SHIPINHAO_SEARCH_MAX_INTERVAL_SECONDS,
            ),
        )

    def search(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        return self._search(query, self.max_results_per_query)

    def search_full_page(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        return self._search(query, None)

    def search_full_page_metadata(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        return self._search(query, None, hydrate_video=False)

    def hydrate_search_result_media(
        self,
        result: dict[str, Any],
        query: dict[str, Any],
    ) -> dict[str, Any]:
        item = result.get("_platform_search_item")
        if not isinstance(item, dict):
            return _strip_transient_fields(result)
        selection = self._select_video_url_with_title_research(item, query)
        hydrated = _normalize_shipinhao_item(
            query,
            item,
            1,
            bool(result.get("has_more")),
            str(result.get("next_cursor") or ""),
            selection,
        )
        merged = {
            **_strip_transient_fields(result),
            "play_url": hydrated.get("play_url"),
            "platform_raw_payload": hydrated.get("platform_raw_payload", {}),
        }
        if hydrated.get("video_url_candidates"):
            merged["video_url_candidates"] = hydrated["video_url_candidates"]
        if selection.get("media_failure_reason"):
            merged["media_failure_reason"] = selection["media_failure_reason"]
        else:
            merged.pop("media_failure_reason", None)
        return merged

    def _search(
        self,
        query: dict[str, Any],
        max_results_per_query: int | None,
        *,
        hydrate_video: bool = True,
    ) -> list[dict[str, Any]]:
        data = self._keyword_search(
            {"keyword": query["search_query"], "cursor": str(query.get("page_cursor") or "")}
        )
        block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
        items = block.get("data", []) if isinstance(block.get("data"), list) else []
        has_more = bool(block.get("has_more", False))
        next_cursor = str(block.get("next_cursor") or "")
        selected = items[:max_results_per_query] if max_results_per_query else items
        results = []
        for index, item in enumerate(selected, start=1):
            selection = (
                self._select_video_url_with_title_research(item, query) if hydrate_video else None
            )
            normalized = _normalize_shipinhao_item(
                query,
                item,
                index,
                has_more,
                next_cursor,
                selection,
            )
            if not hydrate_video:
                normalized["_platform_search_item"] = item
            results.append(normalized)
        return results

    def _keyword_search(self, payload: dict[str, Any]) -> dict[str, Any]:
        def _call() -> dict[str, Any]:
            return post_crawapi_json(
                http_client=self.http_client,
                base_url=self.base_url,
                path=self.keyword_path,
                payload=payload,
                operation="keyword_search",
                timeout_seconds=self.timeout_seconds,
                rate_limiter=self.rate_limiter,
                rate_limit_bucket=SEARCH_RATE_LIMIT_BUCKET,
                business_codes=set(),
                transient_business_codes=TRANSIENT_BUSINESS_CODES,
            )

        try:
            return _retry_transient(
                _call,
                attempts=self.max_attempts,
                backoff_seconds=self.backoff_seconds,
                sleep_fn=self.sleep_fn,
            )
        except CrawapiTransientError as exc:
            raise ContentAgentError(
                ErrorCode.PLATFORM_REQUEST_FAILED,
                "shipinhao search exhausted after retries",
                {"operation": "keyword_search", "max_attempts": self.max_attempts},
            ) from exc

    def _select_video_url(self, sources: list[tuple[str, dict[str, Any]]]) -> dict[str, Any]:
        return platform_video_url.select_video_url(
            "shipinhao",
            sources,
            probe_fn=self.video_url_probe_fn or self._probe_video_url,
        )

    def _select_video_url_with_title_research(
        self,
        item: dict[str, Any],
        query: dict[str, Any],
    ) -> dict[str, Any]:
        search_selection = self._select_video_url([("search", item)])
        if search_selection.get("play_url"):
            return search_selection
        content_id = str(item.get("channel_content_id") or "")
        title_keyword = _research_keyword(item, query)
        if not content_id or not title_keyword:
            return search_selection
        try:
            data = self._keyword_search({"keyword": title_keyword, "cursor": ""})
        except Exception as exc:  # noqa: BLE001 - retain original no-valid diagnostics.
            payload = dict(search_selection.get("platform_raw_payload") or {})
            payload.update(
                {
                    "shipinhao_research_source": "title",
                    "shipinhao_research_status": "failed",
                    "shipinhao_research_exception_type": type(exc).__name__,
                    "shipinhao_research_error": str(exc)[:300],
                }
            )
            return {**search_selection, "platform_raw_payload": payload}
        block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
        rows = block.get("data", []) if isinstance(block.get("data"), list) else []
        matched = next(
            (row for row in rows if str(row.get("channel_content_id") or "") == content_id),
            None,
        )
        if not matched:
            payload = dict(search_selection.get("platform_raw_payload") or {})
            payload.update(
                {
                    "shipinhao_research_source": "title",
                    "shipinhao_research_status": "not_matched",
                    "shipinhao_research_item_count": len(rows),
                }
            )
            return {**search_selection, "platform_raw_payload": payload}
        selection = self._select_video_url([("search", item), ("research", matched)])
        payload = dict(selection.get("platform_raw_payload") or {})
        payload.update(
            {
                "shipinhao_research_source": "title",
                "shipinhao_research_status": (
                    "used" if selection.get("play_url") else "no_valid_play_url"
                ),
                "shipinhao_research_item_count": len(rows),
            }
        )
        return {**selection, "platform_raw_payload": payload}

    def _probe_video_url(self, url: str, platform: str) -> dict[str, Any]:
        return platform_video_url.probe_url_with_httpx(
            url,
            platform,
            http_client=self.http_client,
        )

    def fetch_author_works(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        # 上游 blogger 接口 blocked(code=25011),不发请求、不抛,游走自然退化。
        return []