Upload .\src\data_collection\rss_collector.py with huggingface_hub

.//src//data_collection//rss_collector.py

@@ -0,0 +1,112 @@
+"""RSS feed collector for news sources."""
+
+import json
+import logging
+import time
+import xml.etree.ElementTree as ET
+from pathlib import Path
+from typing import Any
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+class RSSCollector:
+    """Collects articles from RSS feeds."""
+
+    def __init__(self, output_dir: str, config: dict[str, Any]):
+        self.output_dir = Path(output_dir)
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        self.config = config
+        self.session = requests.Session()
+        self.session.headers.update({
+            "User-Agent": "BwengeAi/0.1 (Kinyarwanda AI Research)"
+        })
+
+        rss_config = config.get("newtimes", {})
+        self.feeds = rss_config.get("rss_feeds", [])
+        self.rate_limit = rss_config.get("rate_limit", 1.0)
+
+    def fetch_feed(self, url: str) -> list[dict[str, Any]]:
+        """Fetch and parse an RSS feed."""
+        items = []
+        try:
+            time.sleep(self.rate_limit)
+            response = self.session.get(url, timeout=30)
+            response.raise_for_status()
+
+            root = ET.fromstring(response.content)
+
+            ns = {"atom": "http://www.w3.org/2005/Atom"}
+            entries = root.findall(".//item") or root.findall(".//atom:entry", ns)
+
+            for entry in entries:
+                item = {}
+
+                title = entry.find("title") or entry.find("atom:title", ns)
+                if title is not None and title.text:
+                    item["title"] = title.text.strip()
+
+                link = entry.find("link") or entry.find("atom:link", ns)
+                if link is not None:
+                    item["url"] = link.get("href", link.text or "").strip()
+
+                description = entry.find("description") or entry.find("atom:summary", ns)
+                if description is not None and description.text:
+                    item["text"] = description.text.strip()
+
+                content = entry.find("content:encoded", {"content": "http://purl.org/rss/1.0/modules/content/"})
+                if content is not None and content.text:
+                    item["text"] = content.text.strip()
+
+                pub_date = entry.find("pubDate") or entry.find("atom:published", ns)
+                if pub_date is not None and pub_date.text:
+                    item["date"] = pub_date.text.strip()
+
+                if item.get("text") and len(item["text"]) > 50:
+                    item["source"] = "newtimes_rss"
+                    items.append(item)
+
+        except Exception as e:
+            logger.error(f"Failed to fetch RSS feed {url}: {e}")
+
+        return items
+
+    def collect_all(self) -> list[dict[str, Any]]:
+        """Collect articles from all configured RSS feeds."""
+        logger.info("Starting RSS feed collection...")
+
+        all_items = []
+
+        for feed_config in self.feeds:
+            feed_name = feed_config.get("name", "unknown")
+            feed_url = feed_config.get("url", "")
+
+            if not feed_url:
+                continue
+
+            logger.info(f"Collecting from RSS feed: {feed_name}")
+            items = self.fetch_feed(feed_url)
+            all_items.extend(items)
+            logger.info(f"  Feed '{feed_name}': {len(items)} items collected")
+
+        output_path = self.output_dir / "newtimes_rss_articles.jsonl"
+        with open(output_path, "w", encoding="utf-8") as f:
+            for item in all_items:
+                f.write(json.dumps(item, ensure_ascii=False) + "\n")
+
+        summary = {
+            "source": "newtimes_rss",
+            "feeds": [f.get("name", "") for f in self.feeds],
+            "total_articles": len(all_items),
+            "output_path": str(output_path),
+            "total_characters": sum(len(item.get("text", "")) for item in all_items),
+        }
+
+        summary_path = self.output_dir / "newtimes_rss_summary.json"
+        with open(summary_path, "w", encoding="utf-8") as f:
+            json.dump(summary, f, indent=2, ensure_ascii=False)
+
+        logger.info(f"Collected {len(all_items)} articles from RSS feeds")
+        return all_items