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

.//src//data_collection//igihe_scraper.py

@@ -0,0 +1,173 @@
+"""Igihe news scraper for Kinyarwanda content."""
+
+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 IgiheScraper:
+    """Scrapes Kinyarwanda news content from Igihe.com."""
+
+    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; contact: info@bwengeai.rw)"
+        })
+        
+        igihe_config = config.get("igihe", {})
+        self.base_url = igihe_config.get("base_url", "https://igihe.com")
+        self.categories = igihe_config.get("categories", ["amakuru", "politiki", "ubuzima"])
+        self.rate_limit = igihe_config.get("rate_limit", 2.0)
+        self.contact_email = igihe_config.get("contact_email", "info@igihe.com")
+
+    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, category: str, max_pages: int = 5) -> list[str]:
+        """Get article links from a category page."""
+        links = []
+        
+        for page in range(1, max_pages + 1):
+            if page == 1:
+                url = f"{self.base_url}/{category}"
+            else:
+                url = f"{self.base_url}/{category}?debut_gh_news={((page - 1) * 20)}"
+            
+            soup = self.fetch_page(url)
+            if not soup:
+                break
+            
+            article_links = soup.find_all("a", href=True)
+            for link in article_links:
+                href = link["href"]
+                if "/article/" in href or "/news/" in href:
+                    if not href.startswith("http"):
+                        href = f"{self.base_url}{href}"
+                    if href not in links:
+                        links.append(href)
+            
+            logger.info(f"  Category '{category}' page {page}: found {len(article_links)} links")
+        
+        return links
+
+    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": "igihe"}
+        
+        title_tag = soup.find("h1") or soup.find("title")
+        if title_tag:
+            article["title"] = title_tag.get_text(strip=True)
+        
+        article["text"] = self._extract_article_text(soup)
+        
+        if not article["text"] or len(article["text"]) < 100:
+            return None
+        
+        author_tag = soup.find("span", class_=lambda x: x and "author" in x.lower()) or \
+                     soup.find("a", class_=lambda x: x and "author" in x.lower())
+        if author_tag:
+            article["author"] = author_tag.get_text(strip=True)
+        
+        date_tag = soup.find("time") or soup.find("span", class_=lambda x: x and "date" in x.lower())
+        if date_tag:
+            article["date"] = date_tag.get("datetime", date_tag.get_text(strip=True))
+        
+        article["length"] = len(article["text"])
+        
+        return article
+
+    def _extract_article_text(self, soup: BeautifulSoup) -> str:
+        """Extract article text content."""
+        for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
+            tag.decompose()
+        
+        article = soup.find("article") or soup.find("div", class_=lambda x: x and "content" in x.lower())
+        
+        if article:
+            paragraphs = article.find_all("p")
+        else:
+            paragraphs = soup.find_all("p")
+        
+        texts = []
+        for p in paragraphs:
+            text = p.get_text(strip=True)
+            if text and len(text) > 20:
+                texts.append(text)
+        
+        return "\n\n".join(texts)
+
+    def collect_category(self, category: str, max_articles: int = 100) -> list[dict[str, Any]]:
+        """Collect articles from a category."""
+        logger.info(f"Collecting articles from category: {category}")
+        
+        links = self.get_article_links(category, max_pages=10)
+        logger.info(f"  Found {len(links)} article links")
+        
+        articles = []
+        for url in links[:max_articles]:
+            article = self.scrape_article(url)
+            if article:
+                articles.append(article)
+                logger.debug(f"  Scraped: {article.get('title', 'untitled')[:50]}")
+        
+        return articles
+
+    def collect_all(self) -> list[dict[str, Any]]:
+        """Collect articles from all configured categories."""
+        logger.info("Starting Igihe collection...")
+        logger.info(f"NOTE: Igihe content is copyrighted. Contact {self.contact_email} for licensing.")
+        
+        all_articles = []
+        
+        for category in self.categories:
+            articles = self.collect_category(category, max_articles=50)
+            all_articles.extend(articles)
+            logger.info(f"  Category '{category}': {len(articles)} articles collected")
+        
+        output_path = self.output_dir / "igihe_articles.jsonl"
+        with open(output_path, "w", encoding="utf-8") as f:
+            for article in all_articles:
+                f.write(json.dumps(article, ensure_ascii=False) + "\n")
+        
+        summary = {
+            "source": "igihe",
+            "categories": self.categories,
+            "total_articles": len(all_articles),
+            "output_path": str(output_path),
+            "total_characters": sum(a.get("length", 0) for a in all_articles),
+            "contact_email": self.contact_email,
+            "copyright_notice": "All Rights Reserved - IGIHE Ltd",
+        }
+        
+        summary_path = self.output_dir / "igihe_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_articles)} articles from Igihe")
+        return all_articles