scripts/migrate-journal-duckdb-to-sqlite.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Offline DuckDB journal migration with verified, resumable publication.
All source columns/rows are checked, not silently filtered through defaults.
SQLite nulls and stored P&L are preserved; the current reader, tag normalisation
and FTS triggers validate the new database before it becomes public. A failed
archive returns 3 (published/archive-pending), never completed or a fresh start.
"""
from __future__ import annotations
import argparse
import logging
import re
from contextlib import closing
from pathlib import Path
from flinttrade_core.legacy_sqlite_migration import (
ArchivePending,
MigrationRefused,
database_digest,
migrate_database,
rows_digest,
sqlite_connection,
sqlite_value,
)
from flinttrade_core.workspace import workspace_dir
from flinttrade_journal.trade_journal import TradeJournal, _coerce_timestamp
_TABLES = ("journal_entries", "daily_notes", "journal_screenshots")
def _legacy_rows(source: Path | None, target: Path) -> dict[str, tuple[list[str], list[tuple]]]:
if source is None:
return {}
import duckdb # noqa: PLC0415 -- migration only; absence is an error, never success
with duckdb.connect(str(source), read_only=True, config={"threads": 1}) as duck:
tables = {r[0] for r in duck.execute("SHOW TABLES").fetchall()}
if "journal_entries" not in tables or tables - set(_TABLES):
raise MigrationRefused("unsupported or incomplete journal source schema")
# Views and non-main schemas could otherwise hide data from SHOW TABLES.
if duck.execute("SELECT count(*) FROM information_schema.tables WHERE table_schema <> 'main' "
"OR table_type <> 'BASE TABLE'").fetchone()[0]:
raise MigrationRefused("unsupported journal source objects")
result = {}
with closing(sqlite_connection(target, readonly=True)) as conn:
for table in _TABLES:
if table not in tables:
continue
info = conn.execute(f"PRAGMA table_info({table})").fetchall()
columns = [r[1] for r in info]
description = duck.execute(f"DESCRIBE {table}").fetchall()
if {r[0] for r in description} != set(columns):
raise MigrationRefused("journal column set mismatch")
if any(r[1] in {"TIMESTAMP_NS", "TIMESTAMP_S", "TIMESTAMP_MS"} for r in description):
raise MigrationRefused("unsupported journal timestamp precision")
rows = duck.execute(f"SELECT {','.join(columns)} FROM {table}").fetchall()
converted = [tuple(sqlite_value(value, info[i][2]) for i, value in enumerate(row)) for row in rows]
for row in converted:
if any(row[i] is None and (col[3] or col[5]) for i, col in enumerate(info)):
raise MigrationRefused("null journal key or required value")
if table == "journal_entries":
for name in ("entry_time", "exit_time", "created_at", "updated_at"):
value = row[columns.index(name)]
if value is not None and (
_coerce_timestamp(value) is None or re.search(r"[.,]\d{7,}", value)
):
raise MigrationRefused("journal timestamp cannot be represented by current reader")
result[table] = (columns, converted)
return result
def _build(source: Path | None, target: Path) -> None:
with TradeJournal(target) as journal:
journal.initialise()
tables = _legacy_rows(source, target)
with closing(sqlite_connection(target)) as conn:
conn.execute("PRAGMA synchronous=FULL")
with conn:
for table, (columns, rows) in tables.items():
conn.executemany(f"INSERT INTO {table} ({','.join(columns)}) VALUES "
f"({','.join('?' for _ in columns)})", rows)
# Preserve stored values (including NULL P&L and tags); add_entry would
# recompute P&L and normalise NULL tags to a different stored value.
with closing(sqlite_connection(target)) as conn, TradeJournal(target) as journal:
with conn:
for (entry_id,) in conn.execute("SELECT id FROM journal_entries").fetchall():
entry = journal.get_entry(entry_id)
if entry is None:
raise MigrationRefused("journal reader lost an entry")
conn.executemany("INSERT INTO journal_tags VALUES (?, ?)",
[(entry_id, tag) for tag in set(entry.tags) if tag])
def _validate(source: Path | None, target: Path) -> None:
expected = _legacy_rows(source, target)
database_digest(target) # full SQLite integrity and foreign-key checks
with closing(sqlite_connection(target, readonly=True)) as conn, TradeJournal(target) as journal:
for table in _TABLES:
columns, rows = expected.get(table, ([], []))
actual = conn.execute(f"SELECT {','.join(columns) if columns else '*'} FROM {table}").fetchall()
if rows_digest(actual) != rows_digest(rows):
raise MigrationRefused("journal data mismatch")
tag_rows = []
for row in conn.execute("SELECT id,side,setup_quality,execution_quality FROM journal_entries"):
entry = journal.get_entry(row[0])
if entry is None or (entry.side, entry.setup_quality, entry.execution_quality) != tuple(row[1:]):
raise MigrationRefused("journal reader normalises invalid source values")
tag_rows.extend((entry.id, tag) for tag in set(entry.tags) if tag)
if rows_digest(conn.execute("SELECT entry_id,tag FROM journal_tags").fetchall()) != rows_digest(tag_rows):
raise MigrationRefused("journal tag relationships mismatch")
fts = conn.execute("SELECT rowid,id,symbol,notes,tags,strategy FROM journal_fts").fetchall()
expected_fts = conn.execute("SELECT rowid,id,symbol,COALESCE(notes,''),COALESCE(tags,''),"
"COALESCE(strategy,'') FROM journal_entries").fetchall()
if rows_digest(fts) != rows_digest(expected_fts):
raise MigrationRefused("journal search index mismatch")
def migrate(workspace: Path) -> Path:
"""Migrate under exclusive backend maintenance, retaining every source byte."""
return migrate_database(workspace, source_name="journal.db", destination_name="journal.sqlite",
build=_build, validate=_validate)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--workspace", type=Path, default=None)
args = parser.parse_args()
# Library errors can carry row contents. Never emit raw exceptions/tracebacks.
logging.basicConfig(level=logging.CRITICAL)
try:
migrate(args.workspace if args.workspace is not None else workspace_dir())
except ArchivePending:
print("published/archive-pending: destination retained; rerun to resume archival")
return 3
except Exception as exc:
print(f"migration refused ({type(exc).__name__}); originals retained; reconcile source/dependencies")
return 2
print("journal migration complete; destination and archive verified")
return 0
if __name__ == "__main__":
raise SystemExit(main())