"""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, sanitize_error_detail
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 CrawapiBusinessError(RuntimeError):
    """Non-rate-limit crawapi business failure with the raw response detail.

    This remains a RuntimeError for compatibility with callers/tests that treat
    ordinary platform business failures as PLATFORM_REQUEST_FAILED.
    """

    def __init__(self, message: str, detail: dict[str, Any]) -> None:
        super().__init__(message)
        self.detail = sanitize_error_detail(detail)


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)}"
            )
        message = (
            f"crawapi {operation} failed: business_error "
            f"code={code} message={_business_message(data)} "
            f"keys={sorted(str(key) for key in data.keys())}"
        )
        raise CrawapiBusinessError(
            message,
            _business_error_detail(
                operation=operation,
                response=response,
                response_json=data,
                request_payload=payload,
            ),
        )
    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_error_detail(
    *,
    operation: str,
    response: httpx.Response | None,
    response_json: dict[str, Any],
    request_payload: dict[str, Any],
) -> dict[str, Any]:
    code = response_json.get("code")
    return {
        "operation": operation,
        "business_code": str(code) if code is not None else None,
        "business_message": _business_message(response_json),
        "response_summary": _response_summary(response),
        "response_json_keys": sorted(str(key) for key in response_json.keys()),
        "business_data": response_json.get("data"),
        "trace_refs": _extract_trace_refs(response_json, response),
        "request_payload_summary": _request_payload_summary(request_payload),
    }


def _extract_trace_refs(
    response_json: dict[str, Any],
    response: httpx.Response | None,
) -> dict[str, str]:
    refs: dict[str, str] = {}
    for source in (response_json, response_json.get("data")):
        if not isinstance(source, dict):
            continue
        for key in (
            "request_id",
            "requestId",
            "trace_id",
            "traceId",
            "log_id",
            "logId",
            "rid",
        ):
            value = source.get(key)
            if value:
                refs[key] = _truncate_text(str(value))
    if response is not None:
        for key in (
            "x-request-id",
            "x-trace-id",
            "x-tt-logid",
            "x-log-id",
            "trace-id",
        ):
            value = response.headers.get(key)
            if value:
                refs[key] = _truncate_text(str(value))
    return refs


def _request_payload_summary(payload: dict[str, Any]) -> dict[str, Any]:
    summary: dict[str, Any] = {}
    for key, value in payload.items():
        if key in {
            "keyword",
            "content_type",
            "sort_type",
            "publish_time",
            "duration",
            "cursor",
            "cookie_batch",
        }:
            summary[key] = value
        elif key.endswith("_id") or key in {"account_id", "sec_uid"}:
            summary[key] = "<redacted>"
    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