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

.//src//data_collection//wikimedia_collector.py

@@ -0,0 +1,257 @@
+"""Wikimedia projects collector for Wikisource and Wiktionary."""
+
+import json
+import logging
+import time
+from pathlib import Path
+from typing import Any
+
+import requests
+from bs4 import BeautifulSoup
+
+logger = logging.getLogger(__name__)
+
+
+class WikisourceCollector:
+    """Collects texts from Kinyarwanda Wikisource."""
+
+    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)"
+        })
+
+        ws_config = config.get("wikisource", {})
+        self.api_url = ws_config.get("api_url", "https://rw.wikisource.org/w/api.php")
+        self.base_url = ws_config.get("base_url", "https://rw.wikisource.org")
+
+    def fetch_page_list(self, namespace: int = 0, limit: int = 500) -> list[str]:
+        """Fetch list of page titles."""
+        pages = []
+        params = {
+            "action": "query",
+            "list": "allpages",
+            "apnamespace": namespace,
+            "aplimit": str(min(limit, 500)),
+            "format": "json",
+        }
+
+        while True:
+            try:
+                time.sleep(0.5)
+                response = self.session.get(self.api_url, params=params, timeout=30)
+                response.raise_for_status()
+                data = response.json()
+
+                allpages = data.get("query", {}).get("allpages", [])
+                pages.extend(p["title"] for p in allpages)
+
+                if "continue" in data and len(pages) < limit:
+                    params["apcontinue"] = data["continue"]["apcontinue"]
+                else:
+                    break
+
+            except Exception as e:
+                logger.error(f"Failed to fetch page list: {e}")
+                break
+
+        return pages[:limit]
+
+    def fetch_page_content(self, title: str) -> dict[str, Any] | None:
+        """Fetch content for a single page."""
+        params = {
+            "action": "parse",
+            "page": title,
+            "prop": "text",
+            "format": "json",
+        }
+
+        try:
+            time.sleep(0.5)
+            response = self.session.get(self.api_url, params=params, timeout=30)
+            response.raise_for_status()
+            data = response.json()
+
+            parse = data.get("parse", {})
+            html = parse.get("text", {}).get("*", "")
+
+            text = self._clean_html(html)
+
+            if text and len(text) > 100:
+                return {
+                    "title": title,
+                    "text": text,
+                    "source": "wikisource",
+                    "length": len(text),
+                }
+
+        except Exception as e:
+            logger.warning(f"Failed to fetch page '{title}': {e}")
+
+        return None
+
+    def _clean_html(self, html: str) -> str:
+        """Clean HTML to plain text."""
+        soup = BeautifulSoup(html, "lxml")
+
+        for tag in soup(["script", "style", "sup", "sub", "table"]):
+            tag.decompose()
+
+        text = soup.get_text(separator=" ", strip=True)
+
+        lines = text.split()
+        clean_lines = [line.strip() for line in lines if line.strip() and len(line) > 2]
+        return " ".join(clean_lines)
+
+    def collect_all(self) -> list[dict[str, Any]]:
+        """Collect all texts from Kinyarwanda Wikisource."""
+        logger.info("Starting Kinyarwanda Wikisource collection...")
+
+        pages = self.fetch_page_list(namespace=0, limit=500)
+        logger.info(f"Found {len(pages)} pages")
+
+        results = []
+        for title in pages:
+            content = self.fetch_page_content(title)
+            if content:
+                results.append(content)
+
+        output_path = self.output_dir / "wikisource_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": "wikisource",
+            "total_pages": len(pages),
+            "collected": len(results),
+            "output_path": str(output_path),
+            "total_characters": sum(r["length"] for r in results),
+        }
+
+        summary_path = self.output_dir / "wikisource_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)} texts from Wikisource")
+        return results
+
+
+class WiktionaryCollector:
+    """Collects dictionary entries from Kinyarwanda Wiktionary."""
+
+    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)"
+        })
+
+        wt_config = config.get("wiktionary", {})
+        self.api_url = wt_config.get("api_url", "https://rw.wiktionary.org/w/api.php")
+
+    def fetch_all_words(self, limit: int = 2000) -> list[str]:
+        """Fetch all word entries."""
+        words = []
+        params = {
+            "action": "query",
+            "list": "allpages",
+            "apnamespace": 0,
+            "aplimit": "500",
+            "format": "json",
+        }
+
+        while len(words) < limit:
+            try:
+                time.sleep(0.5)
+                response = self.session.get(self.api_url, params=params, timeout=30)
+                response.raise_for_status()
+                data = response.json()
+
+                allpages = data.get("query", {}).get("allpages", [])
+                words.extend(p["title"] for p in allpages)
+
+                if "continue" in data:
+                    params["apcontinue"] = data["continue"]["apcontinue"]
+                else:
+                    break
+
+            except Exception as e:
+                logger.error(f"Failed to fetch words: {e}")
+                break
+
+        return words[:limit]
+
+    def fetch_word_entry(self, word: str) -> dict[str, Any] | None:
+        """Fetch a single word entry."""
+        params = {
+            "action": "parse",
+            "page": word,
+            "prop": "text",
+            "format": "json",
+        }
+
+        try:
+            time.sleep(0.3)
+            response = self.session.get(self.api_url, params=params, timeout=30)
+            response.raise_for_status()
+            data = response.json()
+
+            html = data.get("parse", {}).get("text", {}).get("*", "")
+
+            soup = BeautifulSoup(html, "lxml")
+            for tag in soup(["script", "style"]):
+                tag.decompose()
+
+            text = soup.get_text(separator=" ", strip=True)
+
+            if text and len(text) > 10:
+                return {
+                    "word": word,
+                    "text": text,
+                    "source": "wiktionary",
+                    "length": len(text),
+                }
+
+        except Exception as e:
+            logger.warning(f"Failed to fetch word '{word}': {e}")
+
+        return None
+
+    def collect_all(self) -> list[dict[str, Any]]:
+        """Collect all dictionary entries."""
+        logger.info("Starting Kinyarwanda Wiktionary collection...")
+
+        words = self.fetch_all_words(limit=2000)
+        logger.info(f"Found {len(words)} words")
+
+        results = []
+        for word in words:
+            entry = self.fetch_word_entry(word)
+            if entry:
+                results.append(entry)
+
+        output_path = self.output_dir / "wiktionary_entries.jsonl"
+        with open(output_path, "w", encoding="utf-8") as f:
+            for entry in results:
+                f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+
+        summary = {
+            "source": "wiktionary",
+            "total_words": len(words),
+            "collected": len(results),
+            "output_path": str(output_path),
+            "total_characters": sum(r["length"] for r in results),
+        }
+
+        summary_path = self.output_dir / "wiktionary_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)} entries from Wiktionary")
+        return results