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

"""Shared crawapi HTTP base (V3-M1A).

Extracted verbatim from douyin.py so multiple platform clients (douyin /
shipinhao) reuse the same HTTP post + rate limiting + rate-limit error
classification + env-file helpers, instead of each duplicating them.
Pure refactor: behaviour is identical to the original douyin implementation.
"""

from __future__ import annotations

import os
import random
import time
from pathlib import Path
from typing import Any, Callable
from urllib.parse import urljoin

import httpx

from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations import timeout_config

RATE_LIMIT_MESSAGE_TOKENS = ("限流", "请求频繁", "rate limit", "too many requests")
_MAX_RESPONSE_SUMMARY_LENGTH = 500


class CrawapiTransientError(RuntimeError):
    """Retryable crawapi failure (network/timeout, or a platform-declared
    transient business code such as 视频号 25011). Subclasses RuntimeError so
    existing `except RuntimeError` handlers keep working unchanged."""


class RateLimiter:
    def __init__(
        self,
        min_interval_seconds: float = 30.0,
        max_interval_seconds: float | None = None,
        now_fn: Callable[[], float] = time.monotonic,
        sleep_fn: Callable[[float], None] = time.sleep,
        random_fn: Callable[[float, float], float] = random.uniform,
    ) -> None:
        self.min_interval_seconds = min_interval_seconds
        self.max_interval_seconds = max_interval_seconds
        self.now_fn = now_fn
        self.sleep_fn = sleep_fn
        self.random_fn = random_fn
        self._last_call_by_bucket: dict[str, float] = {}

    def wait(self, bucket: str) -> None:
        last = self._last_call_by_bucket.get(bucket)
        if last is not None:
            remaining = self._next_interval_seconds() - (self.now_fn() - last)
            if remaining > 0:
                self.sleep_fn(remaining)
        self._last_call_by_bucket[bucket] = self.now_fn()

    def _next_interval_seconds(self) -> float:
        if self.max_interval_seconds is None or self.max_interval_seconds <= self.min_interval_seconds:
            return self.min_interval_seconds
        return self.random_fn(self.min_interval_seconds, self.max_interval_seconds)


def is_rate_limit_business_error(
    code: Any, data: dict[str, Any], *, business_codes: set[str]
) -> bool:
    if str(code) in business_codes:
        return True
    message = str(data.get("msg") or data.get("message") or "").lower()
    return any(token in message for token in RATE_LIMIT_MESSAGE_TOKENS)


def post_crawapi_json(
    *,
    http_client: Any,
    base_url: str,
    path: str,
    payload: dict[str, Any],
    operation: str,
    timeout_seconds: float,
    rate_limiter: RateLimiter | None = None,
    rate_limit_bucket: str | None = None,
    business_codes: set[str],
    transient_business_codes: set[str] = frozenset(),
) -> dict[str, Any]:
    if rate_limit_bucket and rate_limiter:
        rate_limiter.wait(rate_limit_bucket)
    url = urljoin(base_url, path)
    try:
        response = http_client.post(
            url,
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=timeout_config.as_httpx_timeout(timeout_seconds, read=timeout_config.read_timeout("crawapi")),
        )
        response.raise_for_status()
        data = response.json()
    except httpx.HTTPStatusError as exc:
        status_code = exc.response.status_code if exc.response is not None else "unknown"
        if status_code == 429:
            raise ContentAgentError(
                ErrorCode.PLATFORM_RATE_LIMITED,
                f"crawapi {operation} failed: rate_limited",
                {
                    "operation": operation,
                    "status_code": 429,
                    "response_summary": _response_summary(exc.response),
                },
            ) from exc
        raise RuntimeError(
            f"crawapi {operation} failed: HTTP {status_code}; "
            f"response={_response_summary(exc.response)}"
        ) from exc
    except httpx.HTTPError as exc:
        raise CrawapiTransientError(
            f"crawapi {operation} failed: network_error "
            f"exception_type={type(exc).__name__} message={_truncate_text(str(exc))}"
        ) from exc
    except ValueError as exc:
        response_summary = _response_summary(response) if "response" in locals() else {}
        raise RuntimeError(
            f"crawapi {operation} failed: bad_json; response={response_summary}"
        ) from exc
    if not isinstance(data, dict):
        raise RuntimeError(f"crawapi {operation} failed: bad_response")
    code = data.get("code")
    if code is not None and code not in (0, "0"):
        if is_rate_limit_business_error(code, data, business_codes=business_codes):
            raise ContentAgentError(
                ErrorCode.PLATFORM_RATE_LIMITED,
                f"crawapi {operation} failed: rate_limited",
                {
                    "operation": operation,
                    "business_code": str(code),
                    "business_message": _business_message(data),
                },
            )
        if str(code) in transient_business_codes:
            raise CrawapiTransientError(
                f"crawapi {operation} failed: transient_business_error "
                f"code={code} message={_business_message(data)}"
            )
        raise RuntimeError(
            f"crawapi {operation} failed: business_error "
            f"code={code} message={_business_message(data)} "
            f"keys={sorted(str(key) for key in data.keys())}"
        )
    return data


def _response_summary(response: httpx.Response | None) -> dict[str, Any]:
    if response is None:
        return {}
    summary: dict[str, Any] = {
        "status_code": response.status_code,
        "content_type": response.headers.get("content-type"),
    }
    try:
        data = response.json()
    except ValueError:
        text = response.text
        if text:
            summary["body_summary"] = _truncate_text(text)
        return summary
    if isinstance(data, dict):
        summary["json_keys"] = sorted(str(key) for key in data.keys())
        for key in ("code", "status", "msg", "message", "error"):
            if key in data:
                summary[key] = _truncate_text(str(data.get(key)))
    else:
        summary["json_type"] = type(data).__name__
    return summary


def _business_message(data: dict[str, Any]) -> str:
    return _truncate_text(str(data.get("msg") or data.get("message") or ""))


def _truncate_text(text: str, limit: int = _MAX_RESPONSE_SUMMARY_LENGTH) -> str:
    if len(text) <= limit:
        return text
    return f"{text[:limit]}..."


def _load_env_file(env_path: str | Path) -> dict[str, str]:
    path = Path(env_path)
    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 _env(
    key: str,
    file_env: dict[str, str],
    default: str | None = None,
    required: bool = False,
) -> str:
    value = file_env.get(key) or os.getenv(key) or default
    if required and not value:
        raise RuntimeError(f"missing required env: {key}")
    return value or ""


def _optional_positive_int(value: str) -> int | None:
    try:
        parsed = int(value)
    except ValueError:
        return None
    return parsed if parsed > 0 else None


def search_previous_discovery_step(query: dict[str, Any]) -> str:
    """搜索结果的来源步:游走标签搜索标 hashtag_to_query,首轮标 search_query_direct。

    修真跑发现的 labeling bug——此前所有搜索结果硬编码 search_query_direct,
    导致游走带回的视频被当成首轮、前端首轮/游走分不开。
    """
    if query.get("search_query_generation_method") == "tag_query":
        return "hashtag_to_query"
    return "search_query_direct"


def content_format(raw_content_type: str) -> str:
    if "图文" in raw_content_type:
        return "image_text"
    if "文本" in raw_content_type:
        return "text"
    if "直播" in raw_content_type:
        return "live"
    return "video"


def score_from_statistics(statistics: dict[str, Any]) -> int:
    digg = int(statistics.get("digg_count") or 0)
    comment = int(statistics.get("comment_count") or 0)
    share = int(statistics.get("share_count") or 0)
    weighted = digg + comment * 3 + share * 4
    if weighted >= 3000:
        return 72
    if weighted >= 1000:
        return 62
    if weighted >= 300:
        return 55
    return 45