scripts/fetch_steam_news.py

#!/usr/bin/env python3
"""
Diagnostic helper for inspecting public Steam News API payloads.

Defaults to Going Medieval (appid 1029780), which recently had a SteamDB patch
note marked as "Major". The goal is to inspect what fields are actually exposed
by the public ISteamNews/GetNewsForApp endpoint.

Examples:
    python scripts/fetch_steam_news.py
    python scripts/fetch_steam_news.py --appid 1029780 --count 20 --raw
    python scripts/fetch_steam_news.py --appid 1029780 --contains "1.0"
"""

from __future__ import annotations

import argparse
import json
import sys
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path

import httpx


STEAM_NEWS_API_URL = "https://api.steampowered.com/ISteamNews/GetNewsForApp/v2/"
DEFAULT_APP_ID = 1029780
DEFAULT_COUNT = 20


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Inspect recent Steam News API payloads for a game."
    )
    parser.add_argument(
        "--appid",
        default=str(DEFAULT_APP_ID),
        help=f"Steam appid to inspect (default: {DEFAULT_APP_ID} for Going Medieval).",
    )
    parser.add_argument(
        "--count",
        type=int,
        default=DEFAULT_COUNT,
        help=f"Number of news items to fetch (default: {DEFAULT_COUNT}).",
    )
    parser.add_argument(
        "--maxlength",
        type=int,
        default=0,
        help="Steam API maxlength parameter (default: 0).",
    )
    parser.add_argument(
        "--contains",
        default=None,
        help="Only print items whose title or contents contain this substring.",
    )
    parser.add_argument(
        "--raw",
        action="store_true",
        help="Dump full raw JSON response after the human-readable summary.",
    )
    parser.add_argument(
        "--output",
        default=None,
        help="Optional path to save the raw JSON response.",
    )
    return parser.parse_args()


def _format_timestamp(value: object) -> str | None:
    if not isinstance(value, int) or value <= 0:
        return None
    return datetime.fromtimestamp(value, tz=timezone.utc).isoformat()


def _matches_filter(item: dict, needle: str | None) -> bool:
    if not needle:
        return True
    haystacks = [
        str(item.get("title", "")),
        str(item.get("contents", "")),
        str(item.get("feedlabel", "")),
        str(item.get("feedname", "")),
    ]
    lowered = needle.lower()
    return any(lowered in text.lower() for text in haystacks)


def _print_summary(app_id: str, news_items: list[dict]) -> None:
    print("=" * 80)
    print(f"Steam News API inspection for appid {app_id}")
    print(f"Items returned: {len(news_items)}")

    all_keys = sorted({key for item in news_items for key in item.keys()})
    print(f"Observed keys: {', '.join(all_keys) if all_keys else '(none)'}")

    feedlabels = Counter(str(item.get("feedlabel", "")) for item in news_items)
    feednames = Counter(str(item.get("feedname", "")) for item in news_items)
    print(f"feedlabel values: {dict(feedlabels)}")
    print(f"feedname values: {dict(feednames)}")
    print("=" * 80)


def _print_items(news_items: list[dict]) -> None:
    for index, item in enumerate(news_items, start=1):
        title = item.get("title", "")
        date_iso = _format_timestamp(item.get("date"))
        print(f"[{index}] {title}")
        print(f"    date: {date_iso}")
        print(f"    gid: {item.get('gid')}")
        print(f"    url: {item.get('url')}")
        print(f"    author: {item.get('author')}")
        print(f"    feedlabel: {item.get('feedlabel')}")
        print(f"    feedname: {item.get('feedname')}")
        print(f"    tags: {item.get('tags')}")
        if "feed_type" in item:
            print(f"    feed_type: {item.get('feed_type')}")
        if "announcement_body" in item:
            print("    announcement_body: present")
        contents = str(item.get("contents", ""))
        preview = contents[:220].replace("\n", " ").strip()
        print(f"    contents_preview: {preview}")
        print(f"    contents_length: {len(contents)}")
        print(f"    keys: {sorted(item.keys())}")
        print()


def main() -> int:
    args = _parse_args()
    params = {
        "appid": args.appid,
        "count": args.count,
        "maxlength": args.maxlength,
    }

    try:
        with httpx.Client(timeout=30.0) as client:
            response = client.get(STEAM_NEWS_API_URL, params=params)
            response.raise_for_status()
            payload = response.json()
    except Exception as exc:  # deliberate broad catch for diagnostics
        print(f"Steam News request failed: {exc}", file=sys.stderr)
        return 1

    appnews = payload.get("appnews", {})
    news_items = appnews.get("newsitems", [])
    filtered_items = [item for item in news_items if _matches_filter(item, args.contains)]

    _print_summary(str(args.appid), news_items)

    if args.contains:
        print(f"Filter substring: {args.contains!r}")
        print(f"Filtered items: {len(filtered_items)}")
        print("-" * 80)

    _print_items(filtered_items)

    if args.output:
        output_path = Path(args.output)
        output_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
        print(f"Raw JSON saved to: {output_path}")

    if args.raw:
        print("=" * 80)
        print("Raw JSON")
        print("=" * 80)
        print(json.dumps(payload, indent=2, ensure_ascii=False))

    return 0


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