File size: 5,741 Bytes
eca5751 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | """
Script chuẩn bị dataset cho training
====================================
Process raw collected data → cleaned, deduplicated, formatted training data.
Steps:
1. Load raw data from ./data/raw/
2. Clean text (TextCleaner)
3. Format code samples (CodeFormatter)
4. Filter by quality (QualityFilter)
5. Deduplicate (Deduplicator)
6. Save processed data to ./data/processed/
Usage:
python scripts/prepare_dataset.py --input ./data/raw --output ./data/processed
"""
import sys
import os
import json
import argparse
import logging
from pathlib import Path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)
def load_raw_data(input_dir: str):
"""Load all JSONL files from input directory."""
files = [
f for f in os.listdir(input_dir)
if f.endswith(".jsonl")
]
total = 0
for fname in files:
fpath = os.path.join(input_dir, fname)
count = 0
with open(fpath, "r", encoding="utf-8") as f:
for line in f:
try:
item = json.loads(line)
yield item
count += 1
except json.JSONDecodeError:
continue
logger.info(f" Loaded {count} from {fname}")
total += count
logger.info(f"Total raw samples: {total}")
def process_data(input_dir: str, output_dir: str, max_samples: int = None):
"""Process raw data through cleaning, dedup, quality filter."""
from nexus.data.processors.cleaner import TextCleaner
from nexus.data.processors.quality_filter import QualityFilter
from nexus.data.processors.code_formatter import CodeFormatter
from nexus.data.processors.deduplicator import Deduplicator
from nexus.data.curriculum import CurriculumLearning
cleaner = TextCleaner()
quality_filter = QualityFilter()
code_formatter = CodeFormatter()
deduplicator = Deduplicator()
curriculum = CurriculumLearning()
os.makedirs(output_dir, exist_ok=True)
# Output files by difficulty
output_files = {
"easy": open(os.path.join(output_dir, "train_easy.jsonl"), "w", encoding="utf-8"),
"medium": open(os.path.join(output_dir, "train_medium.jsonl"), "w", encoding="utf-8"),
"hard": open(os.path.join(output_dir, "train_hard.jsonl"), "w", encoding="utf-8"),
"expert": open(os.path.join(output_dir, "train_expert.jsonl"), "w", encoding="utf-8"),
}
stats = {
"total_input": 0,
"cleaned": 0,
"quality_passed": 0,
"deduplicated": 0,
"by_difficulty": {"easy": 0, "medium": 0, "hard": 0, "expert": 0},
}
logger.info("Processing samples...")
for sample in load_raw_data(input_dir):
if max_samples and stats["total_input"] >= max_samples:
break
stats["total_input"] += 1
# Step 1: Clean
sample = cleaner.process(sample)
if sample is None:
continue
stats["cleaned"] += 1
# Step 2: Format code
sample = code_formatter.process(sample)
# Step 3: Quality filter
if not quality_filter.filter(sample):
continue
sample = next(quality_filter.process([sample]), None)
if sample is None:
continue
stats["quality_passed"] += 1
# Step 4: Dedup
if deduplicator.is_duplicate(sample.get("text", "")):
continue
deduplicator.add(sample["text"], sample)
stats["deduplicated"] += 1
# Step 5: Classify by difficulty
difficulty = curriculum.classify_sample(sample).value
output_files[difficulty].write(json.dumps(sample, ensure_ascii=False) + "\n")
stats["by_difficulty"][difficulty] += 1
if stats["deduplicated"] % 1000 == 0:
logger.info(f" Processed {stats['deduplicated']} unique samples...")
# Close files
for f in output_files.values():
f.close()
# Print stats
print("\n" + "=" * 60)
print(" PROCESSING COMPLETE")
print("=" * 60)
print(f" Input samples: {stats['total_input']:,}")
print(f" After cleaning: {stats['cleaned']:,}")
print(f" Quality passed: {stats['quality_passed']:,}")
print(f" After dedup: {stats['deduplicated']:,}")
print("-" * 60)
print(" By difficulty:")
for level, count in stats["by_difficulty"].items():
print(f" {level:8s}: {count:,}")
print("-" * 60)
print(f" Output dir: {output_dir}")
print("=" * 60)
# Save stats
stats_path = os.path.join(output_dir, "processing_stats.json")
with open(stats_path, "w", encoding="utf-8") as f:
json.dump(stats, f, indent=2)
return stats
def main():
parser = argparse.ArgumentParser(description="Nexus Coder Dataset Processor")
parser.add_argument("--input", type=str, default="./data/raw")
parser.add_argument("--output", type=str, default="./data/processed")
parser.add_argument("--max-samples", type=int, default=None)
args = parser.parse_args()
print("=" * 70)
print(" NEXUS CODER v0.2 - DATASET PROCESSOR")
print(" Tác giả: Hieu Louis")
print("=" * 70)
if not os.path.exists(args.input):
print(f"\n❌ Input dir not found: {args.input}")
print("Run scripts/collect_data.py first to collect raw data.")
return 1
process_data(args.input, args.output, args.max_samples)
return 0
if __name__ == "__main__":
sys.exit(main())
|