Last30days

interface:
  display_name: "Last 30 Days"
  short_description: "Research any topic across Reddit, X, YouTube, and the web from the last 30 days. Returns synthesized expert answers and copy-paste prompts."
  default_prompt: "Research this topic from the last 30 days across Reddit, X, YouTube, and web. Synthesize what people are actually saying, upvoting, and sharing right now."
  brand_color: "#FF6B35"

policy:
  allow_implicit_invocation: true

#!/usr/bin/env python3
"""
last30days-cn - 研究过去30天内中国平台上的热门话题。

Author: Jesse (https://github.com/Jesseovo)

Usage:
    python3 last30days.py <topic> [options]

Options:
    --emit=MODE         输出模式: compact|json|md|context|path (default: compact)
    --quick             快速搜索，减少数据源
    --deep              深度搜索，更多数据源
    --debug             启用调试日志
    --days N            回溯天数 (1-30, default: 30)
    --search SOURCES    指定搜索源 (逗号分隔): weibo,xiaohongshu,bilibili,zhihu,douyin,wechat,baidu,toutiao
    --diagnose          显示数据源可用性诊断
"""

import argparse
import atexit
import json
import os
import signal
import sys
import threading
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from pathlib import Path

SCRIPT_DIR = Path(__file__).parent.resolve()
sys.path.insert(0, str(SCRIPT_DIR))

_child_pids: set = set()
_child_pids_lock = threading.Lock()

TIMEOUT_PROFILES = {
    "quick":   {"global": 90,  "future": 30, "weibo_future": 30, "bilibili_future": 30, "zhihu_future": 30, "douyin_future": 60, "xiaohongshu_future": 30, "wechat_future": 30, "baidu_future": 30, "toutiao_future": 15, "http": 15},
    "default": {"global": 180, "future": 60, "weibo_future": 60, "bilibili_future": 60, "zhihu_future": 60, "douyin_future": 90, "xiaohongshu_future": 60, "wechat_future": 60, "baidu_future": 60, "toutiao_future": 30, "http": 30},
    "deep":    {"global": 300, "future": 90, "weibo_future": 90, "bilibili_future": 90, "zhihu_future": 90, "douyin_future": 120, "xiaohongshu_future": 90, "wechat_future": 90, "baidu_future": 90, "toutiao_future": 45, "http": 30},
}

VALID_SEARCH_SOURCES = {
    "weibo", "xiaohongshu", "xhs", "bilibili", "zhihu",
    "douyin", "wechat", "baidu", "toutiao",
}


def parse_search_flag(search_str: str) -> set:
    sources = set()
    for s in search_str.split(","):
        s = s.strip().lower()
        if not s:
            continue
        if s == "xhs":
            s = "xiaohongshu"
        if s not in VALID_SEARCH_SOURCES:
            print(f"错误: 未知搜索源 '{s}'。可用: {', '.join(sorted(VALID_SEARCH_SOURCES))}", file=sys.stderr)
            sys.exit(1)
        sources.add(s)
    if not sources:
        print("错误: --search 需要至少一个搜索源。", file=sys.stderr)
        sys.exit(1)
    return sources


def _cleanup_children():
    with _child_pids_lock:
        pids = list(_child_pids)
    for pid in pids:
        try:
            os.kill(pid, signal.SIGTERM)
        except (ProcessLookupError, PermissionError, OSError):
            pass

atexit.register(_cleanup_children)


def _install_global_timeout(timeout_seconds: int):
    if hasattr(signal, 'SIGALRM'):
        def _handler(signum, frame):
            sys.stderr.write(f"\n[超时] 全局超时 ({timeout_seconds}s) 已超过。正在清理。\n")
            sys.stderr.flush()
            _cleanup_children()
            sys.exit(1)
        signal.signal(signal.SIGALRM, _handler)
        signal.alarm(timeout_seconds)
    else:
        def _watchdog():
            sys.stderr.write(f"\n[超时] 全局超时 ({timeout_seconds}s) 已超过。正在清理。\n")
            sys.stderr.flush()
            _cleanup_children()
            os._exit(1)
        timer = threading.Timer(timeout_seconds, _watchdog)
        timer.daemon = True
        timer.start()


from lib import (
    weibo,
    xiaohongshu,
    bilibili,
    zhihu,
    douyin,
    wechat,
    baidu,
    toutiao,
    dates,
    dedupe,
    env,
    normalize,
    query,
    render,
    schema,
    score,
    setup_wizard,
    query_type as qt,
    crawler_bridge,
)


def _search_weibo(topic, config, from_date, to_date, depth):
    try:
        token = config.get("WEIBO_ACCESS_TOKEN")
        items = weibo.search_weibo(topic, from_date, to_date, depth=depth, token=token)
        return items, None
    except Exception as e:
        return [], f"{type(e).__name__}: {e}"

def _search_xiaohongshu(topic, config, from_date, to_date, depth):
    try:
        token = config.get("SCRAPECREATORS_API_KEY")
        api_base = env.get_xiaohongshu_api_base(config)
        items = xiaohongshu.search_xiaohongshu(topic, from_date, to_date, depth=depth, token=token, api_base=api_base)
        return items, None
    except Exception as e:
        return [], f"{type(e).__name__}: {e}"

def _search_bilibili(topic, from_date, to_date, depth):
    try:
        items = bilibili.search_bilibili(topic, from_date, to_date, depth=depth)
        return items, None
    except Exception as e:
        return [], f"{type(e).__name__}: {e}"

def _search_zhihu(topic, config, from_date, to_date, depth):
    try:
        cookie = config.get("ZHIHU_COOKIE")
        items = zhihu.search_zhihu(topic, from_date, to_date, depth=depth, cookie=cookie)
        return items, None
    except Exception as e:
        return [], f"{type(e).__name__}: {e}"

def _search_douyin(topic, config, from_date, to_date, depth):
    try:
        token = config.get("TIKHUB_API_KEY") or config.get("DOUYIN_API_KEY")
        items = douyin.search_douyin(topic, from_date, to_date, depth=depth, token=token)
        return items, None
    except Exception as e:
        return [], f"{type(e).__name__}: {e}"

def _search_wechat(topic, config, from_date, to_date, depth):
    try:
        api_key = config.get("WECHAT_API_KEY")
        items = wechat.search_wechat(topic, from_date, to_date, depth=depth, api_key=api_key)
        return items, None
    except Exception as e:
        return [], f"{type(e).__name__}: {e}"

def _search_baidu(topic, config, from_date, to_date, depth):
    try:
        api_key = config.get("BAIDU_API_KEY")
        secret_key = config.get("BAIDU_SECRET_KEY")
        items = baidu.search_baidu(topic, from_date, to_date, depth=depth, api_key=api_key, secret_key=secret_key)
        return items, None
    except Exception as e:
        return [], f"{type(e).__name__}: {e}"

def _search_toutiao(topic, from_date, to_date, depth):
    try:
        items = toutiao.search_toutiao(topic, from_date, to_date, depth=depth)
        return items, None
    except Exception as e:
        return [], f"{type(e).__name__}: {e}"


def run_research(
    topic: str,
    config: dict,
    from_date: str,
    to_date: str,
    depth: str = "default",
    timeouts: dict = None,
    search_sources: set = None,
    query_type: str = "breaking_news",
) -> dict:
    if timeouts is None:
        timeouts = TIMEOUT_PROFILES[depth]
    future_timeout = timeouts["future"]

    all_sources = {"weibo", "xiaohongshu", "bilibili", "zhihu", "douyin", "wechat", "baidu", "toutiao"}
    if search_sources:
        active = search_sources & all_sources
    else:
        active = {s for s in all_sources if qt.is_source_enabled(s, query_type)}

    results = {src: {"items": [], "error": None} for src in all_sources}

    futures = {}
    max_workers = len(active)

    with ThreadPoolExecutor(max_workers=max(max_workers, 1)) as executor:
        if "weibo" in active:
            sys.stderr.write("[微博] 搜索中...\n")
            futures["weibo"] = executor.submit(_search_weibo, topic, config, from_date, to_date, depth)
        if "xiaohongshu" in active:
            sys.stderr.write("[小红书] 搜索中...\n")
            futures["xiaohongshu"] = executor.submit(_search_xiaohongshu, topic, config, from_date, to_date, depth)
        if "bilibili" in active:
            sys.stderr.write("[B站] 搜索中...\n")
            futures["bilibili"] = executor.submit(_search_bilibili, topic, from_date, to_date, depth)
        if "zhihu" in active:
            sys.stderr.write("[知乎] 搜索中...\n")
            futures["zhihu"] = executor.submit(_search_zhihu, topic, config, from_date, to_date, depth)
        if "douyin" in active:
            sys.stderr.write("[抖音] 搜索中...\n")
            futures["douyin"] = executor.submit(_search_douyin, topic, config, from_date, to_date, depth)
        if "wechat" in active:
            sys.stderr.write("[微信] 搜索中...\n")
            futures["wechat"] = executor.submit(_search_wechat, topic, config, from_date, to_date, depth)
        if "baidu" in active:
            sys.stderr.write("[百度] 搜索中...\n")
            futures["baidu"] = executor.submit(_search_baidu, topic, config, from_date, to_date, depth)
        if "toutiao" in active:
            sys.stderr.write("[头条] 搜索中...\n")
            futures["toutiao"] = executor.submit(_search_toutiao, topic, from_date, to_date, depth)

        for source, future in futures.items():
            timeout = timeouts.get(f"{source}_future", future_timeout)
            try:
                items, error = future.result(timeout=timeout)
                results[source]["items"] = items
                results[source]["error"] = error
                if error:
                    sys.stderr.write(f"[{source}] 错误: {error}\n")
                else:
                    sys.stderr.write(f"[{source}] {len(items)} 条结果\n")
            except TimeoutError:
                results[source]["error"] = f"{source} 搜索超时 ({timeout}s)"
                sys.stderr.write(f"[{source}] 超时 ({timeout}s)\n")
            except Exception as e:
                results[source]["error"] = f"{type(e).__name__}: {e}"
                sys.stderr.write(f"[{source}] 错误: {e}\n")

    sys.stderr.flush()
    return results


def main():
    if sys.platform == "win32":
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")

    parser = argparse.ArgumentParser(description="研究过去N天内中国平台上的热门话题")
    parser.add_argument("topic", nargs="*", help="研究主题")
    parser.add_argument("--emit", choices=["compact", "json", "md", "context", "path"], default="compact", help="输出模式")
    parser.add_argument("--quick", action="store_true", help="快速搜索")
    parser.add_argument("--deep", action="store_true", help="深度搜索")
    parser.add_argument("--debug", action="store_true", help="启用调试日志")
    parser.add_argument("--days", type=int, default=30, choices=range(1, 31), metavar="N", help="回溯天数 (1-30)")
    parser.add_argument("--diagnose", action="store_true", help="显示数据源诊断")
    parser.add_argument("--timeout", type=int, default=None, metavar="SECS", help="全局超时秒数")
    parser.add_argument("--search", type=str, default=None, metavar="SOURCES", help="逗号分隔的搜索源列表")
    parser.add_argument("--save-dir", type=str, default=None, metavar="DIR", help="自动保存原始输出")

    args = parser.parse_args()
    args.topic = " ".join(args.topic) if args.topic else None

    if args.debug:
        os.environ["LAST30DAYS_DEBUG"] = "1"

    if args.quick and args.deep:
        print("错误: 不能同时使用 --quick 和 --deep", file=sys.stderr)
        sys.exit(1)
    elif args.quick:
        depth = "quick"
    elif args.deep:
        depth = "deep"
    else:
        depth = "default"

    timeouts = TIMEOUT_PROFILES[depth]
    global_timeout = args.timeout or timeouts["global"]
    _install_global_timeout(global_timeout)

    config = env.get_config()

    if args.diagnose:
        crawler_status = crawler_bridge.get_crawler_status()
        diag = {
            "weibo": env.is_weibo_available(config),
            "xiaohongshu": env.is_xiaohongshu_available(config),
            "bilibili": True,
            "zhihu": True,
            "douyin": env.is_douyin_available(config),
            "wechat": env.is_wechat_available(config),
            "baidu_api": env.is_baidu_api_available(config),
            "toutiao": True,
            "xiaohongshu_api_base": env.get_xiaohongshu_api_base(config),
            "crawler_engine": {
                "playwright_available": crawler_status["playwright_available"],
                "cached_logins": crawler_status["cached_logins"],
                "note": "安装 Playwright 后，微博/小红书/抖音/B站/知乎可无需 API Key 使用爬虫模式",
            },
        }
        print(json.dumps(diag, indent=2, ensure_ascii=False))
        sys.exit(0)

    if args.topic and args.topic.strip().lower() == "setup":
        results = setup_wizard.run_auto_setup(config)
        env_path = env.CONFIG_FILE
        if env_path:
            written = setup_wizard.write_setup_config(env_path)
            results["env_written"] = written
        else:
            results["env_written"] = False
        print(setup_wizard.get_setup_status_text(results))
        sys.exit(0)

    if not args.topic:
        print("错误: 请提供研究主题。", file=sys.stderr)
        print("用法: python3 last30days.py <topic> [options]", file=sys.stderr)
        sys.exit(1)

    from_date, to_date = dates.get_date_range(args.days)

    search_sources = None
    if args.search:
        search_sources = parse_search_flag(args.search)

    query_type = qt.detect_query_type(args.topic)
    search_topic = query.extract_core_subject(args.topic)

    sys.stderr.write(f"正在搜索: {args.topic}\n")
    if search_topic != args.topic:
        sys.stderr.write(f"提纯关键词: {search_topic}\n")
    sys.stderr.write(f"查询类型: {query_type} | 日期范围: {from_date} 至 {to_date}\n")
    sys.stderr.flush()

    raw_results = run_research(
        search_topic, config, from_date, to_date, depth,
        timeouts=timeouts, search_sources=search_sources,
        query_type=query_type,
    )

    sys.stderr.write("正在处理结果...\n")
    sys.stderr.flush()

    norm_weibo = normalize.normalize_weibo_items(raw_results["weibo"]["items"], from_date, to_date)
    norm_xhs = normalize.normalize_xiaohongshu_items(raw_results["xiaohongshu"]["items"], from_date, to_date)
    norm_bili = normalize.normalize_bilibili_items(raw_results["bilibili"]["items"], from_date, to_date)
    norm_zhihu = normalize.normalize_zhihu_items(raw_results["zhihu"]["items"], from_date, to_date)
    norm_douyin = normalize.normalize_douyin_items(raw_results["douyin"]["items"], from_date, to_date)
    norm_wechat = normalize.normalize_wechat_items(raw_results["wechat"]["items"], from_date, to_date)
    norm_baidu = normalize.normalize_baidu_items(raw_results["baidu"]["items"], from_date, to_date)
    norm_toutiao = normalize.normalize_toutiao_items(raw_results["toutiao"]["items"], from_date, to_date)

    filt_weibo = normalize.filter_by_date_range(norm_weibo, from_date, to_date)
    filt_xhs = normalize.filter_by_date_range(norm_xhs, from_date, to_date)
    filt_bili = normalize.filter_by_date_range(norm_bili, from_date, to_date)
    filt_zhihu = normalize.filter_by_date_range(norm_zhihu, from_date, to_date)
    filt_douyin = normalize.filter_by_date_range(norm_douyin, from_date, to_date)
    filt_wechat = normalize.filter_by_date_range(norm_wechat, from_date, to_date)
    filt_baidu = normalize.filter_by_date_range(norm_baidu, from_date, to_date)
    filt_toutiao = normalize.filter_by_date_range(norm_toutiao, from_date, to_date)

    scored_weibo = score.score_weibo_items(filt_weibo)
    scored_xhs = score.score_xiaohongshu_items(filt_xhs)
    scored_bili = score.score_bilibili_items(filt_bili)
    scored_zhihu = score.score_zhihu_items(filt_zhihu)
    scored_douyin = score.score_douyin_items(filt_douyin)
    scored_wechat = score.score_wechat_items(filt_wechat, query_type=query_type)
    scored_baidu = score.score_baidu_items(filt_baidu, query_type=query_type)
    scored_toutiao = score.score_toutiao_items(filt_toutiao)

    sorted_weibo = score.sort_items(scored_weibo, query_type=query_type)
    sorted_xhs = score.sort_items(scored_xhs, query_type=query_type)
    sorted_bili = score.sort_items(scored_bili, query_type=query_type)
    sorted_zhihu = score.sort_items(scored_zhihu, query_type=query_type)
    sorted_douyin = score.sort_items(scored_douyin, query_type=query_type)
    sorted_wechat = score.sort_items(scored_wechat, query_type=query_type)
    sorted_baidu = score.sort_items(scored_baidu, query_type=query_type)
    sorted_toutiao = score.sort_items(scored_toutiao, query_type=query_type)

    deduped_weibo = dedupe.dedupe_weibo(sorted_weibo)
    deduped_xhs = dedupe.dedupe_xiaohongshu(sorted_xhs)
    deduped_bili = dedupe.dedupe_bilibili(sorted_bili)
    deduped_zhihu = dedupe.dedupe_zhihu(sorted_zhihu)
    deduped_douyin = dedupe.dedupe_douyin(sorted_douyin)
    deduped_wechat = dedupe.dedupe_wechat(sorted_wechat)
    deduped_baidu = dedupe.dedupe_baidu(sorted_baidu)
    deduped_toutiao = dedupe.dedupe_toutiao(sorted_toutiao)

    deduped_weibo = score.relevance_filter(deduped_weibo, "WEIBO")
    deduped_xhs = score.relevance_filter(deduped_xhs, "XIAOHONGSHU")
    deduped_bili = score.relevance_filter(deduped_bili, "BILIBILI")
    deduped_zhihu = score.relevance_filter(deduped_zhihu, "ZHIHU")
    deduped_douyin = score.relevance_filter(deduped_douyin, "DOUYIN")
    deduped_wechat = score.relevance_filter(deduped_wechat, "WECHAT")
    deduped_baidu = score.relevance_filter(deduped_baidu, "BAIDU")
    deduped_toutiao = score.relevance_filter(deduped_toutiao, "TOUTIAO")

    dedupe.cross_source_link(
        deduped_weibo, deduped_xhs, deduped_bili, deduped_zhihu,
        deduped_douyin, deduped_wechat, deduped_baidu, deduped_toutiao,
    )

    report = schema.create_report(args.topic, from_date, to_date, "all")
    report.weibo = deduped_weibo
    report.xiaohongshu = deduped_xhs
    report.bilibili = deduped_bili
    report.zhihu = deduped_zhihu
    report.douyin = deduped_douyin
    report.wechat = deduped_wechat
    report.baidu = deduped_baidu
    report.toutiao = deduped_toutiao
    report.weibo_error = raw_results["weibo"]["error"]
    report.xiaohongshu_error = raw_results["xiaohongshu"]["error"]
    report.bilibili_error = raw_results["bilibili"]["error"]
    report.zhihu_error = raw_results["zhihu"]["error"]
    report.douyin_error = raw_results["douyin"]["error"]
    report.wechat_error = raw_results["wechat"]["error"]
    report.baidu_error = raw_results["baidu"]["error"]
    report.toutiao_error = raw_results["toutiao"]["error"]

    report.context_snippet_md = render.render_context_snippet(report)
    render.write_outputs(report)

    total = sum(len(getattr(report, src, [])) for src in ["weibo", "xiaohongshu", "bilibili", "zhihu", "douyin", "wechat", "baidu", "toutiao"])
    sys.stderr.write(f"\n完成! 共 {total} 条结果\n")
    sys.stderr.flush()

    if args.emit == "compact":
        print(render.render_compact(report))
        print(render.render_source_status(report))
    elif args.emit == "json":
        print(json.dumps(report.to_dict(), indent=2, ensure_ascii=False))
    elif args.emit == "md":
        print(render.render_full_report(report))
    elif args.emit == "context":
        print(report.context_snippet_md)
    elif args.emit == "path":
        print(render.get_context_path())

    if args.save_dir:
        import re as re_mod
        save_dir = Path(args.save_dir).expanduser()
        save_dir.mkdir(parents=True, exist_ok=True)
        slug = re_mod.sub(r'[^a-z0-9\u4e00-\u9fff]+', '-', args.topic.lower()).strip('-')[:60]
        save_path = save_dir / f"{slug}-raw.md"
        if save_path.exists():
            save_path = save_dir / f"{slug}-raw-{datetime.now().strftime('%Y-%m-%d')}.md"
        content = render.render_compact(report)
        content += "\n" + render.render_source_status(report)
        save_path.write_text(content, encoding="utf-8")
        print(f"已保存: {save_path}", file=sys.stderr)


if __name__ == "__main__":
    main()

# last30days library modules

"""百度搜索模块 - 替代 Exa/Brave 等 Web 搜索后端。

Author: Jesse (https://github.com/Jesseovo)

v2.1 改动：
- 更真实的浏览器头（Accept-Language/Referer/Sec-Fetch-*）+ UA 轮换
- 若检测到「百度安全验证」拦截页，主动记录日志并降级
- 新增 Bing 国内版兜底（`https://cn.bing.com/search`）
- 同时兼容百度新版 HTML 的 `result c-container` 块级选择器
"""

import json
import random
import re
import sys
import time
import urllib.parse
import urllib.request
from typing import Any, Dict, List, Optional

from . import relevance

_UA_POOL = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36 Edg/125.0.0.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
]

_ANTIBOT_SIGNATURES = (
    "wappass.baidu.com",
    "百度安全验证",
    "百度智能云验证",
    "verify.baidu.com",
    "/static/verify",
)


def search_baidu(
    topic: str,
    from_date: str,
    to_date: str,
    depth: str = "default",
    api_key: Optional[str] = None,
    secret_key: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """搜索百度网页。

    Args:
        topic: 搜索关键词
        from_date: 起始日期
        to_date: 结束日期
        depth: 搜索深度
        api_key: 百度 API Key（可选）
        secret_key: 百度 Secret Key（可选）

    Returns:
        网页搜索结果列表
    """
    limit_map = {"quick": 10, "default": 20, "deep": 30}
    limit = limit_map.get(depth, 20)

    items: List[Dict[str, Any]] = []

    if api_key and secret_key:
        items = _search_via_api(topic, limit, api_key, secret_key)

    if not items:
        items = _search_via_public(topic, limit)

    if not items:
        items = _search_via_bing(topic, limit)

    scored = []
    for i, item in enumerate(items):
        title = item.get("title", "")
        snippet = item.get("snippet", "")
        combined = f"{title} {snippet}"
        rel = relevance.token_overlap_relevance(topic, combined)
        item["id"] = f"BD{i+1}"
        item["relevance"] = rel
        item["why_relevant"] = f"{item.get('source_tag', '百度搜索')}：{title[:50]}"
        scored.append(item)

    scored.sort(key=lambda x: x.get("relevance", 0), reverse=True)
    return scored[:limit]


def _build_headers(referer: str) -> Dict[str, str]:
    ua = random.choice(_UA_POOL)
    return {
        "User-Agent": ua,
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
        "Accept-Encoding": "identity",
        "Referer": referer,
        "Sec-Ch-Ua": '"Chromium";v="124", "Not-A.Brand";v="99"',
        "Sec-Ch-Ua-Mobile": "?0",
        "Sec-Ch-Ua-Platform": '"Windows"',
        "Sec-Fetch-Dest": "document",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Site": "same-origin",
        "Sec-Fetch-User": "?1",
        "Upgrade-Insecure-Requests": "1",
    }


def _is_antibot_page(html: str) -> bool:
    if not html:
        return True
    if len(html) < 2000:
        return any(sig in html for sig in _ANTIBOT_SIGNATURES)
    return any(sig in html for sig in _ANTIBOT_SIGNATURES)


def _fetch(url: str, headers: Dict[str, str], timeout: int = 15) -> str:
    req = urllib.request.Request(url, headers=headers)
    with urllib.request.urlopen(req, timeout=timeout) as response:
        return response.read().decode("utf-8", errors="replace")


def _search_via_api(
    topic: str, limit: int, api_key: str, secret_key: str
) -> List[Dict[str, Any]]:
    """通过百度搜索 API 进行搜索。"""
    items = []
    try:
        encoded = urllib.parse.quote(topic)
        url = f"https://api.baidu.com/search/v1?q={encoded}&rn={limit}&key={api_key}"
        req = urllib.request.Request(url, headers={"User-Agent": _UA_POOL[0]})
        with urllib.request.urlopen(req, timeout=15) as response:
            data = json.loads(response.read().decode("utf-8"))

        for r in data.get("result", []):
            items.append({
                "title": _clean_html(r.get("title", "")),
                "snippet": _clean_html(r.get("abstract", "") or r.get("description", "")),
                "url": r.get("url", ""),
                "source_domain": _extract_domain(r.get("url", "")),
                "date": r.get("date"),
                "date_confidence": "med" if r.get("date") else "low",
                "source_tag": "百度API",
            })
    except Exception as e:
        sys.stderr.write(f"[百度] API 搜索失败: {e}\n")
    return items


def _search_via_public(topic: str, limit: int) -> List[Dict[str, Any]]:
    """通过百度公开搜索（HTML 解析）。"""
    items: List[Dict[str, Any]] = []
    try:
        encoded = urllib.parse.quote(topic)
        now_ts = int(time.time())
        ago_ts = now_ts - 30 * 86400
        url = (
            f"https://www.baidu.com/s?wd={encoded}&rn={min(limit, 20)}"
            f"&gpc=stf%3D{ago_ts}%2C{now_ts}%7Cstftype%3D1"
        )
        headers = _build_headers("https://www.baidu.com/")
        html = _fetch(url, headers)

        if _is_antibot_page(html):
            sys.stderr.write(
                "[百度] 公开搜索被安全验证拦截，已自动降级到 Bing 兜底。"
                "建议配置 BAIDU_API_KEY/BAIDU_SECRET_KEY 以获得稳定结果。\n"
            )
            return []

        container_blocks = re.findall(
            r'<div[^>]*class="[^"]*result[^"]*c-container[^"]*"[^>]*>([\s\S]*?)</div>\s*(?=<div[^>]*class="[^"]*result|<div[^>]*id="content_bottom)',
            html,
        )

        for block in container_blocks[:limit]:
            title_match = re.search(
                r'<h3[^>]*>\s*<a[^>]*href="([^"]+)"[^>]*>(.*?)</a>',
                block,
                re.S,
            )
            if not title_match:
                continue
            href = title_match.group(1)
            title = _clean_html(title_match.group(2))

            snippet = ""
            snip_match = re.search(
                r'<span[^>]*class="[^"]*content-right_[^"]*"[^>]*>([\s\S]*?)</span>',
                block,
            )
            if not snip_match:
                snip_match = re.search(
                    r'<span[^>]*class="[^"]*c-font-normal[^"]*"[^>]*>([\s\S]*?)</span>',
                    block,
                )
            if snip_match:
                snippet = _clean_html(snip_match.group(1))

            items.append({
                "title": title,
                "snippet": snippet,
                "url": href,
                "source_domain": _extract_domain(href),
                "date": None,
                "date_confidence": "low",
                "source_tag": "百度",
            })

        if not items:
            results = re.findall(
                r'<h3[^>]*class="[^"]*t[^"]*"[^>]*>\s*<a[^>]*href="([^"]*)"[^>]*>(.*?)</a>',
                html,
                re.S,
            )
            snippets = re.findall(
                r'<span class="content-right_8Zs40">(.*?)</span>', html, re.S
            )
            for idx, (href, title_html) in enumerate(results[:limit]):
                title = _clean_html(title_html)
                snippet = _clean_html(snippets[idx]) if idx < len(snippets) else ""
                items.append({
                    "title": title,
                    "snippet": snippet,
                    "url": href,
                    "source_domain": _extract_domain(href),
                    "date": None,
                    "date_confidence": "low",
                    "source_tag": "百度",
                })
    except Exception as e:
        sys.stderr.write(f"[百度] 公开搜索失败: {e}\n")
    return items


def _search_via_bing(topic: str, limit: int) -> List[Dict[str, Any]]:
    """Bing 国内版兜底搜索，避免百度完全失效时返回 0 条。"""
    items: List[Dict[str, Any]] = []
    try:
        encoded = urllib.parse.quote(topic)
        url = f"https://cn.bing.com/search?q={encoded}&setmkt=zh-CN&ensearch=0"
        headers = _build_headers("https://cn.bing.com/")
        html = _fetch(url, headers)

        blocks = re.findall(
            r'<li class="b_algo"[^>]*>([\s\S]*?)</li>',
            html,
        )
        for block in blocks[:limit]:
            title_match = re.search(
                r'<h2[^>]*>\s*<a[^>]*href="([^"]+)"[^>]*>(.*?)</a>\s*</h2>',
                block,
                re.S,
            )
            if not title_match:
                continue
            href = title_match.group(1)
            title = _clean_html(title_match.group(2))

            snippet = ""
            snip_match = re.search(
                r'<p[^>]*class="b_lineclamp\d*\s*b_algoSlug"[^>]*>([\s\S]*?)</p>',
                block,
            ) or re.search(r"<p[^>]*>([\s\S]*?)</p>", block)
            if snip_match:
                snippet = _clean_html(snip_match.group(1))

            items.append({
                "title": title,
                "snippet": snippet,
                "url": href,
                "source_domain": _extract_domain(href),
                "date": None,
                "date_confidence": "low",
                "source_tag": "Bing兜底",
            })
    except Exception as e:
        sys.stderr.write(f"[百度] Bing 兜底搜索失败: {e}\n")
    return items


def _clean_html(text: str) -> str:
    """清除 HTML 标签。"""
    return re.sub(r"<[^>]+>", "", text or "").strip()


def _extract_domain(url: str) -> str:
    """从 URL 中提取域名。"""
    try:
        from urllib.parse import urlparse
        return urlparse(url).netloc
    except Exception:
        return ""

"""B站搜索模块 - 搜索哔哩哔哩视频内容。

Author: Jesse (https://github.com/Jesseovo)

支持两种模式（自动切换）：
1. B站公开搜索 API（无需 API Key）
2. MediaCrawler 浏览器爬虫（备用方案）
"""

import json
import re
import sys
import urllib.parse
import urllib.request
from typing import Any, Dict, List, Optional

from . import relevance

_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"


def search_bilibili(
    topic: str,
    from_date: str,
    to_date: str,
    depth: str = "default",
) -> List[Dict[str, Any]]:
    """搜索B站视频。

    Args:
        topic: 搜索关键词
        from_date: 起始日期 YYYY-MM-DD
        to_date: 结束日期 YYYY-MM-DD
        depth: 搜索深度 quick/default/deep

    Returns:
        B站视频列表
    """
    limit_map = {"quick": 10, "default": 20, "deep": 40}
    limit = limit_map.get(depth, 20)
    pages = 1 if depth == "quick" else (2 if depth == "default" else 3)

    items: List[Dict[str, Any]] = []

    for page_num in range(1, pages + 1):
        try:
            page_items = _search_page(topic, page_num)
            items.extend(page_items)
        except Exception as e:
            sys.stderr.write(f"[B站] 搜索第 {page_num} 页失败: {e}\n")
            break

    if not items:
        try:
            from . import crawler_bridge
            if crawler_bridge.is_playwright_available():
                sys.stderr.write("[B站] API 无结果，尝试 MediaCrawler 爬虫模式...\n")
                items = crawler_bridge.crawl_bilibili(topic, limit)
                if items:
                    sys.stderr.write(f"[B站] 爬虫模式获取 {len(items)} 条结果\n")
        except Exception as e:
            sys.stderr.write(f"[B站] 爬虫模式失败: {e}\n")

    scored = []
    for i, item in enumerate(items):
        title = _clean_html(item.get("title", ""))
        rel = relevance.token_overlap_relevance(topic, title)
        item["id"] = f"BL{i+1}"
        item["title"] = title
        item["relevance"] = rel
        item["why_relevant"] = f"B站视频：{title[:50]}"
        scored.append(item)

    scored.sort(key=lambda x: x.get("relevance", 0), reverse=True)
    return scored[:limit]


def _search_page(topic: str, page: int = 1) -> List[Dict[str, Any]]:
    """搜索B站单页结果。"""
    encoded = urllib.parse.quote(topic)
    url = (
        f"https://api.bilibili.com/x/web-interface/search/type"
        f"?search_type=video&keyword={encoded}&page={page}&page_size=20"
        f"&order=totalrank"
    )
    headers = {
        "User-Agent": _UA,
        "Referer": "https://search.bilibili.com/",
    }
    req = urllib.request.Request(url, headers=headers)
    with urllib.request.urlopen(req, timeout=15) as response:
        data = json.loads(response.read().decode("utf-8"))

    items = []
    results = data.get("data", {}).get("result", [])
    if not results:
        return items

    for r in results:
        items.append(_parse_video(r))

    return items


def _parse_video(v: dict) -> Dict[str, Any]:
    """解析B站视频搜索结果。"""
    bvid = v.get("bvid", "")
    pubdate = v.get("pubdate", 0)
    date_str = None
    if pubdate:
        try:
            from datetime import datetime
            date_str = datetime.fromtimestamp(pubdate).strftime("%Y-%m-%d")
        except Exception:
            pass

    return {
        "title": v.get("title", ""),
        "url": f"https://www.bilibili.com/video/{bvid}" if bvid else v.get("arcurl", ""),
        "bvid": bvid,
        "channel_name": v.get("author", ""),
        "author_mid": v.get("mid", ""),
        "date": date_str,
        "duration": v.get("duration", ""),
        "description": v.get("description", ""),
        "engagement": {
            "views": v.get("play", 0),
            "danmaku": v.get("danmaku", 0),
            "comments": v.get("review", 0) or v.get("comment", 0),
            "favorites": v.get("favorites", 0),
            "likes": v.get("like", 0),
        },
    }


def _clean_html(text: str) -> str:
    """清除搜索结果中的 HTML 高亮标签。"""
    text = re.sub(r"<[^>]+>", "", text)
    return text.strip()

"""Caching utilities for last30days skill.

Author: Jesse (https://github.com/Jesseovo)
"""

import hashlib
import json
import os
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional

CACHE_DIR = Path.home() / ".cache" / "last30days-cn"
DEFAULT_TTL_HOURS = 24
MODEL_CACHE_TTL_DAYS = 7
MODEL_CACHE_FILE = CACHE_DIR / "model_selection.json"


def ensure_cache_dir():
    """Ensure cache directory exists. Supports env override and sandbox fallback."""
    global CACHE_DIR, MODEL_CACHE_FILE
    env_dir = os.environ.get("LAST30DAYS_CACHE_DIR")
    if env_dir:
        CACHE_DIR = Path(env_dir)
        MODEL_CACHE_FILE = CACHE_DIR / "model_selection.json"

    try:
        CACHE_DIR.mkdir(parents=True, exist_ok=True)
    except PermissionError:
        CACHE_DIR = Path(tempfile.gettempdir()) / "last30days-cn" / "cache"
        MODEL_CACHE_FILE = CACHE_DIR / "model_selection.json"
        CACHE_DIR.mkdir(parents=True, exist_ok=True)


def get_cache_key(topic: str, from_date: str, to_date: str, sources: str) -> str:
    """Generate a cache key from query parameters."""
    key_data = f"{topic}|{from_date}|{to_date}|{sources}"
    return hashlib.sha256(key_data.encode()).hexdigest()[:16]


def get_cache_path(cache_key: str) -> Path:
    """Get path to cache file."""
    return CACHE_DIR / f"{cache_key}.json"


def is_cache_valid(cache_path: Path, ttl_hours: int = DEFAULT_TTL_HOURS) -> bool:
    """Check if cache file exists and is within TTL."""
    if not cache_path.exists():
        return False

    try:
        stat = cache_path.stat()
        mtime = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)
        now = datetime.now(timezone.utc)
        age_hours = (now - mtime).total_seconds() / 3600
        return age_hours < ttl_hours
    except OSError:
        return False


def load_cache(cache_key: str, ttl_hours: int = DEFAULT_TTL_HOURS) -> Optional[dict]:
    """Load data from cache if valid."""
    cache_path = get_cache_path(cache_key)

    if not is_cache_valid(cache_path, ttl_hours):
        return None

    try:
        with open(cache_path, 'r') as f:
            return json.load(f)
    except (json.JSONDecodeError, OSError):
        return None


def get_cache_age_hours(cache_path: Path) -> Optional[float]:
    """Get age of cache file in hours."""
    if not cache_path.exists():
        return None
    try:
        stat = cache_path.stat()
        mtime = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)
        now = datetime.now(timezone.utc)
        return (now - mtime).total_seconds() / 3600
    except OSError:
        return None


def load_cache_with_age(cache_key: str, ttl_hours: int = DEFAULT_TTL_HOURS) -> tuple:
    """Load data from cache with age info.

    Returns:
        Tuple of (data, age_hours) or (None, None) if invalid
    """
    cache_path = get_cache_path(cache_key)

    if not is_cache_valid(cache_path, ttl_hours):
        return None, None

    age = get_cache_age_hours(cache_path)

    try:
        with open(cache_path, 'r') as f:
            return json.load(f), age
    except (json.JSONDecodeError, OSError):
        return None, None


def save_cache(cache_key: str, data: dict):
    """Save data to cache."""
    ensure_cache_dir()
    cache_path = get_cache_path(cache_key)

    try:
        with open(cache_path, 'w') as f:
            json.dump(data, f)
    except OSError:
        pass  # Silently fail on cache write errors


def clear_cache():
    """Clear all cache files."""
    if CACHE_DIR.exists():
        for f in CACHE_DIR.glob("*.json"):
            try:
                f.unlink()
            except OSError:
                pass


# Model selection cache (longer TTL) — MODEL_CACHE_FILE is set at module level
# and updated by ensure_cache_dir() if env override or fallback is needed.


def load_model_cache() -> dict:
    """Load model selection cache."""
    if not is_cache_valid(MODEL_CACHE_FILE, MODEL_CACHE_TTL_DAYS * 24):
        return {}

    try:
        with open(MODEL_CACHE_FILE, 'r') as f:
            return json.load(f)
    except (json.JSONDecodeError, OSError):
        return {}


def save_model_cache(data: dict):
    """Save model selection cache."""
    ensure_cache_dir()
    try:
        with open(MODEL_CACHE_FILE, 'w') as f:
            json.dump(data, f)
    except OSError:
        pass


def get_cached_model(provider: str) -> Optional[str]:
    """Get cached model selection for a provider."""
    cache = load_model_cache()
    return cache.get(provider)


def set_cached_model(provider: str, model: str):
    """Cache model selection for a provider."""
    cache = load_model_cache()
    cache[provider] = model
    cache['updated_at'] = datetime.now(timezone.utc).isoformat()
    save_model_cache(cache)

"""MediaCrawler 爬虫桥接模块 — 基于 Playwright 浏览器自动化的数据采集。

灵感来源: https://github.com/NanmiCoder/MediaCrawler
技术原理: 利用 Playwright 控制浏览器，通过保留登录态的上下文获取平台数据，
         无需逆向复杂加密算法，无需付费 API Key。

v2.1 改动：
- 抽取 `_launch_browser_context` 公共函数，统一 locale/viewport/UA
- 小红书/抖音改用 `page.on("response")` XHR 拦截，避开 Virtual DOM 渲染
- `page.goto` 统一使用 `wait_until="domcontentloaded"`，配合条件轮询替代固定 sleep

⚠️ 免责声明:
本模块仅供学习和研究目的。使用者必须遵守相关法律法规及各平台的服务条款。
禁止用于商业用途、大规模数据采集或任何非法活动。

Author: Jesse (https://github.com/Jesseovo)
"""

import json
import os
import re
import sys
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Dict, List, Optional
from datetime import datetime

COOKIE_DIR = Path.home() / ".config" / "last30days-cn" / "browser_cookies"
_playwright_available: Optional[bool] = None

_DESKTOP_UA = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
)
_MOBILE_UA = (
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) "
    "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1"
)


def is_playwright_available() -> bool:
    """检查 Playwright 是否已安装并可用。"""
    global _playwright_available
    if _playwright_available is not None:
        return _playwright_available
    try:
        from playwright.sync_api import sync_playwright  # noqa: F401
        _playwright_available = True
    except ImportError:
        _playwright_available = False
    return _playwright_available


def _ensure_cookie_dir():
    COOKIE_DIR.mkdir(parents=True, exist_ok=True)


def _get_cookie_path(platform: str) -> Path:
    _ensure_cookie_dir()
    return COOKIE_DIR / f"{platform}_cookies.json"


def save_cookies(platform: str, cookies: list):
    path = _get_cookie_path(platform)
    path.write_text(json.dumps(cookies, ensure_ascii=False, indent=2), encoding="utf-8")


def load_cookies(platform: str) -> Optional[list]:
    path = _get_cookie_path(platform)
    if path.exists():
        try:
            return json.loads(path.read_text(encoding="utf-8"))
        except Exception:
            return None
    return None


def _clean_html(text: str) -> str:
    if not text:
        return ""
    text = re.sub(r"<[^>]+>", "", str(text))
    return re.sub(r"\s+", " ", text).strip()


@contextmanager
def _launch_browser_context(platform: str, mobile: bool = False, headless: bool = True):
    """统一构造 Playwright 浏览器上下文，自动加载并回写 cookies。

    Yields:
        (browser, context, page) 三元组
    """
    from playwright.sync_api import sync_playwright

    ua = _MOBILE_UA if mobile else _DESKTOP_UA
    viewport = {"width": 390, "height": 844} if mobile else {"width": 1280, "height": 800}

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=headless)
        try:
            context = browser.new_context(
                user_agent=ua,
                locale="zh-CN",
                timezone_id="Asia/Shanghai",
                viewport=viewport,
            )
            cookies = load_cookies(platform)
            if cookies:
                try:
                    context.add_cookies(cookies)
                except Exception as e:
                    sys.stderr.write(f"[爬虫-{platform}] 加载 Cookie 失败: {e}\n")

            page = context.new_page()
            try:
                yield browser, context, page
            finally:
                try:
                    save_cookies(platform, context.cookies())
                except Exception:
                    pass
        finally:
            try:
                browser.close()
            except Exception:
                pass


def _wait_for(page, predicate, timeout_ms: int = 8000, interval_ms: int = 500) -> bool:
    """轮询等待条件满足，返回是否在超时前命中。"""
    elapsed = 0
    while elapsed < timeout_ms:
        try:
            if predicate():
                return True
        except Exception:
            pass
        page.wait_for_timeout(interval_ms)
        elapsed += interval_ms
    return False


def crawl_weibo(topic: str, limit: int = 20) -> List[Dict[str, Any]]:
    """通过浏览器自动化爬取微博搜索结果。"""
    if not is_playwright_available():
        return []

    items: List[Dict[str, Any]] = []
    try:
        with _launch_browser_context("weibo", mobile=True) as (browser, context, page):
            search_url = (
                f"https://m.weibo.cn/api/container/getIndex?"
                f"containerid=100103type%3D1%26q%3D{topic}&page_type=searchall"
            )
            page.goto(
                f"https://m.weibo.cn/search?containerid=100103type%3D1%26q%3D{topic}",
                wait_until="domcontentloaded",
                timeout=20000,
            )
            page.wait_for_timeout(1500)

            response = page.evaluate(
                """
                async (url) => {
                    const resp = await fetch(url);
                    return await resp.json();
                }
                """,
                search_url,
            )

            if isinstance(response, dict):
                cards = response.get("data", {}).get("cards", [])
                for card in cards[:limit]:
                    if card.get("card_type") == 9:
                        mblog = card.get("mblog", {})
                        if mblog:
                            items.append(_parse_weibo_mblog(mblog))
                    elif card.get("card_type") == 11:
                        for group in card.get("card_group", []):
                            mblog = group.get("mblog", {})
                            if mblog:
                                items.append(_parse_weibo_mblog(mblog))
    except Exception as e:
        sys.stderr.write(f"[爬虫-微博] 浏览器爬取失败: {e}\n")
    return items


def _parse_weibo_mblog(mblog: dict) -> Dict[str, Any]:
    text = _clean_html(mblog.get("text", ""))
    user = mblog.get("user", {})
    mid = mblog.get("mid", "") or mblog.get("id", "")

    return {
        "text": text,
        "url": f"https://weibo.com/{user.get('id', '')}/{mid}",
        "author_handle": user.get("screen_name", ""),
        "author_id": str(user.get("id", "")),
        "date": _parse_relative_date(mblog.get("created_at", "")),
        "engagement": {
            "reposts": mblog.get("reposts_count", 0),
            "comments": mblog.get("comments_count", 0),
            "likes": mblog.get("attitudes_count", 0),
        },
        "source": "crawler",
    }


def crawl_xiaohongshu(topic: str, limit: int = 20) -> List[Dict[str, Any]]:
    """通过浏览器自动化爬取小红书搜索结果。

    v2.1：参考 MediaCrawler 思路，优先拦截 XHR `/api/sns/web/v1/search/notes`
    避开 Virtual DOM。失败时回退到 DOM 选择器（命中率较低，仅作兜底）。
    """
    if not is_playwright_available():
        return []

    items: List[Dict[str, Any]] = []
    try:
        with _launch_browser_context("xiaohongshu") as (browser, context, page):
            captured: Dict[str, Any] = {"payload": None}

            def _on_response(resp):
                try:
                    url = resp.url
                    if "/api/sns/web/v1/search/notes" in url and resp.status == 200:
                        captured["payload"] = resp.json()
                except Exception:
                    pass

            page.on("response", _on_response)

            search_url = (
                f"https://www.xiaohongshu.com/search_result?"
                f"keyword={topic}&source=web_search_result_notes"
            )
            try:
                page.goto(search_url, wait_until="domcontentloaded", timeout=30000)
            except Exception as e:
                sys.stderr.write(f"[爬虫-小红书] 页面加载失败: {e}\n")

            for _ in range(5):
                if captured["payload"]:
                    break
                try:
                    page.mouse.wheel(0, 2000)
                except Exception:
                    pass
                page.wait_for_timeout(1500)

            payload = captured["payload"]
            if isinstance(payload, dict):
                data = payload.get("data") or {}
                raw_items = data.get("items") or []
                for raw in raw_items[:limit]:
                    note_card = raw.get("note_card") or {}
                    if not note_card:
                        continue
                    note_id = raw.get("id") or note_card.get("note_id", "")
                    user = note_card.get("user") or {}
                    interact = note_card.get("interact_info") or {}
                    items.append({
                        "title": note_card.get("display_title") or note_card.get("title", ""),
                        "desc": note_card.get("desc", ""),
                        "url": f"https://www.xiaohongshu.com/explore/{note_id}" if note_id else "",
                        "author_name": user.get("nickname") or user.get("nick_name", ""),
                        "author_id": user.get("user_id") or user.get("userid", ""),
                        "date": None,
                        "engagement": {
                            "likes": _parse_count(str(interact.get("liked_count", "0"))),
                            "collects": _parse_count(str(interact.get("collected_count", "0"))),
                            "comments": _parse_count(str(interact.get("comment_count", "0"))),
                            "shares": _parse_count(str(interact.get("share_count", "0"))),
                        },
                        "hashtags": [],
                        "images": [
                            img.get("url_default") or img.get("url", "")
                            for img in (note_card.get("image_list") or [])
                            if isinstance(img, dict)
                        ],
                        "source": "crawler-xhr",
                    })

            if not items:
                note_elements = page.query_selector_all(
                    "section.note-item, div[class*='note-item'], a[class*='cover'], a[href*='/explore/']"
                )
                for elem in note_elements[:limit]:
                    try:
                        title_el = elem.query_selector("span[class*='title'], div[class*='title']")
                        title = title_el.inner_text() if title_el else ""
                        link = elem.get_attribute("href") or ""
                        if link and not link.startswith("http"):
                            link = f"https://www.xiaohongshu.com{link}"

                        author_el = elem.query_selector("span[class*='name'], div[class*='author']")
                        author = author_el.inner_text() if author_el else ""

                        likes_el = elem.query_selector("span[class*='like'], span[class*='count']")
                        likes_text = likes_el.inner_text() if likes_el else "0"
                        likes = _parse_count(likes_text)

                        items.append({
                            "title": title,
                            "desc": "",
                            "url": link,
                            "author_name": author,
                            "author_id": "",
                            "date": None,
                            "engagement": {
                                "likes": likes,
                                "collects": 0,
                                "comments": 0,
                                "shares": 0,
                            },
                            "hashtags": [],
                            "images": [],
                            "source": "crawler-dom",
                        })
                    except Exception:
                        continue
    except Exception as e:
        sys.stderr.write(f"[爬虫-小红书] 浏览器爬取失败: {e}\n")
    return items


def crawl_douyin(topic: str, limit: int = 20) -> List[Dict[str, Any]]:
    """通过浏览器自动化爬取抖音搜索结果。

    v2.1：优先拦截 XHR `/aweme/v1/web/search/item/`，DOM 解析作为兜底。
    """
    if not is_playwright_available():
        return []

    items: List[Dict[str, Any]] = []
    try:
        with _launch_browser_context("douyin") as (browser, context, page):
            captured: Dict[str, Any] = {"payload": None}

            def _on_response(resp):
                try:
                    url = resp.url
                    if "/aweme/v1/web/search/item" in url and resp.status == 200:
                        data = resp.json()
                        if isinstance(data, dict) and data.get("data"):
                            captured["payload"] = data
                except Exception:
                    pass

            page.on("response", _on_response)

            search_url = f"https://www.douyin.com/search/{topic}?type=video"
            try:
                page.goto(search_url, wait_until="domcontentloaded", timeout=30000)
            except Exception as e:
                sys.stderr.write(f"[爬虫-抖音] 页面加载失败: {e}\n")

            for _ in range(5):
                if captured["payload"]:
                    break
                try:
                    page.mouse.wheel(0, 2000)
                except Exception:
                    pass
                page.wait_for_timeout(1500)

            payload = captured["payload"]
            if isinstance(payload, dict):
                for entry in (payload.get("data") or [])[:limit]:
                    aweme = entry.get("aweme_info") or entry
                    if not isinstance(aweme, dict):
                        continue
                    aweme_id = aweme.get("aweme_id", "")
                    author = aweme.get("author") or {}
                    stats = aweme.get("statistics") or {}
                    items.append({
                        "text": aweme.get("desc", ""),
                        "url": f"https://www.douyin.com/video/{aweme_id}" if aweme_id else "",
                        "author_name": author.get("nickname", ""),
                        "author_id": author.get("uid", ""),
                        "date": None,
                        "engagement": {
                            "views": stats.get("play_count", 0),
                            "likes": stats.get("digg_count", 0),
                            "comments": stats.get("comment_count", 0),
                            "shares": stats.get("share_count", 0),
                        },
                        "hashtags": [],
                        "duration": (aweme.get("duration") or 0) // 1000,
                        "source": "crawler-xhr",
                    })

            if not items:
                video_elements = page.query_selector_all(
                    "div[class*='video-card'], li[class*='search-result']"
                )
                for elem in video_elements[:limit]:
                    try:
                        title_el = elem.query_selector(
                            "a[class*='title'], span[class*='title'], p[class*='desc']"
                        )
                        title = title_el.inner_text() if title_el else ""

                        link_el = elem.query_selector("a[href*='/video/']")
                        link = ""
                        if link_el:
                            href = link_el.get_attribute("href") or ""
                            if href.startswith("/"):
                                link = f"https://www.douyin.com{href}"
                            else:
                                link = href

                        author_el = elem.query_selector(
                            "span[class*='author'], span[class*='nickname']"
                        )
                        author = author_el.inner_text() if author_el else ""

                        likes_el = elem.query_selector("span[class*='like'], span[class*='digg']")
                        likes = _parse_count(likes_el.inner_text() if likes_el else "0")

                        items.append({
                            "text": title,
                            "url": link,
                            "author_name": author,
                            "author_id": "",
                            "date": None,
                            "engagement": {
                                "views": 0,
                                "likes": likes,
                                "comments": 0,
                                "shares": 0,
                            },
                            "hashtags": [],
                            "duration": 0,
                            "source": "crawler-dom",
                        })
                    except Exception:
                        continue
    except Exception as e:
        sys.stderr.write(f"[爬虫-抖音] 浏览器爬取失败: {e}\n")
    return items


def crawl_bilibili(topic: str, limit: int = 20) -> List[Dict[str, Any]]:
    """通过浏览器自动化爬取B站搜索结果。"""
    if not is_playwright_available():
        return []

    items: List[Dict[str, Any]] = []
    try:
        with _launch_browser_context("bilibili") as (browser, context, page):
            search_url = f"https://search.bilibili.com/all?keyword={topic}&order=totalrank"
            try:
                page.goto(search_url, wait_until="domcontentloaded", timeout=30000)
            except Exception as e:
                sys.stderr.write(f"[爬虫-B站] 页面加载失败: {e}\n")

            _wait_for(
                page,
                lambda: bool(page.query_selector("div.bili-video-card, div[class*='video-list-item']")),
                timeout_ms=8000,
            )

            video_elements = page.query_selector_all(
                "div.bili-video-card, div[class*='video-list-item']"
            )
            for elem in video_elements[:limit]:
                try:
                    title_el = elem.query_selector("h3[class*='title'], a[class*='title']")
                    title = _clean_html(title_el.inner_text()) if title_el else ""

                    link_el = elem.query_selector("a[href*='/video/']")
                    link = ""
                    if link_el:
                        href = link_el.get_attribute("href") or ""
                        if href.startswith("//"):
                            link = f"https:{href}"
                        elif href.startswith("/"):
                            link = f"https://www.bilibili.com{href}"
                        else:
                            link = href

                    author_el = elem.query_selector(
                        "span[class*='name'], span.bili-video-card__info--author"
                    )
                    author = author_el.inner_text() if author_el else ""

                    views_el = elem.query_selector("span[class*='play'], span[class*='view']")
                    views = _parse_count(views_el.inner_text() if views_el else "0")

                    items.append({
                        "title": title,
                        "url": link,
                        "bvid": "",
                        "channel_name": author,
                        "author_mid": "",
                        "date": None,
                        "duration": "",
                        "description": "",
                        "engagement": {
                            "views": views,
                            "danmaku": 0,
                            "comments": 0,
                            "favorites": 0,
                            "likes": 0,
                        },
                        "source": "crawler",
                    })
                except Exception:
                    continue
    except Exception as e:
        sys.stderr.write(f"[爬虫-B站] 浏览器爬取失败: {e}\n")
    return items


def crawl_zhihu(topic: str, limit: int = 20) -> List[Dict[str, Any]]:
    """通过浏览器自动化爬取知乎搜索结果。"""
    if not is_playwright_available():
        return []

    items: List[Dict[str, Any]] = []
    try:
        with _launch_browser_context("zhihu") as (browser, context, page):
            search_url = f"https://www.zhihu.com/search?type=content&q={topic}"
            try:
                page.goto(search_url, wait_until="domcontentloaded", timeout=30000)
            except Exception as e:
                sys.stderr.write(f"[爬虫-知乎] 页面加载失败: {e}\n")

            _wait_for(
                page,
                lambda: bool(page.query_selector("div.SearchResult-Card, div[class*='List-item']")),
                timeout_ms=8000,
            )

            result_elements = page.query_selector_all(
                "div.SearchResult-Card, div[class*='List-item']"
            )
            for elem in result_elements[:limit]:
                try:
                    title_el = elem.query_selector(
                        "h2[class*='ContentItem-title'], span[class*='Highlight']"
                    )
                    title = _clean_html(title_el.inner_text()) if title_el else ""

                    link_el = elem.query_selector("a[href*='/question/'], a[href*='/p/']")
                    link = ""
                    if link_el:
                        href = link_el.get_attribute("href") or ""
                        if href.startswith("/"):
                            link = f"https://www.zhihu.com{href}"
                        else:
                            link = href

                    excerpt_el = elem.query_selector(
                        "span[class*='RichText'], div[class*='RichText']"
                    )
                    excerpt = _clean_html(excerpt_el.inner_text()[:300]) if excerpt_el else ""

                    author_el = elem.query_selector(
                        "span[class*='AuthorInfo'] a, a[class*='UserLink']"
                    )
                    author = author_el.inner_text() if author_el else ""

                    voteup_el = elem.query_selector(
                        "button[class*='VoteButton'] span, span[class*='vote']"
                    )
                    voteups = _parse_count(voteup_el.inner_text() if voteup_el else "0")

                    items.append({
                        "title": title,
                        "excerpt": excerpt,
                        "url": link,
                        "author": author,
                        "date": None,
                        "content_type": "search_result",
                        "engagement": {
                            "voteups": voteups,
                            "comments": 0,
                            "collects": 0,
                        },
                        "source": "crawler",
                    })
                except Exception:
                    continue
    except Exception as e:
        sys.stderr.write(f"[爬虫-知乎] 浏览器爬取失败: {e}\n")
    return items


def _parse_count(text: str) -> int:
    """解析数量文本，支持 '1.2万' / '1.2w' / '12k' 等格式。"""
    if not text:
        return 0
    text = str(text).strip().replace(",", "")
    try:
        if "万" in text or text.lower().endswith("w"):
            num = float(re.sub(r"[万wW]", "", text))
            return int(num * 10000)
        elif "亿" in text:
            num = float(text.replace("亿", ""))
            return int(num * 100000000)
        elif text.lower().endswith("k"):
            num = float(text[:-1])
            return int(num * 1000)
        return int(float(re.sub(r"[^\d.]", "", text or "0")))
    except (ValueError, TypeError):
        return 0


def _parse_relative_date(date_str: str) -> Optional[str]:
    """解析微博的相对时间格式。"""
    if not date_str:
        return None

    from datetime import timedelta
    now = datetime.now()

    try:
        dt = datetime.strptime(date_str, "%a %b %d %H:%M:%S %z %Y")
        return dt.strftime("%Y-%m-%d")
    except ValueError:
        pass

    if "分钟前" in date_str:
        m = re.search(r"(\d+)", date_str)
        if m:
            return (now - timedelta(minutes=int(m.group(1)))).strftime("%Y-%m-%d")
    if "小时前" in date_str:
        m = re.search(r"(\d+)", date_str)
        if m:
            return (now - timedelta(hours=int(m.group(1)))).strftime("%Y-%m-%d")
    if "昨天" in date_str:
        return (now - timedelta(days=1)).strftime("%Y-%m-%d")

    m = re.match(r"(\d{2})-(\d{2})", date_str)
    if m:
        return f"{now.year}-{m.group(1)}-{m.group(2)}"
    return None


def get_crawler_status() -> Dict[str, Any]:
    """获取爬虫引擎的状态信息。"""
    pw_available = is_playwright_available()
    cached_platforms = []

    if COOKIE_DIR.exists():
        for f in COOKIE_DIR.glob("*_cookies.json"):
            platform = f.stem.replace("_cookies", "")
            cached_platforms.append(platform)

    return {
        "playwright_available": pw_available,
        "cached_logins": cached_platforms,
        "cookie_dir": str(COOKIE_DIR),
    }