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

.//src//data_collection//data_processor.py

@@ -0,0 +1,323 @@
+"""Data processor for combining and cleaning collected Kinyarwanda data."""
+
+import json
+import logging
+import re
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+KINYARWANDACOMMON_WORDS = {
+    "ni", "ye", "yo", "ra", "ku", "mu", "gu", "ka", "ke", "ki", "ko",
+    "nka", "nk", "icyo", "kandi", "noneho", "ubwo", "icyumweru",
+    "uwo", "iyo", "ari", "cyangwa", "kubera", "bitewe", "mu buryo",
+    "ubukorikori", "ubuzima", "uburezi", "ubukungu", "politiki",
+    "amakuru", "imikino", "imyidagaduro", "ikoranabuhanga",
+    "abana", "abantu", "igihugu", "u Rwanda", "Rwandese",
+    "ndetse", "kubw", "tw", "mw", "bw", "twegerereje",
+    "ibyo", "ubwo", "ari", "yari", "twari", "mwari",
+    "ko", "ngo", "ko", "ncy", "nko",
+}
+
+
+def is_likely_kinyarwanda(text: str) -> bool:
+    """Simple heuristic check if text is likely Kinyarwanda.
+
+    Uses character patterns and common word detection without external packages.
+    Not perfect, but catches obvious non-Kinyarwanda text.
+    """
+    if not text or len(text) < 10:
+        return True
+
+    text_lower = text.lower()
+    words = set(text_lower.split())
+
+    kr_matches = sum(1 for w in KINYARWANDACOMMON_WORDS if w in words)
+    if kr_matches >= 2:
+        return True
+
+    latin_chars = sum(1 for c in text if c.isascii() and c.isalpha())
+    total_chars = sum(1 for c in text if c.isalpha())
+    if total_chars == 0:
+        return False
+
+    latin_ratio = latin_chars / total_chars
+    if latin_ratio < 0.85:
+        return False
+
+    arabic_chars = sum(1 for c in text if '\u0600' <= c <= '\u06FF')
+    if arabic_chars > total_chars * 0.1:
+        return False
+
+    cjk_chars = sum(1 for c in text if '\u4E00' <= c <= '\u9FFF')
+    if cjk_chars > 0:
+        return False
+
+    cyrillic_chars = sum(1 for c in text if '\u0400' <= c <= '\u04FF')
+    if cyrillic_chars > total_chars * 0.1:
+        return False
+
+    return True
+
+
+def count_tokens(text: str) -> int:
+    """Count tokens using whitespace splitting (approximate)."""
+    return len(text.split())
+
+
+class DataProcessor:
+    """Processes and cleans collected Kinyarwanda data for training."""
+
+    def __init__(self, raw_dir: str, processed_dir: str):
+        self.raw_dir = Path(raw_dir)
+        self.processed_dir = Path(processed_dir)
+        self.processed_dir.mkdir(parents=True, exist_ok=True)
+
+    def load_jsonl(self, filepath: Path) -> list[dict[str, Any]]:
+        """Load a JSONL file."""
+        data = []
+        with open(filepath, "r", encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if line:
+                    data.append(json.loads(line))
+        return data
+
+    def clean_text(self, text: str) -> str:
+        """Clean and normalize Kinyarwanda text."""
+        text = re.sub(r"<[^>]+>", "", text)
+        text = re.sub(r"http\S+|www\.\S+", "", text)
+        text = re.sub(r"\S+@\S+\.\S+", "", text)
+        text = re.sub(
+            r"[^\w\s.,;:!?'\-\u00C0-\u024F]",
+            " ",
+            text,
+        )
+        text = re.sub(r"\s+", " ", text)
+        text = text.strip()
+
+        if len(text) < 5:
+            return ""
+
+        return text
+
+    def deduplicate(self, texts: list[str]) -> list[str]:
+        """Remove duplicate texts, keeping first occurrence."""
+        seen = set()
+        unique = []
+
+        for text in texts:
+            normalized = text.lower().strip()
+            if normalized not in seen:
+                seen.add(normalized)
+                unique.append(text)
+
+        return unique
+
+    def create_training_format(self, texts: list[str]) -> list[dict[str, str]]:
+        """Create training format for language modeling."""
+        training_data = []
+
+        for text in texts:
+            if text and len(text) > 10:
+                training_data.append({"text": text})
+
+        return training_data
+
+    def create_instruction_format(self, data: list[dict[str, Any]]) -> list[dict[str, str]]:
+        """Create instruction-following format."""
+        formatted = []
+
+        for item in data:
+            if "instruction" in item and "output" in item:
+                formatted.append({
+                    "instruction": item["instruction"],
+                    "input": item.get("input", ""),
+                    "output": item["output"],
+                })
+            elif "question" in item and "answer" in item:
+                formatted.append({
+                    "instruction": item["question"],
+                    "input": "",
+                    "output": item["answer"],
+                })
+            elif "prompt" in item and "response" in item:
+                formatted.append({
+                    "instruction": item["prompt"],
+                    "input": "",
+                    "output": item["response"],
+                })
+
+        return formatted
+
+    def create_chat_format(self, data: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Create chat/conversation format."""
+        formatted = []
+
+        for item in data:
+            if "messages" in item:
+                formatted.append({"messages": item["messages"]})
+            elif "conversation" in item:
+                formatted.append({"messages": item["conversation"]})
+            elif "input" in item and "output" in item:
+                messages = [
+                    {"role": "user", "content": item["input"]},
+                    {"role": "assistant", "content": item["output"]},
+                ]
+                formatted.append({"messages": messages})
+
+        return formatted
+
+    def process_all(self) -> dict[str, Any]:
+        """Process all collected data."""
+        logger.info("Processing collected data...")
+
+        all_texts = []
+        all_instruction_data = []
+        all_chat_data = []
+
+        file_stats = []
+        total_tokens = 0
+        total_before = 0
+        total_after_lang = 0
+        total_after_clean = 0
+        total_after_dedup = 0
+
+        jsonl_files = list(self.raw_dir.glob("*.jsonl"))
+        logger.info(f"Found {len(jsonl_files)} JSONL files to process")
+
+        for jsonl_file in jsonl_files:
+            logger.info(f"Processing {jsonl_file.name}...")
+
+            data = self.load_jsonl(jsonl_file)
+            file_rows = len(data)
+            total_before += file_rows
+
+            file_texts = []
+            file_instruction = 0
+            file_chat = 0
+            file_lang_removed = 0
+            file_clean_removed = 0
+
+            for item in data:
+                text = self._extract_text(item)
+                if text:
+                    if not is_likely_kinyarwanda(text):
+                        file_lang_removed += 1
+                        continue
+
+                    cleaned = self.clean_text(text)
+                    if cleaned:
+                        file_texts.append(cleaned)
+                        total_tokens += count_tokens(cleaned)
+                    else:
+                        file_clean_removed += 1
+                else:
+                    file_clean_removed += 1
+
+                if self._is_instruction(item):
+                    all_instruction_data.append(item)
+                    file_instruction += 1
+
+                if self._is_chat(item):
+                    all_chat_data.append(item)
+                    file_chat += 1
+
+            total_after_lang += len(file_texts) + file_lang_removed
+            all_texts.extend(file_texts)
+            total_after_clean += len(file_texts)
+
+            stats = {
+                "file": jsonl_file.name,
+                "loaded": file_rows,
+                "lang_filtered": file_lang_removed,
+                "clean_filtered": file_clean_removed,
+                "kept": len(file_texts),
+                "tokens": sum(count_tokens(t) for t in file_texts),
+                "instruction_samples": file_instruction,
+                "chat_samples": file_chat,
+            }
+            file_stats.append(stats)
+            logger.info(
+                f"  {jsonl_file.name}: loaded={file_rows}, "
+                f"lang_removed={file_lang_removed}, clean_removed={file_clean_removed}, "
+                f"kept={len(file_texts)}"
+            )
+
+        logger.info(f"Total texts before dedup: {len(all_texts)}")
+        all_texts = self.deduplicate(all_texts)
+        total_after_dedup = len(all_texts)
+        logger.info(f"After deduplication: {len(all_texts)} (removed {total_before - len(all_texts)} duplicates)")
+
+        training_data = self.create_training_format(all_texts)
+        training_path = self.processed_dir / "training_data.jsonl"
+        with open(training_path, "w", encoding="utf-8") as f:
+            for item in training_data:
+                f.write(json.dumps(item, ensure_ascii=False) + "\n")
+
+        instruction_data = self.create_instruction_format(all_instruction_data)
+        instruction_path = self.processed_dir / "instruction_data.jsonl"
+        with open(instruction_path, "w", encoding="utf-8") as f:
+            for item in instruction_data:
+                f.write(json.dumps(item, ensure_ascii=False) + "\n")
+
+        chat_data = self.create_chat_format(all_chat_data)
+        chat_path = self.processed_dir / "chat_data.jsonl"
+        with open(chat_path, "w", encoding="utf-8") as f:
+            for item in chat_data:
+                f.write(json.dumps(item, ensure_ascii=False) + "\n")
+
+        summary = {
+            "total_loaded": total_before,
+            "total_after_lang_filter": total_after_clean,
+            "total_after_dedup": total_after_dedup,
+            "total_tokens": total_tokens,
+            "total_training_samples": len(training_data),
+            "total_instruction": len(instruction_data),
+            "total_chat": len(chat_data),
+            "per_file_stats": file_stats,
+            "output_files": {
+                "training": str(training_path),
+                "instruction": str(instruction_path),
+                "chat": str(chat_path),
+            },
+        }
+
+        summary_path = self.processed_dir / "processing_summary.json"
+        with open(summary_path, "w", encoding="utf-8") as f:
+            json.dump(summary, f, indent=2, ensure_ascii=False)
+
+        logger.info(f"Processing complete:")
+        logger.info(f"  Loaded: {total_before:,} rows")
+        logger.info(f"  After language filter: {total_after_clean:,}")
+        logger.info(f"  After dedup: {total_after_dedup:,}")
+        logger.info(f"  Tokens: {total_tokens:,}")
+        logger.info(f"  Training samples: {len(training_data):,}")
+        logger.info(f"  Instruction samples: {len(instruction_data):,}")
+        logger.info(f"  Chat samples: {len(chat_data):,}")
+
+        return summary
+
+    def _extract_text(self, item: dict[str, Any]) -> str | None:
+        """Extract text from a data item."""
+        for key in ["text", "content", "article", "document", "body"]:
+            if key in item and isinstance(item[key], str):
+                return item[key]
+
+        for key in item:
+            if isinstance(item[key], str) and len(item[key]) > 50:
+                return item[key]
+
+        return None
+
+    def _is_instruction(self, item: dict[str, Any]) -> bool:
+        """Check if item is instruction-following data."""
+        instruction_keys = ["instruction", "question", "prompt", "input"]
+        return any(key in item for key in instruction_keys)
+
+    def _is_chat(self, item: dict[str, Any]) -> bool:
+        """Check if item is chat/conversation data."""
+        chat_keys = ["messages", "conversation", "chat", "dialogue"]
+        return any(key in item for key in chat_keys)