X Research

{
  "accounts": [
    { "username": "anthropic", "note": "Anthropic", "addedAt": "2026-02-08T00:00:00Z" },
    { "username": "OpenAI", "note": "OpenAI", "addedAt": "2026-02-08T00:00:00Z" }
  ]
}

[project]
name = "x-research-scripts"
version = "0.0.0"
requires-python = ">=3.11"
dependencies = ["httpx>=0.27"]

[tool.ruff]
line-length = 88
target-version = "py311"

[tool.ruff.lint]
select = ["E", "F", "I", "W", "UP", "B", "SIM", "A"]

"""X API wrapper -- search, threads, profiles, single tweets.

Uses Bearer token from env: X_BEARER_TOKEN or XAI_API_KEY.
"""

from __future__ import annotations

import os
import re
import time
from datetime import UTC, datetime
from urllib.parse import quote

import httpx

BASE = "https://api.x.com/2"
RATE_DELAY_S = 0.35
FIELDS = (
    "tweet.fields=created_at,public_metrics,author_id,conversation_id,entities"
    "&expansions=author_id"
    "&user.fields=username,name,public_metrics"
)


def get_token() -> str:
    """Get bearer token from X_BEARER_TOKEN or XAI_API_KEY env var."""
    token = os.environ.get("X_BEARER_TOKEN") or os.environ.get("XAI_API_KEY")
    if not token:
        msg = "X_BEARER_TOKEN or XAI_API_KEY not found in environment"
        raise RuntimeError(msg)
    return token


def _parse_tweets(raw: dict) -> list[dict]:
    """Parse API response into tweet dicts."""
    data = raw.get("data")
    if not data:
        return []

    users: dict[str, dict] = {}
    for u in raw.get("includes", {}).get("users", []):
        users[u["id"]] = u

    tweets = []
    for t in data:
        u = users.get(t.get("author_id", ""), {})
        m = t.get("public_metrics", {})
        entities = t.get("entities", {})
        username = u.get("username", "?")

        tweets.append(
            {
                "id": t["id"],
                "text": t.get("text", ""),
                "author_id": t.get("author_id", ""),
                "username": username,
                "name": u.get("name", "?"),
                "created_at": t.get("created_at", ""),
                "conversation_id": t.get("conversation_id", ""),
                "metrics": {
                    "likes": m.get("like_count", 0),
                    "retweets": m.get("retweet_count", 0),
                    "replies": m.get("reply_count", 0),
                    "quotes": m.get("quote_count", 0),
                    "impressions": m.get("impression_count", 0),
                    "bookmarks": m.get("bookmark_count", 0),
                },
                "urls": [
                    e["expanded_url"]
                    for e in entities.get("urls", [])
                    if e.get("expanded_url")
                ],
                "mentions": [
                    e["username"]
                    for e in entities.get("mentions", [])
                    if e.get("username")
                ],
                "hashtags": [
                    h["tag"] for h in entities.get("hashtags", []) if h.get("tag")
                ],
                "tweet_url": f"https://x.com/{username}/status/{t['id']}",
            }
        )

    return tweets


def _parse_since(since: str) -> str | None:
    """Parse '1h', '3d', etc. into ISO 8601 timestamp."""
    match = re.match(r"^(\d+)(m|h|d)$", since)
    if match:
        num = int(match.group(1))
        unit = match.group(2)
        seconds = num * {"m": 60, "h": 3600, "d": 86400}[unit]
        start = datetime.now(tz=UTC).timestamp() - seconds
        return datetime.fromtimestamp(start, tz=UTC).isoformat()

    if "T" in since or "-" in since:
        try:
            return datetime.fromisoformat(since).isoformat()
        except ValueError:
            return None

    return None


def _api_get(url: str) -> dict:
    """Make authenticated GET request to X API."""
    token = get_token()
    response = httpx.get(
        url,
        headers={"Authorization": f"Bearer {token}"},
        timeout=30,
    )

    if response.status_code == 429:
        reset = response.headers.get("x-rate-limit-reset")
        wait = max(int(reset) - int(time.time()), 1) if reset else 60
        msg = f"Rate limited. Resets in {wait}s"
        raise RuntimeError(msg)

    if not response.is_success:
        msg = f"X API {response.status_code}: {response.text[:200]}"
        raise RuntimeError(msg)

    return response.json()


def search(
    query: str,
    *,
    max_results: int = 100,
    pages: int = 1,
    sort_order: str = "relevancy",
    since: str | None = None,
) -> list[dict]:
    """Search recent tweets (last 7 days)."""
    max_results = max(min(max_results, 100), 10)
    encoded = quote(query, safe="")

    time_filter = ""
    if since:
        start_time = _parse_since(since)
        if start_time:
            time_filter = f"&start_time={start_time}"

    all_tweets: list[dict] = []
    next_token: str | None = None

    for page in range(pages):
        pagination = f"&pagination_token={next_token}" if next_token else ""
        url = (
            f"{BASE}/tweets/search/recent?query={encoded}"
            f"&max_results={max_results}&{FIELDS}"
            f"&sort_order={sort_order}{time_filter}{pagination}"
        )

        raw = _api_get(url)
        all_tweets.extend(_parse_tweets(raw))

        next_token = raw.get("meta", {}).get("next_token")
        if not next_token:
            break
        if page < pages - 1:
            time.sleep(RATE_DELAY_S)

    return all_tweets


def thread(conversation_id: str, *, pages: int = 2) -> list[dict]:
    """Fetch full conversation thread by root tweet ID."""
    query = f"conversation_id:{conversation_id}"
    tweets = search(query, pages=pages, sort_order="recency")

    try:
        raw = _api_get(f"{BASE}/tweets/{conversation_id}?{FIELDS}")
        if raw.get("data") and not isinstance(raw["data"], list):
            root = _parse_tweets({**raw, "data": [raw["data"]]})
            if root:
                tweets.insert(0, root[0])
    except RuntimeError:
        pass

    return tweets


def profile(
    username: str,
    *,
    count: int = 20,
    include_replies: bool = False,
) -> tuple[dict, list[dict]]:
    """Get recent tweets from a specific user."""
    user_url = (
        f"{BASE}/users/by/username/{username}"
        "?user.fields=public_metrics,description,created_at"
    )
    user_data = _api_get(user_url)

    if not user_data.get("data"):
        msg = f"User @{username} not found"
        raise RuntimeError(msg)

    user = user_data["data"]
    time.sleep(RATE_DELAY_S)

    reply_filter = "" if include_replies else " -is:reply"
    query = f"from:{username} -is:retweet{reply_filter}"
    tweets = search(
        query,
        max_results=min(count, 100),
        sort_order="recency",
    )

    return user, tweets


def get_tweet(tweet_id: str) -> dict | None:
    """Fetch a single tweet by ID."""
    raw = _api_get(f"{BASE}/tweets/{tweet_id}?{FIELDS}")
    if raw.get("data") and not isinstance(raw["data"], list):
        parsed = _parse_tweets({**raw, "data": [raw["data"]]})
        return parsed[0] if parsed else None
    return None


def sort_by(
    tweets: list[dict],
    metric: str = "likes",
) -> list[dict]:
    """Sort tweets by engagement metric."""
    return sorted(
        tweets,
        key=lambda t: t["metrics"].get(metric, 0),
        reverse=True,
    )


def filter_engagement(
    tweets: list[dict],
    *,
    min_likes: int | None = None,
    min_impressions: int | None = None,
) -> list[dict]:
    """Filter tweets by minimum engagement."""
    result = []
    for t in tweets:
        if min_likes and t["metrics"]["likes"] < min_likes:
            continue
        if min_impressions and t["metrics"]["impressions"] < min_impressions:
            continue
        result.append(t)
    return result


def dedupe(tweets: list[dict]) -> list[dict]:
    """Deduplicate tweets by ID."""
    seen: set[str] = set()
    result = []
    for t in tweets:
        if t["id"] not in seen:
            seen.add(t["id"])
            result.append(t)
    return result

"""File-based cache for X API results.

Avoids re-fetching identical queries within a TTL window.
"""

from __future__ import annotations

import hashlib
import json
import time
from pathlib import Path

CACHE_DIR = Path(__file__).resolve().parent.parent / "data" / "cache"
DEFAULT_TTL_S = 15 * 60


def _ensure_dir() -> None:
    CACHE_DIR.mkdir(parents=True, exist_ok=True)


def _cache_key(query: str, params: str = "") -> str:
    return hashlib.md5(
        f"{query}|{params}".encode(),
        usedforsecurity=False,
    ).hexdigest()[:12]


def get(
    query: str,
    params: str = "",
    ttl_s: float = DEFAULT_TTL_S,
) -> list[dict] | None:
    """Return cached tweets or None if expired/missing."""
    _ensure_dir()
    path = CACHE_DIR / f"{_cache_key(query, params)}.json"
    if not path.exists():
        return None

    try:
        entry = json.loads(path.read_text())
        if time.time() - entry["timestamp"] > ttl_s:
            path.unlink(missing_ok=True)
            return None
        return entry["tweets"]
    except (json.JSONDecodeError, KeyError):
        return None


def set(  # noqa: A001
    query: str,
    params: str = "",
    tweets: list[dict] | None = None,
) -> None:
    """Cache tweet results."""
    _ensure_dir()
    path = CACHE_DIR / f"{_cache_key(query, params)}.json"
    entry = {
        "query": query,
        "params": params,
        "timestamp": time.time(),
        "tweets": tweets or [],
    }
    path.write_text(json.dumps(entry, indent=2))


def prune(ttl_s: float = DEFAULT_TTL_S) -> int:
    """Remove expired cache entries. Returns count removed."""
    _ensure_dir()
    removed = 0
    for path in CACHE_DIR.glob("*.json"):
        if time.time() - path.stat().st_mtime > ttl_s:
            path.unlink(missing_ok=True)
            removed += 1
    return removed


def clear() -> int:
    """Remove all cache entries. Returns count removed."""
    _ensure_dir()
    files = list(CACHE_DIR.glob("*.json"))
    for path in files:
        path.unlink(missing_ok=True)
    return len(files)

"""Format tweets for terminal or markdown output."""

from __future__ import annotations

import time as _time
from datetime import UTC, datetime
from urllib.parse import urlparse


def _compact(n: int) -> str:
    if n >= 1_000_000:
        return f"{n / 1_000_000:.1f}M"
    if n >= 1_000:
        return f"{n / 1_000:.1f}K"
    return str(n)


def _time_ago(date_str: str) -> str:
    try:
        dt = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
    except ValueError:
        return "?"
    diff = _time.time() - dt.timestamp()
    mins = int(diff // 60)
    if mins < 60:
        return f"{mins}m"
    hours = mins // 60
    if hours < 24:
        return f"{hours}h"
    return f"{hours // 24}d"


def _clean_text(text: str) -> str:
    """Remove t.co links from tweet text."""
    import re

    return re.sub(r"https://t\.co/\S+", "", text).strip()


def format_tweet_terminal(
    tweet: dict,
    index: int | None = None,
    *,
    full: bool = False,
) -> str:
    """Format a single tweet for terminal display."""
    prefix = f"{index + 1}. " if index is not None else ""
    m = tweet["metrics"]
    engagement = f"{_compact(m['likes'])}L {_compact(m['impressions'])}I"
    age = _time_ago(tweet["created_at"])

    text = tweet["text"]
    if not full and len(text) > 200:
        text = text[:197] + "..."
    text = _clean_text(text)

    out = f"{prefix}@{tweet['username']} ({engagement} | {age})\n{text}"

    if tweet.get("urls"):
        out += f"\n  {tweet['urls'][0]}"
    out += f"\n  {tweet['tweet_url']}"

    return out


def format_results_terminal(
    tweets: list[dict],
    query: str = "",
    limit: int = 15,
) -> str:
    """Format a list of tweets for terminal display."""
    shown = tweets[:limit]

    parts = []
    if query:
        parts.append(f'"{query}" -- {len(tweets)} results\n')

    for i, t in enumerate(shown):
        parts.append(format_tweet_terminal(t, i))

    if len(tweets) > limit:
        parts.append(f"\n... +{len(tweets) - limit} more")

    return "\n\n".join(parts)


def format_tweet_markdown(tweet: dict) -> str:
    """Format a single tweet for markdown research docs."""
    m = tweet["metrics"]
    engagement = f"{m['likes']}L {m['impressions']}I"
    text = _clean_text(tweet["text"]).replace("\n", "\n  > ")

    out = (
        f"- **@{tweet['username']}** ({engagement})"
        f" [Tweet]({tweet['tweet_url']})\n  > {text}"
    )

    if tweet.get("urls"):
        links = ", ".join(f"[{urlparse(u).hostname}]({u})" for u in tweet["urls"])
        out += f"\n  Links: {links}"

    return out


def format_research_markdown(
    query: str,
    tweets: list[dict],
    queries: list[str] | None = None,
) -> str:
    """Format results as a markdown research document."""
    date = datetime.now(tz=UTC).strftime("%Y-%m-%d")

    lines = [
        f"# X Research: {query}\n",
        f"**Date:** {date}",
        f"**Tweets found:** {len(tweets)}\n",
        "## Top Results (by engagement)\n",
    ]

    for t in tweets[:30]:
        lines.append(format_tweet_markdown(t))
        lines.append("")

    lines.append("---\n")
    lines.append("## Research Metadata\n")
    lines.append(f"- **Query:** {query}")
    lines.append(f"- **Date:** {date}")
    lines.append(f"- **Tweets scanned:** {len(tweets)}")
    lines.append(f"- **Est. cost:** ~${len(tweets) * 0.005:.2f}")

    if queries:
        lines.append("- **Search queries:**")
        for q in queries:
            lines.append(f"  - `{q}`")

    return "\n".join(lines) + "\n"


def format_profile_terminal(user: dict, tweets: list[dict]) -> str:
    """Format a user profile for terminal display."""
    m = user.get("public_metrics", {})
    followers = _compact(m.get("followers_count", 0))
    tweet_count = _compact(m.get("tweet_count", 0))

    lines = [
        f"@{user.get('username', '?')} -- {user.get('name', '?')}",
        f"{followers} followers | {tweet_count} tweets",
    ]

    desc = user.get("description", "")
    if desc:
        lines.append(desc[:150])

    lines.append("\nRecent:\n")

    for i, t in enumerate(tweets[:10]):
        lines.append(format_tweet_terminal(t, i))
        lines.append("")

    return "\n".join(lines)

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = ["httpx>=0.27"]
# ///
"""x-search -- CLI for X/Twitter research.

Commands:
  search <query> [options]    Search recent tweets
  thread <tweet_id>           Fetch full conversation thread
  profile <username>          Recent tweets from a user
  tweet <tweet_id>            Fetch a single tweet
  watchlist                   Show watchlist
  watchlist add <user>        Add user to watchlist
  watchlist remove <user>     Remove user from watchlist
  watchlist check             Check recent tweets from all watchlist accounts
  cache clear                 Clear search cache
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

# Allow importing sibling modules when run via uv
sys.path.insert(0, str(Path(__file__).resolve().parent))

import os
from datetime import UTC

import x_api as api
import x_cache as cache
import x_format as fmt
import x_xai as xai


def _use_xai() -> bool:
    """True if only XAI_API_KEY is available (no X_BEARER_TOKEN)."""
    return not os.environ.get("X_BEARER_TOKEN") and bool(os.environ.get("XAI_API_KEY"))


SKILL_DIR = Path(__file__).resolve().parent.parent
DATA_DIR = SKILL_DIR / "data"
WATCHLIST_PATH = DATA_DIR / "watchlist.json"


def _load_watchlist() -> dict:
    if not WATCHLIST_PATH.exists():
        return {"accounts": []}
    return json.loads(WATCHLIST_PATH.read_text())


def _save_watchlist(wl: dict) -> None:
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    WATCHLIST_PATH.write_text(json.dumps(wl, indent=2) + "\n")


def cmd_search(args: argparse.Namespace) -> None:
    """Search recent tweets."""
    query = args.query
    limit = args.limit

    if args.quick:
        limit = min(limit, 10)

    if args.from_user and "from:" not in query.lower():
        query += f" from:{args.from_user.lstrip('@')}"

    if _use_xai():
        print(xai.search(query, limit=limit, sort=args.sort, since=args.since))
        return

    pages = args.pages
    if args.quick:
        pages = 1

    if "is:retweet" not in query:
        query += " -is:retweet"
    if (args.quick or args.no_replies) and "is:reply" not in query:
        query += " -is:reply"

    cache_ttl = 3600 if args.quick else 900
    cache_params = f"sort={args.sort}&pages={pages}&since={args.since or '7d'}"

    cached = cache.get(query, cache_params, cache_ttl)
    if cached is not None:
        tweets = cached
        print(f"(cached -- {len(tweets)} tweets)", file=sys.stderr)
    else:
        sort_order = "recency" if args.sort == "recent" else "relevancy"
        tweets = api.search(
            query,
            pages=pages,
            sort_order=sort_order,
            since=args.since,
        )
        cache.set(query, cache_params, tweets)

    raw_count = len(tweets)

    if args.min_likes > 0 or args.min_impressions > 0:
        tweets = api.filter_engagement(
            tweets,
            min_likes=args.min_likes or None,
            min_impressions=args.min_impressions or None,
        )

    if args.quality:
        tweets = api.filter_engagement(tweets, min_likes=10)

    if args.sort != "recent":
        tweets = api.sort_by(tweets, args.sort)

    tweets = api.dedupe(tweets)

    if args.json:
        print(json.dumps(tweets[:limit], indent=2))
    elif args.markdown:
        print(fmt.format_research_markdown(query, tweets, queries=[query]))
    else:
        print(fmt.format_results_terminal(tweets, query=query, limit=limit))

    if args.save:
        import re
        from datetime import datetime

        slug = re.sub(r"[^a-zA-Z0-9]+", "-", query).strip("-")[:40].lower()
        date = datetime.now(tz=UTC).strftime("%Y-%m-%d")
        save_dir = Path.home() / "x-research-output"
        save_dir.mkdir(parents=True, exist_ok=True)
        save_path = save_dir / f"x-research-{slug}-{date}.md"
        md = fmt.format_research_markdown(query, tweets, queries=[query])
        save_path.write_text(md)
        print(f"\nSaved to {save_path}", file=sys.stderr)

    cost = f"{raw_count * 0.005:.2f}"
    if args.quick:
        print(
            f"\nquick mode | {raw_count} tweets read (~${cost})",
            file=sys.stderr,
        )
    else:
        print(f"\n{raw_count} tweets read | est. cost ~${cost}", file=sys.stderr)

    filtered = f" -> {len(tweets)} after filters" if raw_count != len(tweets) else ""
    since_label = f" | since {args.since}" if args.since else ""
    print(
        f"{raw_count} tweets{filtered}"
        f" | sorted by {args.sort} | {pages} page(s){since_label}",
        file=sys.stderr,
    )


def cmd_thread(args: argparse.Namespace) -> None:
    """Fetch full conversation thread."""
    if _use_xai():
        print(xai.thread(args.tweet_id))
        return

    tweets = api.thread(args.tweet_id, pages=min(args.pages, 5))
    if not tweets:
        print("No tweets found in thread.")
        return

    print(f"Thread ({len(tweets)} tweets)\n")
    for t in tweets:
        print(fmt.format_tweet_terminal(t, full=True))
        print()


def cmd_profile(args: argparse.Namespace) -> None:
    """Recent tweets from a user."""
    username = args.username.lstrip("@")

    if _use_xai():
        print(xai.profile(username, count=args.count))
        return

    user, tweets = api.profile(
        username,
        count=args.count,
        include_replies=args.replies,
    )

    if args.json:
        print(json.dumps({"user": user, "tweets": tweets}, indent=2))
    else:
        print(fmt.format_profile_terminal(user, tweets))


def cmd_tweet(args: argparse.Namespace) -> None:
    """Fetch a single tweet."""
    if _use_xai():
        print(xai.tweet(args.tweet_id))
        return

    tweet = api.get_tweet(args.tweet_id)
    if not tweet:
        print("Tweet not found.")
        return

    if args.json:
        print(json.dumps(tweet, indent=2))
    else:
        print(fmt.format_tweet_terminal(tweet, full=True))


def cmd_watchlist(args: argparse.Namespace) -> None:
    """Manage watchlist."""
    wl = _load_watchlist()
    sub = args.watchlist_action

    if sub == "add":
        username = args.watchlist_user.lstrip("@")
        existing = [
            a for a in wl["accounts"] if a["username"].lower() == username.lower()
        ]
        if existing:
            print(f"@{username} already on watchlist.")
            return
        from datetime import datetime

        entry = {
            "username": username,
            "addedAt": datetime.now(tz=UTC).isoformat(),
        }
        if args.watchlist_note:
            entry["note"] = " ".join(args.watchlist_note)
        wl["accounts"].append(entry)
        _save_watchlist(wl)
        note = f" ({entry.get('note', '')})" if entry.get("note") else ""
        print(f"Added @{username} to watchlist.{note}")
        return

    if sub in ("remove", "rm"):
        username = args.watchlist_user.lstrip("@")
        before = len(wl["accounts"])
        wl["accounts"] = [
            a for a in wl["accounts"] if a["username"].lower() != username.lower()
        ]
        _save_watchlist(wl)
        if len(wl["accounts"]) < before:
            print(f"Removed @{username} from watchlist.")
        else:
            print(f"@{username} not found on watchlist.")
        return

    if sub == "check":
        if not wl["accounts"]:
            print("Watchlist is empty. Add accounts with: watchlist add <user>")
            return
        print(f"Checking {len(wl['accounts'])} watchlist accounts...\n")
        for acct in wl["accounts"]:
            note = f" ({acct['note']})" if acct.get("note") else ""
            print(f"\n--- @{acct['username']}{note} ---")
            try:
                if _use_xai():
                    print(xai.profile(acct["username"], count=3))
                else:
                    _, tweets = api.profile(acct["username"], count=5)
                    if not tweets:
                        print("  No recent tweets.")
                    else:
                        for t in tweets[:3]:
                            print(fmt.format_tweet_terminal(t))
                            print()
            except RuntimeError as exc:
                print(
                    f"  Error checking @{acct['username']}: {exc}",
                    file=sys.stderr,
                )
        return

    if not wl["accounts"]:
        print("Watchlist is empty. Add accounts with: watchlist add <user>")
        return
    print(f"Watchlist ({len(wl['accounts'])} accounts)\n")
    for acct in wl["accounts"]:
        note = f" -- {acct['note']}" if acct.get("note") else ""
        added = acct.get("addedAt", "?").split("T")[0]
        print(f"  @{acct['username']}{note} (added {added})")


def cmd_cache(args: argparse.Namespace) -> None:
    """Manage cache."""
    if args.cache_action == "clear":
        removed = cache.clear()
        print(f"Cleared {removed} cached entries.")
    else:
        removed = cache.prune()
        print(f"Pruned {removed} expired entries.")


def build_parser() -> argparse.ArgumentParser:
    """Build the CLI argument parser."""
    parser = argparse.ArgumentParser(
        prog="x-search",
        description="X/Twitter research CLI",
    )
    sub = parser.add_subparsers(dest="command")

    # search
    sp = sub.add_parser("search", aliases=["s"], help="Search recent tweets")
    sp.add_argument("query", help="Search query")
    sp.add_argument(
        "--sort",
        default="likes",
        choices=["likes", "impressions", "retweets", "recent"],
    )
    sp.add_argument("--since", help="Time filter: 1h, 3h, 12h, 1d, 7d")
    sp.add_argument("--min-likes", type=int, default=0)
    sp.add_argument("--min-impressions", type=int, default=0)
    sp.add_argument("--pages", type=int, default=1)
    sp.add_argument("--limit", type=int, default=15)
    sp.add_argument("--quick", action="store_true")
    sp.add_argument("--from-user", metavar="USER", dest="from_user")
    sp.add_argument("--quality", action="store_true")
    sp.add_argument("--no-replies", action="store_true")
    sp.add_argument("--save", action="store_true")
    sp.add_argument("--json", action="store_true")
    sp.add_argument("--markdown", action="store_true")

    # thread
    sp = sub.add_parser("thread", aliases=["t"], help="Fetch conversation")
    sp.add_argument("tweet_id", help="Root tweet ID")
    sp.add_argument("--pages", type=int, default=2)

    # profile
    sp = sub.add_parser("profile", aliases=["p"], help="User profile")
    sp.add_argument("username", help="X username")
    sp.add_argument("--count", type=int, default=20)
    sp.add_argument("--replies", action="store_true")
    sp.add_argument("--json", action="store_true")

    # tweet
    sp = sub.add_parser("tweet", help="Fetch single tweet")
    sp.add_argument("tweet_id", help="Tweet ID")
    sp.add_argument("--json", action="store_true")

    # watchlist
    sp = sub.add_parser("watchlist", aliases=["wl"], help="Manage watchlist")
    sp.add_argument(
        "watchlist_action",
        nargs="?",
        default="show",
        choices=["show", "add", "remove", "rm", "check"],
    )
    sp.add_argument("watchlist_user", nargs="?")
    sp.add_argument("watchlist_note", nargs="*")

    # cache
    sp = sub.add_parser("cache", help="Manage cache")
    sp.add_argument(
        "cache_action",
        nargs="?",
        default="prune",
        choices=["clear", "prune"],
    )

    return parser


def main() -> None:
    """Entry point."""
    parser = build_parser()
    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        return

    commands = {
        "search": cmd_search,
        "s": cmd_search,
        "thread": cmd_thread,
        "t": cmd_thread,
        "profile": cmd_profile,
        "p": cmd_profile,
        "tweet": cmd_tweet,
        "watchlist": cmd_watchlist,
        "wl": cmd_watchlist,
        "cache": cmd_cache,
    }

    handler = commands.get(args.command)
    if handler:
        handler(args)
    else:
        parser.print_help()


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(1)

"""xAI Responses API backend for X search.

Uses Grok's x_search server-side tool to search X/Twitter
when only XAI_API_KEY is available (no X_BEARER_TOKEN).
"""

from __future__ import annotations

import os

import httpx

XAI_BASE = "https://api.x.ai/v1"
XAI_MODEL = "grok-4-fast"


def get_xai_key() -> str | None:
    """Return XAI_API_KEY if set, else None."""
    return os.environ.get("XAI_API_KEY")


def _responses_api(
    prompt: str,
    *,
    x_search_opts: dict | None = None,
) -> dict:
    """Call the xAI Responses API with x_search tool."""
    key = get_xai_key()
    if not key:
        msg = "XAI_API_KEY not found in environment"
        raise RuntimeError(msg)

    tool: dict = {"type": "x_search"}
    if x_search_opts:
        tool.update(x_search_opts)

    response = httpx.post(
        f"{XAI_BASE}/responses",
        headers={
            "Authorization": f"Bearer {key}",
            "Content-Type": "application/json",
        },
        json={
            "model": XAI_MODEL,
            "input": [{"role": "user", "content": prompt}],
            "tools": [tool],
        },
        timeout=120,
    )

    if not response.is_success:
        msg = f"xAI API {response.status_code}: {response.text[:200]}"
        raise RuntimeError(msg)

    return response.json()


def _extract_text(data: dict) -> str:
    """Extract text content from xAI Responses API output."""
    for item in data.get("output", []):
        if item.get("type") == "message":
            for content in item.get("content", []):
                if content.get("type") == "output_text":
                    return content.get("text", "")
    return "(no results)"


def search(
    query: str,
    *,
    limit: int = 10,
    sort: str = "likes",
    since: str | None = None,
) -> str:
    """Search X via Grok's x_search tool. Returns formatted text."""
    time_hint = f" from the last {since}" if since else ""
    sort_hint = f" sorted by {sort}" if sort != "recent" else " most recent"

    prompt = (
        f"Search X for tweets about: {query}\n"
        f"Find up to {limit} results{time_hint},{sort_hint}.\n"
        "For each tweet, include:\n"
        "- @username\n"
        "- Tweet text (first 200 chars)\n"
        "- Engagement: likes, reposts, views\n"
        "- Tweet URL\n"
        "Format as a numbered list. Include engagement numbers."
    )

    data = _responses_api(prompt)
    text = _extract_text(data)

    usage = data.get("usage", {})
    tool_details = usage.get("server_side_tool_usage_details", {})
    x_calls = tool_details.get("x_search_calls", 0)

    return f"{text}\n\n[xAI backend | {x_calls} x_search call(s)]"


def thread(tweet_id: str) -> str:
    """Fetch a thread via Grok's x_search tool."""
    prompt = (
        f"Find and display the full conversation thread for tweet ID {tweet_id}. "
        "Show each tweet in order with @username, text, and engagement metrics."
    )
    data = _responses_api(prompt)
    return _extract_text(data)


def profile(username: str, *, count: int = 10) -> str:
    """Fetch recent tweets from a user via Grok's x_search tool."""
    username = username.lstrip("@")
    prompt = (
        f"Show the {count} most recent tweets from @{username}. "
        "For each tweet include: text, engagement metrics (likes, views), "
        "and tweet URL. Also include a brief profile summary at the top."
    )
    opts = {"allowed_x_handles": [username]}
    data = _responses_api(prompt, x_search_opts=opts)
    return _extract_text(data)


def tweet(tweet_id: str) -> str:
    """Fetch a single tweet via Grok's x_search tool."""
    prompt = (
        f"Find and display the tweet with ID {tweet_id}. "
        "Show: @username, full text, engagement metrics, and tweet URL."
    )
    data = _responses_api(prompt)
    return _extract_text(data)