import logging import os import sys from dataclasses import dataclass, field from pathlib import Path from typing import List from bot import MastodonRSSBot # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger(__name__) @dataclass class Config: """Configuration loaded from environment variables.""" instance_url: str client_id: str client_secret: str access_token: str feed_urls: List[str] = field(default_factory=list) toot_visibility: str = "public" check_interval: int = 300 state_file: Path = field(default_factory=lambda: Path("/state/processed_entries.txt")) @classmethod def from_env(cls) -> "Config": """Load configuration from environment variables.""" instance_url = os.environ.get("MASTODON_INSTANCE_URL") client_id = os.environ.get("MASTODON_CLIENT_ID") client_secret = os.environ.get("MASTODON_CLIENT_SECRET") access_token = os.environ.get("MASTODON_ACCESS_TOKEN") if not all([instance_url, client_id, client_secret, access_token]): missing = [ k for k, v in { "MASTODON_INSTANCE_URL": instance_url, "MASTODON_CLIENT_ID": client_id, "MASTODON_CLIENT_SECRET": client_secret, "MASTODON_ACCESS_TOKEN": access_token, }.items() if not v ] raise ValueError(f"Missing required environment variables: {', '.join(missing)}") # Parse feeds feed_urls = [] # 1. Legacy single feed URL if os.environ.get("RSS_FEED_URL"): feed_urls.append(os.environ["RSS_FEED_URL"]) # 2. Comma-separated list of feeds if os.environ.get("RSS_FEEDS"): feeds = [url.strip() for url in os.environ["RSS_FEEDS"].split(",") if url.strip()] feed_urls.extend(feeds) # 3. File containing list of feeds feeds_file = os.environ.get("FEEDS_FILE") if feeds_file: path = Path(feeds_file) if path.exists(): try: content = path.read_text().splitlines() file_feeds = [line.strip() for line in content if line.strip() and not line.startswith("#")] feed_urls.extend(file_feeds) except Exception as e: logger.error(f"Error reading feeds file {feeds_file}: {e}") else: logger.warning(f"Feeds file configured but not found: {feeds_file}") # Deduplicate while preserving order unique_feed_urls = list(dict.fromkeys(feed_urls)) if not unique_feed_urls: raise ValueError("No RSS feeds configured. Please set RSS_FEED_URL, RSS_FEEDS, or FEEDS_FILE.") return cls( instance_url=instance_url, # type: ignore # checked above client_id=client_id, # type: ignore client_secret=client_secret, # type: ignore access_token=access_token, # type: ignore feed_urls=unique_feed_urls, toot_visibility=os.environ.get("TOOT_VISIBILITY", "public"), check_interval=int(os.environ.get("CHECK_INTERVAL", "300")), state_file=Path(os.environ.get("PROCESSED_ENTRIES_FILE", "/state/processed_entries.txt")), ) def main(): """Initialize and run the bot with environment configuration""" logger.info("Starting Mastodon RSS Bot...") try: config = Config.from_env() except ValueError as e: logger.critical(str(e)) sys.exit(1) except Exception as e: logger.critical(f"Failed to load configuration: {e}") sys.exit(1) logger.info("Bot configured successfully:") logger.info(f" Instance: {config.instance_url}") logger.info(f" Monitoring {len(config.feed_urls)} feed(s):") for url in config.feed_urls: logger.info(f" - {url}") logger.info(f" Visibility: {config.toot_visibility}") logger.info(f" Check interval: {config.check_interval} seconds") logger.info(f" State file: {config.state_file}") bot = MastodonRSSBot( client_id=config.client_id, client_secret=config.client_secret, access_token=config.access_token, instance_url=config.instance_url, feed_urls=config.feed_urls, toot_visibility=config.toot_visibility, check_interval=config.check_interval, state_file=config.state_file, ) # Start the bot bot.run() if __name__ == "__main__": main()