SupplyAgent/find_agent_v2/providers.py

"""Independent external-data clients for find_agent_v2.

This module owns request validation, HTTP calls and response normalization;
database persistence remains in :mod:`find_agent_v2.service`.
"""

from __future__ import annotations

import asyncio
import os
import re
import time
from typing import Any

import httpx
from dotenv import load_dotenv

from supply_agent.paths import find_project_root

INTERNAL_SEARCH_ENDPOINT = os.getenv(
    "FIND_AGENT_V2_INTERNAL_SEARCH_ENDPOINT",
    "http://crawapi.piaoquantv.com/crawler/dou_yin/keyword",
)
TIKHUB_SEARCH_ENDPOINT = os.getenv(
    "FIND_AGENT_V2_TIKHUB_SEARCH_ENDPOINT",
    "https://api.tikhub.io/api/v1/douyin/search/fetch_video_search_v2",
)
DETAIL_ENDPOINT = os.getenv(
    "FIND_AGENT_V2_DETAIL_ENDPOINT",
    "http://8.217.190.241:8888/crawler/dou_yin/detail",
)
CONTENT_PORTRAIT_ENDPOINT = os.getenv(
    "FIND_AGENT_V2_CONTENT_PORTRAIT_ENDPOINT",
    "http://crawapi.piaoquantv.com/crawler/dou_yin/re_dian_bao/video_like_portrait",
)
ACCOUNT_PORTRAIT_ENDPOINT = os.getenv(
    "FIND_AGENT_V2_ACCOUNT_PORTRAIT_ENDPOINT",
    "http://crawapi.piaoquantv.com/crawler/dou_yin/re_dian_bao/account_fans_portrait",
)

DEFAULT_TIMEOUT = 60.0
DEFAULT_ACCOUNT_ID = "771431222"
MAX_BATCH_ITEMS = 8
_env_loaded = False


def _ensure_env_loaded() -> None:
    global _env_loaded
    if _env_loaded:
        return
    load_dotenv(find_project_root() / ".env")
    _env_loaded = True


class _RateLimiter:
    def __init__(self, interval_seconds: float) -> None:
        self.interval_seconds = interval_seconds
        self._lock = asyncio.Lock()
        self._last_request = 0.0

    async def wait(self) -> None:
        async with self._lock:
            remaining = self.interval_seconds - (time.monotonic() - self._last_request)
            if remaining > 0:
                await asyncio.sleep(remaining)
            self._last_request = time.monotonic()


_internal_search_limiter = _RateLimiter(10.1)
_tikhub_limiter = _RateLimiter(1.0)
_detail_limiter = _RateLimiter(10.1)

_CONTENT_TYPES = {"不限": "0", "视频": "1", "图片": "2", "图文": "2", "文章": "3"}
_SORT_TYPES = {"综合排序": "0", "最多点赞": "1", "最新发布": "2"}
_PUBLISH_TIMES = {"不限": "0", "一天内": "1", "最近一天": "1", "一周内": "7", "最近一周": "7", "半年内": "180", "最近半年": "180"}
_DURATIONS = {"不限": "0", "一分钟内": "0-1", "1分钟以内": "0-1", "1-5分钟": "1-5", "五分钟以上": "5-10000", "5分钟以上": "5-10000"}


def _safe_int(value: Any, default: int = 0) -> int:
    if value is None or isinstance(value, bool):
        return default
    try:
        return int(float(str(value).strip()))
    except (TypeError, ValueError):
        return default


def _optional_int(value: Any) -> int | None:
    """Preserve a missing upstream metric instead of conflating it with a real zero."""
    if value is None or isinstance(value, bool) or str(value).strip() == "":
        return None
    try:
        return int(float(str(value).strip()))
    except (TypeError, ValueError):
        return None


def _error(exc: Exception | str) -> dict[str, Any]:
    return {"error": str(exc), "search_results": [], "has_more": False}


def _request_error(exc: Exception) -> str:
    if isinstance(exc, httpx.HTTPStatusError):
        return f"HTTP {exc.response.status_code}: {exc.response.text[:1000]}"
    if isinstance(exc, httpx.TimeoutException):
        return "请求超时"
    if isinstance(exc, httpx.RequestError):
        return f"网络错误: {exc}"
    return f"未知错误: {exc}"


def _topics(aweme: dict[str, Any]) -> list[str]:
    result: list[str] = []
    for collection in (aweme.get("topic_list"), aweme.get("text_extra"), aweme.get("cha_list")):
        for item in collection or []:
            if isinstance(item, str):
                value = item
            elif isinstance(item, dict):
                value = item.get("topic_name") or item.get("cha_name") or item.get("hashtag_name") or item.get("name")
            else:
                value = None
            if value and str(value).strip() not in result:
                result.append(str(value).strip())
    return result


def _normalize_search_item(item: dict[str, Any]) -> dict[str, Any] | None:
    aweme_id = str(item.get("aweme_id") or "").strip()
    if not aweme_id:
        return None
    author = item.get("author") if isinstance(item.get("author"), dict) else {}
    stats = item.get("statistics") if isinstance(item.get("statistics"), dict) else {}
    video = item.get("video") if isinstance(item.get("video"), dict) else {}
    return {
        "aweme_id": aweme_id,
        "desc": str(item.get("desc") or item.get("item_title") or "无标题")[:200],
        "url": f"https://www.douyin.com/video/{aweme_id}",
        "author": {
            "nickname": str(author.get("nickname") or "未知作者"),
            "sec_uid": str(author.get("sec_uid") or ""),
        },
        "statistics": {
            "digg_count": _optional_int(stats.get("digg_count")),
            "comment_count": _optional_int(stats.get("comment_count")),
            "share_count": _optional_int(stats.get("share_count")),
            "collect_count": _optional_int(stats.get("collect_count")),
            "play_count": _optional_int(stats.get("play_count")),
        },
        "duration_ms": _safe_int(item.get("duration_ms") or item.get("duration") or video.get("duration")),
        "publish_at": item.get("publish_at") or item.get("create_time") or item.get("create_timestamp") or item.get("publish_timestamp"),
        "topics": _topics(item),
    }


async def search_internal(
    *, keyword: str, content_type: str = "视频", sort_type: str = "综合排序",
    publish_time: str = "不限", cursor: str = "0", min_duration_seconds: int = 30,
    timeout: float = DEFAULT_TIMEOUT,
) -> dict[str, Any]:
    """Search one page through the internal crawler and return normalized candidates."""
    started = time.monotonic()
    try:
        await _internal_search_limiter.wait()
        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.post(INTERNAL_SEARCH_ENDPOINT, json={
                "keyword": keyword,
                "content_type": content_type,
                "sort_type": sort_type,
                "publish_time": publish_time,
                "cursor": cursor,
                "account_id": DEFAULT_ACCOUNT_ID,
            }, headers={"Content-Type": "application/json"})
            response.raise_for_status()
            body = response.json()
        data = body.get("data") if isinstance(body.get("data"), dict) else {}
        items = data.get("data") if isinstance(data.get("data"), list) else []
        minimum_ms = max(30, int(min_duration_seconds)) * 1000
        normalized = [value for item in items if (value := _normalize_search_item(item))]
        results = [item for item in normalized if not item["duration_ms"] or item["duration_ms"] >= minimum_ms]
        return {
            "provider": "internal_keyword", "keyword": keyword,
            "search_results": results, "results_count": len(results),
            "filtered_count": len(normalized) - len(results),
            "has_more": bool(data.get("has_more")),
            "next_cursor": str(data.get("next_cursor") or ""),
            "duration_ms": int((time.monotonic() - started) * 1000),
            "_raw_response": body,
        }
    except Exception as exc:
        return _error(_request_error(exc))


def _enum(value: str, mapping: dict[str, str], field: str) -> str:
    text = str(value).strip()
    if text in mapping.values():
        return text
    if text not in mapping:
        raise ValueError(f"{field} 不支持: {value}")
    return mapping[text]


async def search_tikhub(
    *, keyword: str, content_type: str = "视频", sort_type: str = "综合排序",
    publish_time: str = "不限", cursor: int = 0, filter_duration: str = "不限",
    search_id: str = "", backtrace: str = "", min_duration_seconds: int = 30,
    timeout: float = DEFAULT_TIMEOUT,
) -> dict[str, Any]:
    """Search one TikHub page and retain its pagination state."""
    _ensure_env_loaded()
    api_key = os.getenv("TIKHUB_API_KEY", "").strip()
    if not api_key:
        return _error("未设置环境变量 TIKHUB_API_KEY")
    try:
        payload = {
            "keyword": str(keyword).strip(), "cursor": max(0, int(cursor)),
            "sort_type": _enum(sort_type, _SORT_TYPES, "sort_type"),
            "publish_time": _enum(publish_time, _PUBLISH_TIMES, "publish_time"),
            "filter_duration": _enum(filter_duration, _DURATIONS, "filter_duration"),
            "content_type": _enum(content_type, _CONTENT_TYPES, "content_type"),
            "search_id": str(search_id or ""), "backtrace": str(backtrace or ""),
        }
        await _tikhub_limiter.wait()
        async with httpx.AsyncClient(timeout=timeout, trust_env=False) as client:
            response = await client.post(TIKHUB_SEARCH_ENDPOINT, json=payload, headers={
                "Content-Type": "application/json", "Authorization": f"Bearer {api_key}",
            })
            response.raise_for_status()
            body = response.json()
        data = body.get("data") if isinstance(body.get("data"), dict) else {}
        raw_items = data.get("business_data") if isinstance(data.get("business_data"), list) else []
        config = data.get("business_config") if isinstance(data.get("business_config"), dict) else {}
        next_page = config.get("next_page") if isinstance(config.get("next_page"), dict) else {}
        minimum_ms = max(30, int(min_duration_seconds)) * 1000
        results: list[dict[str, Any]] = []
        seen: set[str] = set()
        for raw in raw_items:
            nested = raw.get("data") if isinstance(raw, dict) and isinstance(raw.get("data"), dict) else {}
            aweme = nested.get("aweme_info") if isinstance(nested.get("aweme_info"), dict) else {}
            item = _normalize_search_item(aweme)
            if item and item["aweme_id"] not in seen and (not item["duration_ms"] or item["duration_ms"] >= minimum_ms):
                seen.add(item["aweme_id"])
                results.append(item)
        return {
            "provider": "tikhub", "keyword": keyword, "search_results": results,
            "results_count": len(results), "has_more": config.get("has_more") in (1, True, "1"),
            "next_cursor": _safe_int(next_page.get("cursor")),
            "search_id": str(next_page.get("search_id") or search_id or ""),
            "backtrace": str(next_page.get("backtrace") or config.get("backtrace") or backtrace or ""),
            "_raw_response": body,
        }
    except Exception as exc:
        return _error(_request_error(exc))


def _detail_payload(raw: dict[str, Any], content_id: str) -> dict[str, Any]:
    video_urls = raw.get("video_url_list") if isinstance(raw.get("video_url_list"), list) else []
    first_video = video_urls[0] if video_urls and isinstance(video_urls[0], dict) else {}
    topics = raw.get("topic_list") if isinstance(raw.get("topic_list"), list) else []
    return {
        "content_id": content_id,
        "content_link": raw.get("content_link") or f"https://www.douyin.com/video/{content_id}",
        "title": raw.get("title"), "body_text": raw.get("body_text"),
        "channel_account_name": raw.get("channel_account_name"),
        "channel_account_id": raw.get("channel_account_id"),
        "topic_list": topics,
        "duration_seconds": _safe_int(first_video.get("video_duration") or raw.get("duration_seconds")),
        "publish_at": raw.get("publish_at") or raw.get("publish_time") or raw.get("publish_timestamp") or raw.get("create_time") or raw.get("create_timestamp"),
        "play_count": raw.get("play_count") if raw.get("play_count") is not None else raw.get("view_count"),
        "like_count": raw.get("like_count"), "comment_count": raw.get("comment_count"),
        "collect_count": raw.get("collect_count"), "share_count": raw.get("share_count"),
        "video_url": first_video.get("video_url"),
    }


async def fetch_details(content_ids: list[str], *, timeout: float = DEFAULT_TIMEOUT) -> dict[str, Any]:
    """Fetch and normalize up to eight video detail records."""
    ids = list(dict.fromkeys(str(value).strip() for value in content_ids if str(value).strip()))
    if not ids or len(ids) > MAX_BATCH_ITEMS:
        return {"error": f"content_ids 必须为 1~{MAX_BATCH_ITEMS} 条", "details": [], "errors": []}
    details: list[dict[str, Any]] = []
    errors: list[dict[str, str]] = []
    async with httpx.AsyncClient(timeout=timeout, trust_env=False, headers={"Accept": "*/*"}) as client:
        for content_id in ids:
            try:
                await _detail_limiter.wait()
                response = await client.post(DETAIL_ENDPOINT, json={"content_id": content_id})
                response.raise_for_status()
                body = response.json()
                if body.get("code") not in (0, None):
                    raise RuntimeError(f"接口返回错误: {body.get('msg') or body.get('code')}")
                outer = body.get("data") if isinstance(body.get("data"), dict) else {}
                raw = outer.get("data") if isinstance(outer.get("data"), dict) else {}
                if not raw:
                    raise RuntimeError("未查到视频详情")
                details.append(_detail_payload(raw, content_id))
            except Exception as exc:
                errors.append({"content_id": content_id, "error": _request_error(exc)})
    return {"details": details, "errors": errors, "success_count": len(details), "failed_count": len(errors)}


def _number(value: Any) -> float | None:
    if value is None or isinstance(value, bool):
        return None
    try:
        return float(str(value).strip().replace("%", ""))
    except ValueError:
        return None


def _age_dimension(portrait: Any) -> dict[str, Any]:
    if not isinstance(portrait, dict):
        return {}
    for key in ("portrait_data", "content", "account"):
        nested = _age_dimension(portrait.get(key))
        if nested:
            return nested
    for key in ("年龄", "age", "Age", "年龄分布"):
        if isinstance(portrait.get(key), dict):
            return portrait[key]
    return portrait if portrait and all(isinstance(value, dict) for value in portrait.values()) else {}


def _normalize_age(portrait: Any) -> dict[str, Any]:
    buckets: list[dict[str, Any]] = []
    older_ratio = 0.0
    mature_ratio = 0.0
    weighted: list[tuple[float, float]] = []
    dimension = _age_dimension(portrait)
    for label, raw in dimension.items():
        metrics = raw if isinstance(raw, dict) else {}
        ratio = _number(metrics.get("percentage", metrics.get("ratio")))
        if ratio is not None and ratio > 1:
            ratio /= 100
        tgi = _number(metrics.get("preference", metrics.get("tgi")))
        numbers = [int(value) for value in re.findall(r"\d+", str(label))]
        lower, upper = (min(numbers), max(numbers)) if numbers else (0, 0)
        kind = "older" if lower >= 50 else "mature" if lower >= 40 or upper >= 50 else "younger"
        buckets.append({"label": str(label), "kind": kind, "ratio": ratio, "tgi": tgi})
        if ratio is not None and kind == "older":
            older_ratio += ratio
            if tgi is not None:
                weighted.append((tgi, ratio))
        elif ratio is not None and kind == "mature":
            mature_ratio += ratio
    weight = sum(item[1] for item in weighted)
    older_tgi = sum(tgi * ratio for tgi, ratio in weighted) / weight if weight else None
    strength = "missing" if not dimension else "strong" if older_ratio >= 0.35 else "moderate" if older_ratio >= 0.20 or mature_ratio >= 0.30 else "weak"
    return {"has_age_portrait": bool(dimension), "older_ratio": round(older_ratio, 6), "older_tgi": older_tgi, "mature_ratio": round(mature_ratio, 6), "strength": strength, "buckets": buckets}


def normalize_age_pair(content: Any, account: Any) -> dict[str, Any]:
    content_age, account_age = _normalize_age(content), _normalize_age(account)
    if content_age["has_age_portrait"] and account_age["has_age_portrait"]:
        consistency, cap = "aligned" if (content_age["strength"] in {"strong", "moderate"}) == (account_age["strength"] in {"strong", "moderate"}) else "conflict", 1.0
    elif account_age["has_age_portrait"]:
        consistency, cap = "account_only", 0.85 if account_age["strength"] == "strong" else 0.75
    elif content_age["has_age_portrait"]:
        consistency, cap = "content_only", 1.0
    else:
        consistency, cap = "missing", 0.5
    return {"content": content_age, "account": account_age, "consistency": consistency, "elder_score_cap": cap}


def _portrait_data(body: dict[str, Any]) -> dict[str, Any]:
    outer = body.get("data") if isinstance(body.get("data"), dict) else {}
    return outer.get("data") if isinstance(outer.get("data"), dict) else {}


async def fetch_portraits(candidates: list[dict[str, Any]], *, timeout: float = DEFAULT_TIMEOUT) -> dict[str, Any]:
    """Fetch content/account portraits and normalize age evidence independently."""
    if not candidates or len(candidates) > MAX_BATCH_ITEMS:
        return {"error": f"candidates 必须为 1~{MAX_BATCH_ITEMS} 条", "results": []}
    flags = {"need_province": False, "need_city": False, "need_city_level": False, "need_gender": False, "need_age": True, "need_phone_brand": False, "need_phone_price": False}
    results: list[dict[str, Any]] = []
    async with httpx.AsyncClient(timeout=timeout) as client:
        for candidate in candidates:
            aweme_id = str(candidate.get("aweme_id") or "").strip()
            author_id = str(candidate.get("author_sec_uid") or "").strip()
            item: dict[str, Any] = {"aweme_id": aweme_id, "author_sec_uid": author_id or None, "content": {}, "account": {}, "error": None}
            try:
                if not aweme_id.isdigit():
                    raise ValueError("aweme_id 必须为纯数字")
                response = await client.post(CONTENT_PORTRAIT_ENDPOINT, json={"content_id": aweme_id, **flags})
                response.raise_for_status()
                portrait = _portrait_data(response.json())
                item["content"] = {"ok": True, "has_portrait": bool(portrait), "portrait_data": portrait}
            except Exception as exc:
                item["content"] = {"ok": False, "has_portrait": False, "portrait_data": {}, "error": _request_error(exc)}
            if author_id:
                try:
                    response = await client.post(ACCOUNT_PORTRAIT_ENDPOINT, json={"account_id": author_id, **flags})
                    response.raise_for_status()
                    portrait = _portrait_data(response.json())
                    item["account"] = {"attempted": True, "has_portrait": bool(portrait), "portrait_data": portrait}
                except Exception as exc:
                    item["account"] = {"attempted": True, "has_portrait": False, "portrait_data": {}, "error": _request_error(exc)}
            else:
                item["account"] = {"attempted": False, "has_portrait": False, "portrait_data": {}, "skipped_reason": "缺少 author_sec_uid"}
            item["age_normalization"] = normalize_age_pair(item["content"].get("portrait_data"), item["account"].get("portrait_data"))
            if item["content"].get("error") and item["account"].get("error"):
                item["error"] = "内容与账号画像均获取失败"
            results.append(item)
    return {"results": results, "count": len(results)}