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

from __future__ import annotations

from pathlib import Path
from typing import Any

import httpx

# 共享 crawapi 基座(V3-M1A):HTTP/限流/限流错误识别/env helper 集中于 crawapi_http,
# 下方 re-export 保持既有外部 import(测试、smoke 脚本)零改。
from content_agent.integrations.crawapi_http import (
    CrawapiBusinessError,
    CrawapiTransientError,
    RATE_LIMIT_MESSAGE_TOKENS,
    RateLimiter,
    _env,
    _load_env_file,
    _optional_positive_int,
    content_format as _content_format,
    is_rate_limit_business_error,
    post_crawapi_json,
    score_from_statistics as _score_from_statistics,
    search_previous_discovery_step as _search_previous_discovery_step,
)
from content_agent.errors import ContentAgentError
from content_agent.integrations import platform_video_url

RAW_CONTENT_ID_KEY = "_".join(["aweme", "id"])
RAW_AUTHOR_ID_KEY = "_".join(["sec", "uid"])
RAW_AUTHOR_ACCOUNT_KEY = "_".join(["account", "id"])

# 已证实的限流 business code 白名单。当前没有任何已证实的限流 code,
# 识别先依靠 HTTP 429 与 message token;live smoke / 真实运行发现新 code 后补入并加用例。
RATE_LIMIT_BUSINESS_CODES: set[str] = set()

SEARCH_RATE_LIMIT_BUCKET = "douyin_search"
BLOGGER_RATE_LIMIT_BUCKET = "douyin_blogger"
DOUYIN_SEARCH_MIN_INTERVAL_SECONDS = 10.0
DOUYIN_SEARCH_MAX_INTERVAL_SECONDS = 12.0
NON_FALLBACK_BUSINESS_CODES = {"10001"}
NON_FALLBACK_BUSINESS_MESSAGE_TOKENS = ("参数异常", "参数校验", "invalid parameter")


class CrawapiDouyinClient:
    requires_progressive_search_rate_limit = True

    def __init__(
        self,
        base_url: str,
        keyword_path: str,
        fallback_base_url: str = "",
        blogger_path: str = "",
        detail_path: str = "",
        portrait_path: str = "",
        timeout_seconds: float = 60.0,
        default_crawapi_account_ref: str = "",
        default_content_type: str = "视频",
        default_sort_type: str = "综合排序",
        default_publish_time: str = "不限",
        default_cursor: str = "0",
        piaoquantv_account_id: str = "",
        piaoquantv_duration: str = "不限",
        piaoquantv_cookie_batch: str = "default",
        crawler_sort_type: str = "最多点赞",
        crawler_cursor_default: str = "",
        default_account_works_sort_type: str = "最新",
        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:
        self.base_url = base_url.rstrip("/") + "/"
        self.fallback_base_url = fallback_base_url.rstrip("/") + "/" if fallback_base_url else ""
        self._crawapi_base_urls = _dedupe_crawapi_base_urls(
            [("primary", self.base_url), ("candidate", self.fallback_base_url)]
        )
        self._legacy_crawapi_base_urls = _legacy_crawapi_base_urls(self._crawapi_base_urls)
        self.keyword_path = keyword_path.lstrip("/")
        self.blogger_path = blogger_path.lstrip("/")
        self.detail_path = detail_path.lstrip("/")
        self.portrait_path = portrait_path.lstrip("/")
        self.timeout_seconds = timeout_seconds
        self.default_crawapi_account_ref = default_crawapi_account_ref
        self.default_content_type = default_content_type
        self.default_sort_type = default_sort_type
        self.default_publish_time = default_publish_time
        self.default_cursor = default_cursor
        self.piaoquantv_account_id = piaoquantv_account_id or default_crawapi_account_ref
        self.piaoquantv_duration = piaoquantv_duration
        self.piaoquantv_cookie_batch = piaoquantv_cookie_batch
        self.crawler_sort_type = crawler_sort_type
        self.crawler_cursor_default = crawler_cursor_default
        self.default_account_works_sort_type = default_account_works_sort_type
        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") -> "CrawapiDouyinClient":
        env = _load_env_file(env_path)
        return cls(
            base_url=_env("CONTENTFIND_API_CRAWAPI_BASE_URL", env, required=True),
            fallback_base_url=_env("CONTENTFIND_API_CRAWAPI_FALLBACK_BASE_URL", env, default=""),
            keyword_path=_env("CONTENTFIND_DOUYIN_KEYWORD_PATH", env, required=True),
            blogger_path=_env("CONTENTFIND_DOUYIN_BLOGGER_PATH", env, required=True),
            detail_path=_env(
                "CONTENTFIND_DOUYIN_DETAIL_PATH", env, default="/crawler/dou_yin/detail"
            ),
            portrait_path=_env(
                "CONTENTFIND_DOUYIN_PORTRAIT_PATH",
                env,
                default="/crawler/dou_yin/re_dian_bao/account_fans_portrait",
            ),
            timeout_seconds=float(
                _env("CONTENTFIND_API_CRAWAPI_TIMEOUT_SECONDS", env, default="180")
            ),
            default_crawapi_account_ref=_env("CONTENTFIND_DOUYIN_DEFAULT_ACCOUNT_ID", env, default=""),
            default_content_type=_env("CONTENTFIND_DOUYIN_DEFAULT_CONTENT_TYPE", env, default="视频"),
            default_sort_type=_env("CONTENTFIND_DOUYIN_DEFAULT_SORT_TYPE", env, default="综合排序"),
            default_publish_time=_env("CONTENTFIND_DOUYIN_DEFAULT_PUBLISH_TIME", env, default="不限"),
            default_cursor=_env("CONTENTFIND_DOUYIN_DEFAULT_CURSOR", env, default="0"),
            piaoquantv_account_id=_env(
                "CONTENTFIND_DOUYIN_PIAOQUANTV_ACCOUNT_ID",
                env,
                default=_env("CONTENTFIND_DOUYIN_DEFAULT_ACCOUNT_ID", env, default=""),
            ),
            piaoquantv_duration=_env("CONTENTFIND_DOUYIN_PIAOQUANTV_DURATION", env, default="不限"),
            piaoquantv_cookie_batch=_env(
                "CONTENTFIND_DOUYIN_PIAOQUANTV_COOKIE_BATCH", env, default="default"
            ),
            crawler_sort_type=_env("CONTENTFIND_DOUYIN_CRAWLER_SORT_TYPE", env, default="最多点赞"),
            crawler_cursor_default=_env("CONTENTFIND_DOUYIN_CRAWLER_CURSOR_DEFAULT", env, default=""),
            default_account_works_sort_type=_env(
                "CONTENTFIND_DOUYIN_ACCOUNT_WORKS_DEFAULT_SORT_TYPE", env, default="最新"
            ),
            max_results_per_query=_optional_positive_int(
                _env("CONTENTFIND_DOUYIN_MAX_RESULTS_PER_QUERY", env, default="5")
            ),
            rate_limiter=RateLimiter(
                min_interval_seconds=DOUYIN_SEARCH_MIN_INTERVAL_SECONDS,
                max_interval_seconds=DOUYIN_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(
        self,
        query: dict[str, Any],
        max_results_per_query: int | None,
    ) -> list[dict[str, Any]]:
        data = self._post_keyword_json(query)
        data_block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
        items = data_block.get("data", []) if isinstance(data_block.get("data"), list) else []
        has_more = bool(data_block.get("has_more", False))
        next_cursor = str(data_block.get("next_cursor") or "")

        results: list[dict[str, Any]] = []
        selected_items = items[:max_results_per_query] if max_results_per_query else items
        for index, item in enumerate(selected_items, start=1):
            selection = self._select_video_url_with_detail_fallback(item)
            results.append(
                self._normalize_content_item(query, item, index, has_more, next_cursor, selection)
            )
        return results

    def fetch_author_works(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        payload = {
            RAW_AUTHOR_ACCOUNT_KEY: str(query.get("platform_author_id") or ""),
            "sort_type": self.default_account_works_sort_type,
            "cursor": str(query.get("page_cursor") or ""),
        }
        data = self._post_json(
            self.blogger_path, payload, operation="author_works",
            rate_limit_bucket=BLOGGER_RATE_LIMIT_BUCKET,
        )
        data_block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
        items = data_block.get("data", []) if isinstance(data_block.get("data"), list) else []
        has_more = bool(data_block.get("has_more", False))
        next_cursor = str(data_block.get("next_cursor") or "")

        selected_items = items[: self.max_results_per_query] if self.max_results_per_query else items
        results: list[dict[str, Any]] = []
        for index, item in enumerate(selected_items, start=1):
            selection = self._select_video_url_with_detail_fallback(item)
            normalized = self._normalize_content_item(query, item, index, has_more, next_cursor, selection)
            normalized["previous_discovery_step"] = "author_works"
            normalized["content_metadata_source"] = "douyin_blogger"
            results.append(normalized)
        return results

    def fetch_account_fans_portrait(self, account_id: str) -> dict[str, Any]:
        """M9A:热点宝作者粉丝画像(account_id=sec_uid)。返回 data.data(含 account/fans/posts)。

        错误自然上抛(post_crawapi_json 已分类 rate_limit/transient/business),不在此吞。
        """
        payload = {
            RAW_AUTHOR_ACCOUNT_KEY: str(account_id or ""),
            "need_age": True,
            "need_gender": True,
            "need_province": True,
        }
        data = self._post_json(
            self.portrait_path, payload, operation="account_fans_portrait",
        )
        outer = data.get("data", {})
        inner = outer.get("data", {}) if isinstance(outer, dict) else {}
        return inner if isinstance(inner, dict) else {}

    def _normalize_content_item(
        self,
        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 self._select_video_url([("search", item)])
        author = item.get("author", {}) if isinstance(item.get("author"), dict) else {}
        statistics = item.get("statistics", {}) if isinstance(item.get("statistics"), dict) else {}
        platform_content_id = str(item.get(RAW_CONTENT_ID_KEY) or "")
        platform_author_id = str(author.get(RAW_AUTHOR_ID_KEY) or "")
        result = {
            "content_discovery_id": f"{query['search_query_id']}_content_{index:03d}",
            "search_query_id": query["search_query_id"],
            "platform": "douyin",
            "platform_content_id": platform_content_id,
            "platform_content_format": _content_format(self.default_content_type),
            "play_url": video_url_selection.get("play_url"),
            "description": item.get("desc") or item.get("item_title") or "",
            "platform_author_id": platform_author_id,
            "author_display_name": author.get("nickname") or "",
            "statistics": {
                "digg_count": int(statistics.get("digg_count") or 0),
                "comment_count": int(statistics.get("comment_count") or 0),
                "share_count": int(statistics.get("share_count") or 0),
                "collect_count": int(statistics.get("collect_count") or 0),
                "play_count": int(statistics.get("play_count") or 0),
            },
            "tags": _extract_tags(item),
            "text_extra": item.get("text_extra") or [],
            "create_time": item.get("create_time"),
            "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": "douyin_keyword_search",
            "platform_auth_mode": "no_bearer",
            "platform_raw_payload": {
                RAW_CONTENT_ID_KEY: platform_content_id,
                "author": {RAW_AUTHOR_ID_KEY: platform_author_id},
                **dict(video_url_selection.get("platform_raw_payload") or {}),
            },
        }
        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 fetch_detail(self, content_id: str) -> dict[str, Any]:
        detail = self._fetch_detail_item(content_id)
        statistics = {
            "digg_count": int(detail.get("like_count") or 0),
            "comment_count": int(detail.get("comment_count") or 0),
            "share_count": int(detail.get("share_count") or 0),
            "collect_count": int(detail.get("collect_count") or 0),
            "play_count": int(detail.get("play_count") or 0),
        }
        topic_list = detail.get("topic_list") or []
        tags = [t if str(t).startswith("#") else f"#{t}" for t in topic_list if t]
        selection = self._select_video_url([("detail", detail)])
        publish_ms = detail.get("publish_timestamp")
        result = {
            "platform": "douyin",
            "platform_content_id": str(detail.get("channel_content_id") or content_id),
            "platform_content_url": detail.get("content_link"),
            "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": tags,
            "play_url": selection.get("play_url"),
            "create_time": int(publish_ms) // 1000 if publish_ms else None,
            "content_metadata_source": "douyin_detail",
            "platform_raw_payload": dict(selection.get("platform_raw_payload") or {}),
        }
        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(
            "douyin",
            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)])
        content_id = str(item.get(RAW_CONTENT_ID_KEY) or "")
        if not self._should_fetch_detail_for_video_url(search_selection) or not self.detail_path or 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(
                {
                    "douyin_detail_fallback_attempted": True,
                    "douyin_detail_fallback_status": "failed",
                    "douyin_detail_fallback_exception_type": type(exc).__name__,
                    "douyin_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(
            {
                "douyin_detail_fallback_attempted": True,
                "douyin_detail_fallback_status": (
                    "used" if detail_selection.get("play_url") else "no_valid_play_url"
                ),
            }
        )
        return {**detail_selection, "platform_raw_payload": payload}

    def _should_fetch_detail_for_video_url(self, selection: dict[str, Any]) -> bool:
        payload = selection.get("platform_raw_payload") if isinstance(selection.get("platform_raw_payload"), dict) else {}
        if not selection.get("play_url"):
            return True
        host = str(payload.get("selected_video_url_host") or "").lower()
        if host in {"v11-weba.douyinvod.com", "v96-hcc.douyinvod.com"}:
            return True
        return str(payload.get("selected_video_url_probe_status") or "") in {"failed", "failed_fallback"}

    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=SEARCH_RATE_LIMIT_BUCKET,
        )
        block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
        return block.get("data", {}) if isinstance(block.get("data"), 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 _post_keyword_json(self, query: dict[str, Any]) -> dict[str, Any]:
        attempts: list[dict[str, Any]] = []
        for index, (role, base_url) in enumerate(self._crawapi_base_urls):
            provider = _keyword_provider_for_base_url(role, base_url)
            payload = self._keyword_payload_for_provider(provider, query)
            try:
                return post_crawapi_json(
                    http_client=self.http_client,
                    base_url=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=RATE_LIMIT_BUSINESS_CODES,
                )
            except ContentAgentError as exc:
                attempt = _crawapi_host_attempt_summary(role, base_url, exc, provider=provider)
                self._raise_crawapi_error(exc, attempts + [attempt])
            except (CrawapiBusinessError, CrawapiTransientError, RuntimeError) as exc:
                attempt = _crawapi_host_attempt_summary(role, base_url, exc, provider=provider)
                if index == len(self._crawapi_base_urls) - 1 or not _should_try_crawapi_candidate(exc):
                    self._raise_crawapi_error(exc, attempts + [attempt])
                attempts.append(attempt)
        raise RuntimeError("crawapi keyword_search failed: no base_url configured")

    def _keyword_payload_for_provider(self, provider: str, query: dict[str, Any]) -> dict[str, Any]:
        if provider == "crawler":
            return {
                "content_type": self.default_content_type,
                "keyword": query["search_query"],
                "cursor": _query_cursor(query, self.crawler_cursor_default),
                "sort_type": self.crawler_sort_type,
            }
        return {
            RAW_AUTHOR_ACCOUNT_KEY: self.piaoquantv_account_id,
            "keyword": query["search_query"],
            "content_type": self.default_content_type,
            "sort_type": self.default_sort_type,
            "publish_time": self.default_publish_time,
            "duration": self.piaoquantv_duration,
            "cursor": _query_cursor(query, self.default_cursor),
            "cookie_batch": self.piaoquantv_cookie_batch,
        }

    def _post_json(
        self,
        path: str,
        payload: dict[str, Any],
        operation: str,
        rate_limit_bucket: str | None = None,
    ) -> dict[str, Any]:
        attempts: list[dict[str, Any]] = []
        base_urls = self._legacy_crawapi_base_urls
        for index, (role, base_url) in enumerate(base_urls):
            try:
                return post_crawapi_json(
                    http_client=self.http_client,
                    base_url=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=RATE_LIMIT_BUSINESS_CODES,
                )
            except ContentAgentError as exc:
                attempt = _crawapi_host_attempt_summary(role, base_url, exc)
                self._raise_crawapi_error(exc, attempts + [attempt])
            except (CrawapiBusinessError, CrawapiTransientError, RuntimeError) as exc:
                attempt = _crawapi_host_attempt_summary(role, base_url, exc)
                if index == len(base_urls) - 1 or not _should_try_crawapi_candidate(exc):
                    self._raise_crawapi_error(exc, attempts + [attempt])
                attempts.append(attempt)
        raise RuntimeError(f"crawapi {operation} failed: no base_url configured")

    def _raise_crawapi_error(self, exc: Exception, attempts: list[dict[str, Any]]) -> None:
        if len(self._crawapi_base_urls) > 1:
            detail = getattr(exc, "detail", None)
            if not isinstance(detail, dict):
                detail = {}
                setattr(exc, "detail", detail)
            detail["crawapi_host_attempts"] = attempts
        raise exc


def _dedupe_crawapi_base_urls(candidates: list[tuple[str, str]]) -> list[tuple[str, str]]:
    result: list[tuple[str, str]] = []
    seen: set[str] = set()
    for role, base_url in candidates:
        if not base_url:
            continue
        normalized = base_url.rstrip("/") + "/"
        if normalized in seen:
            continue
        seen.add(normalized)
        result.append((role, normalized))
    return result


def _legacy_crawapi_base_urls(candidates: list[tuple[str, str]]) -> list[tuple[str, str]]:
    aiddit_candidates = [
        (role, base_url) for role, base_url in candidates if _is_aiddit_crawapi_base_url(base_url)
    ]
    return aiddit_candidates or candidates


def _is_aiddit_crawapi_base_url(base_url: str) -> bool:
    return "crawler.aiddit.com" in base_url.lower()


def _keyword_provider_for_base_url(role: str, base_url: str) -> str:
    if _is_aiddit_crawapi_base_url(base_url):
        return "crawler"
    normalized = base_url.lower()
    if "piaoquantv" in normalized:
        return "piaoquantv"
    return "crawler" if role == "candidate" else "piaoquantv"


def _query_cursor(query: dict[str, Any], default: str) -> str:
    if "page_cursor" in query:
        return str(query.get("page_cursor") or "")
    return str(default)


def _should_try_crawapi_candidate(exc: Exception) -> bool:
    if not isinstance(exc, CrawapiBusinessError):
        return True
    detail = getattr(exc, "detail", None)
    if not isinstance(detail, dict):
        return True
    business_code = str(detail.get("business_code") or "")
    business_message = str(detail.get("business_message") or "").lower()
    if business_code in NON_FALLBACK_BUSINESS_CODES:
        return False
    return not any(
        token.lower() in business_message
        for token in NON_FALLBACK_BUSINESS_MESSAGE_TOKENS
    )


def _crawapi_host_attempt_summary(
    role: str,
    base_url: str,
    exc: Exception,
    provider: str | None = None,
) -> dict[str, Any]:
    summary: dict[str, Any] = {
        "role": role,
        "provider": provider or _keyword_provider_for_base_url(role, base_url),
        "base_url": base_url.rstrip("/"),
        "exception_type": type(exc).__name__,
    }
    message = str(exc)
    if message:
        summary["exception_message"] = message[:300]
        http_status = _http_status_from_exception_message(message)
        if http_status is not None:
            summary["status_code"] = http_status
    if isinstance(exc, ContentAgentError):
        summary["error_code"] = exc.error_code.value
    detail = getattr(exc, "detail", None)
    if isinstance(detail, dict):
        for key in ("operation", "business_code", "business_message", "status_code"):
            if key in detail:
                summary[key] = detail.get(key)
        response_summary = detail.get("response_summary")
        if isinstance(response_summary, dict):
            for key in ("status_code", "content_type"):
                if key in response_summary:
                    summary[key] = response_summary.get(key)
        request_payload_summary = detail.get("request_payload_summary")
        if isinstance(request_payload_summary, dict):
            summary["request_payload_summary"] = request_payload_summary
    return summary


def _http_status_from_exception_message(message: str) -> int | None:
    if "HTTP " not in message:
        return None
    status_text = message.split("HTTP ", 1)[1].split(";", 1)[0].strip()
    return int(status_text) if status_text.isdigit() else None


def _extract_play_url(item: dict[str, Any]) -> str | None:
    video = item.get("video") if isinstance(item.get("video"), dict) else {}
    play_addr = video.get("play_addr") if isinstance(video.get("play_addr"), dict) else {}
    url_list = play_addr.get("url_list") or []
    return str(url_list[0]) if url_list else None


def _extract_tags(item: dict[str, Any]) -> list[str]:
    tags: list[str] = []
    for tag in item.get("cha_list") or []:
        if isinstance(tag, str):
            tags.append(tag if tag.startswith("#") else f"#{tag}")
        elif isinstance(tag, dict):
            name = tag.get("cha_name") or tag.get("hashtag_name") or tag.get("name")
            if name:
                tags.append(str(name) if str(name).startswith("#") else f"#{name}")
    for text in item.get("text_extra") or []:
        if isinstance(text, dict) and text.get("hashtag_name"):
            tags.append(f"#{text['hashtag_name']}")
    return list(dict.fromkeys(tags))