99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Load blog-training-alpaca.jsonl into PostgreSQL learning_corpus table via psql."""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
JSONL_FILE = Path(__file__).parent.parent / "data" / "blog-training-alpaca.jsonl"
|
|
DB_URL = os.environ.get("FT_DB_URL", "postgresql://llm@localhost:15432/llm_gateway")
|
|
PSQL_BIN = os.environ.get("PSQL_BIN", "psql")
|
|
TASK_TYPE = "tip_blog"
|
|
|
|
# Parse connection details
|
|
parsed = urlparse(DB_URL)
|
|
db_user = parsed.username or "llm"
|
|
db_pass = parsed.password or os.environ.get("PGPASSWORD")
|
|
host = parsed.hostname or "localhost"
|
|
port = str(parsed.port or 5432)
|
|
db_name = parsed.path.lstrip("/") or "llm_gateway"
|
|
|
|
print("🔄 Loading blog training data into PostgreSQL...")
|
|
print(f" File: {JSONL_FILE}")
|
|
print(f" Host: {host}:{port}")
|
|
print(f" Database: {db_name}")
|
|
print()
|
|
|
|
# Generate SQL statements
|
|
sql_statements = []
|
|
|
|
# Read JSONL and convert to SQL
|
|
try:
|
|
with open(JSONL_FILE) as f:
|
|
samples = [json.loads(line.strip()) for line in f if line.strip()]
|
|
except FileNotFoundError:
|
|
print(f"❌ File not found: {JSONL_FILE}")
|
|
sys.exit(1)
|
|
|
|
# Start transaction
|
|
sql_statements.append("BEGIN;")
|
|
sql_statements.append(f"SELECT COUNT(*) as before_count FROM learning_corpus WHERE task_type = '{TASK_TYPE}';")
|
|
|
|
# Insert samples
|
|
for sample in samples:
|
|
prompt_text = sample.get("instruction", "")
|
|
if sample.get("input"):
|
|
prompt_text += f"\n{sample['input']}"
|
|
|
|
# Escape single quotes for SQL
|
|
prompt_text = prompt_text.replace("'", "''")
|
|
completion_text = sample.get("output", "").replace("'", "''")
|
|
quality_score = sample.get("quality_score", 8.0)
|
|
source = sample.get("source", "unknown").replace("'", "''")
|
|
|
|
tags = f"ARRAY['{source}', '{TASK_TYPE}']"
|
|
|
|
sql = f"INSERT INTO learning_corpus (task_type, prompt_text, completion_text, quality_score, tags) VALUES ('{TASK_TYPE}', '{prompt_text}', '{completion_text}', {quality_score}, {tags});"
|
|
sql_statements.append(sql)
|
|
|
|
sql_statements.append(f"SELECT COUNT(*) as after_count FROM learning_corpus WHERE task_type = '{TASK_TYPE}';")
|
|
sql_statements.append("COMMIT;")
|
|
|
|
# Write to temporary SQL file
|
|
tmpfile = Path("/tmp/load_blog_data.sql")
|
|
with open(tmpfile, "w") as f:
|
|
f.write("\n".join(sql_statements))
|
|
|
|
# Execute via psql
|
|
try:
|
|
env = os.environ.copy()
|
|
if db_pass:
|
|
env["PGPASSWORD"] = db_pass
|
|
result = subprocess.run(
|
|
[
|
|
PSQL_BIN,
|
|
"-h", host,
|
|
"-p", port,
|
|
"-U", db_user,
|
|
"-d", db_name,
|
|
"-f", str(tmpfile),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
print(result.stdout)
|
|
print(f"\n✅ Loaded {len(samples)} samples into learning_corpus (task_type={TASK_TYPE})")
|
|
print("🎯 Next: python3 scripts/manual_trigger.py --general --force")
|
|
else:
|
|
print("❌ Error loading data:")
|
|
print(result.stderr)
|
|
sys.exit(1)
|
|
finally:
|
|
tmpfile.unlink(missing_ok=True)
|