steam_achievements_rss.py

#!/usr/bin/env python3
"""
Steam Achievements RSS Feed Generator

A production-ready script that generates RSS feeds from Steam user achievements
using the official Steam Web API.

Author: Generated by Claude
License: MIT
"""

import argparse
import json
import logging
import os
import sys
import time
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional, Tuple
from urllib.parse import quote
import xml.etree.ElementTree as ET

import requests
from dotenv import load_dotenv


# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler('steam_rss.log', mode='a')
    ]
)
logger = logging.getLogger(__name__)


class SteamAPIError(Exception):
    """Custom exception for Steam API related errors."""
    pass


class SteamAPIClient:
    """Steam Web API client with rate limiting and error handling."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.steampowered.com"
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Steam-Achievements-RSS/1.0'
        })
        self._last_request_time = 0
        self._min_request_interval = 0.1  # 10 requests per second max

    def _rate_limit(self):
        """Enforce rate limiting between API requests."""
        elapsed = time.time() - self._last_request_time
        if elapsed < self._min_request_interval:
            time.sleep(self._min_request_interval - elapsed)
        self._last_request_time = time.time()

    def _make_request(self, endpoint: str, params: Dict, max_retries: int = 3) -> Dict:
        """Make an API request with exponential backoff retry logic."""
        self._rate_limit()

        url = f"{self.base_url}/{endpoint}"
        params['key'] = self.api_key

        for attempt in range(max_retries):
            try:
                response = self.session.get(url, params=params, timeout=30)

                if response.status_code == 429:
                    # Rate limited - exponential backoff
                    wait_time = 2 ** attempt
                    logger.warning(f"Rate limited. Waiting {wait_time}s before retry {attempt + 1}/{max_retries}")
                    time.sleep(wait_time)
                    continue
                elif response.status_code == 400:
                    # Bad request - don't retry, this is usually a game with no achievements
                    raise SteamAPIError(f"Bad request (400) - likely no achievements available")
                elif response.status_code == 403:
                    # Forbidden - don't retry, this is a permissions issue
                    raise SteamAPIError(f"Access forbidden (403) - profile may be private")

                response.raise_for_status()
                return response.json()

            except requests.exceptions.Timeout:
                if attempt < max_retries - 1:
                    logger.debug(f"Request timeout on attempt {attempt + 1}/{max_retries}, retrying...")
                    time.sleep(2 ** attempt)
                    continue
                else:
                    raise SteamAPIError("Request timed out after multiple attempts")

            except requests.exceptions.RequestException as e:
                # For 400/403 errors, don't retry
                if hasattr(e, 'response') and e.response is not None:
                    if e.response.status_code in [400, 403]:
                        raise SteamAPIError(f"HTTP {e.response.status_code}")

                if attempt < max_retries - 1:
                    logger.debug(f"Request failed on attempt {attempt + 1}/{max_retries}: {e}")
                    time.sleep(2 ** attempt)
                    continue
                else:
                    raise SteamAPIError(f"Request failed: {e}")

        raise SteamAPIError("Max retries exceeded")

    def resolve_vanity_url(self, vanity_url: str) -> str:
        """Resolve a vanity URL to SteamID64."""
        logger.info(f"Resolving vanity URL: {vanity_url}")

        try:
            data = self._make_request("ISteamUser/ResolveVanityURL/v0001/", {
                'vanityurl': vanity_url
            })

            if data.get('response', {}).get('success') == 1:
                steam_id = data['response']['steamid']
                logger.info(f"Resolved {vanity_url} to {steam_id}")
                return steam_id
            else:
                raise SteamAPIError(f"Could not resolve vanity URL '{vanity_url}'. Please check the username.")

        except KeyError:
            raise SteamAPIError("Invalid response format from Steam API")

    def get_player_achievements(self, steam_id: str, app_id: str) -> Dict:
        """Get player achievements for a specific game."""
        return self._make_request("ISteamUserStats/GetPlayerAchievements/v0001/", {
            'steamid': steam_id,
            'appid': app_id
        })

    def get_owned_games(self, steam_id: str) -> Dict:
        """Get list of games owned by the player."""
        return self._make_request("IPlayerService/GetOwnedGames/v0001/", {
            'steamid': steam_id,
            'include_appinfo': 1,
            'include_played_free_games': 1
        })

    def get_game_schema(self, app_id: str) -> Dict:
        """Get game schema including achievement definitions."""
        return self._make_request("ISteamUserStats/GetSchemaForGame/v2/", {
            'appid': app_id
        })

    def get_global_achievement_percentages(self, app_id: str) -> Dict:
        """Get global achievement percentages for a game."""
        try:
            return self._make_request("ISteamUserStats/GetGlobalAchievementPercentagesForApp/v0002/", {
                'gameid': app_id
            })
        except SteamAPIError:
            # Some games don't have global stats available
            logger.debug(f"No global achievement stats available for app {app_id}")
            return {}


class AchievementProcessor:
    """Process and filter Steam achievements."""

    def __init__(self, api_client: SteamAPIClient):
        self.api_client = api_client
        self.game_cache = {}
        self.schema_cache = {}
        self.global_stats_cache = {}

    def get_recent_achievements(self, steam_id: str, days: int = 30) -> List[Dict]:
        """Get recent achievements for a Steam user."""
        logger.info(f"Fetching recent achievements for user {steam_id} (last {days} days)")

        # Get owned games
        try:
            games_data = self.api_client.get_owned_games(steam_id)
            games = games_data.get('response', {}).get('games', [])
        except SteamAPIError as e:
            if "403" in str(e) or "private" in str(e).lower():
                raise SteamAPIError("User profile is private. Cannot access game data.")
            raise

        if not games:
            raise SteamAPIError("No games found or profile is private")

        logger.info(f"Found {len(games)} owned games")

        # Cache game info
        for game in games:
            self.game_cache[str(game['appid'])] = {
                'name': game['name'],
                'appid': game['appid']
            }

        # Calculate cutoff date
        cutoff_date = datetime.now(timezone.utc) - timedelta(days=days)
        cutoff_timestamp = int(cutoff_date.timestamp())

        all_achievements = []
        processed_games = 0
        games_with_achievements = 0

        for game in games:
            app_id = str(game['appid'])

            try:
                # Get player achievements for this game
                achievements_data = self.api_client.get_player_achievements(steam_id, app_id)

                if 'playerstats' not in achievements_data:
                    continue

                playerstats = achievements_data['playerstats']

                if not playerstats.get('success', True):
                    continue

                achievements = playerstats.get('achievements', [])
                if not achievements:
                    continue

                games_with_achievements += 1

                # Filter for recent achievements
                recent_achievements = [
                    ach for ach in achievements
                    if ach.get('achieved', 0) == 1 and
                    ach.get('unlocktime', 0) >= cutoff_timestamp
                ]

                if recent_achievements:
                    # Get game schema for achievement details
                    schema = self._get_game_schema(app_id)
                    global_stats = self._get_global_stats(app_id)

                    # Enrich achievement data
                    for achievement in recent_achievements:
                        enriched = self._enrich_achievement(
                            achievement, app_id, schema, global_stats
                        )
                        if enriched:
                            all_achievements.append(enriched)

                processed_games += 1

                # Progress logging
                if processed_games % 20 == 0:
                    logger.info(f"Processed {processed_games}/{len(games)} games ({games_with_achievements} with achievements), found {len(all_achievements)} recent achievements")

            except SteamAPIError as e:
                # Log only unexpected errors, not common 400s for games without achievements
                if "400" not in str(e):
                    logger.debug(f"Could not get achievements for game {app_id} ({game.get('name', 'Unknown')}): {e}")
                continue

        # Sort by unlock time (newest first)
        all_achievements.sort(key=lambda x: x['unlocktime'], reverse=True)

        logger.info(f"Scan complete: {processed_games} games processed ({games_with_achievements} had achievements)")
        logger.info(f"Found {len(all_achievements)} recent achievements in the last {days} days")
        return all_achievements

    def _get_game_schema(self, app_id: str) -> Dict:
        """Get and cache game schema."""
        if app_id not in self.schema_cache:
            try:
                schema_data = self.api_client.get_game_schema(app_id)
                self.schema_cache[app_id] = schema_data.get('game', {})
            except SteamAPIError:
                self.schema_cache[app_id] = {}

        return self.schema_cache[app_id]

    def _get_global_stats(self, app_id: str) -> Dict:
        """Get and cache global achievement statistics."""
        if app_id not in self.global_stats_cache:
            global_data = self.api_client.get_global_achievement_percentages(app_id)
            stats_dict = {}

            for achievement in global_data.get('achievementpercentages', {}).get('achievements', []):
                stats_dict[achievement['name']] = achievement['percent']

            self.global_stats_cache[app_id] = stats_dict

        return self.global_stats_cache[app_id]

    def _enrich_achievement(self, achievement: Dict, app_id: str, schema: Dict, global_stats: Dict) -> Optional[Dict]:
        """Enrich achievement data with additional information."""
        api_name = achievement['apiname']

        # Find achievement details in schema
        achievement_details = None
        for ach in schema.get('availableGameStats', {}).get('achievements', []):
            if ach['name'] == api_name:
                achievement_details = ach
                break

        if not achievement_details:
            return None

        # Get global percentage
        global_percent = global_stats.get(api_name, 0)

        # Determine rarity category
        if global_percent == 0:
            rarity = "Unknown"
        elif global_percent < 1:
            rarity = "Ultra Rare"
        elif global_percent < 5:
            rarity = "Very Rare"
        elif global_percent < 15:
            rarity = "Rare"
        elif global_percent < 50:
            rarity = "Uncommon"
        else:
            rarity = "Common"

        return {
            'name': achievement_details.get('displayName', api_name),
            'description': achievement_details.get('description', ''),
            'icon': achievement_details.get('icon', ''),
            'unlocktime': achievement['unlocktime'],
            'app_id': app_id,
            'game_name': self.game_cache[app_id]['name'],
            'global_percent': global_percent,
            'rarity': rarity
        }


class RSSGenerator:
    """Generate RSS 2.0 feeds from achievement data."""

    def __init__(self, title: str = "Steam Achievements", description: str = "Recent Steam achievements",
                 link: str = "https://steamcommunity.com"):
        self.title = title
        self.description = description
        self.link = link

    def generate_feed(self, achievements: List[Dict], steam_id: str) -> str:
        """Generate RSS 2.0 XML feed from achievements."""
        # Create root RSS element
        rss = ET.Element("rss", version="2.0")
        rss.set("xmlns:atom", "http://www.w3.org/2005/Atom")

        # Create channel
        channel = ET.SubElement(rss, "channel")

        # Add channel metadata
        ET.SubElement(channel, "title").text = self.title
        ET.SubElement(channel, "description").text = self.description
        ET.SubElement(channel, "link").text = self.link
        ET.SubElement(channel, "language").text = "en-us"
        ET.SubElement(channel, "lastBuildDate").text = self._format_rfc2822(datetime.now(timezone.utc))
        ET.SubElement(channel, "generator").text = "Steam Achievements RSS Generator"

        # Add atom:link for self-reference
        atom_link = ET.SubElement(channel, "atom:link")
        atom_link.set("href", self.link)
        atom_link.set("rel", "self")
        atom_link.set("type", "application/rss+xml")

        # Add achievement items
        for achievement in achievements:
            item = ET.SubElement(channel, "item")

            # Title
            title = f"🏆 {achievement['name']} - {achievement['game_name']}"
            ET.SubElement(item, "title").text = title

            # Description with rich HTML
            description = self._generate_achievement_html(achievement)
            ET.SubElement(item, "description").text = description

            # Link to game store page
            link = f"https://store.steampowered.com/app/{achievement['app_id']}"
            ET.SubElement(item, "link").text = link

            # GUID (unique identifier)
            guid = f"steam-achievement-{steam_id}-{achievement['app_id']}-{achievement['unlocktime']}"
            guid_elem = ET.SubElement(item, "guid")
            guid_elem.text = guid
            guid_elem.set("isPermaLink", "false")

            # Publication date
            pub_date = datetime.fromtimestamp(achievement['unlocktime'], timezone.utc)
            ET.SubElement(item, "pubDate").text = self._format_rfc2822(pub_date)

            # Category
            ET.SubElement(item, "category").text = "Gaming"

        # Convert to string with XML declaration
        rough_string = ET.tostring(rss, encoding='unicode')
        return f'<?xml version="1.0" encoding="UTF-8"?>\n{rough_string}'

    def _generate_achievement_html(self, achievement: Dict) -> str:
        """Generate rich HTML description for an achievement."""
        unlock_date = datetime.fromtimestamp(achievement['unlocktime'], timezone.utc)
        formatted_date = unlock_date.strftime("%B %d, %Y at %I:%M %p UTC")

        # Format percentage
        if achievement['global_percent'] > 0:
            percent_str = f"{achievement['global_percent']:.2f}%"
        else:
            percent_str = "Unknown"

        html = f"""
<img src="{achievement['icon']}" alt="Achievement Icon"
     style="float: left; margin-right: 10px; margin-bottom: 5px; width: 64px; height: 64px;" />

<p>{achievement['description']}</p>

<hr>

<div style="margin-top: 10px; padding: 18px; background-color: #f5f5f5;
            border-radius: 4px; font-size: 0.9em;">
    <strong>🎮 Game:</strong>
    <a href="https://store.steampowered.com/app/{achievement['app_id']}">{achievement['game_name']}</a><br/>
    <strong>🏆 Rarity:</strong> {achievement['rarity']} ({percent_str} of players)<br/>
    <strong>📅 Unlocked:</strong> {formatted_date}
</div>

<div style="clear: both;"></div>
        """.strip()

        return html

    def _format_rfc2822(self, dt: datetime) -> str:
        """Format datetime as RFC 2822 string for RSS."""
        return dt.strftime("%a, %d %b %Y %H:%M:%S %z")


def setup_argument_parser() -> argparse.ArgumentParser:
    """Set up command line argument parser."""
    parser = argparse.ArgumentParser(
        description="Generate RSS feed from Steam achievements",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --steam-id 76561198024399018 --days 7
  %(prog)s --vanity-url myusername --days 30 --output feed.xml
  %(prog)s --vanity-url myusername --title "My Gaming Achievements"
        """
    )

    # Steam ID options (mutually exclusive)
    id_group = parser.add_mutually_exclusive_group(required=True)
    id_group.add_argument('--steam-id', help='Steam ID64')
    id_group.add_argument('--vanity-url', help='Steam vanity URL (custom username)')

    # Configuration options
    parser.add_argument('--days', type=int, default=30,
                       help='Number of days to look back for achievements (default: 30)')
    parser.add_argument('--output', default='steam_achievements.xml',
                       help='Output RSS file path (default: steam_achievements.xml)')

    # RSS metadata
    parser.add_argument('--title', default='Steam Achievements',
                       help='RSS feed title (default: Steam Achievements)')
    parser.add_argument('--description', default='Recent Steam achievements',
                       help='RSS feed description (default: Recent Steam achievements)')
    parser.add_argument('--link', default='https://steamcommunity.com',
                       help='RSS feed link (default: https://steamcommunity.com)')

    # API key option
    parser.add_argument('--api-key', help='Steam Web API key (can also use .env file)')

    # Logging
    parser.add_argument('--verbose', '-v', action='store_true',
                       help='Enable verbose logging')

    return parser


def load_configuration(args: argparse.Namespace) -> Tuple[str, str]:
    """Load configuration from arguments and environment."""
    # Load .env file
    load_dotenv()

    # Get API key
    api_key = args.api_key or os.getenv('STEAM_API_KEY')
    if not api_key:
        raise ValueError(
            "Steam API key is required. Provide it via --api-key argument or "
            "STEAM_API_KEY environment variable in .env file.\n"
            "Get your API key at: https://steamcommunity.com/dev/apikey"
        )

    # Get Steam ID
    steam_id = args.steam_id or os.getenv('STEAM_ID')
    if not steam_id and not args.vanity_url:
        vanity_url = os.getenv('STEAM_VANITY_URL')
        if vanity_url:
            args.vanity_url = vanity_url

    return api_key, steam_id


def main():
    """Main application entry point."""
    parser = setup_argument_parser()
    args = parser.parse_args()

    # Configure logging level
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)

    try:
        # Load configuration
        api_key, steam_id = load_configuration(args)

        # Initialize API client
        api_client = SteamAPIClient(api_key)

        # Resolve vanity URL if needed
        if args.vanity_url:
            steam_id = api_client.resolve_vanity_url(args.vanity_url)

        if not steam_id:
            raise ValueError("Steam ID is required via --steam-id argument or environment variable")

        # Process achievements
        processor = AchievementProcessor(api_client)
        achievements = processor.get_recent_achievements(steam_id, args.days)

        if not achievements:
            logger.info(f"No achievements found in the last {args.days} days")
            logger.info("The RSS feed will be created but will be empty")
            # Create empty feed
            achievements = []
        else:
            logger.info(f"Creating RSS feed with {len(achievements)} achievements")

        # Generate RSS feed
        rss_generator = RSSGenerator(
            title=args.title,
            description=args.description,
            link=args.link
        )

        rss_content = rss_generator.generate_feed(achievements, steam_id)

        # Write to file
        with open(args.output, 'w', encoding='utf-8') as f:
            f.write(rss_content)

        logger.info(f"RSS feed generated successfully: {args.output}")
        logger.info(f"Feed contains {len(achievements)} achievement(s)")

    except ValueError as e:
        logger.error(f"Configuration error: {e}")
        sys.exit(1)
    except SteamAPIError as e:
        logger.error(f"Steam API error: {e}")
        sys.exit(1)
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        if args.verbose:
            logger.exception("Full traceback:")
        sys.exit(1)


if __name__ == "__main__":
    main()