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

.//src//data_collection//cc100_oscar_collector.py

@@ -0,0 +1,185 @@
+"""CC-100 and OSCAR massive monolingual corpus collector."""
+
+import json
+import logging
+from pathlib import Path
+from typing import Any
+
+from datasets import load_dataset
+
+logger = logging.getLogger(__name__)
+
+CC100_VALID_LANGUAGES = {
+    "am", "ar", "bg", "bn", "br", "bs", "ca", "cs", "cy", "da", "de",
+    "el", "en", "eo", "es", "et", "eu", "fa", "fi", "fr", "fy", "ga",
+    "gd", "gl", "gu", "ha", "he", "hi", "hr", "hu", "hy", "id", "is",
+    "it", "ja", "jv", "ka", "kk", "km", "kn", "ko", "ku", "ky", "la",
+    "lt", "lv", "mg", "mk", "ml", "mn", "mr", "ms", "my", "ne", "nl",
+    "no", "om", "or", "pa", "pl", "ps", "pt", "ro", "ru", "sa", "sd",
+    "si", "sk", "sl", "so", "sq", "sr", "su", "sv", "sw", "ta", "te",
+    "th", "tl", "tr", "ug", "uk", "ur", "uz", "vi", "vo", "xh", "yi",
+    "yo", "zh", "zu",
+}
+
+OSCAR_VALID_LANGUAGES = {
+    "ab", "af", "am", "ar", "as", "az", "be", "bg", "bn", "br", "bs",
+    "ca", "cs", "cy", "da", "de", "el", "en", "eo", "es", "et", "eu",
+    "fa", "fi", "fr", "fy", "ga", "gd", "gl", "gu", "ha", "he", "hi",
+    "hr", "hu", "hy", "id", "is", "it", "ja", "jv", "ka", "kk", "km",
+    "kn", "ko", "ku", "ky", "la", "lt", "lv", "mg", "mk", "ml", "mn",
+    "mr", "ms", "my", "ne", "nl", "no", "om", "or", "pa", "pl", "ps",
+    "pt", "ro", "ru", "sa", "sd", "si", "sk", "sl", "so", "sq", "sr",
+    "su", "sv", "sw", "ta", "te", "th", "tl", "tr", "ug", "uk", "ur",
+    "uz", "vi", "vo", "xh", "yi", "yo", "zh", "zu",
+}
+
+
+class CC100OSCARCollector:
+    """Collects massive monolingual corpora from CC-100 and OSCAR."""
+
+    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
+
+    def collect_cc100(self, language: str = "rw", max_rows: int | None = None) -> dict[str, Any]:
+        """Collect CC-100 data for a specific language."""
+        logger.info(f"Collecting CC-100 for language: {language}")
+
+        if language not in CC100_VALID_LANGUAGES:
+            logger.warning(
+                f"CC-100 does not support language '{language}'. "
+                f"Supported: {sorted(CC100_VALID_LANGUAGES)}. Skipping."
+            )
+            return {
+                "name": f"cc100_{language}",
+                "language": language,
+                "rows": 0,
+                "status": "skipped",
+                "error": f"Language '{language}' not in CC-100 dataset",
+            }
+
+        done_marker = self.output_dir / f"cc100_{language}.jsonl.done"
+        output_path = self.output_dir / f"cc100_{language}.jsonl"
+
+        if done_marker.exists():
+            logger.info(f"CC-100 {language} already collected, skipping")
+            return {
+                "name": f"cc100_{language}",
+                "language": language,
+                "rows": sum(1 for _ in open(output_path, encoding="utf-8")),
+                "output_path": str(output_path),
+                "status": "success",
+            }
+
+        try:
+            dataset = load_dataset("statmt/cc100", language, split="train", streaming=True)
+
+            count = 0
+            with open(output_path, "w", encoding="utf-8") as f:
+                for item in dataset:
+                    f.write(json.dumps(item, ensure_ascii=False) + "\n")
+                    count += 1
+                    if max_rows and count >= max_rows:
+                        break
+                    if count % 100000 == 0:
+                        logger.info(f"  CC-100 {language}: {count} rows...")
+
+            done_marker.write_text(str(count))
+            return {
+                "name": f"cc100_{language}",
+                "language": language,
+                "rows": count,
+                "output_path": str(output_path),
+                "status": "success",
+            }
+
+        except Exception as e:
+            logger.error(f"Failed to collect CC-100 {language}: {e}")
+            return {"name": f"cc100_{language}", "status": "failed", "error": str(e)}
+
+    def collect_oscar(self, language: str = "rw", max_rows: int | None = None) -> dict[str, Any]:
+        """Collect OSCAR data for a specific language."""
+        logger.info(f"Collecting OSCAR for language: {language}")
+
+        if language not in OSCAR_VALID_LANGUAGES:
+            logger.warning(
+                f"OSCAR-2201 does not support language '{language}'. "
+                f"Supported: {sorted(OSCAR_VALID_LANGUAGES)}. Skipping."
+            )
+            return {
+                "name": f"oscar_{language}",
+                "language": language,
+                "rows": 0,
+                "status": "skipped",
+                "error": f"Language '{language}' not in OSCAR-2201 dataset",
+            }
+
+        done_marker = self.output_dir / f"oscar_{language}.jsonl.done"
+        output_path = self.output_dir / f"oscar_{language}.jsonl"
+
+        if done_marker.exists():
+            logger.info(f"OSCAR {language} already collected, skipping")
+            return {
+                "name": f"oscar_{language}",
+                "language": language,
+                "rows": sum(1 for _ in open(output_path, encoding="utf-8")),
+                "output_path": str(output_path),
+                "status": "success",
+            }
+
+        try:
+            dataset = load_dataset("oscar-corpus/OSCAR-2201", language, split="train", streaming=True)
+
+            count = 0
+            with open(output_path, "w", encoding="utf-8") as f:
+                for item in dataset:
+                    f.write(json.dumps(item, ensure_ascii=False) + "\n")
+                    count += 1
+                    if max_rows and count >= max_rows:
+                        break
+                    if count % 100000 == 0:
+                        logger.info(f"  OSCAR {language}: {count} rows...")
+
+            done_marker.write_text(str(count))
+            return {
+                "name": f"oscar_{language}",
+                "language": language,
+                "rows": count,
+                "output_path": str(output_path),
+                "status": "success",
+            }
+
+        except Exception as e:
+            logger.error(f"Failed to collect OSCAR {language}: {e}")
+            return {"name": f"oscar_{language}", "status": "failed", "error": str(e)}
+
+    def collect_all(self, languages: list[str] | None = None) -> list[dict[str, Any]]:
+        """Collect all configured monolingual corpora."""
+        if languages is None:
+            languages = ["rw"]
+
+        results = []
+        for lang in languages:
+            results.append(self.collect_cc100(lang))
+            results.append(self.collect_oscar(lang))
+
+        successful = sum(1 for r in results if r["status"] == "success")
+        skipped = sum(1 for r in results if r["status"] == "skipped")
+        failed = sum(1 for r in results if r["status"] == "failed")
+        total_rows = sum(r.get("rows", 0) for r in results if r["status"] == "success")
+
+        summary = {
+            "total": len(results),
+            "successful": successful,
+            "skipped": skipped,
+            "failed": failed,
+            "total_rows": total_rows,
+        }
+
+        summary_path = self.output_dir / "cc100_oscar_summary.json"
+        with open(summary_path, "w", encoding="utf-8") as f:
+            json.dump(summary, f, indent=2, ensure_ascii=False)
+
+        logger.info(f"CC-100/OSCAR collection: {summary}")
+        return results