diff --git a/aggregators/kagi-news/.gitignore b/aggregators/kagi-news/.gitignore new file mode 100644 index 0000000..783188a --- /dev/null +++ b/aggregators/kagi-news/.gitignore @@ -0,0 +1,41 @@ +# Environment and config +.env +config.yaml +venv/ + +# State files +data/*.json +data/world.xml + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo diff --git a/aggregators/kagi-news/src/__init__.py b/aggregators/kagi-news/src/__init__.py new file mode 100644 index 0000000..850a393 --- /dev/null +++ b/aggregators/kagi-news/src/__init__.py @@ -0,0 +1,3 @@ +"""Kagi News RSS Aggregator for Coves.""" + +__version__ = "0.1.0" diff --git a/aggregators/kagi-news/src/config.py b/aggregators/kagi-news/src/config.py new file mode 100644 index 0000000..0ca2b3f --- /dev/null +++ b/aggregators/kagi-news/src/config.py @@ -0,0 +1,165 @@ +""" +Configuration Loader for Kagi News Aggregator. + +Loads and validates configuration from YAML files. +""" +import os +import logging +from pathlib import Path +from typing import Dict, Any +import yaml +from urllib.parse import urlparse + +from src.models import AggregatorConfig, FeedConfig + +logger = logging.getLogger(__name__) + + +class ConfigError(Exception): + """Configuration error.""" + pass + + +class ConfigLoader: + """ + Loads and validates aggregator configuration. + + Supports: + - Loading from YAML file + - Environment variable overrides + - Validation of required fields + - URL validation + """ + + def __init__(self, config_path: Path): + """ + Initialize config loader. + + Args: + config_path: Path to config.yaml file + """ + self.config_path = Path(config_path) + + def load(self) -> AggregatorConfig: + """ + Load and validate configuration. + + Returns: + AggregatorConfig object + + Raises: + ConfigError: If config is invalid or missing + """ + # Check file exists + if not self.config_path.exists(): + raise ConfigError(f"Configuration file not found: {self.config_path}") + + # Load YAML + try: + with open(self.config_path, 'r') as f: + config_data = yaml.safe_load(f) + except yaml.YAMLError as e: + raise ConfigError(f"Failed to parse YAML: {e}") + + if not config_data: + raise ConfigError("Configuration file is empty") + + # Validate and parse + try: + return self._parse_config(config_data) + except Exception as e: + raise ConfigError(f"Invalid configuration: {e}") + + def _parse_config(self, data: Dict[str, Any]) -> AggregatorConfig: + """ + Parse and validate configuration data. + + Args: + data: Parsed YAML data + + Returns: + AggregatorConfig object + + Raises: + ConfigError: If validation fails + """ + # Get coves_api_url (with env override) + coves_api_url = os.getenv('COVES_API_URL', data.get('coves_api_url')) + if not coves_api_url: + raise ConfigError("Missing required field: coves_api_url") + + # Validate URL + if not self._is_valid_url(coves_api_url): + raise ConfigError(f"Invalid URL for coves_api_url: {coves_api_url}") + + # Get log level (default to info) + log_level = data.get('log_level', 'info') + + # Parse feeds + feeds_data = data.get('feeds', []) + if not feeds_data: + raise ConfigError("Configuration must include at least one feed") + + feeds = [] + for feed_data in feeds_data: + feed = self._parse_feed(feed_data) + feeds.append(feed) + + logger.info(f"Loaded configuration with {len(feeds)} feeds ({sum(1 for f in feeds if f.enabled)} enabled)") + + return AggregatorConfig( + coves_api_url=coves_api_url, + feeds=feeds, + log_level=log_level + ) + + def _parse_feed(self, data: Dict[str, Any]) -> FeedConfig: + """ + Parse and validate a single feed configuration. + + Args: + data: Feed configuration data + + Returns: + FeedConfig object + + Raises: + ConfigError: If validation fails + """ + # Required fields + required_fields = ['name', 'url', 'community_handle'] + for field in required_fields: + if field not in data: + raise ConfigError(f"Missing required field in feed config: {field}") + + name = data['name'] + url = data['url'] + community_handle = data['community_handle'] + enabled = data.get('enabled', True) # Default to True + + # Validate URL + if not self._is_valid_url(url): + raise ConfigError(f"Invalid URL for feed '{name}': {url}") + + return FeedConfig( + name=name, + url=url, + community_handle=community_handle, + enabled=enabled + ) + + def _is_valid_url(self, url: str) -> bool: + """ + Validate URL format. + + Args: + url: URL to validate + + Returns: + True if valid, False otherwise + """ + try: + result = urlparse(url) + return all([result.scheme, result.netloc]) + except Exception: + return False diff --git a/aggregators/kagi-news/src/coves_client.py b/aggregators/kagi-news/src/coves_client.py new file mode 100644 index 0000000..4f2d786 --- /dev/null +++ b/aggregators/kagi-news/src/coves_client.py @@ -0,0 +1,175 @@ +""" +Coves API Client for posting to communities. + +Handles authentication and posting via XRPC. +""" +import logging +import requests +from typing import Dict, List, Optional +from atproto import Client + +logger = logging.getLogger(__name__) + + +class CovesClient: + """ + Client for posting to Coves communities via XRPC. + + Handles: + - Authentication with aggregator credentials + - Creating posts in communities (social.coves.post.create) + - External embed formatting + """ + + def __init__(self, api_url: str, handle: str, password: str, pds_url: Optional[str] = None): + """ + Initialize Coves client. + + Args: + api_url: Coves AppView URL for posting (e.g., "http://localhost:8081") + handle: Aggregator handle (e.g., "kagi-news.coves.social") + password: Aggregator password/app password + pds_url: Optional PDS URL for authentication (defaults to api_url) + """ + self.api_url = api_url + self.pds_url = pds_url or api_url # Auth through PDS, post through AppView + self.handle = handle + self.password = password + self.client = Client(base_url=self.pds_url) # Use PDS for auth + self._authenticated = False + + def authenticate(self): + """ + Authenticate with Coves API. + + Uses com.atproto.server.createSession directly to avoid + Bluesky-specific endpoints that don't exist on Coves PDS. + + Raises: + Exception: If authentication fails + """ + try: + logger.info(f"Authenticating as {self.handle}") + + # Use createSession directly (avoid app.bsky.actor.getProfile) + session = self.client.com.atproto.server.create_session( + {"identifier": self.handle, "password": self.password} + ) + + # Manually set session (skip profile fetch) + self.client._session = session + self._authenticated = True + self.did = session.did + + logger.info(f"Authentication successful (DID: {self.did})") + except Exception as e: + logger.error(f"Authentication failed: {e}") + raise + + def create_post( + self, + community_handle: str, + content: str, + facets: List[Dict], + embed: Optional[Dict] = None + ) -> str: + """ + Create a post in a community. + + Args: + community_handle: Community handle (e.g., "world-news.coves.social") + content: Post content (rich text) + facets: Rich text facets (formatting, links) + embed: Optional external embed + + Returns: + AT Proto URI of created post (e.g., "at://did:plc:.../social.coves.post/...") + + Raises: + Exception: If post creation fails + """ + if not self._authenticated: + self.authenticate() + + try: + # Prepare post data for social.coves.post.create endpoint + post_data = { + "community": community_handle, + "content": content, + "facets": facets + } + + # Add embed if provided + if embed: + post_data["embed"] = embed + + # Use Coves-specific endpoint (not direct PDS write) + # This provides validation, authorization, and business logic + logger.info(f"Creating post in community: {community_handle}") + + # Make direct HTTP request to XRPC endpoint + url = f"{self.api_url}/xrpc/social.coves.post.create" + headers = { + "Authorization": f"Bearer {self.client._session.access_jwt}", + "Content-Type": "application/json" + } + + response = requests.post(url, json=post_data, headers=headers, timeout=30) + + # Log detailed error if request fails + if not response.ok: + error_body = response.text + logger.error(f"Post creation failed ({response.status_code}): {error_body}") + response.raise_for_status() + + result = response.json() + post_uri = result["uri"] + logger.info(f"Post created: {post_uri}") + return post_uri + + except Exception as e: + logger.error(f"Failed to create post: {e}") + raise + + def create_external_embed( + self, + uri: str, + title: str, + description: str, + thumb: Optional[str] = None + ) -> Dict: + """ + Create external embed object for hot-linked content. + + Args: + uri: External URL (story link) + title: Story title + description: Story description/summary + thumb: Optional thumbnail image URL + + Returns: + External embed dictionary + """ + embed = { + "$type": "social.coves.embed.external", + "external": { + "uri": uri, + "title": title, + "description": description + } + } + + if thumb: + embed["external"]["thumb"] = thumb + + return embed + + def _get_timestamp(self) -> str: + """ + Get current timestamp in ISO 8601 format. + + Returns: + ISO timestamp string + """ + from datetime import datetime, timezone + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") diff --git a/aggregators/kagi-news/src/html_parser.py b/aggregators/kagi-news/src/html_parser.py new file mode 100644 index 0000000..9d04b22 --- /dev/null +++ b/aggregators/kagi-news/src/html_parser.py @@ -0,0 +1,300 @@ +""" +Kagi News HTML description parser. + +Parses the HTML content from RSS feed item descriptions +into structured data. +""" +import re +import logging +from typing import Dict, List, Optional +from datetime import datetime +from bs4 import BeautifulSoup +from urllib.parse import urlparse + +from src.models import KagiStory, Perspective, Quote, Source + +logger = logging.getLogger(__name__) + + +class KagiHTMLParser: + """Parses Kagi News HTML descriptions into structured data.""" + + def parse(self, html_description: str) -> Dict: + """ + Parse HTML description into structured data. + + Args: + html_description: HTML content from RSS item description + + Returns: + Dictionary with extracted data: + - summary: str + - image_url: Optional[str] + - image_alt: Optional[str] + - highlights: List[str] + - quote: Optional[Dict[str, str]] + - perspectives: List[Dict] + - sources: List[Dict] + """ + soup = BeautifulSoup(html_description, 'html.parser') + + return { + 'summary': self._extract_summary(soup), + 'image_url': self._extract_image_url(soup), + 'image_alt': self._extract_image_alt(soup), + 'highlights': self._extract_highlights(soup), + 'quote': self._extract_quote(soup), + 'perspectives': self._extract_perspectives(soup), + 'sources': self._extract_sources(soup), + } + + def parse_to_story( + self, + title: str, + link: str, + guid: str, + pub_date: datetime, + categories: List[str], + html_description: str + ) -> KagiStory: + """ + Parse HTML and create a KagiStory object. + + Args: + title: Story title + link: Story URL + guid: Unique identifier + pub_date: Publication date + categories: List of categories + html_description: HTML content from description + + Returns: + KagiStory object + """ + parsed = self.parse(html_description) + + # Convert parsed data to model objects + perspectives = [ + Perspective( + actor=p['actor'], + description=p['description'], + source_url=p['source_url'] + ) + for p in parsed['perspectives'] + ] + + sources = [ + Source( + title=s['title'], + url=s['url'], + domain=s['domain'] + ) + for s in parsed['sources'] + ] + + quote = None + if parsed['quote']: + quote = Quote( + text=parsed['quote']['text'], + attribution=parsed['quote']['attribution'] + ) + + return KagiStory( + title=title, + link=link, + guid=guid, + pub_date=pub_date, + categories=categories, + summary=parsed['summary'], + highlights=parsed['highlights'], + perspectives=perspectives, + quote=quote, + sources=sources, + image_url=parsed['image_url'], + image_alt=parsed['image_alt'] + ) + + def _extract_summary(self, soup: BeautifulSoup) -> str: + """Extract summary from first

tag.""" + p_tag = soup.find('p') + if p_tag: + return p_tag.get_text(strip=True) + return "" + + def _extract_image_url(self, soup: BeautifulSoup) -> Optional[str]: + """Extract image URL from tag.""" + img_tag = soup.find('img') + if img_tag and img_tag.get('src'): + return img_tag['src'] + return None + + def _extract_image_alt(self, soup: BeautifulSoup) -> Optional[str]: + """Extract image alt text from tag.""" + img_tag = soup.find('img') + if img_tag and img_tag.get('alt'): + return img_tag['alt'] + return None + + def _extract_highlights(self, soup: BeautifulSoup) -> List[str]: + """Extract highlights list from H3 section.""" + highlights = [] + + # Find "Highlights:" h3 tag + h3_tags = soup.find_all('h3') + for h3 in h3_tags: + if 'Highlights' in h3.get_text(): + # Get the