mirror of
https://codeberg.org/PostERG/xamxam.git
synced 2026-06-25 16:19:19 +02:00
222 lines
9.0 KiB
Python
222 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate a clean schema.sql from the fully-migrated local database."""
|
|
|
|
import sqlite3
|
|
import re
|
|
import sys
|
|
|
|
DB_PATH = sys.argv[1] if len(sys.argv) > 1 else 'app/storage/xamxam.db'
|
|
|
|
db = sqlite3.connect(DB_PATH)
|
|
|
|
# Get all objects
|
|
tables = db.execute(
|
|
"SELECT name, sql FROM sqlite_master WHERE type='table' "
|
|
"AND name NOT LIKE 'sqlite_%' AND name != '_migrations' ORDER BY name"
|
|
).fetchall()
|
|
|
|
indexes = db.execute(
|
|
"SELECT sql FROM sqlite_master WHERE type='index' "
|
|
"AND name NOT LIKE 'sqlite_autoindex%' AND sql IS NOT NULL ORDER BY name"
|
|
).fetchall()
|
|
|
|
views = db.execute(
|
|
"SELECT sql FROM sqlite_master WHERE type='view' ORDER BY name"
|
|
).fetchall()
|
|
|
|
triggers = db.execute(
|
|
"SELECT sql FROM sqlite_master WHERE type='trigger' ORDER BY name"
|
|
).fetchall()
|
|
|
|
print("-- ============================================================================")
|
|
print("-- XAMXAM Database Schema — complete, fully migrated")
|
|
print(f"-- Generated from local database on {db.execute('SELECT date()').fetchone()[0]}")
|
|
print("-- All 28 migrations merged into this single file.")
|
|
print("-- ============================================================================")
|
|
print()
|
|
|
|
# We define the ideal CREATE TABLE statements by querying PRAGMA table_info
|
|
# and reconstructing clean DDL, because SQLite stores ALTER TABLE artifacts.
|
|
|
|
def build_clean_table_ddl(name):
|
|
"""Build a clean CREATE TABLE statement from PRAGMA table_info."""
|
|
cols = db.execute(f"PRAGMA table_info('{name}')").fetchall()
|
|
pk_cols = [c[1] for c in cols if c[5] > 0]
|
|
|
|
lines = [f"CREATE TABLE IF NOT EXISTS {name} ("]
|
|
col_defs = []
|
|
for cid, col_name, col_type, not_null, default, is_pk in cols:
|
|
part = f" {col_name} {col_type}"
|
|
if is_pk:
|
|
if len(pk_cols) == 1:
|
|
part += " PRIMARY KEY"
|
|
if col_type.upper() == 'INTEGER':
|
|
part += " AUTOINCREMENT"
|
|
if not_null:
|
|
part += " NOT NULL"
|
|
if default is not None:
|
|
if default.upper() in ('CURRENT_TIMESTAMP', 'CURRENT_TIME', 'CURRENT_DATE'):
|
|
part += f" DEFAULT {default}"
|
|
elif any(default.lower().startswith(fn + '(') for fn in ('datetime', 'date', 'time', 'strftime')):
|
|
part += f" DEFAULT ({default})"
|
|
elif default.startswith("'") or default.startswith('"'):
|
|
part += f" DEFAULT {default}"
|
|
else:
|
|
part += f" DEFAULT {default}"
|
|
col_defs.append(part)
|
|
|
|
# Multi-column PK
|
|
if len(pk_cols) > 1:
|
|
col_defs.append(f" PRIMARY KEY ({', '.join(pk_cols)})")
|
|
|
|
# Foreign keys
|
|
fks = db.execute(f"PRAGMA foreign_key_list('{name}')").fetchall()
|
|
for fk in fks:
|
|
fk_id, seq, ref_table, from_col, to_col, on_update, on_delete, match = fk
|
|
if seq == 0:
|
|
col_defs.append(
|
|
f" FOREIGN KEY ({from_col}) REFERENCES {ref_table}({to_col})"
|
|
+ (f" ON DELETE {on_delete}" if on_delete and on_delete != 'NO ACTION' else "")
|
|
)
|
|
|
|
lines.append(",\n".join(col_defs))
|
|
lines.append(");")
|
|
|
|
# Add CHECK constraints if any from original DDL
|
|
orig = db.execute(
|
|
"SELECT sql FROM sqlite_master WHERE type='table' AND name=?", (name,)
|
|
).fetchone()
|
|
if orig and orig[0]:
|
|
for m in re.finditer(r'CHECK\s*\([^)]+\)', orig[0]):
|
|
constraint = m.group(0)
|
|
# INSERT OR IGNORE into site_settings uses ON CONFLICT — keep that pattern
|
|
pass # CHECK is already in the column definition from PRAGMA
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# Sort tables in dependency order (rough)
|
|
table_order = [
|
|
'orientations', 'ap_programs', 'finality_types', 'languages',
|
|
'format_types', 'tags', 'access_types', 'license_types',
|
|
'authors', 'supervisors',
|
|
'theses',
|
|
'thesis_authors', 'thesis_supervisors', 'thesis_languages',
|
|
'thesis_formats', 'thesis_tags', 'thesis_files',
|
|
'site_settings', 'system_cache', 'pages', 'share_links',
|
|
'smtp_settings', 'apropos_contents',
|
|
'file_access_requests', 'file_access_tokens', 'file_access_sessions',
|
|
'file_access_audit', 'form_help_blocks', 'admin_audit_log',
|
|
'peertube_settings', 'audit_log',
|
|
]
|
|
|
|
seen = set()
|
|
for name in table_order:
|
|
if name in seen:
|
|
continue
|
|
seen.add(name)
|
|
try:
|
|
ddl = build_clean_table_ddl(name)
|
|
print(ddl)
|
|
print()
|
|
except Exception as e:
|
|
print(f"-- ERROR building DDL for {name}: {e}", file=sys.stderr)
|
|
|
|
print("-- ============================================================================")
|
|
print("-- INDEXES")
|
|
print("-- ============================================================================")
|
|
print()
|
|
|
|
for (sql,) in indexes:
|
|
clean = sql.strip()
|
|
if not clean.endswith(';'):
|
|
clean += ';'
|
|
# Make idempotent — SQLite doesn't store IF NOT EXISTS in sqlite_master.sql
|
|
if clean.upper().startswith('CREATE INDEX ') and 'IF NOT EXISTS' not in clean.upper():
|
|
clean = clean.replace('CREATE INDEX ', 'CREATE INDEX IF NOT EXISTS ', 1)
|
|
if clean.upper().startswith('CREATE UNIQUE INDEX ') and 'IF NOT EXISTS' not in clean.upper():
|
|
clean = clean.replace('CREATE UNIQUE INDEX ', 'CREATE UNIQUE INDEX IF NOT EXISTS ', 1)
|
|
print(clean)
|
|
print()
|
|
|
|
print("-- ============================================================================")
|
|
print("-- VIEWS")
|
|
print("-- ============================================================================")
|
|
print()
|
|
|
|
for (sql,) in views:
|
|
clean = sql.strip()
|
|
if not clean.endswith(';'):
|
|
clean += ';'
|
|
# Make idempotent — SQLite doesn't store IF NOT EXISTS in sqlite_master.sql
|
|
if clean.upper().startswith('CREATE VIEW ') and 'IF NOT EXISTS' not in clean.upper():
|
|
clean = clean.replace('CREATE VIEW ', 'CREATE VIEW IF NOT EXISTS ', 1)
|
|
print(clean)
|
|
print()
|
|
|
|
print("-- ============================================================================")
|
|
print("-- TRIGGERS")
|
|
print("-- ============================================================================")
|
|
print()
|
|
|
|
for (sql,) in triggers:
|
|
clean = sql.strip()
|
|
if not clean.endswith(';'):
|
|
clean += ';'
|
|
# Make idempotent — SQLite doesn't store IF NOT EXISTS in sqlite_master.sql
|
|
if clean.upper().startswith('CREATE TRIGGER ') and 'IF NOT EXISTS' not in clean.upper():
|
|
clean = clean.replace('CREATE TRIGGER ', 'CREATE TRIGGER IF NOT EXISTS ', 1)
|
|
print(clean)
|
|
print()
|
|
|
|
print("-- ============================================================================")
|
|
print("-- SEED DATA (reference data + initial settings)")
|
|
print("-- ============================================================================")
|
|
print()
|
|
|
|
# Extract seed data from the local database
|
|
# (table, query, explicit_column_names)
|
|
seed_tables = [
|
|
('orientations', "SELECT name FROM orientations", ['name']),
|
|
('ap_programs', "SELECT name, code FROM ap_programs", ['name', 'code']),
|
|
('finality_types', "SELECT name FROM finality_types", ['name']),
|
|
('languages', "SELECT name FROM languages WHERE deleted_at IS NULL", ['name']),
|
|
('format_types', "SELECT name, sort_order FROM format_types", ['name', 'sort_order']),
|
|
('access_types', "SELECT name, description FROM access_types", ['name', 'description']),
|
|
('license_types', "SELECT name FROM license_types", ['name']),
|
|
('site_settings', "SELECT key, value FROM site_settings WHERE key IN ('access_type_interdit_enabled', 'access_type_interne_enabled', 'access_type_libre_enabled', 'objet_these_enabled', 'objet_frart_enabled', 'restricted_files_enabled', 'peertube_upload_enabled')", ['key', 'value']),
|
|
('pages', "SELECT slug, title, content, is_published FROM pages WHERE slug IN ('charte', 'about', 'licenses')", ['slug', 'title', 'content', 'is_published']),
|
|
('form_help_blocks', "SELECT key, name, content, enabled, sort_order FROM form_help_blocks", ['key', 'name', 'content', 'enabled', 'sort_order']),
|
|
('apropos_contents', "SELECT key, value FROM apropos_contents WHERE key = 'contacts'", ['key', 'value']),
|
|
]
|
|
|
|
for table, query, col_names in seed_tables:
|
|
rows = db.execute(query).fetchall()
|
|
if not rows:
|
|
continue
|
|
|
|
for row in rows:
|
|
vals = []
|
|
for i, val in enumerate(row):
|
|
if val is None:
|
|
vals.append('NULL')
|
|
elif isinstance(val, bool):
|
|
vals.append('1' if val else '0')
|
|
elif isinstance(val, (int, float)):
|
|
vals.append(str(val))
|
|
else:
|
|
escaped = str(val).replace("'", "''")
|
|
vals.append(f"'{escaped}'")
|
|
|
|
print(f"INSERT OR IGNORE INTO {table} ({', '.join(col_names)}) VALUES ({', '.join(vals)});")
|
|
print()
|
|
|
|
# Singleton table inserts
|
|
print("-- Singleton table placeholders")
|
|
print("INSERT OR IGNORE INTO smtp_settings (id) VALUES (1);")
|
|
print("INSERT OR IGNORE INTO peertube_settings (id) VALUES (1);")
|
|
print()
|
|
print("-- ============================================================================")
|
|
print("-- END OF SCHEMA")
|
|
print("-- ============================================================================")
|