#!/usr/bin/env python3
"""
Bulk generate Python training data via 3B model on port 8085.
Uses delimiter-based output format to avoid JSON escaping issues.
Target: 500+ entries.
"""
import json
import os
import re
import sys
import time
import requests
from pathlib import Path
BASE = Path(__file__).parent.parent / "data" / "processed"
MODEL = "Qwen2.5-3B-Instruct-Q4_K_M.gguf"
API = "http://localhost:8085/v1/chat/completions"
SYSTEM = """You are a Python code review expert generating training data.
Output format - one entry per block:
---ENTRY---
ISSUE: Short title of the problem
BAD:
[the bad python code here]
GOOD:
[the fixed python code here]
Rules:
- Each entry fixes ONE specific Python anti-pattern or bug
- BAD shows real code a developer would write
- GOOD shows the corrected, modern, idiomatic fix
- Keep code examples focused and realistic (3-15 lines)
- Generate exactly the number requested
- Do NOT add explanation text outside the entry blocks"""
CATEGORIES = [
("SQL injection, command injection, eval() on user input, path traversal, pickle deserialization, log injection, hardcoded secrets, insecure random, CSRF missing, sensitive data in URLs, XSS in templates, weak password hashing, missing HTTPS redirect, missing input validation", 15),
("String concatenation in loop, unnecessary list copy, range(len()) instead of enumerate, global variable lookup in hot path, double dict lookup, missing __slots__, linear search on large list, materializing generator into list, string concat instead of f-string, missing lru_cache, blocking I/O in main thread, import inside loop, uncompiled regex in loop, unnecessary deepcopy, threading for CPU work", 15),
("Missing type hints, old % string formatting, missing context manager, bare except, mutable default argument, list instead of deque for queue, manual isinstance chain instead of match/case, class instead of dataclass, dict.get with manual check instead of defaultdict, integer constants instead of Enum, os.path instead of pathlib, manual retry instead of tenacity, missing Optional return type, missing __future__ annotations, if/elif chain instead of match/case", 15),
("Flask missing error handler, FastAPI sync endpoint blocking, Django N+1 query, Django extra() raw SQL, missing select_for_update, FastAPI missing Pydantic validation, Flask missing Blueprint, missing pagination, DRF fields __all__, FastAPI missing dependency injection, Flask missing app factory, Django missing CSRF for AJAX, Flask missing pytest fixture, FastAPI missing response_model", 14),
("Sync requests.get in async function, unawaited coroutine, asyncio.gather without return_exceptions, missing asyncio.Semaphore, time.sleep in async, race condition without asyncio.Lock, task cancellation without cleanup, missing contextvars propagation, asyncio.Queue without timeout, blocking file I/O in async handler", 10),
("Testing private implementation details, missing pytest fixtures, testing with real dependencies, missing parameterized tests, flaky time-dependent test, missing test isolation, missing pytest.raises, missing pytest.mark.asyncio, over-mocking, missing hypothesis property tests", 10),
("Manual word count instead of Counter, list comprehension abuse, list as FIFO queue, manual binary search, manual groupby, missing itertools.combinations, inefficient dedup with list, manual chunking, pandas for simple CSV, readlines() loading entire file", 10),
("God class, missing Strategy pattern, missing Observer pattern, hardcoded dependencies, missing Builder pattern, manual singleton, missing Factory pattern, missing Repository pattern, missing Middleware pattern, missing Unit of Work", 10),
("Manual flag instead of any/all, manual min/max loop, range(len) instead of zip, manual dict merge, missing itertools.chain, sorted with unnecessary key, manual enum with integers, len(items) > 0 check, missing walrus operator, missing itertools.chain.from_iterable", 10),
("Missing requirements pinning, root in Docker, missing health check, app.run in production, hardcoded DB URL, no venv, missing logging config, hardcoded database URI, missing graceful shutdown, missing pre-commit hooks", 10),
("Missing return type hints, using Any everywhere, missing Protocol, Optional vs Union confusion, missing Literal, missing TypedDict, forward reference issues, missing Generic, type() instead of isinstance, missing TypeAlias", 10),
("Missing encoding on open(), readlines() instead of iterate, missing tempfile, reading binary as text, missing newline for CSV, missing shutil, file without context manager, missing buffering parameter, not using pathlib.iterdir, missing tempfile.NamedTemporaryFile", 10),
("Bare except, silent except pass, exceptions for control flow, missing exception chaining, catching Exception broadly, missing custom exceptions, swallowed stack traces, except OSError not catching all, raising bare Exception, except not including TypeError", 10),
("SQLAlchemy missing eager loading, Django missing select_related, missing connection pooling, raw SQL in QuerySet, missing transaction.atomic, N+1 in serializer, missing database indexes, not using bulk_create, missing migration rollback test, ORM loading full objects for scalar", 10),
("Missing input validation on API, returning 200 on error, missing pagination on list endpoint, exposing internal IDs, missing rate limiting, not using OpenAPI docs, returning ORM models directly, missing request schemas, missing CORS handling, missing API versioning", 10),
("Hardcoded config values, missing env var validation, config in source control, missing config schema, not separating dev/prod, missing secrets management, config as globals, missing config reload, not using pydantic-settings", 9),
("Logging sensitive data, missing log levels, print instead of logging, missing structured logging, logger without name, missing correlation IDs, logging without exception context, missing metrics endpoint, missing request ID", 9),
("Missing argparse, using sys.argv directly, missing subcommands, missing help descriptions, hardcoded output format, missing version flag, missing argparse choices, missing store_true, missing argument validation", 9),
("Circular imports, missing __init__ exports, relative import errors, module-level import side effects, missing __all__, import in function for perf, missing __main__ guard, duplicate imports, missing namespace package", 9),
("Missing encoding parameter, bytes vs str confusion, str.encode misuse, missing utf-8-sig BOM, latin-1 fallback, f-string with bytes, CSV encoding issues, JSON ensure_ascii overhead", 8),
]
def generate_batch(category: str, count: int, attempt: int = 0) -> list:
"""Request a batch from the 3B model using delimiter format."""
user_msg = (
f"Generate exactly {count} Python code review training entries.\n\n"
f"Topics: {category}\n\n"
f"Output format (one per block):\n"
f"---ENTRY---\n"
f"ISSUE: title\n"
f"BAD:\n"
f"[bad code]\n"
f"GOOD:\n"
f"[good code]\n\n"
f"Keep each entry focused. No extra text outside the blocks."
)
resp = requests.post(
API,
json={
"model": MODEL,
"messages": [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": user_msg},
],
"temperature": 0.8,
"max_tokens": 8192,
"top_p": 0.9,
},
timeout=180,
)
text = resp.json()["choices"][0]["message"]["content"].strip()
# Parse delimiter format
entries = []
blocks = re.split(r'---ENTRY---', text)
for block in blocks:
block = block.strip()
if not block:
continue
issue_m = re.search(r'ISSUE:\s*(.+)', block)
bad_m = re.search(r'BAD:\s*\n(.*?)(?:\nGOOD:|\Z)', block, re.DOTALL)
good_m = re.search(r'GOOD:\s*\n(.*?)(?:\n---|\Z)', block, re.DOTALL)
if issue_m and bad_m and good_m:
issue = issue_m.group(1).strip().strip('"')
bad = bad_m.group(1).strip()
good = good_m.group(1).strip()
if bad and good and bad != good:
entries.append({
"issue": issue,
"bad": bad,
"good": good,
})
if not entries and attempt < 2:
return generate_batch(category, count, attempt + 1)
return entries
def validate_entry(entry: dict) -> bool:
"""Check entry quality."""
if not isinstance(entry, dict):
return False
for key in ("issue", "bad", "good"):
if key not in entry or not str(entry[key]).strip():
return False
# Good code should differ from bad
if str(entry["bad"]).strip() == str(entry["good"]).strip():
return False
return True
def to_training_entry(entry: dict) -> dict:
"""Convert to training format."""
issue = entry["issue"]
bad_code = entry["bad"]
good_code = entry["good"]
tag = re.sub(r"[^\w]+", "_", issue.lower()).strip("_")
return {
"messages": [
{
"role": "user",
"content": f"Review this Python code:\n\n```python\n{bad_code}\n```",
},
{
"role": "assistant",
"content": f"**Issue: {issue}**\n\n**Fix:**\n```python\n{good_code}\n```",
},
],
"source": "python_bulk_gen",
"tags": ["python", tag],
}
def deduplicate(entries: list) -> list:
"""Remove entries with similar bad code."""
seen = set()
unique = []
for e in entries:
raw = e["messages"][0]["content"]
key = re.sub(r"\s+", " ", raw[:200]).lower().strip()
if key not in seen:
seen.add(key)
unique.append(e)
return unique
def main():
existing_path = BASE / "dev_finetuning_python.json"
# Load existing
existing = []
if existing_path.exists():
with open(existing_path) as f:
existing = json.load(f)
print(f"Existing entries: {len(existing)}")
all_raw = []
total_cats = len(CATEGORIES)
start_time = time.time()
for i, (category, count) in enumerate(CATEGORIES, 1):
cat_name = category[:50] + "..." if len(category) > 50 else category
print(f"\n[{i}/{total_cats}] {cat_name} ({count} entries)", flush=True)
batch = generate_batch(category, count)
valid = [e for e in batch if validate_entry(e)]
elapsed = time.time() - start_time
rate = elapsed / max(i, 1)
eta = rate * (total_cats - i)
print(f" → Got {len(batch)}, valid: {len(valid)} | "
f"Elapsed: {elapsed:.0f}s | ETA: {eta:.0f}s", flush=True)
all_raw.extend(valid)
print(f"\nTotal raw entries collected: {len(all_raw)}")
# Convert
new_entries = [to_training_entry(e) for e in all_raw]
# Merge + dedup
combined = existing + new_entries
combined = deduplicate(combined)
print(f"After merge + dedup: {len(combined)}")
# Shuffle
import random
random.seed(42)
random.shuffle(combined)
# Save
os.makedirs(BASE, exist_ok=True)
with open(existing_path, "w") as f:
json.dump(combined, f, indent=2)
total_time = time.time() - start_time
size_kb = round(os.path.getsize(existing_path) / 1024, 1)
print(f"\n{'='*60}")
print(f"Saved {len(combined)} entries ({size_kb} KB)")
print(f"Time: {total_time:.0f}s ({total_time/60:.1f} min)")
# Tag breakdown
tag_counts = {}
for e in combined:
for tag in e.get("tags", []):
if tag != "python":
tag_counts[tag] = tag_counts.get(tag, 0) + 1
print(f"\nTop 15 tags:")
for tag, count in sorted(tag_counts.items(), key=lambda x: -x[1])[:15]:
print(f" {tag}: {count}")
if __name__ == "__main__":
main()