Upload .\scripts\collect_wikipedia.py with huggingface_hub

.//scripts//collect_wikipedia.py

@@ -0,0 +1,151 @@
+"""Standalone Wikipedia collector - no torch/transformers dependency."""
+
+import bz2
+import json
+import logging
+import re
+import time
+import xml.etree.ElementTree as ET
+from pathlib import Path
+from typing import Any
+
+import requests
+from tqdm import tqdm
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+WIKI_API = "https://rw.wikipedia.org/w/api.php"
+SESSION = requests.Session()
+SESSION.headers.update({"User-Agent": "BwengeAi/0.1 (Kinyarwanda AI Research; contact: bwengeai@research.rw)"})
+
+REQUEST_DELAY = 0.5  # seconds between requests
+
+OUTPUT_DIR = Path("C:/Users/admin/BwengeAi/data/raw/wikipedia")
+OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def api_request(params: dict, max_retries: int = 5) -> dict:
+    """Make an API request with retry and rate limiting."""
+    for attempt in range(max_retries):
+        time.sleep(REQUEST_DELAY)
+        try:
+            r = SESSION.get(WIKI_API, params=params, timeout=30)
+            if r.status_code == 429:
+                wait = min(30, 2 ** (attempt + 1))
+                logger.warning(f"Rate limited, waiting {wait}s...")
+                time.sleep(wait)
+                continue
+            r.raise_for_status()
+            return r.json()
+        except requests.exceptions.HTTPError as e:
+            if "429" in str(e):
+                wait = min(30, 2 ** (attempt + 1))
+                logger.warning(f"Rate limited (attempt {attempt + 1}), waiting {wait}s...")
+                time.sleep(wait)
+            else:
+                raise
+    raise Exception(f"Failed after {max_retries} retries")
+
+
+def get_site_stats() -> dict:
+    params = {"action": "query", "meta": "siteinfo", "siprop": "statistics", "format": "json"}
+    data = api_request(params)
+    return data.get("query", {}).get("statistics", {})
+
+
+def fetch_article_list() -> list[str]:
+    articles = []
+    params = {"action": "query", "list": "allpages", "apnamespace": 0, "aplimit": "max", "format": "json"}
+    while True:
+        data = api_request(params)
+        pages = data.get("query", {}).get("allpages", [])
+        articles.extend(p["title"] for p in pages)
+        logger.info(f"  Fetched {len(articles)} article titles so far...")
+        if "continue" in data:
+            params["apcontinue"] = data["continue"]["apcontinue"]
+        else:
+            break
+    logger.info(f"Found {len(articles)} articles")
+    return articles
+
+
+def clean_html(html: str) -> str:
+    from bs4 import BeautifulSoup
+    soup = BeautifulSoup(html, "lxml")
+    for tag in soup(["script", "style", "sup", "sub"]):
+        tag.decompose()
+    text = soup.get_text(separator=" ", strip=True)
+    lines = [l.strip() for l in text.split() if l.strip() and len(l.strip()) > 2]
+    return " ".join(lines)
+
+
+def clean_mediawiki(text: str) -> str:
+    text = re.sub(r"\[\[([^|\]]*\|)?([^\]]*)\]\]", r"\2", text)
+    text = re.sub(r"\{\{[^}]*\}\}", "", text)
+    text = re.sub(r"'''?", "", text)
+    text = re.sub(r"<[^>]+>", "", text)
+    text = re.sub(r"==+\s*[^=]*\s*==+", "", text)
+    text = re.sub(r"\s+", " ", text)
+    return text.strip()
+
+
+def fetch_article_content(title: str) -> dict | None:
+    params = {"action": "parse", "page": title, "prop": "wikitext|text|categories|links", "format": "json"}
+    try:
+        data = api_request(params)
+        parse = data.get("parse", {})
+        text = parse.get("text", {}).get("*", "")
+        clean = clean_html(text)
+        return {
+            "title": title,
+            "text": clean,
+            "wikitext": parse.get("wikitext", {}).get("*", ""),
+            "categories": [c["*"] for c in parse.get("categories", [])],
+            "links": [l["*"] for l in parse.get("links", [])],
+            "length": len(clean),
+        }
+    except Exception as e:
+        logger.warning(f"Failed to fetch '{title}': {e}")
+        return None
+
+
+def main():
+    logger.info("=" * 60)
+    logger.info("BwengeAi - Kinyarwanda Wikipedia Collection")
+    logger.info("=" * 60)
+
+    stats = get_site_stats()
+    logger.info(f"Wikipedia stats: {stats}")
+
+    articles = fetch_article_list()
+    logger.info(f"Collecting {len(articles)} articles...")
+
+    results = []
+    output_path = OUTPUT_DIR / "wikipedia_articles.jsonl"
+
+    with open(output_path, "w", encoding="utf-8") as f:
+        for title in tqdm(articles, desc="Fetching articles"):
+            article = fetch_article_content(title)
+            if article and article["text"] and len(article["text"]) > 50:
+                f.write(json.dumps(article, ensure_ascii=False) + "\n")
+                results.append({"title": article["title"], "length": article["length"]})
+
+    summary = {
+        "source": "kinyarwanda_wikipedia",
+        "total_articles": len(articles),
+        "collected": len(results),
+        "output_path": str(output_path),
+        "total_characters": sum(r["length"] for r in results),
+    }
+
+    summary_path = OUTPUT_DIR / "wikipedia_summary.json"
+    with open(summary_path, "w", encoding="utf-8") as f:
+        json.dump(summary, f, indent=2, ensure_ascii=False)
+
+    logger.info(f"Done! Collected {len(results)} articles ({summary['total_characters']} characters)")
+    logger.info(f"Saved to: {output_path}")
+
+
+if __name__ == "__main__":
+    main()