tools/import-sqlite-dump.py - master - grigorin/ITLeaderTrack

tools/import-sqlite-dump.py
188 строк9 KB
okv
COREBS-2604 Актуализация GIGACODE.md и импорта под персональные оценки
23 июл 2026, 10:22
23 июл 2026, 10:224fc2afb
#!/usr/bin/env python3
"""Перенос данных из исторической SQLite-БД в PostgreSQL.

Одноразовый инструмент для наполнения локальной базы: до перехода на PostgreSQL проект
работал на файловой SQLite, и накопленный справочник компетенций (узлы, ресурсы, группы
вопросов, шаблоны интервью) удобнее перенести, чем набивать заново.

Особенности, ради которых это скрипт, а не `pgloader`:

* порядок вставки задан явно — иначе внешние ключи отвергнут строки;
* значения `id` переносятся как есть, поэтому после вставки нужно подвинуть sequence,
  иначе следующий INSERT наткнётся на занятый идентификатор;
* SQLite хранит булевы значения как 0/1 и всё подряд как TEXT — колонки приводятся
  к типам целевой схемы по метаданным `information_schema`;
* `interview_sessions.version` в SQLite нет: колонка появилась вместе с оптимистичной
  блокировкой, у неё есть DEFAULT, и в списке переносимых колонок она отсутствует.

Запуск:

    python3 tools/import-sqlite-dump.py <путь-к-sqlite.db> [--truncate]

Параметры подключения берутся из окружения (POSTGRES_HOST/PORT/DB/USER/PASSWORD/SCHEMA),
то есть достаточно `set -a && . ./.env && set +a`.
"""

import argparse
import os
import sqlite3
import subprocess
import sys

# Порядок важен: родители раньше детей, иначе FK отвергнет вставку.
TABLES = [
    "skill_nodes",
    "resources",
    "assessments",
    "question_groups",
    "interview_templates",
    "interview_scale_items",
    "interview_knowledge_levels",
    "interview_directions",
    "interview_questions",
    "interview_result_ranges",
    "interview_sessions",
    "interview_answers",
    "interview_direction_scores",
    "interview_dev_plan_items",
]


def psql(sql: str, *, capture: bool = True) -> str:
    """Выполняет SQL в контейнере. Используется docker exec, чтобы не требовать psql на хосте."""
    container = os.environ.get("PG_CONTAINER", "leadertrack-postgres")
    result = subprocess.run(
        [
            "docker", "exec", "-i", container,
            "psql", "-U", os.environ.get("POSTGRES_USER", "leadertrack"),
            "-d", os.environ.get("POSTGRES_DB", "leadertrack"),
            "-v", "ON_ERROR_STOP=1", "-t", "-A", "-c", sql,
        ],
        capture_output=capture,
        text=True,
    )
    if result.returncode != 0:
        sys.exit(f"psql упал:\n{result.stderr}")
    return result.stdout.strip() if capture else ""


def pg_columns(schema: str, table: str) -> dict[str, str]:
    raw = psql(
        "select column_name || '|' || data_type from information_schema.columns "
        f"where table_schema = '{schema}' and table_name = '{table}' order by ordinal_position"
    )
    return dict(line.split("|", 1) for line in raw.splitlines() if line)


# Колонки, которых нет в исторической SQLite-БД: их значения подставляются при переносе.
# assessments стали персональными (миграция 004), а в старой схеме владельца не было —
# оценки достаются пользователю из --assessments-owner.
SYNTHETIC = {
    "assessments": {"username": None, "id": None},
}


def literal(value, data_type: str) -> str:
    """SQL-литерал с приведением типа: SQLite отдаёт почти всё как TEXT."""
    if value is None or value == "":
        return "NULL"
    if data_type == "boolean":
        return "true" if str(value) in ("1", "true", "True") else "false"
    if data_type in ("integer", "bigint", "smallint", "double precision", "numeric", "real"):
        return str(value)
    escaped = str(value).replace("'", "''")
    return f"'{escaped}'"


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("sqlite_path")
    parser.add_argument(
        "--assessments-owner",
        default="admin",
        help="кому достанутся самооценки: в старой схеме владельца не было (по умолчанию admin)",
    )
    parser.add_argument(
        "--truncate",
        action="store_true",
        help="очистить целевые таблицы перед вставкой (иначе конфликт по существующим id)",
    )
    args = parser.parse_args()

    schema = os.environ.get("POSTGRES_SCHEMA", "leadtrack")
    src = sqlite3.connect(args.sqlite_path)
    src.row_factory = sqlite3.Row

    if args.truncate:
        # CASCADE — из-за внешних ключей между сессиями, ответами и баллами.
        psql(f"truncate {', '.join(f'{schema}.{t}' for t in TABLES)} cascade")
        print("целевые таблицы очищены")

    total = 0
    for table in TABLES:
        rows = src.execute(f'select * from "{table}"').fetchall()
        if not rows:
            print(f"{table}: пусто, пропуск")
            continue

        target_types = pg_columns(schema, table)
        columns = [c for c in rows[0].keys() if c in target_types]

        # Досочиняем колонки, появившиеся после того, как снималась SQLite-БД.
        synthetic = {}
        if table == "assessments":
            if "username" in target_types:
                synthetic["username"] = lambda _row, _i: f"'{args.assessments_owner}'"
            if "id" in target_types:
                synthetic["id"] = lambda _row, i: str(i + 1)
        columns += [c for c in synthetic if c not in columns]
        skipped = [c for c in rows[0].keys() if c not in target_types]
        if skipped:
            print(f"{table}: колонки нет в целевой схеме, пропускаю: {skipped}")

        def cell(row, column, index):
            if column in synthetic:
                return synthetic[column](row, index)
            return literal(row[column], target_types[column])

        values = ",\n".join(
            "(" + ", ".join(cell(row, c, i) for c in columns) + ")"
            for i, row in enumerate(rows)
        )
        column_list = ", ".join(f'"{c}"' for c in columns)
        psql(f'insert into {schema}."{table}" ({column_list}) values\n{values}')
        print(f"{table}: перенесено {len(rows)}")
        total += len(rows)

    # Явные id не двигают sequence — следующий INSERT иначе упрётся в занятый идентификатор.
    print("\nсинхронизация последовательностей:")
    for table in TABLES:
        # У части таблиц (assessments) первичный ключ не `id`, и sequence там нет вовсе.
        has_id = psql(
            "select 1 from information_schema.columns "
            f"where table_schema = '{schema}' and table_name = '{table}' and column_name = 'id'"
        )
        if not has_id:
            print(f"  {table}: без колонки id, пропуск")
            continue
        # pg_get_serial_sequence находит только последовательности, ПРИВЯЗАННЫЕ к колонке
        # (SERIAL/IDENTITY). assessments_id_seq создана отдельным CREATE SEQUENCE и такой
        # привязки не имеет — её приходится искать по имени, иначе счётчик остаётся на 1
        # и первая же вставка через приложение падает на занятом идентификаторе.
        seq = psql(f"select pg_get_serial_sequence('{schema}.{table}', 'id')")
        if not seq:
            seq = psql(
                "select schemaname || '.' || sequencename from pg_sequences "
                f"where schemaname = '{schema}' and sequencename = '{table}_id_seq'"
            )
        if seq:
            psql(
                f"select setval('{seq}', coalesce((select max(id) from {schema}.{table}), 0) + 1, false)"
            )
            print(f"  {table}: ok")

    print(f"\nитого перенесено строк: {total}")


if __name__ == "__main__":
    main()