+41

aggregators/reddit-highlights/.gitignore

··· 1 1 + # Environment and config 2 2 + .env 3 3 + config.yaml 4 4 + venv/ 5 5 + 6 6 + # State files 7 7 + data/*.json 8 8 + data/world.xml 9 9 + 10 10 + # Python 11 11 + __pycache__/ 12 12 + *.py[cod] 13 13 + *$py.class 14 14 + *.so 15 15 + .Python 16 16 + build/ 17 17 + develop-eggs/ 18 18 + dist/ 19 19 + downloads/ 20 20 + eggs/ 21 21 + .eggs/ 22 22 + lib/ 23 23 + lib64/ 24 24 + parts/ 25 25 + sdist/ 26 26 + var/ 27 27 + wheels/ 28 28 + *.egg-info/ 29 29 + .installed.cfg 30 30 + *.egg 31 31 + 32 32 + # Testing 33 33 + .pytest_cache/ 34 34 + .coverage 35 35 + htmlcov/ 36 36 + 37 37 + # IDE 38 38 + .vscode/ 39 39 + .idea/ 40 40 + *.swp 41 41 + *.swo

+140

aggregators/reddit-highlights/README.md

··· 1 1 + # Reddit Highlights Aggregator 2 2 + 3 3 + Aggregates video highlights from Reddit subreddits (e.g., r/nba) and posts them to Coves communities. 4 4 + 5 5 + ## Features 6 6 + 7 7 + - Fetches posts from Reddit via RSS (no API key required) 8 8 + - Extracts streamable.com video links 9 9 + - Posts to configured Coves communities with proper attribution 10 10 + - Anti-detection jitter (randomized polling intervals) 11 11 + - State tracking for deduplication 12 12 + - Docker deployment with cron scheduler 13 13 + 14 14 + ## Quick Start 15 15 + 16 16 + 1. **Copy environment file:** 17 17 + ```bash 18 18 + cp .env.example .env 19 19 + ``` 20 20 + 21 21 + 2. **Configure your Coves API key:** 22 22 + ```bash 23 23 + # Edit .env and set your API key 24 24 + COVES_API_KEY=ckapi_your_key_here 25 25 + ``` 26 26 + 27 27 + 3. **Build and run:** 28 28 + ```bash 29 29 + docker-compose up -d 30 30 + ``` 31 31 + 32 32 + 4. **View logs:** 33 33 + ```bash 34 34 + docker-compose logs -f 35 35 + ``` 36 36 + 37 37 + ## Configuration 38 38 + 39 39 + ### config.yaml 40 40 + 41 41 + ```yaml 42 42 + coves_api_url: "https://coves.social" 43 43 + 44 44 + subreddits: 45 45 + - name: "nba" 46 46 + community_handle: "nba.coves.social" 47 47 + enabled: true 48 48 + 49 49 + allowed_domains: 50 50 + - streamable.com 51 51 + ``` 52 52 + 53 53 + ### Adding More Subreddits 54 54 + 55 55 + 1. Add entry to `config.yaml`: 56 56 + ```yaml 57 57 + - name: "soccer" 58 58 + community_handle: "soccer.coves.social" 59 59 + enabled: true 60 60 + ``` 61 61 + 62 62 + 2. Authorize the aggregator for the new community in Coves 63 63 + 64 64 + 3. Restart the container: 65 65 + ```bash 66 66 + docker-compose restart 67 67 + ``` 68 68 + 69 69 + ## Polling Schedule 70 70 + 71 71 + - Cron runs every **10 minutes** 72 72 + - Python script adds **0-10 minutes random jitter** 73 73 + - Effective polling interval: **10-20 minutes** (varies each run) 74 74 + 75 75 + This randomization helps avoid bot detection patterns. 76 76 + 77 77 + ## Development 78 78 + 79 79 + ### Setup 80 80 + 81 81 + ```bash 82 82 + # Create virtual environment 83 83 + python -m venv venv 84 84 + source venv/bin/activate 85 85 + 86 86 + # Install dependencies 87 87 + pip install -r requirements.txt 88 88 + ``` 89 89 + 90 90 + ### Run Tests 91 91 + 92 92 + ```bash 93 93 + pytest 94 94 + ``` 95 95 + 96 96 + ### Run Manually 97 97 + 98 98 + ```bash 99 99 + # Set environment variables 100 100 + export COVES_API_KEY=ckapi_your_key 101 101 + export SKIP_JITTER=true # Skip delay for testing 102 102 + 103 103 + # Run 104 104 + python -m src.main 105 105 + ``` 106 106 + 107 107 + ## Architecture 108 108 + 109 109 + ``` 110 110 + src/ 111 111 + ├── main.py # Orchestration (CRON entry point) 112 112 + ├── rss_fetcher.py # RSS feed fetching with retry 113 113 + ├── link_extractor.py # Streamable URL detection 114 114 + ├── coves_client.py # Coves API client 115 115 + ├── state_manager.py # Deduplication state tracking 116 116 + ├── config.py # YAML config loader 117 117 + └── models.py # Data models 118 118 + ``` 119 119 + 120 120 + ## Post Format 121 121 + 122 122 + Posts are created with: 123 123 + - **Title**: Reddit post title 124 124 + - **Embed**: Streamable video link with metadata 125 125 + - **Sources**: Link back to original Reddit post 126 126 + 127 127 + Example embed: 128 128 + ```json 129 129 + { 130 130 + "$type": "social.coves.embed.external", 131 131 + "external": { 132 132 + "uri": "https://streamable.com/abc123", 133 133 + "title": "LeBron with the chase-down block!", 134 134 + "description": "From r/nba", 135 135 + "sources": [ 136 136 + {"uri": "https://reddit.com/...", "title": "r/nba", "domain": "reddit.com"} 137 137 + ] 138 138 + } 139 139 + } 140 140 + ```

+39

aggregators/reddit-highlights/config.example.yaml

··· 1 1 + # Reddit Highlights Aggregator Configuration 2 2 + # 3 3 + # Copy this file to config.yaml and customize for your setup. 4 4 + # 5 5 + # This file configures which subreddits to fetch video highlights from 6 6 + # and which Coves communities to post them to. 7 7 + 8 8 + # Coves API endpoint (can be overridden with COVES_API_URL env var) 9 9 + coves_api_url: "https://coves.social" 10 10 + 11 11 + # Subreddit-to-community mappings 12 12 + # Add entries here for each subreddit you want to aggregate 13 13 + subreddits: 14 14 + # NBA highlights 15 15 + - name: "nba" 16 16 + community_handle: "nba.coves.social" 17 17 + enabled: true 18 18 + 19 19 + # Example: Soccer/football highlights (disabled by default) 20 20 + # - name: "soccer" 21 21 + # community_handle: "soccer.coves.social" 22 22 + # enabled: false 23 23 + 24 24 + # Example: NHL highlights (disabled by default) 25 25 + # - name: "hockey" 26 26 + # community_handle: "hockey.coves.social" 27 27 + # enabled: false 28 28 + 29 29 + # Allowed video hosting domains 30 30 + # Only posts with links to these domains will be imported 31 31 + # Add more domains here as needed 32 32 + allowed_domains: 33 33 + - streamable.com 34 34 + # Future options: 35 35 + # - streamff.com 36 36 + # - streamja.com 37 37 + 38 38 + # Logging level (debug, info, warning, error) 39 39 + log_level: "info"

+6

aggregators/reddit-highlights/pytest.ini

··· 1 1 + [pytest] 2 2 + testpaths = tests 3 3 + python_files = test_*.py 4 4 + python_classes = Test* 5 5 + python_functions = test_* 6 6 + addopts = -v --tb=short

+15

aggregators/reddit-highlights/requirements.txt

··· 1 1 + # Core dependencies 2 2 + feedparser==6.0.11 3 3 + requests==2.31.0 4 4 + pyyaml==6.0.1 5 5 + 6 6 + # Testing 7 7 + pytest==8.1.1 8 8 + pytest-cov==5.0.0 9 9 + responses==0.25.0 10 10 + 11 11 + # Development 12 12 + black==24.3.0 13 13 + mypy==1.9.0 14 14 + types-PyYAML==6.0.12.12 15 15 + types-requests==2.31.0.20240311

+1

aggregators/reddit-highlights/src/__init__.py

··· 1 1 + """Reddit Highlights Aggregator for Coves."""

+181

aggregators/reddit-highlights/src/config.py

··· 1 1 + """ 2 2 + Configuration Loader for Reddit Highlights Aggregator. 3 3 + 4 4 + Loads and validates configuration from YAML files. 5 5 + """ 6 6 + import os 7 7 + import logging 8 8 + from pathlib import Path 9 9 + from typing import Dict, Any, List 10 10 + import yaml 11 11 + from urllib.parse import urlparse 12 12 + 13 13 + from src.models import AggregatorConfig, SubredditConfig, LogLevel 14 14 + 15 15 + logger = logging.getLogger(__name__) 16 16 + 17 17 + 18 18 + class ConfigError(Exception): 19 19 + """Configuration error.""" 20 20 + 21 21 + pass 22 22 + 23 23 + 24 24 + class ConfigLoader: 25 25 + """ 26 26 + Loads and validates aggregator configuration. 27 27 + 28 28 + Supports: 29 29 + - Loading from YAML file 30 30 + - Environment variable overrides 31 31 + - Validation of required fields 32 32 + """ 33 33 + 34 34 + def __init__(self, config_path: Path): 35 35 + """ 36 36 + Initialize config loader. 37 37 + 38 38 + Args: 39 39 + config_path: Path to config.yaml file 40 40 + """ 41 41 + self.config_path = Path(config_path) 42 42 + 43 43 + def load(self) -> AggregatorConfig: 44 44 + """ 45 45 + Load and validate configuration. 46 46 + 47 47 + Returns: 48 48 + AggregatorConfig object 49 49 + 50 50 + Raises: 51 51 + ConfigError: If config is invalid or missing 52 52 + """ 53 53 + if not self.config_path.exists(): 54 54 + raise ConfigError(f"Configuration file not found: {self.config_path}") 55 55 + 56 56 + try: 57 57 + with open(self.config_path, "r") as f: 58 58 + config_data = yaml.safe_load(f) 59 59 + except yaml.YAMLError as e: 60 60 + raise ConfigError(f"Failed to parse YAML: {e}") 61 61 + 62 62 + if not config_data: 63 63 + raise ConfigError("Configuration file is empty") 64 64 + 65 65 + try: 66 66 + return self._parse_config(config_data) 67 67 + except ConfigError: 68 68 + raise 69 69 + except Exception as e: 70 70 + raise ConfigError(f"Invalid configuration: {e}") 71 71 + 72 72 + def _parse_config(self, data: Dict[str, Any]) -> AggregatorConfig: 73 73 + """ 74 74 + Parse and validate configuration data. 75 75 + 76 76 + Args: 77 77 + data: Parsed YAML data 78 78 + 79 79 + Returns: 80 80 + AggregatorConfig object 81 81 + 82 82 + Raises: 83 83 + ConfigError: If validation fails 84 84 + """ 85 85 + coves_api_url = os.getenv("COVES_API_URL", data.get("coves_api_url")) 86 86 + if not coves_api_url: 87 87 + raise ConfigError("Missing required field: coves_api_url") 88 88 + 89 89 + if not self._is_valid_url(coves_api_url): 90 90 + raise ConfigError(f"Invalid URL for coves_api_url: {coves_api_url}") 91 91 + 92 92 + # Parse log level with validation 93 93 + log_level_str = data.get("log_level", "info").lower() 94 94 + try: 95 95 + log_level = LogLevel(log_level_str) 96 96 + except ValueError: 97 97 + valid_levels = [level.value for level in LogLevel] 98 98 + raise ConfigError(f"Invalid log_level '{log_level_str}'. Valid values: {valid_levels}") 99 99 + 100 100 + subreddits_data = data.get("subreddits", []) 101 101 + if not subreddits_data: 102 102 + raise ConfigError("Configuration must include at least one subreddit") 103 103 + 104 104 + subreddits = [] 105 105 + for sub_data in subreddits_data: 106 106 + subreddit = self._parse_subreddit(sub_data) 107 107 + subreddits.append(subreddit) 108 108 + 109 109 + allowed_domains = tuple(data.get("allowed_domains", ["streamable.com"])) 110 110 + 111 111 + enabled_count = sum(1 for s in subreddits if s.enabled) 112 112 + logger.info( 113 113 + f"Loaded configuration with {len(subreddits)} subreddits ({enabled_count} enabled)" 114 114 + ) 115 115 + 116 116 + return AggregatorConfig( 117 117 + coves_api_url=coves_api_url, 118 118 + subreddits=tuple(subreddits), # Convert to tuple for immutability 119 119 + allowed_domains=allowed_domains, 120 120 + log_level=log_level, 121 121 + ) 122 122 + 123 123 + def _parse_subreddit(self, data: Dict[str, Any]) -> SubredditConfig: 124 124 + """ 125 125 + Parse and validate a single subreddit configuration. 126 126 + 127 127 + Args: 128 128 + data: Subreddit configuration data 129 129 + 130 130 + Returns: 131 131 + SubredditConfig object 132 132 + 133 133 + Raises: 134 134 + ConfigError: If validation fails 135 135 + """ 136 136 + required_fields = ["name", "community_handle"] 137 137 + for field in required_fields: 138 138 + if field not in data: 139 139 + raise ConfigError( 140 140 + f"Missing required field in subreddit config: {field}" 141 141 + ) 142 142 + 143 143 + name = data["name"] 144 144 + community_handle = data["community_handle"] 145 145 + enabled = data.get("enabled", True) 146 146 + 147 147 + if not name or not name.strip(): 148 148 + raise ConfigError("Subreddit name cannot be empty") 149 149 + 150 150 + if not community_handle or not community_handle.strip(): 151 151 + raise ConfigError(f"Community handle cannot be empty for subreddit '{name}'") 152 152 + 153 153 + return SubredditConfig( 154 154 + name=name.strip().lower(), 155 155 + community_handle=community_handle.strip(), 156 156 + enabled=enabled, 157 157 + ) 158 158 + 159 159 + def _is_valid_url(self, url: str) -> bool: 160 160 + """ 161 161 + Validate URL format. 162 162 + 163 163 + Only allows http and https schemes to prevent dangerous schemes 164 164 + like file://, javascript://, or data:// URIs. 165 165 + 166 166 + Args: 167 167 + url: URL to validate 168 168 + 169 169 + Returns: 170 170 + True if valid HTTP/HTTPS URL, False otherwise 171 171 + """ 172 172 + try: 173 173 + result = urlparse(url) 174 174 + # Only allow http and https schemes 175 175 + if result.scheme not in ("http", "https"): 176 176 + logger.warning(f"URL has invalid scheme '{result.scheme}': {url}") 177 177 + return False 178 178 + return bool(result.netloc) 179 179 + except ValueError as e: 180 180 + logger.warning(f"Failed to parse URL '{url}': {e}") 181 181 + return False

+285

aggregators/reddit-highlights/src/coves_client.py

··· 1 1 + """ 2 2 + Coves API Client for posting to communities. 3 3 + 4 4 + Handles API key authentication and posting via XRPC. 5 5 + """ 6 6 + import logging 7 7 + import requests 8 8 + from typing import Dict, List, Optional 9 9 + 10 10 + logger = logging.getLogger(__name__) 11 11 + 12 12 + 13 13 + class CovesAPIError(Exception): 14 14 + """Base exception for Coves API errors.""" 15 15 + 16 16 + def __init__(self, message: str, status_code: int = None, response_body: str = None): 17 17 + super().__init__(message) 18 18 + self.status_code = status_code 19 19 + self.response_body = response_body 20 20 + 21 21 + 22 22 + class CovesAuthenticationError(CovesAPIError): 23 23 + """Raised when authentication fails (401 Unauthorized).""" 24 24 + pass 25 25 + 26 26 + 27 27 + class CovesNotFoundError(CovesAPIError): 28 28 + """Raised when a resource is not found (404 Not Found).""" 29 29 + pass 30 30 + 31 31 + 32 32 + class CovesRateLimitError(CovesAPIError): 33 33 + """Raised when rate limit is exceeded (429 Too Many Requests).""" 34 34 + pass 35 35 + 36 36 + 37 37 + class CovesForbiddenError(CovesAPIError): 38 38 + """Raised when access is forbidden (403 Forbidden).""" 39 39 + pass 40 40 + 41 41 + 42 42 + class CovesClient: 43 43 + """ 44 44 + Client for posting to Coves communities via XRPC. 45 45 + 46 46 + Handles: 47 47 + - API key authentication 48 48 + - Creating posts in communities (social.coves.community.post.create) 49 49 + - External embed formatting 50 50 + """ 51 51 + 52 52 + # API key format constants (must match Go constants in apikey_service.go) 53 53 + API_KEY_PREFIX = "ckapi_" 54 54 + API_KEY_TOTAL_LENGTH = 70 # 6 (prefix) + 64 (32 bytes hex-encoded) 55 55 + 56 56 + def __init__(self, api_url: str, api_key: str): 57 57 + """ 58 58 + Initialize Coves client with API key authentication. 59 59 + 60 60 + Args: 61 61 + api_url: Coves API URL for posting (e.g., "https://coves.social") 62 62 + api_key: Coves API key, 70 characters total (6-char prefix + 64-char hex token) 63 63 + 64 64 + Raises: 65 65 + ValueError: If api_key is empty, has wrong prefix, or wrong length 66 66 + """ 67 67 + # Validate API key format for early failure with clear error 68 68 + if not api_key: 69 69 + raise ValueError("API key cannot be empty") 70 70 + if not api_key.startswith(self.API_KEY_PREFIX): 71 71 + raise ValueError(f"API key must start with '{self.API_KEY_PREFIX}'") 72 72 + if len(api_key) != self.API_KEY_TOTAL_LENGTH: 73 73 + raise ValueError( 74 74 + f"API key must be {self.API_KEY_TOTAL_LENGTH} characters " 75 75 + f"(got {len(api_key)})" 76 76 + ) 77 77 + 78 78 + self.api_url = api_url.rstrip('/') 79 79 + self.api_key = api_key 80 80 + self.session = requests.Session() 81 81 + self.session.headers['Authorization'] = f'Bearer {api_key}' 82 82 + self.session.headers['Content-Type'] = 'application/json' 83 83 + 84 84 + def authenticate(self): 85 85 + """ 86 86 + No-op for API key authentication. 87 87 + 88 88 + API key is set in the session headers during initialization. 89 89 + This method is kept for backward compatibility with existing code 90 90 + that calls authenticate() before making requests. 91 91 + """ 92 92 + logger.info("Using API key authentication (no session creation needed)") 93 93 + 94 94 + def create_post( 95 95 + self, 96 96 + community_handle: str, 97 97 + content: str, 98 98 + facets: List[Dict], 99 99 + title: Optional[str] = None, 100 100 + embed: Optional[Dict] = None, 101 101 + thumbnail_url: Optional[str] = None 102 102 + ) -> str: 103 103 + """ 104 104 + Create a post in a community. 105 105 + 106 106 + Args: 107 107 + community_handle: Community handle (e.g., "world-news.coves.social") 108 108 + content: Post content (rich text) 109 109 + facets: Rich text facets (formatting, links) 110 110 + title: Optional post title 111 111 + embed: Optional external embed 112 112 + thumbnail_url: Optional thumbnail URL (for trusted aggregators only) 113 113 + 114 114 + Returns: 115 115 + AT Proto URI of created post (e.g., "at://did:plc:.../social.coves.post/...") 116 116 + 117 117 + Raises: 118 118 + CovesAuthenticationError: If authentication fails (401) 119 119 + CovesForbiddenError: If access is denied (403) 120 120 + CovesNotFoundError: If community not found (404) 121 121 + CovesRateLimitError: If rate limit exceeded (429) 122 122 + CovesAPIError: For other API errors or invalid responses 123 123 + requests.RequestException: For network-level errors 124 124 + """ 125 125 + try: 126 126 + # Prepare post data for social.coves.community.post.create endpoint 127 127 + post_data = { 128 128 + "community": community_handle, 129 129 + "content": content, 130 130 + "facets": facets 131 131 + } 132 132 + 133 133 + # Add title if provided 134 134 + if title: 135 135 + post_data["title"] = title 136 136 + 137 137 + # Add embed if provided 138 138 + if embed: 139 139 + post_data["embed"] = embed 140 140 + 141 141 + # Add thumbnail URL at top level if provided (for trusted aggregators) 142 142 + if thumbnail_url: 143 143 + post_data["thumbnailUrl"] = thumbnail_url 144 144 + 145 145 + # Use Coves-specific endpoint (not direct PDS write) 146 146 + # This provides validation, authorization, and business logic 147 147 + logger.info(f"Creating post in community: {community_handle}") 148 148 + 149 149 + # Make HTTP request to XRPC endpoint using session with API key 150 150 + url = f"{self.api_url}/xrpc/social.coves.community.post.create" 151 151 + response = self.session.post(url, json=post_data, timeout=30) 152 152 + 153 153 + # Handle specific error cases 154 154 + if not response.ok: 155 155 + # Log status code but not full response body (may contain sensitive data) 156 156 + logger.error(f"Post creation failed with status {response.status_code}") 157 157 + self._raise_for_status(response) 158 158 + 159 159 + try: 160 160 + result = response.json() 161 161 + post_uri = result["uri"] 162 162 + except (ValueError, KeyError) as e: 163 163 + # ValueError for invalid JSON, KeyError for missing 'uri' field 164 164 + logger.error(f"Failed to parse post creation response: {e}") 165 165 + raise CovesAPIError( 166 166 + f"Invalid response from server: {e}", 167 167 + status_code=response.status_code, 168 168 + response_body=response.text 169 169 + ) 170 170 + 171 171 + logger.info(f"Post created: {post_uri}") 172 172 + return post_uri 173 173 + 174 174 + except requests.RequestException as e: 175 175 + # Network errors, timeouts, etc. 176 176 + logger.error(f"Network error creating post: {e}") 177 177 + raise 178 178 + except CovesAPIError: 179 179 + # Re-raise our custom exceptions as-is 180 180 + raise 181 181 + 182 182 + def create_external_embed( 183 183 + self, 184 184 + uri: str, 185 185 + title: str, 186 186 + description: str, 187 187 + sources: Optional[List[Dict]] = None, 188 188 + embed_type: Optional[str] = None, 189 189 + provider: Optional[str] = None, 190 190 + domain: Optional[str] = None 191 191 + ) -> Dict: 192 192 + """ 193 193 + Create external embed object for hot-linked content. 194 194 + 195 195 + Args: 196 196 + uri: URL of the external content 197 197 + title: Title of the content 198 198 + description: Description/summary 199 199 + sources: Optional list of source dicts with uri, title, domain 200 200 + embed_type: Type hint for rendering (article, image, video, website) 201 201 + provider: Service provider name (e.g., streamable, imgur) 202 202 + domain: Domain of the linked content (e.g., streamable.com) 203 203 + 204 204 + Returns: 205 205 + Embed dictionary ready for post creation 206 206 + """ 207 207 + external = { 208 208 + "uri": uri, 209 209 + "title": title, 210 210 + "description": description 211 211 + } 212 212 + 213 213 + if sources: 214 214 + external["sources"] = sources 215 215 + 216 216 + if embed_type: 217 217 + external["embedType"] = embed_type 218 218 + 219 219 + if provider: 220 220 + external["provider"] = provider 221 221 + 222 222 + if domain: 223 223 + external["domain"] = domain 224 224 + 225 225 + return { 226 226 + "$type": "social.coves.embed.external", 227 227 + "external": external 228 228 + } 229 229 + 230 230 + def _raise_for_status(self, response: requests.Response) -> None: 231 231 + """ 232 232 + Raise specific exceptions based on HTTP status code. 233 233 + 234 234 + Args: 235 235 + response: The HTTP response object 236 236 + 237 237 + Raises: 238 238 + CovesAuthenticationError: For 401 Unauthorized 239 239 + CovesNotFoundError: For 404 Not Found 240 240 + CovesRateLimitError: For 429 Too Many Requests 241 241 + CovesAPIError: For other 4xx/5xx errors 242 242 + """ 243 243 + status_code = response.status_code 244 244 + error_body = response.text 245 245 + 246 246 + if status_code == 401: 247 247 + raise CovesAuthenticationError( 248 248 + f"Authentication failed: {error_body}", 249 249 + status_code=status_code, 250 250 + response_body=error_body 251 251 + ) 252 252 + elif status_code == 403: 253 253 + raise CovesForbiddenError( 254 254 + f"Access forbidden: {error_body}", 255 255 + status_code=status_code, 256 256 + response_body=error_body 257 257 + ) 258 258 + elif status_code == 404: 259 259 + raise CovesNotFoundError( 260 260 + f"Resource not found: {error_body}", 261 261 + status_code=status_code, 262 262 + response_body=error_body 263 263 + ) 264 264 + elif status_code == 429: 265 265 + raise CovesRateLimitError( 266 266 + f"Rate limit exceeded: {error_body}", 267 267 + status_code=status_code, 268 268 + response_body=error_body 269 269 + ) 270 270 + else: 271 271 + raise CovesAPIError( 272 272 + f"API request failed ({status_code}): {error_body}", 273 273 + status_code=status_code, 274 274 + response_body=error_body 275 275 + ) 276 276 + 277 277 + def _get_timestamp(self) -> str: 278 278 + """ 279 279 + Get current timestamp in ISO 8601 format. 280 280 + 281 281 + Returns: 282 282 + ISO timestamp string 283 283 + """ 284 284 + from datetime import datetime, timezone 285 285 + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

+212

aggregators/reddit-highlights/src/link_extractor.py

··· 1 1 + """ 2 2 + Link extractor for detecting streamable.com URLs from Reddit RSS entries. 3 3 + """ 4 4 + import re 5 5 + import logging 6 6 + from typing import List, Optional 7 7 + from urllib.parse import urlparse 8 8 + 9 9 + logger = logging.getLogger(__name__) 10 10 + 11 11 + 12 12 + class LinkExtractor: 13 13 + """ 14 14 + Extracts video links from Reddit RSS entries. 15 15 + 16 16 + Focuses on streamable.com but can be extended to other video hosts. 17 17 + """ 18 18 + 19 19 + # Supported video hosting domains 20 20 + DEFAULT_ALLOWED_DOMAINS = ["streamable.com"] 21 21 + 22 22 + # Pattern to find URLs in text/HTML content. 23 23 + # Matches http:// or https:// followed by any non-whitespace characters 24 24 + # that aren't typically URL delimiters in HTML/text (< > " ' ) ]). 25 25 + # This is intentionally permissive to catch URLs in various contexts, 26 26 + # with further validation done by is_allowed_url(). 27 27 + URL_PATTERN = re.compile( 28 28 + r'https?://[^\s<>"\')\]]+', 29 29 + re.IGNORECASE, 30 30 + ) 31 31 + 32 32 + def __init__(self, allowed_domains: Optional[List[str]] = None): 33 33 + """ 34 34 + Initialize link extractor. 35 35 + 36 36 + Args: 37 37 + allowed_domains: List of allowed video hosting domains. 38 38 + Defaults to ["streamable.com"] 39 39 + """ 40 40 + self.allowed_domains = allowed_domains or self.DEFAULT_ALLOWED_DOMAINS 41 41 + # Normalize domains to lowercase 42 42 + self.allowed_domains = [d.lower() for d in self.allowed_domains] 43 43 + 44 44 + def extract_video_url(self, entry) -> Optional[str]: 45 45 + """ 46 46 + Extract video URL from a Reddit RSS entry. 47 47 + 48 48 + Checks: 49 49 + 1. Direct link (entry.link is a video URL) 50 50 + 2. Entry content/description for embedded video URLs 51 51 + 52 52 + Args: 53 53 + entry: feedparser entry object from Reddit RSS 54 54 + 55 55 + Returns: 56 56 + Video URL if found, None otherwise 57 57 + """ 58 58 + # Check 1: Direct link 59 59 + if hasattr(entry, "link") and entry.link: 60 60 + if self.is_allowed_url(entry.link): 61 61 + logger.debug(f"Found video URL in direct link: {entry.link}") 62 62 + return self._normalize_url(entry.link) 63 63 + 64 64 + # Check 2: Entry content (Reddit RSS uses 'content' field) 65 65 + if hasattr(entry, "content") and entry.content: 66 66 + for content_item in entry.content: 67 67 + if hasattr(content_item, "value"): 68 68 + url = self._find_url_in_text(content_item.value) 69 69 + if url: 70 70 + logger.debug(f"Found video URL in content: {url}") 71 71 + return url 72 72 + 73 73 + # Check 3: Entry description/summary 74 74 + if hasattr(entry, "description") and entry.description: 75 75 + url = self._find_url_in_text(entry.description) 76 76 + if url: 77 77 + logger.debug(f"Found video URL in description: {url}") 78 78 + return url 79 79 + 80 80 + if hasattr(entry, "summary") and entry.summary: 81 81 + url = self._find_url_in_text(entry.summary) 82 82 + if url: 83 83 + logger.debug(f"Found video URL in summary: {url}") 84 84 + return url 85 85 + 86 86 + return None 87 87 + 88 88 + def _find_url_in_text(self, text: str) -> Optional[str]: 89 89 + """ 90 90 + Find first allowed video URL in text. 91 91 + 92 92 + Args: 93 93 + text: Text/HTML content to search 94 94 + 95 95 + Returns: 96 96 + First video URL found, or None 97 97 + """ 98 98 + if not text: 99 99 + return None 100 100 + 101 101 + urls = self.URL_PATTERN.findall(text) 102 102 + for url in urls: 103 103 + # Clean up common trailing characters 104 104 + url = url.rstrip(".,;:!?") 105 105 + if self.is_allowed_url(url): 106 106 + return self._normalize_url(url) 107 107 + 108 108 + return None 109 109 + 110 110 + def is_allowed_url(self, url: str) -> bool: 111 111 + """ 112 112 + Check if URL is from an allowed video hosting domain. 113 113 + 114 114 + Args: 115 115 + url: URL to check 116 116 + 117 117 + Returns: 118 118 + True if URL is from allowed domain 119 119 + """ 120 120 + if not url: 121 121 + return False 122 122 + 123 123 + try: 124 124 + parsed = urlparse(url) 125 125 + domain = parsed.netloc.lower() 126 126 + 127 127 + # Remove www. prefix for comparison 128 128 + if domain.startswith("www."): 129 129 + domain = domain[4:] 130 130 + 131 131 + return domain in self.allowed_domains 132 132 + 133 133 + except ValueError as e: 134 134 + logger.debug(f"Failed to parse URL '{url}': {e}") 135 135 + return False 136 136 + 137 137 + def _normalize_url(self, url: str) -> str: 138 138 + """ 139 139 + Normalize URL for consistent storage/comparison. 140 140 + 141 141 + Args: 142 142 + url: URL to normalize 143 143 + 144 144 + Returns: 145 145 + Normalized URL 146 146 + """ 147 147 + # Remove trailing slashes 148 148 + url = url.rstrip("/") 149 149 + 150 150 + # Ensure https 151 151 + if url.startswith("http://"): 152 152 + url = "https://" + url[7:] 153 153 + 154 154 + return url 155 155 + 156 156 + def get_video_id(self, url: str) -> Optional[str]: 157 157 + """ 158 158 + Extract video ID from streamable URL for deduplication. 159 159 + 160 160 + Args: 161 161 + url: Streamable URL 162 162 + 163 163 + Returns: 164 164 + Video ID or None 165 165 + """ 166 166 + if not url: 167 167 + return None 168 168 + 169 169 + try: 170 170 + parsed = urlparse(url) 171 171 + # Streamable URLs are like: https://streamable.com/abc123 172 172 + path = parsed.path.strip("/") 173 173 + if path: 174 174 + # Return first path segment as video ID 175 175 + return path.split("/")[0] 176 176 + except ValueError as e: 177 177 + logger.debug(f"Failed to extract video ID from URL '{url}': {e}") 178 178 + 179 179 + return None 180 180 + 181 181 + def get_thumbnail_url(self, url: str, timeout: int = 10) -> Optional[str]: 182 182 + """ 183 183 + Fetch thumbnail URL from Streamable's oembed API. 184 184 + 185 185 + Args: 186 186 + url: Streamable video URL 187 187 + timeout: Request timeout in seconds 188 188 + 189 189 + Returns: 190 190 + Thumbnail URL or None if fetch fails 191 191 + """ 192 192 + if not url or not self.is_allowed_url(url): 193 193 + return None 194 194 + 195 195 + import requests 196 196 + 197 197 + oembed_url = f"https://api.streamable.com/oembed?url={url}" 198 198 + 199 199 + try: 200 200 + response = requests.get(oembed_url, timeout=timeout) 201 201 + response.raise_for_status() 202 202 + data = response.json() 203 203 + thumbnail = data.get("thumbnail_url") 204 204 + if thumbnail: 205 205 + logger.debug(f"Fetched thumbnail for {url}: {thumbnail[:50]}...") 206 206 + return thumbnail 207 207 + except requests.RequestException as e: 208 208 + logger.warning(f"Failed to fetch oembed for {url}: {e}") 209 209 + return None 210 210 + except (ValueError, KeyError) as e: 211 211 + logger.warning(f"Failed to parse oembed response for {url}: {e}") 212 212 + return None

+364

aggregators/reddit-highlights/src/main.py

··· 1 1 + """ 2 2 + Main Orchestration Script for Reddit Highlights Aggregator. 3 3 + 4 4 + Coordinates all components to: 5 5 + 1. Apply anti-detection jitter delay 6 6 + 2. Fetch Reddit RSS feeds 7 7 + 3. Extract streamable video links 8 8 + 4. Deduplicate via state tracking 9 9 + 5. Post to Coves communities 10 10 + 11 11 + Designed to run via CRON (single execution, then exit). 12 12 + """ 13 13 + import os 14 14 + import re 15 15 + import sys 16 16 + import time 17 17 + import random 18 18 + import logging 19 19 + from pathlib import Path 20 20 + from datetime import datetime 21 21 + from typing import Optional 22 22 + 23 23 + from src.config import ConfigLoader 24 24 + from src.rss_fetcher import RSSFetcher 25 25 + from src.link_extractor import LinkExtractor 26 26 + from src.state_manager import StateManager 27 27 + from src.coves_client import CovesClient 28 28 + from src.models import RedditPost 29 29 + 30 30 + # Setup logging 31 31 + logging.basicConfig( 32 32 + level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" 33 33 + ) 34 34 + logger = logging.getLogger(__name__) 35 35 + 36 36 + # Reddit RSS URL template 37 37 + REDDIT_RSS_URL = "https://www.reddit.com/r/{subreddit}/.rss" 38 38 + 39 39 + # Anti-detection jitter range (0-10 minutes in seconds) 40 40 + JITTER_MIN_SECONDS = 0 41 41 + JITTER_MAX_SECONDS = 600 42 42 + 43 43 + 44 44 + class Aggregator: 45 45 + """ 46 46 + Main aggregator orchestration. 47 47 + 48 48 + Coordinates all components to fetch, filter, and post video highlights. 49 49 + """ 50 50 + 51 51 + def __init__( 52 52 + self, 53 53 + config_path: Path, 54 54 + state_file: Path, 55 55 + coves_client: Optional[CovesClient] = None, 56 56 + skip_jitter: bool = False, 57 57 + ): 58 58 + """ 59 59 + Initialize aggregator. 60 60 + 61 61 + Args: 62 62 + config_path: Path to config.yaml 63 63 + state_file: Path to state.json 64 64 + coves_client: Optional CovesClient (for testing) 65 65 + skip_jitter: Skip anti-detection delay (for testing) 66 66 + """ 67 67 + self.skip_jitter = skip_jitter 68 68 + 69 69 + # Load configuration 70 70 + logger.info("Loading configuration...") 71 71 + config_loader = ConfigLoader(config_path) 72 72 + self.config = config_loader.load() 73 73 + 74 74 + # Initialize components 75 75 + logger.info("Initializing components...") 76 76 + self.rss_fetcher = RSSFetcher() 77 77 + self.link_extractor = LinkExtractor( 78 78 + allowed_domains=self.config.allowed_domains 79 79 + ) 80 80 + self.state_manager = StateManager(state_file) 81 81 + self.state_file = state_file 82 82 + 83 83 + # Initialize Coves client (or use provided one for testing) 84 84 + if coves_client: 85 85 + self.coves_client = coves_client 86 86 + else: 87 87 + api_key = os.getenv("COVES_API_KEY") 88 88 + if not api_key: 89 89 + raise ValueError("COVES_API_KEY environment variable required") 90 90 + 91 91 + self.coves_client = CovesClient( 92 92 + api_url=self.config.coves_api_url, api_key=api_key 93 93 + ) 94 94 + 95 95 + def run(self): 96 96 + """ 97 97 + Run aggregator: apply jitter, fetch, filter, post, and update state. 98 98 + 99 99 + This is the main entry point for CRON execution. 100 100 + """ 101 101 + logger.info("=" * 60) 102 102 + logger.info("Starting Reddit Highlights Aggregator") 103 103 + logger.info("=" * 60) 104 104 + 105 105 + # Anti-detection jitter: random delay before starting 106 106 + if not self.skip_jitter: 107 107 + jitter_seconds = random.uniform(JITTER_MIN_SECONDS, JITTER_MAX_SECONDS) 108 108 + logger.info( 109 109 + f"Applying anti-detection jitter: sleeping for {jitter_seconds:.1f} seconds " 110 110 + f"({jitter_seconds/60:.1f} minutes)" 111 111 + ) 112 112 + time.sleep(jitter_seconds) 113 113 + 114 114 + # Get enabled subreddits only 115 115 + enabled_subreddits = [s for s in self.config.subreddits if s.enabled] 116 116 + logger.info(f"Processing {len(enabled_subreddits)} enabled subreddits") 117 117 + 118 118 + # Authenticate once at the start 119 119 + try: 120 120 + self.coves_client.authenticate() 121 121 + except Exception as e: 122 122 + logger.error(f"Failed to authenticate: {e}") 123 123 + logger.error("Cannot continue without authentication") 124 124 + raise RuntimeError("Authentication failed") from e 125 125 + 126 126 + # Process each subreddit 127 127 + for subreddit_config in enabled_subreddits: 128 128 + try: 129 129 + self._process_subreddit(subreddit_config) 130 130 + except KeyboardInterrupt: 131 131 + # Re-raise interrupt signals - don't suppress user abort 132 132 + logger.info("Received interrupt signal, stopping...") 133 133 + raise 134 134 + except Exception as e: 135 135 + # Log error but continue with other subreddits 136 136 + logger.error( 137 137 + f"Error processing subreddit '{subreddit_config.name}': {e}", 138 138 + exc_info=True, 139 139 + ) 140 140 + continue 141 141 + 142 142 + logger.info("=" * 60) 143 143 + logger.info("Aggregator run completed") 144 144 + logger.info("=" * 60) 145 145 + 146 146 + def _process_subreddit(self, subreddit_config): 147 147 + """ 148 148 + Process a single subreddit. 149 149 + 150 150 + Args: 151 151 + subreddit_config: SubredditConfig object 152 152 + """ 153 153 + subreddit_name = subreddit_config.name 154 154 + community_handle = subreddit_config.community_handle 155 155 + 156 156 + # Sanitize subreddit name to prevent URL injection 157 157 + # Only allow alphanumeric, underscores, and hyphens 158 158 + if not re.match(r'^[a-zA-Z0-9_-]+$', subreddit_name): 159 159 + raise ValueError(f"Invalid subreddit name: {subreddit_name}") 160 160 + 161 161 + logger.info(f"Processing subreddit: r/{subreddit_name} -> {community_handle}") 162 162 + 163 163 + # Build RSS URL 164 164 + rss_url = REDDIT_RSS_URL.format(subreddit=subreddit_name) 165 165 + 166 166 + # Fetch RSS feed 167 167 + try: 168 168 + feed = self.rss_fetcher.fetch_feed(rss_url) 169 169 + except Exception as e: 170 170 + logger.error(f"Failed to fetch feed for r/{subreddit_name}: {e}") 171 171 + raise 172 172 + 173 173 + # Check for feed errors 174 174 + if feed.bozo: 175 175 + bozo_exception = getattr(feed, 'bozo_exception', None) 176 176 + logger.warning( 177 177 + f"Feed for r/{subreddit_name} has parsing issues (bozo flag set): {bozo_exception}" 178 178 + ) 179 179 + 180 180 + # Process entries 181 181 + new_posts = 0 182 182 + skipped_posts = 0 183 183 + no_video_count = 0 184 184 + 185 185 + for entry in feed.entries: 186 186 + try: 187 187 + # Extract video URL 188 188 + video_url = self.link_extractor.extract_video_url(entry) 189 189 + if not video_url: 190 190 + no_video_count += 1 191 191 + continue # Skip posts without video links 192 192 + 193 193 + # Get entry ID for deduplication 194 194 + entry_id = self._get_entry_id(entry) 195 195 + if self.state_manager.is_posted(subreddit_name, entry_id): 196 196 + skipped_posts += 1 197 197 + logger.debug(f"Skipping already-posted entry: {entry_id}") 198 198 + continue 199 199 + 200 200 + # Parse entry into RedditPost 201 201 + reddit_post = self._parse_entry(entry, subreddit_name, video_url) 202 202 + 203 203 + # Create embed with sources and video metadata 204 204 + # Note: Thumbnail is fetched by backend via unfurl service 205 205 + embed = self.coves_client.create_external_embed( 206 206 + uri=reddit_post.streamable_url, 207 207 + title=reddit_post.title, 208 208 + description=f"From r/{subreddit_name}", 209 209 + sources=[ 210 210 + { 211 211 + "uri": reddit_post.reddit_url, 212 212 + "title": f"r/{subreddit_name}", 213 213 + "domain": "reddit.com", 214 214 + } 215 215 + ], 216 216 + embed_type="video", 217 217 + provider="streamable", 218 218 + domain="streamable.com", 219 219 + ) 220 220 + 221 221 + # Post to community 222 222 + try: 223 223 + post_uri = self.coves_client.create_post( 224 224 + community_handle=community_handle, 225 225 + title=reddit_post.title, 226 226 + content="", # No additional content needed 227 227 + facets=[], 228 228 + embed=embed, 229 229 + ) 230 230 + 231 231 + # Mark as posted (only if successful) 232 232 + self.state_manager.mark_posted(subreddit_name, entry_id, post_uri) 233 233 + new_posts += 1 234 234 + logger.info(f"Posted: {reddit_post.title[:50]}... -> {post_uri}") 235 235 + 236 236 + except Exception as e: 237 237 + # Don't update state if posting failed 238 238 + logger.error(f"Failed to post '{reddit_post.title}': {e}") 239 239 + continue 240 240 + 241 241 + except Exception as e: 242 242 + # Log error but continue with other entries 243 243 + logger.error(f"Error processing entry: {e}", exc_info=True) 244 244 + continue 245 245 + 246 246 + # Update last run timestamp 247 247 + self.state_manager.update_last_run(subreddit_name, datetime.now()) 248 248 + 249 249 + logger.info( 250 250 + f"r/{subreddit_name}: {new_posts} new posts, {skipped_posts} duplicates, " 251 251 + f"{no_video_count} without video" 252 252 + ) 253 253 + 254 254 + def _get_entry_id(self, entry) -> str: 255 255 + """ 256 256 + Get unique identifier for RSS entry. 257 257 + 258 258 + Args: 259 259 + entry: feedparser entry 260 260 + 261 261 + Returns: 262 262 + Unique ID string 263 263 + """ 264 264 + # Reddit RSS uses 'id' field with format like 't3_abc123' 265 265 + if hasattr(entry, "id") and entry.id: 266 266 + return entry.id 267 267 + 268 268 + # Fallback to link 269 269 + if hasattr(entry, "link") and entry.link: 270 270 + return entry.link 271 271 + 272 272 + # Last resort: title hash (using SHA-256 for security) 273 273 + if hasattr(entry, "title") and entry.title: 274 274 + import hashlib 275 275 + 276 276 + logger.warning(f"Using fallback hash for entry ID (no id or link found)") 277 277 + return hashlib.sha256(entry.title.encode()).hexdigest() 278 278 + 279 279 + raise ValueError("Cannot determine entry ID") 280 280 + 281 281 + def _parse_entry(self, entry, subreddit: str, video_url: str) -> RedditPost: 282 282 + """ 283 283 + Parse RSS entry into RedditPost object. 284 284 + 285 285 + Args: 286 286 + entry: feedparser entry 287 287 + subreddit: Subreddit name 288 288 + video_url: Extracted video URL 289 289 + 290 290 + Returns: 291 291 + RedditPost object 292 292 + """ 293 293 + # Get entry ID 294 294 + entry_id = self._get_entry_id(entry) 295 295 + 296 296 + # Get title 297 297 + title = entry.title if hasattr(entry, "title") else "Untitled" 298 298 + 299 299 + # Get Reddit permalink 300 300 + reddit_url = entry.link if hasattr(entry, "link") else "" 301 301 + 302 302 + # Get author (Reddit RSS uses 'author' field) 303 303 + author = "" 304 304 + if hasattr(entry, "author"): 305 305 + author = entry.author 306 306 + elif hasattr(entry, "author_detail") and hasattr(entry.author_detail, "name"): 307 307 + author = entry.author_detail.name 308 308 + 309 309 + # Get published date 310 310 + published = None 311 311 + if hasattr(entry, "published_parsed") and entry.published_parsed: 312 312 + try: 313 313 + published = datetime(*entry.published_parsed[:6]) 314 314 + except (TypeError, ValueError) as e: 315 315 + logger.warning(f"Failed to parse published date for entry: {e}") 316 316 + 317 317 + return RedditPost( 318 318 + id=entry_id, 319 319 + title=title, 320 320 + link=entry.link if hasattr(entry, "link") else "", 321 321 + reddit_url=reddit_url, 322 322 + subreddit=subreddit, 323 323 + author=author, 324 324 + published=published, 325 325 + streamable_url=video_url, 326 326 + ) 327 327 + 328 328 + 329 329 + def main(): 330 330 + """ 331 331 + Main entry point for command-line execution. 332 332 + 333 333 + Usage: 334 334 + python -m src.main 335 335 + """ 336 336 + # Get paths from environment or use defaults 337 337 + config_path = Path(os.getenv("CONFIG_PATH", "config.yaml")) 338 338 + state_file = Path(os.getenv("STATE_FILE", "data/state.json")) 339 339 + 340 340 + # Check for skip jitter flag (for testing) 341 341 + skip_jitter = os.getenv("SKIP_JITTER", "").lower() in ("true", "1", "yes") 342 342 + 343 343 + # Validate config file exists 344 344 + if not config_path.exists(): 345 345 + logger.error(f"Configuration file not found: {config_path}") 346 346 + logger.error("Please create config.yaml (see config.example.yaml)") 347 347 + sys.exit(1) 348 348 + 349 349 + # Create aggregator and run 350 350 + try: 351 351 + aggregator = Aggregator( 352 352 + config_path=config_path, 353 353 + state_file=state_file, 354 354 + skip_jitter=skip_jitter, 355 355 + ) 356 356 + aggregator.run() 357 357 + sys.exit(0) 358 358 + except Exception as e: 359 359 + logger.error(f"Aggregator failed: {e}", exc_info=True) 360 360 + sys.exit(1) 361 361 + 362 362 + 363 363 + if __name__ == "__main__": 364 364 + main()

+90

aggregators/reddit-highlights/src/models.py

··· 1 1 + """ 2 2 + Data models for Reddit Highlights Aggregator. 3 3 + """ 4 4 + from dataclasses import dataclass, field 5 5 + from enum import Enum 6 6 + from typing import List, Optional, Tuple 7 7 + from datetime import datetime 8 8 + import re 9 9 + 10 10 + 11 11 + class LogLevel(Enum): 12 12 + """Valid log levels for aggregator configuration.""" 13 13 + DEBUG = "debug" 14 14 + INFO = "info" 15 15 + WARNING = "warning" 16 16 + ERROR = "error" 17 17 + CRITICAL = "critical" 18 18 + 19 19 + 20 20 + @dataclass 21 21 + class RedditPost: 22 22 + """ 23 23 + Represents a Reddit post with video content. 24 24 + 25 25 + Parsed from Reddit RSS feed entries. 26 26 + """ 27 27 + 28 28 + id: str # Reddit post ID (e.g., "t3_1abc123" or just the rkey) 29 29 + title: str # Post title 30 30 + link: str # Direct link to content (may be streamable URL) 31 31 + reddit_url: str # Permalink to Reddit post 32 32 + subreddit: str # Subreddit name (without r/) 33 33 + author: str # Reddit username 34 34 + published: Optional[datetime] = None # Post publication time 35 35 + streamable_url: Optional[str] = None # Extracted streamable URL (if found) 36 36 + 37 37 + def __post_init__(self): 38 38 + """Validate required fields.""" 39 39 + if not self.id: 40 40 + raise ValueError("RedditPost.id cannot be empty") 41 41 + if not self.title: 42 42 + raise ValueError("RedditPost.title cannot be empty") 43 43 + if not self.subreddit: 44 44 + raise ValueError("RedditPost.subreddit cannot be empty") 45 45 + 46 46 + 47 47 + @dataclass(frozen=True) 48 48 + class SubredditConfig: 49 49 + """ 50 50 + Configuration for a single subreddit source. 51 51 + 52 52 + Maps a subreddit to a Coves community. 53 53 + Immutable (frozen) to prevent accidental modification. 54 54 + """ 55 55 + 56 56 + name: str # Subreddit name (e.g., "nba") 57 57 + community_handle: str # Coves community (e.g., "nba.coves.social") 58 58 + enabled: bool = True # Whether to fetch from this subreddit 59 59 + 60 60 + def __post_init__(self): 61 61 + """Validate configuration fields.""" 62 62 + if not self.name or not self.name.strip(): 63 63 + raise ValueError("SubredditConfig.name cannot be empty") 64 64 + if not self.community_handle or not self.community_handle.strip(): 65 65 + raise ValueError("SubredditConfig.community_handle cannot be empty") 66 66 + # Validate subreddit name format (alphanumeric, underscores, hyphens only) 67 67 + if not re.match(r'^[a-zA-Z0-9_-]+$', self.name): 68 68 + raise ValueError(f"Invalid subreddit name format: {self.name}") 69 69 + 70 70 + 71 71 + @dataclass(frozen=True) 72 72 + class AggregatorConfig: 73 73 + """ 74 74 + Full aggregator configuration. 75 75 + 76 76 + Loaded from config.yaml. 77 77 + Immutable (frozen) to prevent accidental modification after loading. 78 78 + """ 79 79 + 80 80 + coves_api_url: str 81 81 + subreddits: Tuple[SubredditConfig, ...] # Use tuple for immutability 82 82 + allowed_domains: Tuple[str, ...] = ("streamable.com",) # Default tuple 83 83 + log_level: LogLevel = LogLevel.INFO 84 84 + 85 85 + def __post_init__(self): 86 86 + """Validate configuration.""" 87 87 + if not self.coves_api_url: 88 88 + raise ValueError("AggregatorConfig.coves_api_url cannot be empty") 89 89 + if not self.subreddits: 90 90 + raise ValueError("AggregatorConfig.subreddits cannot be empty")

+84

aggregators/reddit-highlights/src/rss_fetcher.py

··· 1 1 + """ 2 2 + RSS feed fetcher with retry logic and error handling. 3 3 + """ 4 4 + import time 5 5 + import logging 6 6 + import requests 7 7 + import feedparser 8 8 + from typing import Optional 9 9 + 10 10 + logger = logging.getLogger(__name__) 11 11 + 12 12 + 13 13 + class RSSFetcher: 14 14 + """ 15 15 + Fetches and parses RSS feeds with retry logic and error handling. 16 16 + 17 17 + Features: 18 18 + - Configurable timeout and retry count 19 19 + - Exponential backoff on failures 20 20 + - Custom User-Agent header (required by Reddit) 21 21 + - Automatic HTTP to HTTPS upgrade handling 22 22 + """ 23 23 + 24 24 + DEFAULT_USER_AGENT = "Coves-Reddit-Aggregator/1.0 (https://coves.social; contact@coves.social)" 25 25 + 26 26 + def __init__(self, timeout: int = 30, max_retries: int = 3, user_agent: Optional[str] = None): 27 27 + """ 28 28 + Initialize RSS fetcher. 29 29 + 30 30 + Args: 31 31 + timeout: Request timeout in seconds 32 32 + max_retries: Maximum number of retry attempts 33 33 + user_agent: Custom User-Agent string (Reddit requires this) 34 34 + """ 35 35 + self.timeout = timeout 36 36 + self.max_retries = max_retries 37 37 + self.user_agent = user_agent or self.DEFAULT_USER_AGENT 38 38 + 39 39 + def fetch_feed(self, url: str) -> feedparser.FeedParserDict: 40 40 + """ 41 41 + Fetch and parse an RSS feed. 42 42 + 43 43 + Args: 44 44 + url: RSS feed URL 45 45 + 46 46 + Returns: 47 47 + Parsed feed object 48 48 + 49 49 + Raises: 50 50 + ValueError: If URL is empty 51 51 + requests.RequestException: If all retry attempts fail 52 52 + """ 53 53 + if not url: 54 54 + raise ValueError("URL cannot be empty") 55 55 + 56 56 + last_error = None 57 57 + 58 58 + for attempt in range(self.max_retries): 59 59 + try: 60 60 + logger.info(f"Fetching feed from {url} (attempt {attempt + 1}/{self.max_retries})") 61 61 + 62 62 + headers = {"User-Agent": self.user_agent} 63 63 + response = requests.get(url, timeout=self.timeout, headers=headers) 64 64 + response.raise_for_status() 65 65 + 66 66 + # Parse with feedparser 67 67 + feed = feedparser.parse(response.content) 68 68 + 69 69 + logger.info(f"Successfully fetched feed: {feed.feed.get('title', 'Unknown')}") 70 70 + return feed 71 71 + 72 72 + except requests.RequestException as e: 73 73 + last_error = e 74 74 + logger.warning(f"Fetch attempt {attempt + 1} failed: {e}") 75 75 + 76 76 + if attempt < self.max_retries - 1: 77 77 + # Exponential backoff 78 78 + sleep_time = 2 ** attempt 79 79 + logger.info(f"Retrying in {sleep_time} seconds...") 80 80 + time.sleep(sleep_time) 81 81 + 82 82 + # All retries exhausted 83 83 + logger.error(f"Failed to fetch feed after {self.max_retries} attempts") 84 84 + raise last_error

+255

aggregators/reddit-highlights/src/state_manager.py

··· 1 1 + """ 2 2 + State Manager for tracking posted stories. 3 3 + 4 4 + Handles deduplication by tracking which stories have already been posted. 5 5 + Uses JSON file for persistence. 6 6 + """ 7 7 + import json 8 8 + import logging 9 9 + from pathlib import Path 10 10 + from datetime import datetime, timedelta 11 11 + from typing import Optional, Dict, List 12 12 + 13 13 + logger = logging.getLogger(__name__) 14 14 + 15 15 + 16 16 + class StateManager: 17 17 + """ 18 18 + Manages aggregator state for deduplication. 19 19 + 20 20 + Tracks posted Reddit entries per subreddit to prevent duplicate posting. 21 21 + 22 22 + Attributes tracked per subreddit: 23 23 + - Posted GUIDs (with timestamps and Coves post URIs) 24 24 + - Last successful run timestamp 25 25 + - Automatic cleanup of old entries to prevent state file bloat 26 26 + 27 27 + Note: The 'feed_url' parameter in methods refers to the subreddit name 28 28 + (e.g., 'nba'), not a full RSS URL. This naming is historical but the 29 29 + functionality uses subreddit names as keys. 30 30 + """ 31 31 + 32 32 + def __init__(self, state_file: Path, max_guids_per_feed: int = 100, max_age_days: int = 30): 33 33 + """ 34 34 + Initialize state manager. 35 35 + 36 36 + Args: 37 37 + state_file: Path to JSON state file 38 38 + max_guids_per_feed: Maximum GUIDs to keep per feed (default: 100) 39 39 + max_age_days: Maximum age in days for GUIDs (default: 30) 40 40 + """ 41 41 + self.state_file = Path(state_file) 42 42 + self.max_guids_per_feed = max_guids_per_feed 43 43 + self.max_age_days = max_age_days 44 44 + self.state = self._load_state() 45 45 + 46 46 + def _load_state(self) -> Dict: 47 47 + """Load state from file, or create new state if file doesn't exist.""" 48 48 + if not self.state_file.exists(): 49 49 + logger.info(f"Creating new state file at {self.state_file}") 50 50 + state = {'feeds': {}} 51 51 + self._save_state(state) 52 52 + return state 53 53 + 54 54 + try: 55 55 + with open(self.state_file, 'r') as f: 56 56 + state = json.load(f) 57 57 + logger.info(f"Loaded state from {self.state_file}") 58 58 + return state 59 59 + except json.JSONDecodeError as e: 60 60 + # Backup corrupted file before overwriting 61 61 + backup_path = self.state_file.with_suffix('.json.corrupted') 62 62 + logger.error(f"State file corrupted: {e}. Backing up to {backup_path}") 63 63 + try: 64 64 + import shutil 65 65 + shutil.copy2(self.state_file, backup_path) 66 66 + logger.info(f"Corrupted state file backed up to {backup_path}") 67 67 + except OSError as backup_error: 68 68 + logger.warning(f"Failed to backup corrupted state file: {backup_error}") 69 69 + state = {'feeds': {}} 70 70 + self._save_state(state) 71 71 + return state 72 72 + 73 73 + def _save_state(self, state: Optional[Dict] = None): 74 74 + """ 75 75 + Save state to file atomically. 76 76 + 77 77 + Uses write-to-temp-then-rename pattern to prevent corruption 78 78 + if the process is interrupted during write. 79 79 + 80 80 + Raises: 81 81 + OSError: If write fails (after logging the error) 82 82 + """ 83 83 + if state is None: 84 84 + state = self.state 85 85 + 86 86 + # Ensure parent directory exists 87 87 + self.state_file.parent.mkdir(parents=True, exist_ok=True) 88 88 + 89 89 + # Write to temp file first for atomic update 90 90 + temp_file = self.state_file.with_suffix('.json.tmp') 91 91 + try: 92 92 + with open(temp_file, 'w') as f: 93 93 + json.dump(state, f, indent=2) 94 94 + # Atomic rename (on POSIX systems) 95 95 + temp_file.rename(self.state_file) 96 96 + except OSError as e: 97 97 + logger.error(f"Failed to save state file: {e}") 98 98 + # Clean up temp file if it exists 99 99 + if temp_file.exists(): 100 100 + try: 101 101 + temp_file.unlink() 102 102 + except OSError: 103 103 + pass 104 104 + raise 105 105 + 106 106 + def _ensure_feed_exists(self, feed_url: str): 107 107 + """Ensure feed entry exists in state.""" 108 108 + if feed_url not in self.state['feeds']: 109 109 + self.state['feeds'][feed_url] = { 110 110 + 'posted_guids': [], 111 111 + 'last_successful_run': None 112 112 + } 113 113 + 114 114 + def is_posted(self, feed_url: str, guid: str) -> bool: 115 115 + """ 116 116 + Check if a story has already been posted. 117 117 + 118 118 + Args: 119 119 + feed_url: RSS feed URL 120 120 + guid: Story GUID 121 121 + 122 122 + Returns: 123 123 + True if already posted, False otherwise 124 124 + """ 125 125 + self._ensure_feed_exists(feed_url) 126 126 + 127 127 + posted_guids = self.state['feeds'][feed_url]['posted_guids'] 128 128 + return any(entry['guid'] == guid for entry in posted_guids) 129 129 + 130 130 + def mark_posted(self, feed_url: str, guid: str, post_uri: str): 131 131 + """ 132 132 + Mark a story as posted. 133 133 + 134 134 + Args: 135 135 + feed_url: RSS feed URL 136 136 + guid: Story GUID 137 137 + post_uri: AT Proto URI of created post 138 138 + """ 139 139 + self._ensure_feed_exists(feed_url) 140 140 + 141 141 + # Add to posted list 142 142 + entry = { 143 143 + 'guid': guid, 144 144 + 'post_uri': post_uri, 145 145 + 'posted_at': datetime.now().isoformat() 146 146 + } 147 147 + self.state['feeds'][feed_url]['posted_guids'].append(entry) 148 148 + 149 149 + # Auto-cleanup to keep state file manageable 150 150 + self.cleanup_old_entries(feed_url) 151 151 + 152 152 + # Save state 153 153 + self._save_state() 154 154 + 155 155 + logger.info(f"Marked as posted: {guid} -> {post_uri}") 156 156 + 157 157 + def get_last_run(self, feed_url: str) -> Optional[datetime]: 158 158 + """ 159 159 + Get last successful run timestamp for a feed. 160 160 + 161 161 + Args: 162 162 + feed_url: RSS feed URL 163 163 + 164 164 + Returns: 165 165 + Datetime of last run, or None if never run 166 166 + """ 167 167 + self._ensure_feed_exists(feed_url) 168 168 + 169 169 + timestamp_str = self.state['feeds'][feed_url]['last_successful_run'] 170 170 + if timestamp_str is None: 171 171 + return None 172 172 + 173 173 + return datetime.fromisoformat(timestamp_str) 174 174 + 175 175 + def update_last_run(self, feed_url: str, timestamp: datetime): 176 176 + """ 177 177 + Update last successful run timestamp. 178 178 + 179 179 + Args: 180 180 + feed_url: RSS feed URL 181 181 + timestamp: Timestamp of successful run 182 182 + """ 183 183 + self._ensure_feed_exists(feed_url) 184 184 + 185 185 + self.state['feeds'][feed_url]['last_successful_run'] = timestamp.isoformat() 186 186 + self._save_state() 187 187 + 188 188 + logger.info(f"Updated last run for {feed_url}: {timestamp}") 189 189 + 190 190 + def cleanup_old_entries(self, feed_url: str): 191 191 + """ 192 192 + Remove old entries from state. 193 193 + 194 194 + Removes entries that are: 195 195 + - Older than max_age_days 196 196 + - Beyond max_guids_per_feed limit (keeps most recent) 197 197 + 198 198 + Args: 199 199 + feed_url: RSS feed URL 200 200 + """ 201 201 + self._ensure_feed_exists(feed_url) 202 202 + 203 203 + posted_guids = self.state['feeds'][feed_url]['posted_guids'] 204 204 + 205 205 + # Filter out entries older than max_age_days 206 206 + cutoff_date = datetime.now() - timedelta(days=self.max_age_days) 207 207 + filtered = [] 208 208 + for entry in posted_guids: 209 209 + try: 210 210 + posted_at = datetime.fromisoformat(entry['posted_at']) 211 211 + if posted_at > cutoff_date: 212 212 + filtered.append(entry) 213 213 + except (KeyError, ValueError) as e: 214 214 + # Skip entries with malformed or missing timestamps 215 215 + logger.warning(f"Skipping entry with invalid timestamp: {e}") 216 216 + continue 217 217 + 218 218 + # Keep only most recent max_guids_per_feed entries 219 219 + # Sort by posted_at (most recent first) 220 220 + filtered.sort(key=lambda x: x['posted_at'], reverse=True) 221 221 + filtered = filtered[:self.max_guids_per_feed] 222 222 + 223 223 + # Update state 224 224 + old_count = len(posted_guids) 225 225 + new_count = len(filtered) 226 226 + self.state['feeds'][feed_url]['posted_guids'] = filtered 227 227 + 228 228 + if old_count != new_count: 229 229 + logger.info(f"Cleaned up {old_count - new_count} old entries for {feed_url}") 230 230 + 231 231 + def get_posted_count(self, feed_url: str) -> int: 232 232 + """ 233 233 + Get count of posted items for a feed. 234 234 + 235 235 + Args: 236 236 + feed_url: RSS feed URL 237 237 + 238 238 + Returns: 239 239 + Number of posted items 240 240 + """ 241 241 + self._ensure_feed_exists(feed_url) 242 242 + return len(self.state['feeds'][feed_url]['posted_guids']) 243 243 + 244 244 + def get_all_posted_guids(self, feed_url: str) -> List[str]: 245 245 + """ 246 246 + Get all posted GUIDs for a feed. 247 247 + 248 248 + Args: 249 249 + feed_url: RSS feed URL 250 250 + 251 251 + Returns: 252 252 + List of GUIDs 253 253 + """ 254 254 + self._ensure_feed_exists(feed_url) 255 255 + return [entry['guid'] for entry in self.state['feeds'][feed_url]['posted_guids']]