search_posts.py

"""Collect a bounded sample of recent public posts with xfetch. Python 3.9+."""

import argparse
from http.client import HTTPException
import json
import os
from pathlib import Path
import sys
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen


ENDPOINT = "https://api.xfetch.io/v1/search/recent"
PAGE_SIZE = 20


def fetch_page(api_key, params):
    request = Request(ENDPOINT + "?" + urlencode(params))
    request.add_unredirected_header("Authorization", "Bearer " + api_key)
    request.add_header("Accept", "application/json")
    try:
        with urlopen(request, timeout=30) as response:
            page = json.load(response)
    except HTTPError as exc:
        raise RuntimeError(f"HTTP {exc.code}; no automatic retry was made.") from None
    except (URLError, TimeoutError, OSError, HTTPException):
        raise RuntimeError("Network failure; the last request's charge is unknown.") from None
    except (ValueError, UnicodeError):
        raise RuntimeError("The server returned an unreadable JSON response.") from None

    try:
        posts = page["data"]
        charged = page["meta"]["credits"]["charged"]
        token = page["meta"].get("pagination", {}).get("next_token")
        valid_posts = isinstance(posts, list) and all(
            isinstance(post, dict) and isinstance(post.get("id"), str) and post["id"]
            for post in posts
        )
        if not valid_posts or type(charged) is not int or charged < 0:
            raise ValueError
        if token is not None and (not isinstance(token, str) or not token):
            raise ValueError
    except (KeyError, TypeError, AttributeError, ValueError):
        raise RuntimeError("Unexpected search response; check the API reference.") from None
    return posts, charged, token


def collect_posts(api_key, query, max_pages):
    # Keep these parameters unchanged when continuing with next_token.
    params = {"query": query, "limit": PAGE_SIZE, "sort_order": "recency"}
    posts_by_id = {}
    seen_tokens = set()
    result = {
        "query": query,
        "pages_fetched": 0,
        "credits_charged_confirmed": 0,
        "stop_reason": "page_limit",
        "data": [],
    }
    for _ in range(max_pages):
        try:
            posts, charged, token = fetch_page(api_key, params)
        except RuntimeError as exc:
            result["stop_reason"] = "error"
            result["error"] = str(exc)
            break
        result["pages_fetched"] += 1
        result["credits_charged_confirmed"] += charged
        for post in posts:
            posts_by_id.setdefault(post["id"], post)

        if token is None:
            result["stop_reason"] = "no_next_token"
            break
        if token in seen_tokens:
            result["stop_reason"] = "repeated_token"
            break
        seen_tokens.add(token)
        params["next_token"] = token

    result["data"] = list(posts_by_id.values())
    return result


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("query", help="Keywords or an xfetch search-query expression")
    parser.add_argument("--max-pages", type=int, choices=range(1, 4), default=3)
    parser.add_argument("--output", type=Path, default=Path("posts.json"))
    args = parser.parse_args(argv)
    query = args.query.strip()
    api_key = os.environ.get("XFETCH_API_KEY", "").strip()
    if not query:
        parser.error("query must not be empty")
    if not api_key:
        parser.error("set XFETCH_API_KEY in your environment before running")
    if args.output.exists() or not args.output.parent.is_dir():
        parser.error("choose a new output filename in an existing directory")

    result = collect_posts(api_key, query, args.max_pages)
    try:
        with args.output.open("x", encoding="utf-8") as output:
            json.dump(result, output, ensure_ascii=False, indent=2)
            output.write("\n")
    except OSError:
        print("Could not write the export; check the path and permissions.", file=sys.stderr)
        return 1

    print(
        f"Saved {len(result['data'])} unique posts from {result['pages_fetched']} pages. "
        f"Confirmed credits: {result['credits_charged_confirmed']}. "
        f"Stop reason: {result['stop_reason']}.",
        file=sys.stderr,
    )
    if result["stop_reason"] in ("error", "repeated_token"):
        print("Run incomplete; the export retains earlier successful pages.", file=sys.stderr)
        return 1
    return 0


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