""" Importa dados do SQLite do app Flask para o banco configurado no Django (MySQL). Preenche colunas extras do modelo User (is_superuser, is_staff, is_active, last_login). """ from __future__ import annotations import sqlite3 from pathlib import Path from django.core.management.base import BaseCommand from django.db import connection, transaction DEFAULT_SQLITE = ( Path(__file__).resolve().parents[3].parent / "sysRevisao" / "instance" / "sysrevisao.db" ) TABLE_ORDER = [ "users", "sources", "reviews", "review_coauthors", "review_sources", "questions", "hypotheses", "selection_criteria", "keywords", "search_sessions", "articles", "quality_questions", "quality_answers", "quality_assessments", "data_extraction_fields", "data_extraction_lookups", "data_extractions", "extraction_select_values", "invites", ] # Colunas legadas (Flask) → Django M2M COLUMN_ALIASES: dict[str, dict[str, str]] = { "extraction_select_values": { "extraction_id": "dataextraction_id", "lookup_id": "dataextractionlookup_id", }, } USER_DEFAULTS = { "last_login": None, "is_superuser": False, "is_staff": False, "is_active": True, } def _normalize_value(table: str, column: str, value): if table == "reviews" and column == "publish_status": if not value or value in ("U", "P"): return value or "U" return "U" return value def _sqlite_columns(conn: sqlite3.Connection, table: str) -> list[str]: cur = conn.execute(f"PRAGMA table_info([{table}])") return [row[1] for row in cur.fetchall()] def _target_columns(cursor, table: str) -> list[str]: vendor = connection.vendor if vendor == "mysql": cursor.execute( """ SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s ORDER BY ORDINAL_POSITION """, [table], ) elif vendor == "sqlite": cursor.execute(f"PRAGMA table_info([{table}])") return [row[1] for row in cursor.fetchall()] else: cursor.execute( """ SELECT column_name FROM information_schema.columns WHERE table_name = %s ORDER BY ordinal_position """, [table], ) return [row[0] for row in cursor.fetchall()] def _table_exists_sqlite(conn: sqlite3.Connection, table: str) -> bool: row = conn.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,), ).fetchone() return row is not None class Command(BaseCommand): help = "Importa dados do SQLite legado (Flask) preservando IDs." def add_arguments(self, parser): parser.add_argument( "--sqlite", type=Path, default=DEFAULT_SQLITE, help=f"Caminho do SQLite (padrão: {DEFAULT_SQLITE})", ) parser.add_argument( "--force", action="store_true", help="Substitui dados existentes nas tabelas importadas.", ) parser.add_argument( "--dry-run", action="store_true", help="Apenas exibe contagens.", ) def handle(self, *args, **options): sqlite_path: Path = options["sqlite"] if not sqlite_path.is_file(): self.stderr.write(self.style.ERROR(f"SQLite não encontrado: {sqlite_path}")) return src = sqlite3.connect(sqlite_path) src.row_factory = sqlite3.Row self.stdout.write(f"Origem: {sqlite_path}") for table in TABLE_ORDER: if _table_exists_sqlite(src, table): n = src.execute(f"SELECT COUNT(*) FROM [{table}]").fetchone()[0] if n: self.stdout.write(f" SQLite {table}: {n}") if options["dry_run"]: src.close() return vendor = connection.vendor with transaction.atomic(): with connection.cursor() as cursor: if vendor == "mysql": cursor.execute("SET FOREIGN_KEY_CHECKS = 0") total = 0 for table in TABLE_ORDER: if not _table_exists_sqlite(src, table): continue tgt_cols = _target_columns(cursor, table) if not tgt_cols: self.stdout.write(self.style.WARNING(f" {table}: ausente no destino")) continue src_cols = _sqlite_columns(src, table) aliases = COLUMN_ALIASES.get(table, {}) insert_cols: list[str] = [] src_to_tgt: list[tuple[str, str]] = [] for tgt_col in tgt_cols: if tgt_col in src_cols: insert_cols.append(tgt_col) src_to_tgt.append((tgt_col, tgt_col)) elif tgt_col in aliases.values(): continue elif tgt_col in USER_DEFAULTS: insert_cols.append(tgt_col) src_to_tgt.append((tgt_col, tgt_col)) for src_col, tgt_col in aliases.items(): if src_col in src_cols and tgt_col in tgt_cols and tgt_col not in insert_cols: insert_cols.append(tgt_col) src_to_tgt.append((src_col, tgt_col)) if not insert_cols: continue rows = [dict(r) for r in src.execute(f"SELECT * FROM [{table}]").fetchall()] if not rows: continue if options["force"]: cursor.execute(f"DELETE FROM `{table}`") placeholders = ", ".join(["%s"] * len(insert_cols)) cols_sql = ", ".join(f"`{c}`" for c in insert_cols) sql = f"INSERT INTO `{table}` ({cols_sql}) VALUES ({placeholders})" seen_usernames: set[str] = set() payload = [] for row in rows: item = [] for src_col, tgt_col in src_to_tgt: col = tgt_col if src_col in row.keys(): val = _normalize_value(table, col, row[src_col]) if table == "users" and col == "username" and val: base = val candidate = base n = 1 while candidate in seen_usernames: candidate = f"{base}_{n}" n += 1 seen_usernames.add(candidate) val = candidate item.append(val) elif tgt_col in USER_DEFAULTS: item.append(USER_DEFAULTS[tgt_col]) else: item.append(None) payload.append(tuple(item)) cursor.executemany(sql, payload) if "id" in insert_cols and payload: max_id = max( r[insert_cols.index("id")] for r in payload if r[insert_cols.index("id")] is not None ) if vendor == "mysql" and max_id: cursor.execute( f"ALTER TABLE `{table}` AUTO_INCREMENT = %s", [max_id + 1], ) self.stdout.write(self.style.SUCCESS(f" {table}: {len(payload)} linha(s)")) total += len(payload) if vendor == "mysql": cursor.execute("SET FOREIGN_KEY_CHECKS = 1") src.close() self.stdout.write(self.style.SUCCESS(f"Importação concluída ({total} linhas)."))