Upload .\scripts\test_pipeline.py with huggingface_hub

.//scripts//test_pipeline.py

@@ -0,0 +1,141 @@
+"""Small end-to-end test for BwengeAi pipeline."""
+
+import json
+import logging
+import sys
+import shutil
+from pathlib import Path
+
+import yaml
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from data_collection.huggingface_collector import HuggingfaceCollector
+from data_collection.data_processor import DataProcessor
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+
+def load_config(config_path: str = None) -> dict:
+    if config_path is None:
+        config_path = str(PROJECT_ROOT / "configs/test_small.yaml")
+    with open(config_path, "r", encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+
+def main():
+    config = load_config()
+    data_config = config.get("data", {})
+
+    raw_dir = str(PROJECT_ROOT / data_config.get("raw_dir", "data/raw_test"))
+    processed_dir = str(PROJECT_ROOT / data_config.get("processed_dir", "data/processed_test"))
+
+    Path(raw_dir).mkdir(parents=True, exist_ok=True)
+    Path(processed_dir).mkdir(parents=True, exist_ok=True)
+
+    logger.info("=" * 60)
+    logger.info("STEP 1: Collect 4 small HuggingFace datasets")
+    logger.info("=" * 60)
+
+    hf_collector = HuggingfaceCollector(
+        output_dir=f"{raw_dir}/huggingface",
+        config=data_config,
+    )
+    hf_results = hf_collector.collect_all()
+
+    successful = sum(1 for r in hf_results if r.get("status") == "success")
+    failed = sum(1 for r in hf_results if r.get("status") == "failed")
+    total_rows = sum(r.get("rows", 0) for r in hf_results if r.get("status") == "success")
+
+    logger.info(f"Collection: {successful} succeeded, {failed} failed, {total_rows} total rows")
+
+    logger.info("=" * 60)
+    logger.info("STEP 2: Process and clean data")
+    logger.info("=" * 60)
+
+    processor = DataProcessor(
+        raw_dir=raw_dir,
+        processed_dir=processed_dir,
+    )
+
+    import shutil as sh
+    seen_names = set()
+    for jsonl_file in Path(raw_dir).rglob("*.jsonl"):
+        if jsonl_file.parent == Path(raw_dir):
+            continue
+        parent_dir = jsonl_file.parent.name
+        prefixed_name = f"{parent_dir}_{jsonl_file.name}"
+        dest = Path(raw_dir) / prefixed_name
+        if prefixed_name not in seen_names:
+            sh.copy2(jsonl_file, dest)
+            seen_names.add(prefixed_name)
+
+    processing_summary = processor.process_all()
+
+    logger.info("=" * 60)
+    logger.info("TEST RESULTS")
+    logger.info("=" * 60)
+
+    raw_path = Path(raw_dir)
+    processed_path = Path(processed_dir)
+
+    raw_files = list(raw_path.rglob("*.jsonl"))
+    raw_total_rows = 0
+    for f in raw_files:
+        with open(f, encoding="utf-8") as fh:
+            raw_total_rows += sum(1 for _ in fh)
+
+    logger.info(f"Raw files: {len(raw_files)}")
+    logger.info(f"Raw total rows: {raw_total_rows:,}")
+
+    training_path = processed_path / "training_data.jsonl"
+    instruction_path = processed_path / "instruction_data.jsonl"
+    chat_path = processed_path / "chat_data.jsonl"
+    summary_path = processed_path / "processing_summary.json"
+
+    for p in [training_path, instruction_path, chat_path, summary_path]:
+        exists = p.exists()
+        size = p.stat().st_size if exists else 0
+        logger.info(f"  {p.name}: {'EXISTS' if exists else 'MISSING'} ({size:,} bytes)")
+
+    if summary_path.exists():
+        with open(summary_path, encoding="utf-8") as f:
+            summary = json.load(f)
+        logger.info(f"\nProcessing Summary:")
+        logger.info(f"  Total loaded:       {summary.get('total_loaded', 0):,}")
+        logger.info(f"  After lang filter:  {summary.get('total_after_lang_filter', 0):,}")
+        logger.info(f"  After dedup:        {summary.get('total_after_dedup', 0):,}")
+        logger.info(f"  Total tokens:       {summary.get('total_tokens', 0):,}")
+        logger.info(f"  Training samples:   {summary.get('total_training_samples', 0):,}")
+        logger.info(f"  Instruction samples: {summary.get('total_instruction', 0):,}")
+        logger.info(f"  Chat samples:       {summary.get('total_chat', 0):,}")
+
+        if summary.get('per_file_stats'):
+            logger.info(f"\nPer-file stats:")
+            for fs in summary['per_file_stats']:
+                logger.info(f"  {fs['file']}: loaded={fs['loaded']}, kept={fs['kept']}, tokens={fs['tokens']}")
+
+    all_ok = (
+        training_path.exists()
+        and summary_path.exists()
+        and raw_total_rows > 0
+    )
+
+    logger.info("\n" + "=" * 60)
+    if all_ok:
+        logger.info("TEST PASSED - Pipeline works end-to-end!")
+    else:
+        logger.info("TEST FAILED - Check errors above")
+    logger.info("=" * 60)
+
+    return 0 if all_ok else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())