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

"""快手(kuaishou)接入 client.

复用 crawapi_http 共享基座。M3 接入关键词搜索、内容详情、账号信息和作者作品。
"""

from __future__ import annotations

import re
from pathlib import Path
from typing import Any

from content_agent.integrations.crawapi_http import (
    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 = "kuaishou_search"
DETAIL_RATE_LIMIT_BUCKET = "kuaishou_detail"
ACCOUNT_RATE_LIMIT_BUCKET = "kuaishou_account_info"
BLOGGER_RATE_LIMIT_BUCKET = "kuaishou_blogger"
KUAISHOU_SEARCH_MIN_INTERVAL_SECONDS = 10.0
KUAISHOU_SEARCH_MAX_INTERVAL_SECONDS = 12.0
_TAG_RE = re.compile(r"#([^\s#@（(]+)")


def _statistics(item: dict[str, Any]) -> dict[str, int]:
    return {
        "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("view_count") or 0),
    }


def _extract_tags(item: dict[str, Any]) -> list[str]:
    topic_list = item.get("topic_list") or []
    tags: list[str] = []
    for topic in topic_list:
        if isinstance(topic, dict):
            topic = topic.get("name") or topic.get("tag_name") or topic.get("topic")
        if topic:
            text = str(topic)
            tags.append(text if text.startswith("#") else f"#{text}")
    if not tags:
        text = " ".join(str(item.get(field) or "") for field in ("body_text", "title"))
        tags = [f"#{match}" for match in _TAG_RE.findall(text)]
    return list(dict.fromkeys(tags))


def _publish_time_seconds(item: dict[str, Any]) -> int | None:
    publish_ms = item.get("publish_timestamp")
    if not publish_ms:
        return None
    return int(publish_ms) // 1000


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_")}


def _normalize_kuaishou_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(
        "kuaishou",
        [("search", item)],
        probe_fn=None,
    )
    statistics = _statistics(item)
    platform_content_id = str(item.get("channel_content_id") or "")
    platform_author_id = str(item.get("channel_account_id") or "")
    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": "kuaishou",
        "platform_content_id": platform_content_id,
        "platform_content_url": item.get("content_link"),
        "platform_content_format": content_format(item.get("content_type") or "video"),
        "play_url": video_url_selection.get("play_url"),
        "description": item.get("body_text") or item.get("title") or "",
        "platform_author_id": platform_author_id,
        "author_display_name": item.get("channel_account_name") or "",
        "statistics": statistics,
        "tags": _extract_tags(item),
        "text_extra": [],
        "create_time": _publish_time_seconds(item),
        "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": "kuaishou_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


class CrawapiKuaishouClient:
    requires_progressive_search_rate_limit = True

    def __init__(
        self,
        base_url: str,
        keyword_path: str = "/crawler/kuai_shou/keyword_v2",
        detail_path: str = "/crawler/kuai_shou/detail",
        account_info_path: str = "/crawler/kuai_shou/account_info",
        blogger_path: str = "/crawler/kuai_shou/blogger",
        timeout_seconds: float = 60.0,
        max_results_per_query: int | None = 5,
        http_client: Any | None = None,
        rate_limiter: RateLimiter | None = None,
        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.detail_path = detail_path.lstrip("/")
        self.account_info_path = account_info_path.lstrip("/")
        self.blogger_path = blogger_path.lstrip("/")
        self.timeout_seconds = timeout_seconds
        self.max_results_per_query = max_results_per_query
        self.http_client = http_client or httpx.Client(timeout=timeout_seconds)
        self.rate_limiter = rate_limiter
        self.video_url_probe_fn = video_url_probe_fn

    @classmethod
    def from_env(cls, env_path: str | Path = ".env") -> "CrawapiKuaishouClient":
        env = _load_env_file(env_path)
        return cls(
            base_url=_env("CONTENTFIND_API_CRAWAPI_BASE_URL", env, required=True),
            keyword_path=_env(
                "CONTENTFIND_KUAISHOU_KEYWORD_PATH",
                env,
                default="/crawler/kuai_shou/keyword_v2",
            ),
            detail_path=_env(
                "CONTENTFIND_KUAISHOU_DETAIL_PATH",
                env,
                default="/crawler/kuai_shou/detail",
            ),
            account_info_path=_env(
                "CONTENTFIND_KUAISHOU_ACCOUNT_INFO_PATH",
                env,
                default="/crawler/kuai_shou/account_info",
            ),
            blogger_path=_env(
                "CONTENTFIND_KUAISHOU_BLOGGER_PATH",
                env,
                default="/crawler/kuai_shou/blogger",
            ),
            timeout_seconds=float(
                _env("CONTENTFIND_API_CRAWAPI_TIMEOUT_SECONDS", env, default="180")
            ),
            max_results_per_query=_optional_positive_int(
                _env("CONTENTFIND_KUAISHOU_MAX_RESULTS_PER_QUERY", env, default="5")
            ),
            rate_limiter=RateLimiter(
                min_interval_seconds=KUAISHOU_SEARCH_MIN_INTERVAL_SECONDS,
                max_interval_seconds=KUAISHOU_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_detail_fallback(item)
        hydrated = _normalize_kuaishou_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._post_json(
            self.keyword_path,
            {"keyword": query["search_query"]},
            operation="keyword_search",
            rate_limit_bucket=SEARCH_RATE_LIMIT_BUCKET,
        )
        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_detail_fallback(item) if hydrate_video else None
            )
            normalized = _normalize_kuaishou_item(
                query,
                item,
                index,
                has_more,
                next_cursor,
                selection,
            )
            if not hydrate_video:
                normalized["_platform_search_item"] = item
            results.append(normalized)
        return results

    def fetch_detail(self, content_id: str) -> dict[str, Any]:
        detail = self._fetch_detail_item(content_id)
        statistics = _statistics(detail)
        selection = self._select_video_url([("detail", detail)])
        platform_raw_payload = {
            "channel_content_id": str(detail.get("channel_content_id") or content_id),
            "channel_account_id": str(detail.get("channel_account_id") or ""),
            **dict(selection.get("platform_raw_payload") or {}),
        }
        result = {
            "platform": "kuaishou",
            "platform_content_id": str(detail.get("channel_content_id") or content_id),
            "platform_content_url": detail.get("content_link"),
            "platform_content_format": content_format(detail.get("content_type") or "video"),
            "description": detail.get("body_text") or detail.get("title") or "",
            "platform_author_id": str(detail.get("channel_account_id") or ""),
            "author_display_name": detail.get("channel_account_name") or "",
            "statistics": statistics,
            "tags": _extract_tags(detail),
            "play_url": selection.get("play_url"),
            "create_time": _publish_time_seconds(detail),
            "content_metadata_source": "kuaishou_detail",
            "platform_raw_payload": platform_raw_payload,
        }
        if selection.get("media_failure_reason"):
            result["media_failure_reason"] = selection["media_failure_reason"]
        if selection.get("video_url_candidates"):
            result["video_url_candidates"] = selection["video_url_candidates"]
        return result

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

    def _select_video_url_with_detail_fallback(self, item: 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 "")
        if not content_id:
            return search_selection
        try:
            detail = self._fetch_detail_item(content_id)
        except Exception as exc:  # noqa: BLE001 - keep search diagnostics if detail is unavailable.
            payload = dict(search_selection.get("platform_raw_payload") or {})
            payload.update(
                {
                    "kuaishou_detail_fallback_attempted": True,
                    "kuaishou_detail_fallback_status": "failed",
                    "kuaishou_detail_fallback_exception_type": type(exc).__name__,
                    "kuaishou_detail_fallback_error": str(exc)[:300],
                }
            )
            return {**search_selection, "platform_raw_payload": payload}
        detail_selection = self._select_video_url([("search", item), ("detail", detail)])
        payload = dict(detail_selection.get("platform_raw_payload") or {})
        payload.update(
            {
                "kuaishou_detail_fallback_attempted": True,
                "kuaishou_detail_fallback_status": (
                    "used" if detail_selection.get("play_url") else "no_valid_play_url"
                ),
            }
        )
        return {**detail_selection, "platform_raw_payload": payload}

    def _fetch_detail_item(self, content_id: str) -> dict[str, Any]:
        data = self._post_json(
            self.detail_path,
            {"content_id": str(content_id)},
            operation="detail",
            rate_limit_bucket=DETAIL_RATE_LIMIT_BUCKET,
        )
        block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
        return block.get("data", block) if isinstance(block, dict) else {}

    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_account_info(self, account_id: str, is_cache: bool = True) -> dict[str, Any]:
        data = self._post_json(
            self.account_info_path,
            {"account_id": str(account_id), "is_cache": is_cache},
            operation="account_info",
            rate_limit_bucket=ACCOUNT_RATE_LIMIT_BUCKET,
        )
        block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
        account = block.get("data", block) if isinstance(block, dict) else {}
        platform_author_id = str(
            account.get("channel_account_id") or account.get("account_id") or account_id
        )
        return {
            "platform": "kuaishou",
            "platform_author_id": platform_author_id,
            "author_display_name": account.get("account_name") or "",
            "profile_snapshot": {
                "channel_account_id": platform_author_id,
                "ks_id": account.get("ks_id"),
                "digit_id": account.get("digit_id"),
                "account_link": account.get("account_link"),
                "avatar_url": account.get("avatar_url"),
                "gender": account.get("gender"),
                "description": account.get("description"),
                "tags": account.get("tags") or [],
                "follower_count": int(account.get("follower_count") or 0),
                "publish_count": int(account.get("publish_count") or 0),
                "like_count": int(account.get("like_count") or 0),
                "update_timestamp": account.get("update_timestamp"),
            },
            "raw_payload": account,
        }

    def fetch_author_works(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        author_id = str(query["platform_author_id"])
        data = self._post_json(
            self.blogger_path,
            {"account_id": author_id},
            operation="blogger",
            rate_limit_bucket=BLOGGER_RATE_LIMIT_BUCKET,
        )
        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 "")
        results: list[dict[str, Any]] = []
        for index, item in enumerate(items, start=1):
            normalized = _normalize_kuaishou_item(
                query,
                item,
                index,
                has_more,
                next_cursor,
                self._select_video_url_with_detail_fallback(item),
            )
            normalized["previous_discovery_step"] = "author_works"
            normalized["content_metadata_source"] = "kuaishou_blogger"
            results.append(normalized)
        return results

    def _post_json(
        self,
        path: str,
        payload: dict[str, Any],
        operation: str,
        rate_limit_bucket: str | None = None,
    ) -> dict[str, Any]:
        return post_crawapi_json(
            http_client=self.http_client,
            base_url=self.base_url,
            path=path,
            payload=payload,
            operation=operation,
            timeout_seconds=self.timeout_seconds,
            rate_limiter=self.rate_limiter,
            rate_limit_bucket=rate_limit_bucket,
            business_codes=set(),
        )