68 lines
3.7 KiB
Python
68 lines
3.7 KiB
Python
"""Copy this project's local learning records into an empty PostgreSQL schema.
|
|
|
|
The SQLite original and a consistent backup remain intact. Connection settings
|
|
come from the owning Node launcher, never command-line password arguments.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
import sqlite3
|
|
|
|
from apps.server.db import PostgresStore
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
TABLES = ("operators", "auth_attempts", "auth_sessions", "learning_tasks", "observations",
|
|
"learning_steps", "candidates", "validations", "quality_observations", "artifacts", "discovery_runs", "reading_runs", "analysis_runs", "business_briefs", "analysis_reviews")
|
|
# These additive tables are optional in older SQLite installations.
|
|
# Missing older core tables still indicate an incomplete/corrupt source.
|
|
OPTIONAL_SOURCE_TABLES = frozenset({"quality_observations", "discovery_runs", "reading_runs", "analysis_runs", "business_briefs", "analysis_reviews"})
|
|
|
|
|
|
def main() -> None:
|
|
database_url = os.environ.get("ASSISTANT_DATABASE_URL", "")
|
|
if not database_url:
|
|
raise SystemExit("PostgreSQL is not configured. Run pnpm db:up first.")
|
|
home = Path(os.environ.get("ASSISTANT_LEARNING_HOME", ROOT / ".runtime" / "learning")).resolve()
|
|
original = home / "learning.sqlite"
|
|
if not original.is_file():
|
|
print(json.dumps({"result": "no_sqlite_source", "message": "Fresh PostgreSQL installation; no local records to copy."}))
|
|
return
|
|
target = PostgresStore(home, database_url, schema=os.environ.get("ASSISTANT_DATABASE_SCHEMA", "assistant_learning"))
|
|
backups = home / "backups"
|
|
backups.mkdir(exist_ok=True)
|
|
backup = backups / ("learning-before-pg-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") + ".sqlite")
|
|
with sqlite3.connect(original.as_uri() + "?mode=ro", uri=True) as live:
|
|
with sqlite3.connect(backup) as copy:
|
|
live.backup(copy)
|
|
# The exact backup is the migration source, so counts and copied records
|
|
# cannot diverge from it even if the original is opened elsewhere later.
|
|
with sqlite3.connect(backup.as_uri() + "?mode=ro", uri=True) as source:
|
|
source.row_factory = sqlite3.Row
|
|
counts = {}
|
|
with target.connect(immediate=True) as destination:
|
|
if any(destination.execute(f"SELECT 1 FROM {table} LIMIT 1").fetchone() for table in TABLES):
|
|
raise SystemExit("PostgreSQL target already contains records. Migration stopped without replacing existing data.")
|
|
for table in TABLES:
|
|
columns = [row["name"] for row in source.execute(f"PRAGMA table_info({table})")]
|
|
if not columns and table in OPTIONAL_SOURCE_TABLES:
|
|
counts[table] = 0
|
|
continue
|
|
rows = source.execute(f"SELECT * FROM {table} ORDER BY rowid").fetchall()
|
|
# Names are from this project's fixed local schema, not HTTP input.
|
|
names = ",".join('"' + name.replace('"', '""') + '"' for name in columns)
|
|
parameters = ",".join(f"${i}" for i in range(1, len(columns) + 1))
|
|
for row in rows:
|
|
destination.execute(f"INSERT INTO {table}({names}) VALUES({parameters})", tuple(row))
|
|
counts[table] = len(rows)
|
|
for table, count in counts.items():
|
|
if destination.execute(f"SELECT count(*) AS n FROM {table}").fetchone()["n"] != count:
|
|
raise RuntimeError("Migration count mismatch; transaction rolled back.")
|
|
print(json.dumps({"result": "copied", "counts": counts, "backup": str(backup), "source_preserved": True}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|