Viral Topic

interface:
  display_name: "Viral Topic"
  short_description: "Route viral topic discovery across platforms"
  default_prompt: "Use $viral-topic to find viral topic references across my target platforms."

interface:
  display_name: "Bilibili Viral Topic"
  short_description: "Find low-follower viral Bilibili topic references"
  default_prompt: "Use $bilibili-viral-topic to find low-follower viral Bilibili videos for my niche."

#!/usr/bin/env python3
"""Find low-follower viral Bilibili videos via public JSON endpoints."""

from __future__ import annotations

import argparse
import datetime as dt
import html
import json
import math
import os
import re
import sys
import time
import urllib.parse
from pathlib import Path
from typing import Any

import requests


BASE = "https://api.bilibili.com"
USER_AGENT = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36"
)


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


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--topic", default="", help="Comma-separated Bilibili keywords.")
    parser.add_argument("--keyword", action="append", default=[], help="Explicit keyword. Can be repeated.")
    parser.add_argument("--days", type=int, default=30)
    parser.add_argument("--max-followers", type=int, default=100_000)
    parser.add_argument("--min-play", type=int, default=10_000)
    parser.add_argument("--min-play-follower-ratio", type=float, default=2.0)
    parser.add_argument("--limit", type=int, default=50)
    parser.add_argument("--pages", type=int, default=3)
    parser.add_argument("--page-size", type=int, default=30)
    parser.add_argument("--orders", default="pubdate,click", help="Search orders, comma-separated.")
    parser.add_argument("--ranking-rids", default="", help="Optional ranking rids, e.g. '0,188'. Empty disables ranking.")
    parser.add_argument("--request-sleep", type=float, default=0.6)
    parser.add_argument("--retries", type=int, default=2)
    parser.add_argument("--cache-path", default=str(Path.home() / ".cache/account-growth/bilibili_followers.json"))
    parser.add_argument("--cache-ttl-hours", type=float, default=24.0)
    parser.add_argument("--include-unknown-followers", action="store_true")
    parser.add_argument("--input-json", help="Load Bilibili API-like payload or normalized list from a local JSON file.")
    parser.add_argument("--format", choices=["json", "markdown"], default="json")
    parser.add_argument("--dry-run", action="store_true")
    return parser.parse_args()


def keywords_from_args(args: argparse.Namespace) -> list[str]:
    values: list[str] = []
    values.extend(args.keyword)
    values.extend(part.strip() for part in args.topic.replace("，", ",").split(","))
    deduped = [value for value in dict.fromkeys(v.strip() for v in values if v.strip())]
    if not deduped:
        raise RuntimeError("Provide --topic or --keyword")
    return deduped[:12]


def strip_html(value: Any) -> str:
    text = html.unescape(str(value or ""))
    text = re.sub(r"<[^>]+>", "", text)
    return text.strip()


def to_iso_from_ts(value: Any) -> str:
    timestamp = as_int(value)
    if not timestamp:
        return ""
    return dt.datetime.fromtimestamp(timestamp, dt.timezone.utc).isoformat()


def is_recent(pubdate: Any, days: int) -> bool:
    if days <= 0:
        return True
    timestamp = as_int(pubdate)
    if not timestamp:
        return False
    cutoff = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=days)
    published = dt.datetime.fromtimestamp(timestamp, dt.timezone.utc)
    return published >= cutoff


def freshness_score(pubdate: Any, days: int) -> float:
    if days <= 0:
        return 0.0
    timestamp = as_int(pubdate)
    if not timestamp:
        return 0.0
    published = dt.datetime.fromtimestamp(timestamp, dt.timezone.utc)
    age_days = max(0.0, (dt.datetime.now(dt.timezone.utc) - published).total_seconds() / 86400)
    return max(0.0, min(100.0, (1.0 - age_days / days) * 100.0))


def make_session() -> requests.Session:
    session = requests.Session()
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/json, text/plain, */*",
        "Accept-Encoding": "gzip, deflate",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
        "Origin": "https://www.bilibili.com",
        "Connection": "keep-alive",
    }
    cookie = os.getenv("BILIBILI_COOKIE", "").strip()
    if cookie:
        headers["Cookie"] = cookie
    session.headers.update(headers)
    return session


def bili_get(
    session: requests.Session,
    path: str,
    params: dict[str, Any],
    referer: str,
    args: argparse.Namespace,
) -> dict[str, Any]:
    url = BASE + path
    last_error = ""
    for attempt in range(args.retries + 1):
        try:
            response = session.get(url, params=params, headers={"Referer": referer}, timeout=20)
            if response.status_code == 412:
                raise RuntimeError("HTTP 412 Precondition Failed")
            response.raise_for_status()
            payload = response.json()
            if as_int(payload.get("code"), 0) != 0:
                raise RuntimeError(f"Bilibili code={payload.get('code')} message={payload.get('message')}")
            return payload
        except Exception as exc:
            last_error = str(exc)
            if attempt >= args.retries:
                break
            time.sleep(args.request_sleep * (attempt + 1) * 2)
    raise RuntimeError(last_error)


def normalize_search_item(item: dict[str, Any], source: str, source_query: str) -> dict[str, Any]:
    return {
        "platform": "bilibili",
        "content_id": str(item.get("bvid") or item.get("aid") or ""),
        "aid": str(item.get("aid") or item.get("id") or ""),
        "url": f"https://www.bilibili.com/video/{item.get('bvid')}" if item.get("bvid") else str(item.get("arcurl") or ""),
        "title": strip_html(item.get("title")),
        "author_id": str(item.get("mid") or ""),
        "author_name": strip_html(item.get("author")),
        "published_at": to_iso_from_ts(item.get("pubdate")),
        "pubdate_ts": as_int(item.get("pubdate")),
        "view_count": as_int(item.get("play")),
        "like_count": as_int(item.get("like")),
        "comment_count": as_int(item.get("review")),
        "share_count": as_int(item.get("share")),
        "save_count": as_int(item.get("favorites")),
        "coin_count": as_int(item.get("coin")),
        "danmaku_count": as_int(item.get("video_review") or item.get("danmaku")),
        "duration": str(item.get("duration") or ""),
        "tag": str(item.get("tag") or ""),
        "source": source,
        "source_query": source_query,
        "raw": item,
    }


def normalize_ranking_item(item: dict[str, Any], source: str, source_query: str) -> dict[str, Any]:
    owner = item.get("owner") if isinstance(item.get("owner"), dict) else {}
    stat = item.get("stat") if isinstance(item.get("stat"), dict) else {}
    bvid = str(item.get("bvid") or "")
    return {
        "platform": "bilibili",
        "content_id": bvid or str(item.get("aid") or ""),
        "aid": str(item.get("aid") or ""),
        "url": f"https://www.bilibili.com/video/{bvid}" if bvid else "",
        "title": strip_html(item.get("title")),
        "author_id": str(owner.get("mid") or ""),
        "author_name": strip_html(owner.get("name")),
        "published_at": to_iso_from_ts(item.get("pubdate")),
        "pubdate_ts": as_int(item.get("pubdate")),
        "view_count": as_int(stat.get("view") or stat.get("vv")),
        "like_count": as_int(stat.get("like")),
        "comment_count": as_int(stat.get("reply")),
        "share_count": as_int(stat.get("share")),
        "save_count": as_int(stat.get("favorite")),
        "coin_count": as_int(stat.get("coin")),
        "danmaku_count": as_int(stat.get("danmaku")),
        "duration": str(item.get("duration") or ""),
        "tag": "",
        "source": source,
        "source_query": source_query,
        "raw": item,
    }


def fetch_search(session: requests.Session, args: argparse.Namespace) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    results: list[dict[str, Any]] = []
    statuses: list[dict[str, Any]] = []
    orders = [order.strip() for order in args.orders.split(",") if order.strip()]
    for keyword in keywords_from_args(args):
        for order in orders:
            for page in range(1, args.pages + 1):
                params = {
                    "search_type": "video",
                    "keyword": keyword,
                    "order": order,
                    "page": page,
                    "page_size": min(args.page_size, 50),
                }
                try:
                    payload = bili_get(
                        session,
                        "/x/web-interface/search/type",
                        params,
                        "https://search.bilibili.com/video?keyword="
                        + urllib.parse.quote(keyword, safe=""),
                        args,
                    )
                    items = (payload.get("data") or {}).get("result") or []
                    normalized = [
                        normalize_search_item(item, f"search:{order}", keyword)
                        for item in items
                        if isinstance(item, dict)
                    ]
                    results.extend(normalized)
                    statuses.append({"source": "bilibili_search", "keyword": keyword, "order": order, "page": page, "ok": True, "count": len(normalized), "error": ""})
                    if not normalized:
                        break
                except Exception as exc:
                    statuses.append({"source": "bilibili_search", "keyword": keyword, "order": order, "page": page, "ok": False, "count": 0, "error": str(exc)})
                    break
                time.sleep(args.request_sleep)
    return results, statuses


def fetch_rankings(session: requests.Session, args: argparse.Namespace) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    results: list[dict[str, Any]] = []
    statuses: list[dict[str, Any]] = []
    rids = [rid.strip() for rid in args.ranking_rids.split(",") if rid.strip()]
    for rid in rids:
        try:
            payload = bili_get(
                session,
                "/x/web-interface/ranking/v2",
                {"rid": rid, "type": "all"},
                "https://www.bilibili.com/v/popular/rank/all",
                args,
            )
            items = (payload.get("data") or {}).get("list") or []
            normalized = [
                normalize_ranking_item(item, "ranking", f"rid:{rid}")
                for item in items
                if isinstance(item, dict)
            ]
            results.extend(normalized)
            statuses.append({"source": "bilibili_ranking", "rid": rid, "ok": True, "count": len(normalized), "error": ""})
        except Exception as exc:
            statuses.append({"source": "bilibili_ranking", "rid": rid, "ok": False, "count": 0, "error": str(exc)})
        time.sleep(args.request_sleep)
    return results, statuses


def load_cache(path: str) -> dict[str, Any]:
    cache_path = Path(path)
    if not cache_path.exists():
        return {}
    try:
        with cache_path.open("r", encoding="utf-8") as handle:
            payload = json.load(handle)
        return payload if isinstance(payload, dict) else {}
    except Exception:
        return {}


def save_cache(path: str, cache: dict[str, Any]) -> None:
    cache_path = Path(path)
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    with cache_path.open("w", encoding="utf-8") as handle:
        json.dump(cache, handle, ensure_ascii=False, indent=2)


def cached_follower(cache: dict[str, Any], mid: str, ttl_hours: float) -> int | None:
    item = cache.get(mid)
    if not isinstance(item, dict):
        return None
    fetched_at = as_int(item.get("fetched_at"))
    if not fetched_at:
        return None
    age_hours = (time.time() - fetched_at) / 3600
    if age_hours > ttl_hours:
        return None
    return as_int(item.get("follower"))


def fetch_follower(session: requests.Session, mid: str, args: argparse.Namespace) -> int:
    payload = bili_get(
        session,
        "/x/relation/stat",
        {"vmid": mid},
        f"https://space.bilibili.com/{mid}",
        args,
    )
    return as_int((payload.get("data") or {}).get("follower"))


def enrich_followers(
    session: requests.Session,
    items: list[dict[str, Any]],
    args: argparse.Namespace,
) -> tuple[dict[str, int], list[dict[str, Any]]]:
    cache = load_cache(args.cache_path)
    followers: dict[str, int] = {}
    statuses: list[dict[str, Any]] = []
    mids = list(dict.fromkeys(str(item.get("author_id") or "") for item in items if item.get("author_id")))
    changed = False
    for mid in mids:
        cached = cached_follower(cache, mid, args.cache_ttl_hours)
        if cached is not None:
            followers[mid] = cached
            statuses.append({"source": "bilibili_relation_stat", "mid": mid, "ok": True, "cached": True, "follower": cached, "error": ""})
            continue
        try:
            follower = fetch_follower(session, mid, args)
            followers[mid] = follower
            cache[mid] = {"follower": follower, "fetched_at": int(time.time())}
            changed = True
            statuses.append({"source": "bilibili_relation_stat", "mid": mid, "ok": True, "cached": False, "follower": follower, "error": ""})
        except Exception as exc:
            statuses.append({"source": "bilibili_relation_stat", "mid": mid, "ok": False, "cached": False, "follower": None, "error": str(exc)})
        time.sleep(args.request_sleep)
    if changed:
        save_cache(args.cache_path, cache)
    return followers, statuses


def score_item(item: dict[str, Any], follower_count: int | None, args: argparse.Namespace) -> dict[str, Any]:
    views = as_int(item.get("view_count"))
    likes = as_int(item.get("like_count"))
    comments = as_int(item.get("comment_count"))
    shares = as_int(item.get("share_count"))
    saves = as_int(item.get("save_count"))
    coins = as_int(item.get("coin_count"))
    danmaku = as_int(item.get("danmaku_count"))
    engagement = likes + coins * 3 + saves * 2 + shares * 3 + comments * 2 + danmaku
    hot_score = views + likes * 10 + coins * 30 + saves * 20 + shares * 30 + comments * 20 + danmaku * 5

    play_follower_ratio = None
    engagement_follower_ratio = None
    if follower_count:
        denominator = max(follower_count, 1000)
        play_follower_ratio = views / denominator
        engagement_follower_ratio = engagement / denominator

    breakout_score = (
        math.log1p(max(views, 0)) * 0.40
        + math.log1p(max(engagement, 0)) * 0.30
        + min(100.0, (play_follower_ratio or 0.0) * 8.0) * 0.25
        + freshness_score(item.get("pubdate_ts"), args.days) * 0.05
    )

    evidence: list[str] = []
    if follower_count is not None and follower_count <= args.max_followers:
        evidence.append(f"low_follower_author:{follower_count}")
    if follower_count and views >= follower_count:
        evidence.append(f"plays_exceed_followers:{views / max(follower_count, 1):.2f}x")
    if play_follower_ratio is not None and play_follower_ratio >= args.min_play_follower_ratio:
        evidence.append(f"high_play_follower_ratio:{play_follower_ratio:.2f}x")
    if engagement_follower_ratio is not None and engagement_follower_ratio >= 0.1:
        evidence.append(f"high_engagement_per_follower:{engagement_follower_ratio:.3f}")
    if (saves + coins) / max(views, 1) >= 0.02 and (saves + coins) >= 100:
        evidence.append("strong_save_coin_signal")
    if follower_count is None:
        evidence.append("unknown_followers")

    output = dict(item)
    output.update(
        {
            "follower_count": follower_count,
            "hot_score": round(float(hot_score), 2),
            "breakout_score": round(float(breakout_score), 2),
            "engagement": round(float(engagement), 2),
            "play_follower_ratio": round(play_follower_ratio, 5) if play_follower_ratio is not None else None,
            "engagement_follower_ratio": round(engagement_follower_ratio, 5) if engagement_follower_ratio is not None else None,
            "evidence": evidence,
        }
    )
    return output


def passes_filters(item: dict[str, Any], args: argparse.Namespace) -> bool:
    follower = item.get("follower_count")
    if follower is None:
        if not args.include_unknown_followers:
            return False
    elif int(follower) > args.max_followers:
        return False
    if as_int(item.get("view_count")) < args.min_play:
        return False
    ratio = item.get("play_follower_ratio")
    if follower is not None and ratio is not None and float(ratio) < args.min_play_follower_ratio:
        return False
    return True


def extract_records(payload: Any) -> list[dict[str, Any]]:
    if isinstance(payload, list):
        return [item for item in payload if isinstance(item, dict)]
    if not isinstance(payload, dict):
        return []
    for key in ("results", "items", "videos"):
        value = payload.get(key)
        if isinstance(value, list):
            return [item for item in value if isinstance(item, dict)]
    data = payload.get("data")
    if isinstance(data, dict):
        if isinstance(data.get("result"), list):
            return [item for item in data["result"] if isinstance(item, dict)]
        if isinstance(data.get("list"), list):
            return [item for item in data["list"] if isinstance(item, dict)]
    return []


def load_input_json(path: str, args: argparse.Namespace) -> list[dict[str, Any]]:
    with open(path, "r", encoding="utf-8") as handle:
        payload = json.load(handle)
    records = extract_records(payload)
    normalized: list[dict[str, Any]] = []
    for record in records:
        if record.get("platform") == "bilibili" and "breakout_score" in record:
            normalized.append(record)
        elif "stat" in record and "owner" in record:
            normalized.append(normalize_ranking_item(record, "input_json", ""))
        else:
            normalized.append(normalize_search_item(record, "input_json", ""))
    return normalized


def collect_live(args: argparse.Namespace) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    session = make_session()
    search_items, search_statuses = fetch_search(session, args)
    ranking_items, ranking_statuses = fetch_rankings(session, args)
    all_items = search_items + ranking_items

    by_id: dict[str, dict[str, Any]] = {}
    for item in all_items:
        content_id = str(item.get("content_id") or item.get("aid") or "")
        if not content_id:
            continue
        existing = by_id.get(content_id)
        if not existing or as_int(item.get("view_count")) > as_int(existing.get("view_count")):
            by_id[content_id] = item

    recent_items = [item for item in by_id.values() if is_recent(item.get("pubdate_ts"), args.days)]
    followers, follower_statuses = enrich_followers(session, recent_items, args)
    scored = [
        score_item(item, followers.get(str(item.get("author_id") or "")), args)
        for item in recent_items
    ]
    statuses = search_statuses + ranking_statuses + follower_statuses
    statuses.append({"source": "dedupe_recent", "ok": True, "count": len(recent_items), "error": ""})
    return scored, statuses


def render_markdown(output: dict[str, Any]) -> str:
    lines = [
        "# Bilibili Viral Topic Results",
        "",
        f"- Run at: {output['run_at']}",
        f"- Window: last {output['config']['days']} days",
        f"- Results: {output['summary']['returned']} / {output['summary']['candidates']}",
        "",
    ]
    for index, item in enumerate(output["results"], 1):
        followers = item.get("follower_count")
        follower_text = "unknown" if followers is None else f"{followers:,}"
        ratio = item.get("play_follower_ratio")
        ratio_text = "n/a" if ratio is None else f"{ratio:.2f}x"
        lines.extend(
            [
                f"## {index}. {item['title']}",
                "",
                f"- URL: {item['url']}",
                f"- UP主: {item['author_name']} ({follower_text} followers)",
                f"- Published: {item['published_at']}",
                f"- Plays: {item['view_count']:,}; Likes: {item['like_count']:,}; Coins: {item['coin_count']:,}; Saves: {item['save_count']:,}; Comments: {item['comment_count']:,}; Danmaku: {item['danmaku_count']:,}",
                f"- Play/follower ratio: {ratio_text}",
                f"- Breakout score: {item['breakout_score']}",
                f"- Evidence: {', '.join(item['evidence'])}",
                "",
            ]
        )
    return "\n".join(lines)


def main() -> int:
    args = parse_args()
    if args.dry_run:
        planned = {
            "dry_run": True,
            "keywords": keywords_from_args(args),
            "orders": [order.strip() for order in args.orders.split(",") if order.strip()],
            "ranking_rids": [rid.strip() for rid in args.ranking_rids.split(",") if rid.strip()],
            "args": vars(args),
            "cookie": "***redacted***" if os.getenv("BILIBILI_COOKIE") else "",
        }
        if "BILIBILI_COOKIE" in planned["args"]:
            planned["args"]["BILIBILI_COOKIE"] = "***redacted***"
        print(json.dumps(planned, ensure_ascii=False, indent=2))
        return 0

    if args.input_json:
        candidates = load_input_json(args.input_json, args)
        scored = [
            item if "breakout_score" in item else score_item(item, item.get("follower_count"), args)
            for item in candidates
            if is_recent(item.get("pubdate_ts") or 0, args.days) or item.get("published_at")
        ]
        statuses = [{"source": "input_json", "ok": True, "count": len(scored), "error": ""}]
    else:
        scored, statuses = collect_live(args)

    filtered = [item for item in scored if passes_filters(item, args)]
    filtered.sort(
        key=lambda item: (
            float(item.get("breakout_score") or 0),
            float(item.get("play_follower_ratio") or 0),
            as_int(item.get("view_count")),
        ),
        reverse=True,
    )
    results = filtered[: args.limit]
    output = {
        "platform": "bilibili",
        "run_at": dt.datetime.now(dt.timezone.utc).isoformat(),
        "version": "v1",
        "config": {
            "keywords": [] if args.input_json else keywords_from_args(args),
            "days": args.days,
            "max_followers": args.max_followers,
            "min_play": args.min_play,
            "min_play_follower_ratio": args.min_play_follower_ratio,
            "limit": args.limit,
            "pages": args.pages,
            "page_size": args.page_size,
            "orders": args.orders,
            "ranking_rids": args.ranking_rids,
        },
        "summary": {
            "candidates": len(scored),
            "matched": len(filtered),
            "returned": len(results),
        },
        "source_status": statuses,
        "results": results,
    }
    if args.format == "markdown":
        print(render_markdown(output))
    else:
        print(json.dumps(output, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())

interface:
  display_name: "WeChat Viral Topic"
  short_description: "Find WeChat viral topic references"
  default_prompt: "Use $wechat-viral-topic to find WeChat articles that beat account average reads in my target category."

#!/usr/bin/env python3
"""Find WeChat public-account articles that beat account average reads."""

from __future__ import annotations

import argparse
import datetime as dt
import json
import math
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any


USER_AGENT = "account-growth-wechat-average-read-viral/0.2"
DEFAULT_EXCLUDED_ACCOUNTS = ["新智元", "机器之心", "差评", "智东西", "极客公园", "量子位", "CSDN"]


def as_int(value: Any, default: int = 0) -> int:
    try:
        if value is None or value == "":
            return default
        return int(float(value))
    except (TypeError, ValueError):
        return default


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default=os.getenv("WECHAT_HOT_API_BASE", ""))
    parser.add_argument("--access-token", default=os.getenv("WECHAT_HOT_ACCESS_TOKEN", ""))
    parser.add_argument("--app-id", default=os.getenv("WECHAT_HOT_APP_ID", ""))
    parser.add_argument("--app-secret", default=os.getenv("WECHAT_HOT_APP_SECRET", ""))
    parser.add_argument("--category", default="ai")
    parser.add_argument("--days", type=int, default=7)
    parser.add_argument("--min-read", type=int, default=10_000)
    parser.add_argument("--max-followers", type=int, default=50_000, help="Deprecated compatibility option; not used for filtering.")
    parser.add_argument("--min-read-follower-ratio", type=float, default=1.0, help="Deprecated compatibility option; not used for filtering.")
    parser.add_argument("--min-read-month-avg-ratio", type=float, default=2.0)
    parser.add_argument("--limit", type=int, default=50)
    parser.add_argument("--max-pages", type=int, default=5)
    parser.add_argument("--account-retry-seconds", type=float, default=0.0)
    parser.add_argument("--include-unknown-followers", action="store_true", help="Deprecated compatibility option; average-read data is required.")
    parser.add_argument(
        "--exclude-account",
        action="append",
        default=[],
        help="Account name to exclude from results. Can be repeated.",
    )
    parser.add_argument(
        "--exclude-accounts",
        default="",
        help="Comma-separated account names to exclude in addition to the default organization/media account list.",
    )
    parser.add_argument(
        "--no-default-excluded-accounts",
        action="store_true",
        help="Disable the built-in organization/media account exclusion list.",
    )
    parser.add_argument("--skip-enrich", action="store_true")
    parser.add_argument("--input-json", help="Load hot-article payload from a local JSON file instead of calling the API.")
    parser.add_argument("--format", choices=["json", "markdown"], default="json")
    parser.add_argument("--dry-run", action="store_true", help="Print planned request parameters without calling the API.")
    return parser.parse_args()


def redacted_args(args: argparse.Namespace) -> dict[str, Any]:
    values = vars(args).copy()
    for key in ("access_token", "app_id", "app_secret"):
        if values.get(key):
            values[key] = "***redacted***"
    return values


def split_account_names(value: str) -> list[str]:
    names: list[str] = []
    for part in re.split(r"[,，\n]", value or ""):
        name = part.strip()
        if name:
            names.append(name)
    return names


def excluded_account_names(args: argparse.Namespace) -> list[str]:
    names: list[str] = []
    if not args.no_default_excluded_accounts:
        names.extend(DEFAULT_EXCLUDED_ACCOUNTS)
    names.extend(split_account_names(args.exclude_accounts))
    names.extend(name.strip() for name in args.exclude_account if name and name.strip())

    seen: set[str] = set()
    unique: list[str] = []
    for name in names:
        key = normalize_account_name(name)
        if key and key not in seen:
            seen.add(key)
            unique.append(name)
    return unique


def normalize_account_name(value: Any) -> str:
    return re.sub(r"\s+", "", str(value or "")).casefold()


def matching_excluded_account(result: dict[str, Any], excluded_names: list[str]) -> str:
    if not excluded_names:
        return ""
    candidates = [
        result.get("author_name"),
        result.get("account_username"),
        (result.get("raw") or {}).get("article", {}).get("nickname"),
        (result.get("raw") or {}).get("account", {}).get("nickname"),
        (result.get("raw") or {}).get("account", {}).get("username"),
    ]
    normalized_candidates = [normalize_account_name(candidate) for candidate in candidates if candidate]
    for excluded in excluded_names:
        normalized_excluded = normalize_account_name(excluded)
        if not normalized_excluded:
            continue
        for candidate in normalized_candidates:
            if candidate == normalized_excluded or normalized_excluded in candidate:
                return excluded
    return ""


def api_url(base_url: str, path: str, query: dict[str, Any] | None = None) -> str:
    if not base_url:
        raise RuntimeError("WECHAT_HOT_API_BASE or --base-url is required for live API calls")
    url = base_url.rstrip("/") + path
    if query:
        url += "?" + urllib.parse.urlencode({k: v for k, v in query.items() if v is not None})
    return url


def post_json(url: str, body: dict[str, Any], timeout: int = 30) -> dict[str, Any]:
    payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    request = urllib.request.Request(
        url,
        data=payload,
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": USER_AGENT,
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            data = response.read().decode("utf-8")
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"HTTP {exc.code} from {url}: {detail[:300]}") from exc
    return json.loads(data)


def get_access_token(args: argparse.Namespace) -> str:
    if args.access_token:
        return args.access_token
    if not args.app_id or not args.app_secret:
        raise RuntimeError("Set WECHAT_HOT_ACCESS_TOKEN or both WECHAT_HOT_APP_ID and WECHAT_HOT_APP_SECRET")
    payload = post_json(
        api_url(args.base_url, "/api/v2/token"),
        {"app_id": args.app_id, "app_secret": args.app_secret},
    )
    token = ((payload.get("data") or {}).get("access_token") or "").strip()
    if not token:
        raise RuntimeError(f"Token response did not include access_token: {payload}")
    return token


def published_after(days: int) -> str:
    cutoff = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=days)
    # The docs call this a date, but the live API rejects YYYY-MM-DD and accepts datetime.
    return cutoff.strftime("%Y-%m-%dT00:00:00")


def extract_items(payload: Any) -> list[dict[str, Any]]:
    if isinstance(payload, list):
        return [item for item in payload if isinstance(item, dict)]
    if not isinstance(payload, dict):
        return []
    data = payload.get("data") if isinstance(payload.get("data"), dict) else payload
    items = data.get("items") if isinstance(data, dict) else None
    if isinstance(items, list):
        return [item for item in items if isinstance(item, dict)]
    return []


def fetch_hot_articles(args: argparse.Namespace, access_token: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    all_items: list[dict[str, Any]] = []
    statuses: list[dict[str, Any]] = []
    last_id: int | None = None
    for page in range(1, args.max_pages + 1):
        body: dict[str, Any] = {
            "category": args.category,
            "read_num": args.min_read,
            "published_at": published_after(args.days),
        }
        if last_id:
            body["last_id"] = last_id
        payload = post_json(api_url(args.base_url, "/api/v2/hot/articles", {"access_token": access_token}), body)
        items = extract_items(payload)
        statuses.append({"source": "wechat_hot_articles", "page": page, "ok": True, "count": len(items), "error": ""})
        all_items.extend(items)
        data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
        next_last_id = as_int(data.get("last_id"), 0)
        if not next_last_id or next_last_id == last_id or len(all_items) >= args.limit * 3:
            break
        last_id = next_last_id
    return all_items, statuses


def extract_biz(content_url: str) -> str:
    if not content_url:
        return ""
    parsed = urllib.parse.urlparse(content_url)
    query = urllib.parse.parse_qs(parsed.query)
    value = (query.get("__biz") or [""])[0]
    if value:
        return value
    match = re.search(r"[?&]__biz=([^&#]+)", content_url)
    return urllib.parse.unquote(match.group(1)) if match else ""


def fetch_account_detail(
    args: argparse.Namespace,
    access_token: str,
    biz: str,
    nickname: str,
) -> tuple[dict[str, Any], str]:
    body: dict[str, Any] = {}
    if biz:
        body["biz"] = biz
    elif nickname:
        body["nickname"] = nickname
    else:
        return {}, "missing biz and nickname"

    url = api_url(args.base_url, "/api/v2/accounts/detail", {"access_token": access_token})
    first = post_json(url, body)
    if as_int(first.get("code"), 0) == 200 and isinstance(first.get("data"), dict):
        return first["data"], ""

    if args.account_retry_seconds > 0:
        time.sleep(args.account_retry_seconds)
        second = post_json(url, body)
        if as_int(second.get("code"), 0) == 200 and isinstance(second.get("data"), dict):
            return second["data"], ""
        return {}, str(second.get("msg") or second)[:300]
    return {}, str(first.get("msg") or first)[:300]


def enrich_accounts(
    args: argparse.Namespace,
    access_token: str,
    items: list[dict[str, Any]],
) -> tuple[dict[str, dict[str, Any]], list[dict[str, Any]]]:
    accounts: dict[str, dict[str, Any]] = {}
    statuses: list[dict[str, Any]] = []
    for item in items:
        biz = extract_biz(str(item.get("content_url") or ""))
        nickname = str(item.get("nickname") or "")
        key = biz or f"nickname:{nickname}"
        if not key or key in accounts:
            continue
        try:
            detail, error = fetch_account_detail(args, access_token, biz, nickname)
            accounts[key] = detail
            statuses.append({"source": "wechat_account_detail", "biz": biz, "nickname": nickname, "ok": not error, "error": error})
        except Exception as exc:  # Keep other articles usable.
            accounts[key] = {}
            statuses.append({"source": "wechat_account_detail", "biz": biz, "nickname": nickname, "ok": False, "error": str(exc)})
    return accounts, statuses


def score_item(item: dict[str, Any], account: dict[str, Any], args: argparse.Namespace) -> dict[str, Any]:
    read_num = as_int(item.get("read_num"))
    like_num = as_int(item.get("like_num"))
    old_like_num = as_int(item.get("old_like_num"))
    share_num = as_int(item.get("share_num"))
    follower_count = as_int(account.get("follower_count"), 0)
    month_read_avg = as_int(account.get("month_read_avg"), 0)
    engagement = like_num + old_like_num + share_num * 3
    hot_score = read_num + like_num * 20 + old_like_num * 8 + share_num * 30
    read_month_avg_ratio = read_num / max(month_read_avg, 1000) if month_read_avg else 0.0
    breakout_score = (
        math.log1p(read_num) * 0.45
        + math.log1p(engagement) * 0.20
        + min(100.0, read_month_avg_ratio * 30.0) * 0.30
        + min(100.0, share_num * 2.0) * 0.05
    )

    evidence: list[str] = []
    if month_read_avg and read_month_avg_ratio >= args.min_read_month_avg_ratio:
        evidence.append(f"above_monthly_average:{read_month_avg_ratio:.2f}x")
    if read_num >= args.min_read:
        evidence.append(f"high_read:{read_num}")
    if share_num >= 20:
        evidence.append(f"strong_share_signal:{share_num}")
    if not month_read_avg:
        evidence.append("unknown_month_read_avg")
    if follower_count:
        evidence.append(f"follower_count_reference:{follower_count}")

    biz = extract_biz(str(item.get("content_url") or "")) or str(account.get("biz") or "")
    return {
        "platform": "wechat",
        "content_id": str(item.get("id") or ""),
        "url": item.get("content_url") or "",
        "title": item.get("title") or "",
        "author_id": biz,
        "author_name": account.get("nickname") or item.get("nickname") or "",
        "account_username": account.get("username") or "",
        "follower_count": follower_count or None,
        "month_read_avg": month_read_avg or None,
        "published_at": item.get("published_at") or "",
        "view_count": read_num,
        "like_count": like_num + old_like_num,
        "share_count": share_num,
        "save_count": None,
        "comment_count": None,
        "hot_score": round(float(hot_score), 2),
        "breakout_score": round(float(breakout_score), 2),
        "read_month_avg_ratio": round(read_month_avg_ratio, 4) if month_read_avg else None,
        "evidence": evidence,
        "raw": {"article": item, "account": account},
    }


def passes_filters(result: dict[str, Any], args: argparse.Namespace) -> bool:
    if as_int(result.get("view_count")) < args.min_read:
        return False
    month_avg = result.get("month_read_avg")
    month_ratio = result.get("read_month_avg_ratio")
    if not month_avg or month_ratio is None:
        return False
    if float(month_ratio) < args.min_read_month_avg_ratio:
        return False
    return True


def format_markdown(payload: dict[str, Any]) -> str:
    lines = [
        "# WeChat Average-Read Viral Results",
        "",
        f"- Category: {payload['query'].get('category')}",
        f"- Window days: {payload['query'].get('days')}",
        f"- Results: {len(payload['results'])}",
        f"- Excluded accounts: {', '.join(payload['query'].get('excluded_accounts') or []) or 'none'}",
        "",
    ]
    for index, item in enumerate(payload["results"], 1):
        followers = item.get("follower_count") if item.get("follower_count") is not None else "unknown"
        lines.extend(
            [
                f"## {index}. {item['title']}",
                "",
                f"- URL: {item['url']}",
                f"- Account: {item.get('author_name') or ''} ({item.get('author_id') or 'unknown biz'})",
                f"- Followers reference: {followers}; month_read_avg: {item.get('month_read_avg') or 'unknown'}",
                f"- Reads: {item['view_count']}; likes: {item['like_count']}; shares: {item['share_count']}",
                f"- Score: {item['breakout_score']}; evidence: {', '.join(item['evidence'])}",
                "",
            ]
        )
    if payload.get("source_status"):
        lines.extend(["## Source Status", ""])
        for status in payload["source_status"]:
            lines.append(f"- {status.get('source')}: ok={status.get('ok')} count={status.get('count', '')} error={status.get('error', '')}")
    if payload.get("excluded_counts"):
        lines.extend(["", "## Excluded Accounts", ""])
        for name, count in sorted(payload["excluded_counts"].items(), key=lambda pair: (-pair[1], pair[0])):
            lines.append(f"- {name}: {count}")
    return "\n".join(lines).strip() + "\n"


def main() -> int:
    args = parse_args()
    excluded_names = excluded_account_names(args)
    if args.dry_run:
        print(json.dumps({"planned": redacted_args(args), "published_at": published_after(args.days), "excluded_accounts": excluded_names}, ensure_ascii=False, indent=2))
        return 0

    source_status: list[dict[str, Any]] = []
    if args.input_json:
        with open(args.input_json, "r", encoding="utf-8") as handle:
            raw_payload = json.load(handle)
        items = extract_items(raw_payload)
        source_status.append({"source": "input_json", "ok": True, "count": len(items), "error": ""})
        access_token = args.access_token
    else:
        access_token = get_access_token(args)
        items, statuses = fetch_hot_articles(args, access_token)
        source_status.extend(statuses)

    accounts: dict[str, dict[str, Any]] = {}
    if not args.skip_enrich:
        if not access_token:
            source_status.append({"source": "wechat_account_detail", "ok": False, "error": "No access token; skipped enrichment"})
        else:
            enriched, statuses = enrich_accounts(args, access_token, items)
            accounts.update(enriched)
            source_status.extend(statuses)

    results: list[dict[str, Any]] = []
    excluded_counts: dict[str, int] = {}
    for item in items:
        biz = extract_biz(str(item.get("content_url") or ""))
        key = biz or f"nickname:{item.get('nickname') or ''}"
        account = accounts.get(key) or {
            "biz": biz,
            "nickname": item.get("nickname"),
            "follower_count": item.get("follower_count"),
            "month_read_avg": item.get("month_read_avg"),
        }
        scored = score_item(item, account, args)
        excluded_by = matching_excluded_account(scored, excluded_names)
        if excluded_by:
            excluded_counts[excluded_by] = excluded_counts.get(excluded_by, 0) + 1
            continue
        if passes_filters(scored, args):
            results.append(scored)

    results.sort(
        key=lambda item: (
            float(item.get("breakout_score") or 0),
            float(item.get("read_month_avg_ratio") or 0),
            int(item.get("view_count") or 0),
        ),
        reverse=True,
    )
    payload = {
        "query": {
            "category": args.category,
            "days": args.days,
            "min_read": args.min_read,
            "min_read_month_avg_ratio": args.min_read_month_avg_ratio,
            "excluded_accounts": excluded_names,
        },
        "source_status": source_status,
        "excluded_counts": excluded_counts,
        "results": results[: args.limit],
    }
    if args.format == "markdown":
        print(format_markdown(payload), end="")
    else:
        print(json.dumps(payload, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

interface:
  display_name: "X Viral Topic"
  short_description: "Find low-follower viral X topic references"
  default_prompt: "Use $x-viral-topic to find low-follower viral X posts for my niche."

#!/usr/bin/env python3
"""Find low-follower viral posts on X/Twitter via twitterapi.io."""

from __future__ import annotations

import argparse
import datetime as dt
import json
import math
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
from typing import Any


API_URL = "https://api.twitterapi.io/twitter/tweet/advanced_search"
USER_AGENT = "account-growth-x-viral-topic/0.1"


def as_int(value: Any, default: int = 0) -> int:
    try:
        if value is None or value == "":
            return default
        return int(float(value))
    except (TypeError, ValueError):
        return default


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--api-key", default=os.getenv("TWITTERAPI_IO_KEY") or os.getenv("TWITTER_API_KEY") or "")
    parser.add_argument("--topic", default="")
    parser.add_argument("--query", action="append", default=[], help="Advanced-search query. Can be repeated.")
    parser.add_argument("--days", type=int, default=7)
    parser.add_argument("--max-followers", type=int, default=50_000)
    parser.add_argument("--min-views", type=int, default=10_000)
    parser.add_argument("--min-engagement", type=float, default=100.0)
    parser.add_argument("--min-faves", type=int, default=20)
    parser.add_argument("--limit", type=int, default=50)
    parser.add_argument("--query-types", default="Top,Latest")
    parser.add_argument("--lang", default="en")
    parser.add_argument("--include-replies", action="store_true")
    parser.add_argument("--include-unknown-followers", action="store_true")
    parser.add_argument("--input-json", help="Load a twitterapi.io response or list of tweets from a local JSON file.")
    parser.add_argument("--format", choices=["json", "markdown"], default="json")
    parser.add_argument("--dry-run", action="store_true")
    return parser.parse_args()


def redacted_args(args: argparse.Namespace) -> dict[str, Any]:
    values = vars(args).copy()
    if values.get("api_key"):
        values["api_key"] = "***redacted***"
    return values


def since_date(days: int) -> str:
    cutoff = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=days)
    return cutoff.date().isoformat()


def topic_terms(topic: str) -> list[str]:
    raw = [part.strip() for part in topic.replace("，", ",").split(",") if part.strip()]
    if raw:
        return raw[:8]
    words = topic.strip()
    return [words] if words else []


def quote_term(term: str) -> str:
    term = term.strip()
    if not term:
        return ""
    if " " in term and not (term.startswith('"') and term.endswith('"')):
        return f'"{term}"'
    return term


def build_queries(args: argparse.Namespace) -> list[str]:
    if args.query:
        return [query.strip() for query in args.query if query.strip()]
    terms = [quote_term(term) for term in topic_terms(args.topic)]
    terms = [term for term in terms if term]
    if not terms:
        raise RuntimeError("Provide --topic or --query")
    term_expr = " OR ".join(terms[:6])
    parts = [f"({term_expr})", f"since:{since_date(args.days)}", f"min_faves:{args.min_faves}"]
    if args.lang:
        parts.insert(1, f"lang:{args.lang}")
    if not args.include_replies:
        parts.append("-filter:replies")
    return [" ".join(parts)]


def fetch_query(api_key: str, query: str, query_type: str) -> dict[str, Any]:
    params = urllib.parse.urlencode({"query": query, "queryType": query_type})
    request = urllib.request.Request(
        f"{API_URL}?{params}",
        headers={"X-API-Key": api_key, "Accept": "application/json", "User-Agent": USER_AGENT},
        method="GET",
    )
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            data = response.read().decode("utf-8")
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"HTTP {exc.code}: {detail[:300]}") from exc
    return json.loads(data)


def extract_tweets(payload: Any) -> list[dict[str, Any]]:
    if isinstance(payload, list):
        return [item for item in payload if isinstance(item, dict)]
    if not isinstance(payload, dict):
        return []
    for key in ("tweets", "data", "results"):
        value = payload.get(key)
        if isinstance(value, list):
            return [item for item in value if isinstance(item, dict)]
        if isinstance(value, dict):
            nested = value.get("tweets") or value.get("items")
            if isinstance(nested, list):
                return [item for item in nested if isinstance(item, dict)]
    return []


def author_handle(tweet: dict[str, Any]) -> str:
    author = tweet.get("author") if isinstance(tweet.get("author"), dict) else {}
    return str(
        author.get("userName")
        or author.get("username")
        or author.get("screen_name")
        or tweet.get("authorUsername")
        or ""
    ).lstrip("@")


def author_followers(tweet: dict[str, Any]) -> int:
    author = tweet.get("author") if isinstance(tweet.get("author"), dict) else {}
    for key in ("followers", "followersCount", "followers_count", "follower_count"):
        value = author.get(key)
        if value is not None:
            return as_int(value)
    public_metrics = author.get("public_metrics") if isinstance(author.get("public_metrics"), dict) else {}
    return as_int(public_metrics.get("followers_count"))


def tweet_id(tweet: dict[str, Any]) -> str:
    return str(tweet.get("id_str") or tweet.get("id") or tweet.get("tweet_id") or "")


def tweet_text(tweet: dict[str, Any]) -> str:
    return str(tweet.get("text") or tweet.get("full_text") or tweet.get("content") or "")


def metric(tweet: dict[str, Any], *keys: str) -> int:
    for key in keys:
        if key in tweet:
            return as_int(tweet.get(key))
    metrics = tweet.get("public_metrics") if isinstance(tweet.get("public_metrics"), dict) else {}
    for key in keys:
        if key in metrics:
            return as_int(metrics.get(key))
    return 0


def normalize_tweet(tweet: dict[str, Any], args: argparse.Namespace) -> dict[str, Any]:
    tid = tweet_id(tweet)
    handle = author_handle(tweet)
    author = tweet.get("author") if isinstance(tweet.get("author"), dict) else {}
    followers = author_followers(tweet)
    likes = metric(tweet, "likeCount", "favorite_count", "like_count")
    reposts = metric(tweet, "retweetCount", "retweet_count", "repost_count")
    replies = metric(tweet, "replyCount", "reply_count")
    views = metric(tweet, "viewCount", "view_count", "impression_count")
    bookmarks = metric(tweet, "bookmarkCount", "bookmark_count")
    engagement = likes + reposts * 3 + replies * 2 + bookmarks * 2 + views / 1000
    engagement_follower_ratio = engagement / max(followers, 500) if followers else 0.0
    view_follower_ratio = views / max(followers, 500) if followers else 0.0
    bookmark_like_ratio = bookmarks / max(likes, 1) if bookmarks else 0.0
    breakout_score = (
        math.log1p(max(engagement, 0)) * 0.45
        + min(100.0, engagement_follower_ratio * 60.0) * 0.25
        + min(100.0, view_follower_ratio * 8.0) * 0.20
        + min(100.0, bookmark_like_ratio * 100.0) * 0.10
    )

    evidence: list[str] = []
    if followers and followers <= args.max_followers:
        evidence.append(f"low_follower_author:{followers}")
    if followers and views >= followers:
        evidence.append(f"views_exceed_followers:{view_follower_ratio:.2f}x")
    if followers and engagement_follower_ratio >= 0.02:
        evidence.append(f"high_engagement_per_follower:{engagement_follower_ratio:.3f}")
    if bookmark_like_ratio >= 0.15:
        evidence.append(f"save_intent:{bookmark_like_ratio:.2f}")
    if not followers:
        evidence.append("unknown_followers")

    url = tweet.get("url") or (f"https://x.com/{handle}/status/{tid}" if handle and tid else f"https://x.com/i/web/status/{tid}")
    text = tweet_text(tweet)
    return {
        "platform": "x",
        "content_id": tid,
        "url": url,
        "title": text[:120],
        "summary": text,
        "author_id": str(author.get("id") or author.get("rest_id") or handle),
        "author_name": author.get("name") or handle,
        "author_handle": handle,
        "follower_count": followers or None,
        "published_at": tweet.get("createdAt") or tweet.get("created_at") or "",
        "view_count": views,
        "like_count": likes,
        "comment_count": replies,
        "share_count": reposts,
        "save_count": bookmarks,
        "hot_score": round(float(engagement), 2),
        "breakout_score": round(float(breakout_score), 2),
        "engagement_follower_ratio": round(engagement_follower_ratio, 5) if followers else None,
        "view_follower_ratio": round(view_follower_ratio, 5) if followers else None,
        "bookmark_like_ratio": round(bookmark_like_ratio, 5) if bookmarks else None,
        "evidence": evidence,
        "raw": tweet,
    }


def passes_filters(item: dict[str, Any], args: argparse.Namespace) -> bool:
    followers = item.get("follower_count")
    if followers is None:
        if not args.include_unknown_followers:
            return False
    elif int(followers) > args.max_followers:
        return False
    if as_int(item.get("view_count")) >= args.min_views:
        return True
    if float(item.get("hot_score") or 0) >= args.min_engagement:
        return True
    return False


def load_input(path: str) -> list[dict[str, Any]]:
    with open(path, "r", encoding="utf-8") as handle:
        payload = json.load(handle)
    return extract_tweets(payload)


def collect_tweets(args: argparse.Namespace) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[str]]:
    queries = build_queries(args)
    if args.input_json:
        tweets = load_input(args.input_json)
        return tweets, [{"source": "input_json", "ok": True, "count": len(tweets), "error": ""}], queries
    if not args.api_key:
        raise RuntimeError("Set TWITTERAPI_IO_KEY or TWITTER_API_KEY, or use --input-json")

    query_types = [item.strip() for item in args.query_types.split(",") if item.strip()]
    statuses: list[dict[str, Any]] = []
    tweets: list[dict[str, Any]] = []
    seen: set[str] = set()
    for query in queries:
        for query_type in query_types:
            try:
                payload = fetch_query(args.api_key, query, query_type)
                batch = extract_tweets(payload)
                statuses.append({"source": "twitterapi.io", "query": query, "query_type": query_type, "ok": True, "count": len(batch), "error": ""})
                for tweet in batch:
                    tid = tweet_id(tweet)
                    key = tid or json.dumps(tweet, sort_keys=True)[:200]
                    if key not in seen:
                        seen.add(key)
                        tweets.append(tweet)
            except Exception as exc:
                statuses.append({"source": "twitterapi.io", "query": query, "query_type": query_type, "ok": False, "count": 0, "error": str(exc)})
    return tweets, statuses, queries


def format_markdown(payload: dict[str, Any]) -> str:
    lines = [
        "# X Viral Topic Results",
        "",
        f"- Queries: {' | '.join(payload['query'].get('queries') or [])}",
        f"- Window days: {payload['query'].get('days')}",
        f"- Results: {len(payload['results'])}",
        "",
    ]
    for index, item in enumerate(payload["results"], 1):
        followers = item.get("follower_count") if item.get("follower_count") is not None else "unknown"
        lines.extend(
            [
                f"## {index}. @{item.get('author_handle') or ''}",
                "",
                f"- URL: {item['url']}",
                f"- Text: {item.get('summary') or item.get('title')}",
                f"- Followers: {followers}; views: {item['view_count']}; likes: {item['like_count']}; reposts: {item['share_count']}; replies: {item['comment_count']}; bookmarks: {item['save_count']}",
                f"- Score: {item['breakout_score']}; evidence: {', '.join(item['evidence'])}",
                "",
            ]
        )
    if payload.get("source_status"):
        lines.extend(["## Source Status", ""])
        for status in payload["source_status"]:
            lines.append(
                f"- {status.get('source')} {status.get('query_type', '')}: ok={status.get('ok')} count={status.get('count', '')} error={status.get('error', '')}"
            )
    return "\n".join(lines).strip() + "\n"


def main() -> int:
    args = parse_args()
    queries = build_queries(args)
    if args.dry_run:
        print(json.dumps({"queries": queries, "args": redacted_args(args)}, ensure_ascii=False, indent=2))
        return 0

    tweets, source_status, queries = collect_tweets(args)
    results = [normalize_tweet(tweet, args) for tweet in tweets]
    results = [item for item in results if passes_filters(item, args)]
    results.sort(
        key=lambda item: (
            float(item.get("breakout_score") or 0),
            float(item.get("engagement_follower_ratio") or 0),
            float(item.get("hot_score") or 0),
        ),
        reverse=True,
    )
    payload = {
        "query": {
            "topic": args.topic,
            "queries": queries,
            "days": args.days,
            "max_followers": args.max_followers,
            "min_views": args.min_views,
            "min_engagement": args.min_engagement,
        },
        "source_status": source_status,
        "results": results[: args.limit],
    }
    if args.format == "markdown":
        print(format_markdown(payload), end="")
    else:
        print(json.dumps(payload, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())