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

.//src//data_collection//rbc_collector.py

@@ -0,0 +1,124 @@
+"""Rwanda Biomedical Centre content collector."""
+
+import json
+import logging
+import time
+from pathlib import Path
+from typing import Any
+
+import requests
+from bs4 import BeautifulSoup
+
+from data_collection.web_utils import check_robots_txt
+
+logger = logging.getLogger(__name__)
+
+
+class RBCCollector:
+    """Collects health-related content from RBC."""
+
+    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)"
+        })
+
+        rbc_config = config.get("rbc", {})
+        self.base_url = rbc_config.get("base_url", "https://rbc.gov.rw")
+        self.rate_limit = rbc_config.get("rate_limit", 2.0)
+
+    def fetch_page(self, url: str) -> BeautifulSoup | None:
+        """Fetch a page and return parsed HTML."""
+        try:
+            if not check_robots_txt(url, user_agent="BwengeAi/0.1", delay=self.rate_limit):
+                return None
+            time.sleep(self.rate_limit)
+            response = self.session.get(url, timeout=30)
+            response.raise_for_status()
+            return BeautifulSoup(response.text, "lxml")
+        except Exception as e:
+            logger.error(f"Failed to fetch {url}: {e}")
+            return None
+
+    def get_article_links(self, max_pages: int = 5) -> list[str]:
+        """Get article links from the site."""
+        links = []
+
+        news_url = f"{self.base_url}/news"
+        soup = self.fetch_page(news_url)
+        if soup:
+            for a_tag in soup.find_all("a", href=True):
+                href = a_tag["href"]
+                if "/news/" in href or "/article/" in href or "/publication/" in href:
+                    if not href.startswith("http"):
+                        href = f"{self.base_url}{href}"
+                    if href not in links:
+                        links.append(href)
+
+        return links[:max_pages * 20]
+
+    def scrape_article(self, url: str) -> dict[str, Any] | None:
+        """Scrape a single article."""
+        soup = self.fetch_page(url)
+        if not soup:
+            return None
+
+        article = {"url": url, "source": "rbc"}
+
+        title_tag = soup.find("h1") or soup.find("title")
+        if title_tag:
+            article["title"] = title_tag.get_text(strip=True)
+
+        for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
+            tag.decompose()
+
+        content_div = soup.find("article") or soup.find("div", class_=lambda x: x and "content" in str(x).lower())
+        if content_div:
+            paragraphs = content_div.find_all("p")
+        else:
+            paragraphs = soup.find_all("p")
+
+        texts = [p.get_text(strip=True) for p in paragraphs if p.get_text(strip=True) and len(p.get_text(strip=True)) > 20]
+        article["text"] = "\n\n".join(texts)
+
+        if not article["text"] or len(article["text"]) < 100:
+            return None
+
+        article["length"] = len(article["text"])
+        return article
+
+    def collect_all(self) -> list[dict[str, Any]]:
+        """Collect all articles from RBC."""
+        logger.info("Starting RBC collection...")
+
+        links = self.get_article_links()
+        logger.info(f"Found {len(links)} article links")
+
+        results = []
+        for url in links:
+            article = self.scrape_article(url)
+            if article:
+                results.append(article)
+
+        output_path = self.output_dir / "rbc_articles.jsonl"
+        with open(output_path, "w", encoding="utf-8") as f:
+            for article in results:
+                f.write(json.dumps(article, ensure_ascii=False) + "\n")
+
+        summary = {
+            "source": "rbc",
+            "total_links": len(links),
+            "collected": len(results),
+            "output_path": str(output_path),
+            "total_characters": sum(a.get("length", 0) for a in results),
+        }
+
+        summary_path = self.output_dir / "rbc_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(results)} articles from RBC")
+        return results