#!/usr/local/bin/perl
#
# import_spz.pl — интерактивный импорт данных из JSON-выгрузки в MySQL
#
# Читает файлы <table>.json из директории экспорта (export_spz.pl),
# сравнивает структуру источника и цели, импортирует только общие колонки.
# Целевая БД может быть более новой версии — лишние колонки сохраняют DEFAULT.
# После импорта автоматически верифицирует корректность данных.
#
# Использование:
#   import_spz.pl                     — интерактивный режим
#   import_spz.pl --all               — импортировать все таблицы
#   import_spz.pl -t spz_rules        — только одна таблица
#   import_spz.pl --dry-run           — показать план без изменений
#   import_spz.pl --truncate          — очистить таблицу перед импортом
#   import_spz.pl --replace           — REPLACE INTO вместо INSERT IGNORE
#   import_spz.pl --no-verify         — пропустить верификацию после импорта
#   import_spz.pl --verify-only       — только верификация, без импорта
#   import_spz.pl --host 127.0.0.1 --port 13306 --user root --pass SpAmAnEt_dev
#   import_spz.pl --help
#

use strict;
use warnings;
use DBI;
use JSON::XS;
use Digest::MD5  qw(md5_hex);
use POSIX        qw(strftime);
use Getopt::Long qw(:config no_ignore_case);
use Encode       qw(encode decode);
use Scalar::Util qw(looks_like_number);

# ── Параметры по умолчанию ────────────────────────────────────────────────────

my $src_dir    = './backup/spzconfig';
my $db_host    = '127.0.0.1';
my $db_port    = 13306;
my $db_user    = 'root';
my $db_pass    = 'SpAmAnEt_dev';
my $db_name    = '';          # пусто = из JSON (multi-db режим)
my $batch_size = 200;         # строк в одной транзакции
my $log_file   = '/tmp/import_spz.log';

my $opt_all         = 0;
my $opt_dry         = 0;
my $opt_truncate    = 0;
my $opt_replace     = 0;
my $opt_no_verify   = 0;
my $opt_verify_only = 0;
my $opt_no_triggers = 0;
my $opt_help        = 0;
my $opt_table       = '';
my $opt_verbose     = 0;
my $opt_extra_db    = 0;   # --extra-db: также импортировать extra_databases из индекса

GetOptions(
    'dir|d=s'     => \$src_dir,
    'host|h=s'    => \$db_host,
    'port|P=i'    => \$db_port,
    'user|u=s'    => \$db_user,
    'pass|p=s'    => \$db_pass,
    'database=s'  => \$db_name,
    'all|a'       => \$opt_all,
    'dry-run|n'   => \$opt_dry,
    'truncate'    => \$opt_truncate,
    'replace'     => \$opt_replace,
    'no-verify'   => \$opt_no_verify,
    'verify-only' => \$opt_verify_only,
    'no-triggers' => \$opt_no_triggers,
    'extra-db'    => \$opt_extra_db,
    'table|t=s'   => \$opt_table,
    'verbose|v'   => \$opt_verbose,
    'batch=i'     => \$batch_size,
    'log=s'       => \$log_file,
    'help'        => \$opt_help,
) or do { usage(); exit 1; };

# --all автоматически включает импорт extra_databases
$opt_extra_db = 1 if $opt_all && !$opt_table;

usage() and exit 0 if $opt_help;

# ── Лог ───────────────────────────────────────────────────────────────────────

open(my $LOG, '>>', $log_file) or die "Не удалось открыть лог $log_file: $!\n";
$LOG->autoflush(1);

sub log_msg {
    my ($level, $msg) = @_;
    $msg =~ s/\n/_/g;
    my $line = strftime('%Y-%m-%d %H:%M:%S', localtime) . " [$level] $msg\n";
    print $LOG $line;
    print $line if $opt_verbose || $level ne 'DEBUG';
}
sub info  { log_msg('INFO',  shift) }
sub warn_ { log_msg('WARN',  shift) }
sub err_  { log_msg('ERROR', shift) }
sub debug { log_msg('DEBUG', shift) if $opt_verbose }

# ── Читаем индекс экспорта ────────────────────────────────────────────────────

my $index_file = "$src_dir/export_index.json";
die "Не найден файл индекса: $index_file\n" unless -f $index_file;

my $json       = JSON::XS->new->utf8->pretty;
my $json_canon = JSON::XS->new->utf8->canonical;
my $index      = $json->decode(slurp($index_file));

my @all_tables = map { $_->{table} } @{ $index->{tables} };
info("Индекс: $index->{exported_at}, таблиц=" . scalar(@all_tables));

# ── Подключение к целевой БД ──────────────────────────────────────────────────

my $dbh = DBI->connect(
    "DBI:mysql:;host=$db_host;port=$db_port",
    $db_user, $db_pass,
    { RaiseError => 0, PrintError => 0, AutoCommit => 1 }
) or die "Не удалось подключиться к БД ${db_host}:${db_port}: $DBI::errstr\n";

$dbh->do("SET NAMES utf8mb4");
$dbh->do("SET SESSION sql_mode = ''");
$dbh->do("SET log_bin_trust_function_creators = 1");

info("Подключён: ${db_host}:${db_port} (${db_user})");

# Получаем список БД в целевом сервере
my %target_dbs = map { $_ => 1 }
    @{ $dbh->selectcol_arrayref("SHOW DATABASES") // [] };

# ── Выбор таблиц ─────────────────────────────────────────────────────────────

my @selected;

if ($opt_table) {
    @selected = ($opt_table);
} elsif ($opt_all || $opt_verify_only) {
    @selected = @all_tables;
} else {
    @selected = interactive_select(\@all_tables, $index, $dbh);
    exit 0 unless @selected;
}

# ── Опции импорта ─────────────────────────────────────────────────────────────

unless ($opt_all || $opt_table || $opt_verify_only) {
    ($opt_truncate, $opt_replace) = ask_import_mode() unless $opt_dry;
}

print_separator();
if ($opt_verify_only) {
    info("Режим: только верификация (импорт пропускается)");
} else {
    my $verb = $opt_dry ? '[DRY-RUN] ' : '';
    info("${verb}Режим: " .
        ($opt_replace  ? 'REPLACE INTO' : 'INSERT IGNORE') .
        ($opt_truncate ? ' + TRUNCATE перед импортом' : ''));
}
info("Таблиц выбрано: " . scalar @selected);
print_separator();

# ── Основной цикл ─────────────────────────────────────────────────────────────

my ($ok, $skipped, $errors, $verify_errors) = (0, 0, 0, 0);
my $total_rows = 0;

$dbh->do("SET foreign_key_checks = 0") unless $opt_verify_only || $opt_dry;
$dbh->do("SET unique_checks = 0")      unless $opt_verify_only || $opt_dry;

for my $tname (@selected) {
    my $json_file = "$src_dir/$tname.json";
    unless (-f $json_file) {
        warn_("$tname: файл не найден ($json_file)");
        $skipped++;
        next;
    }

    # Читаем JSON один раз — используем и для импорта и для верификации
    my $data = eval { $json->decode(slurp($json_file)) };
    if ($@) {
        err_("$tname: ошибка разбора JSON: $@");
        $errors++;
        next;
    }

    my $use_db = $db_name || db_for_table($tname, $index);

    # Импорт (если не verify-only)
    unless ($opt_verify_only) {
        my $result = import_table($dbh, $tname, $data, $use_db);
        if ($result->{error}) {
            err_("$tname: ОШИБКА — $result->{error}");
            $errors++;
            next;
        }
        $total_rows += $result->{rows};
        $ok++;
    }

    # Верификация
    unless ($opt_no_verify || $opt_dry) {
        my $v_result = verify_import($dbh, $tname, $use_db,
            $data->{validation}, $data->{rows}, $data->{columns});
        $verify_errors++ if $v_result->{failed} > 0;
    }
}

$dbh->do("SET foreign_key_checks = 1") unless $opt_verify_only || $opt_dry;
$dbh->do("SET unique_checks = 1")      unless $opt_verify_only || $opt_dry;

# ── Дополнительные БД (extra_databases из индекса) ────────────────────────────

if ($opt_extra_db && $index->{extra_databases} && @{ $index->{extra_databases} }) {
    my @extra = @{ $index->{extra_databases} };
    print_separator();
    info("Импорт дополнительных БД: " . scalar(@extra) . " таблиц");

    $dbh->do("SET foreign_key_checks = 0") unless $opt_verify_only || $opt_dry;
    $dbh->do("SET unique_checks = 0")      unless $opt_verify_only || $opt_dry;

    for my $entry (@extra) {
        my $edb    = $entry->{db};
        my $tname  = $entry->{table};
        my $file   = $entry->{file} // "${edb}__${tname}.json";
        my $json_file = "$src_dir/$file";

        unless (-f $json_file) {
            warn_("$edb.$tname: файл не найден ($json_file)");
            $skipped++;
            next;
        }

        unless ($target_dbs{$edb}) {
            warn_("$edb.$tname: БД '$edb' не существует на целевом сервере — пропуск");
            $skipped++;
            next;
        }

        my $data = eval { $json->decode(slurp($json_file)) };
        if ($@) {
            err_("$edb.$tname: ошибка разбора JSON: $@");
            $errors++;
            next;
        }

        # Подменяем имя таблицы в data для корректной работы import_table
        $data->{table} = $tname;

        unless ($opt_verify_only) {
            my $result = import_table($dbh, $tname, $data, $edb);
            if ($result->{error}) {
                err_("$edb.$tname: ОШИБКА — $result->{error}");
                $errors++;
                next;
            }
            $total_rows += $result->{rows};
            $ok++;
        }

        unless ($opt_no_verify || $opt_dry) {
            my $v_result = verify_import($dbh, $tname, $edb,
                $data->{validation}, $data->{rows}, $data->{columns});
            $verify_errors++ if $v_result->{failed} > 0;
        }
    }

    $dbh->do("SET foreign_key_checks = 1") unless $opt_verify_only || $opt_dry;
    $dbh->do("SET unique_checks = 1")      unless $opt_verify_only || $opt_dry;
}

# ── Проверка пользователя spz_web ────────────────────────────────────────────

print_separator();
{
    # 1. Пользователь существует в mysql.user?
    my ($web_exists) = $dbh->selectrow_array(
        "SELECT COUNT(*) FROM mysql.user WHERE User='spz_web'"
    );

    if (!$web_exists) {
        warn_("spz_web: пользователь НЕ НАЙДЕН в БД — запустите init или создайте вручную");
    } else {
        info("spz_web: пользователь EXISTS в mysql.user");

        # 2. Пробуем подключиться с паролем из credentials
        # Ищем spz_web.cnf в нескольких возможных местах
        my $cred_file = '';
        for my $candidate (
            "$src_dir/../../data/credentials/spz_web.cnf",  # ./backup/spzconfig/../../data/credentials/
            "$src_dir/../data/credentials/spz_web.cnf",     # ./backup/../data/credentials/
            './data/credentials/spz_web.cnf',               # относительно CWD
            '/credentials/spz_web.cnf',                     # внутри контейнера
            "$src_dir/../credentials/spz_web.cnf",          # legacy
        ) {
            if (-f $candidate) { $cred_file = $candidate; last; }
        }

        if (-f $cred_file) {
            my $web_pass = '';
            open(my $cf, '<', $cred_file) or do {
                warn_("spz_web: не удалось прочитать $cred_file: $!");
                goto SKIP_WEB_CHECK;
            };
            while (<$cf>) {
                chomp;
                if (/^password=(.*)$/) { $web_pass = $1; last; }
            }
            close $cf;

            if ($web_pass) {
                my $web_dbh = DBI->connect(
                    "DBI:mysql:spz;host=$db_host;port=$db_port",
                    'spz_web', $web_pass,
                    { RaiseError => 0, PrintError => 0, AutoCommit => 1 }
                );
                if ($web_dbh) {
                    info("spz_web: подключение к БД spz — OK (пароль из $cred_file совпадает)");
                    # Проверяем доступ к status
                    my ($sc) = $web_dbh->selectrow_array("SELECT COUNT(*) FROM status.domains");
                    if (defined $sc) {
                        info("spz_web: доступ к status.domains — OK ($sc строк)");
                    } else {
                        warn_("spz_web: нет доступа к status.domains — " . ($web_dbh->errstr // '?'));
                    }
                    $web_dbh->disconnect();
                } else {
                    warn_("spz_web: подключение НЕУДАЧНО — пароль не совпадает? $DBI::errstr");
                    $verify_errors++;
                }
            } else {
                warn_("spz_web: пароль не найден в $cred_file");
            }
        } else {
            warn_("spz_web: файл credentials не найден ($cred_file) — проверка пароля пропущена");
        }
    }
    SKIP_WEB_CHECK:;
}

print_separator();
if ($opt_verify_only) {
    info(sprintf("Верификация: таблиц=%d ошибок_данных=%d пропущено=%d",
        scalar(@selected) - $skipped, $verify_errors, $skipped));
} else {
    info(sprintf("Готово: импортировано=%d пропущено=%d ошибок=%d строк=%d verify_err=%d",
        $ok, $skipped, $errors, $total_rows, $verify_errors));
}

$dbh->disconnect();
close $LOG;
exit(($errors || $verify_errors) ? 1 : 0);


# ═══════════════════════════════════════════════════════════════════════════════
# Импорт
# ═══════════════════════════════════════════════════════════════════════════════

sub import_table {
    my ($dbh, $tname, $data, $use_db) = @_;

    my @src_cols = map { $_->{name} } @{ $data->{columns} };

    unless ($use_db && $target_dbs{$use_db}) {
        return { error => "БД '$use_db' не существует на целевом сервере" };
    }

    my $tbl_exists = $dbh->selectrow_array(
        "SELECT COUNT(*) FROM information_schema.TABLES
         WHERE TABLE_SCHEMA=? AND TABLE_NAME=?", undef, $use_db, $tname
    );
    return { error => "Таблица $use_db.$tname не существует в целевой БД" }
        unless $tbl_exists;

    # Колонки целевой таблицы
    my %tgt_cols = map { $_ => 1 }
        @{ $dbh->selectcol_arrayref("DESCRIBE `$use_db`.`$tname`") // [] };

    my @common_cols = grep { $tgt_cols{$_} } @src_cols;
    my @only_src    = grep { !$tgt_cols{$_} } @src_cols;
    my @only_tgt    = grep { my $c = $_; !grep { $_ eq $c } @src_cols } keys %tgt_cols;

    warn_("$tname: колонки только в источнике (пропускаются): " . join(', ', @only_src))
        if @only_src;
    debug("$tname: новые колонки в целевой БД (DEFAULT): " . join(', ', sort @only_tgt))
        if @only_tgt;

    return { error => "Нет общих колонок между источником и целью" }
        unless @common_cols;

    if ($opt_truncate) {
        $dbh->do("TRUNCATE TABLE `$use_db`.`$tname`")
            or return { error => "TRUNCATE: " . $dbh->errstr };
        debug("$tname: TRUNCATE выполнен");
    }

    # -- Отключаем триггеры на время импорта -----------------------------------
    my @saved_triggers;
    if ($opt_no_triggers && !$opt_dry) {
        @saved_triggers = drop_triggers($dbh, $use_db, $tname);
        debug("$tname: сброшено триггеров: " . scalar(@saved_triggers)) if @saved_triggers;
    }

    my $verb     = $opt_replace ? 'REPLACE' : 'INSERT IGNORE';
    my $cols_sql = join(', ', map { "`$_`" } @common_cols);
    my $vals_sql = join(', ', ('?') x scalar @common_cols);
    my $sql      = "$verb INTO `$use_db`.`$tname` ($cols_sql) VALUES ($vals_sql)";

    # Индексы для быстрого извлечения значений
    my %col_idx; $col_idx{$src_cols[$_]} = $_ for 0..$#src_cols;
    my @src_idx = map { $col_idx{$_} } @common_cols;

    my $rows    = $data->{rows} // [];
    my $total   = scalar @$rows;
    my $t0      = time();

    if ($total == 0) {
        info(sprintf("  %-40s — нет данных", $tname));
        return { rows => 0 };
    }

    info(sprintf("  %-40s %6s строк → %s (%d/%d кол.)%s",
        $tname, commify($total), $use_db,
        scalar @common_cols, scalar @src_cols,
        $opt_dry ? ' [DRY-RUN]' : ''));

    return { rows => $total } if $opt_dry;

    my $sth = $dbh->prepare($sql)
        or return { error => "prepare: " . $dbh->errstr };

    my $imported = 0;
    $dbh->begin_work();

    for my $row (@$rows) {
        my @vals = map { $row->{ $src_cols[$_] } } @src_idx;

        unless ($sth->execute(@vals)) {
            my $e = $sth->errstr;
            $dbh->rollback();
            return { error => "execute строка ${\($imported+1)}: $e" };
        }
        $imported++;

        if ($imported % $batch_size == 0) {
            $dbh->commit();
            $dbh->begin_work();
            printf("\r    прогресс: %d/%d (%.0f%%)",
                $imported, $total, $imported/$total*100) if -t STDOUT;
        }
    }
    $dbh->commit();
    print "\r" . ' ' x 60 . "\r" if -t STDOUT;

    # -- Восстанавливаем триггеры ----------------------------------------------
    if (@saved_triggers) {
        restore_triggers($dbh, \@saved_triggers, $use_db);
        debug("$tname: триггеры восстановлены: " . scalar(@saved_triggers));
    }

    debug(sprintf("    %d строк за %ds", $imported, time()-$t0));
    return { rows => $imported };
}


# ═══════════════════════════════════════════════════════════════════════════════
# Управление триггерами
# ═══════════════════════════════════════════════════════════════════════════════

# Удаляет все триггеры таблицы, возвращает список хешей для последующего
# восстановления: { name, event, timing, statement, definer, sql_mode, charset, collation, db_collation }
sub drop_triggers {
    my ($dbh, $db, $tname) = @_;
    my $rows = $dbh->selectall_arrayref(
        "SELECT TRIGGER_NAME, EVENT_MANIPULATION, ACTION_TIMING,
                ACTION_STATEMENT, DEFINER, SQL_MODE,
                CHARACTER_SET_CLIENT, COLLATION_CONNECTION, DATABASE_COLLATION
         FROM information_schema.TRIGGERS
         WHERE TRIGGER_SCHEMA = ? AND EVENT_OBJECT_TABLE = ?
         ORDER BY ACTION_ORDER",
        { Slice => {} }, $db, $tname
    ) // [];
    for my $t (@$rows) {
        $dbh->do("DROP TRIGGER IF EXISTS `$db`.`$t->{TRIGGER_NAME}`")
            or warn_("drop_triggers: " . $dbh->errstr);
    }
    return @$rows;
}

# Восстанавливает триггеры из сохранённых дефиниций
sub restore_triggers {
    my ($dbh, $triggers, $db) = @_;
    # Убеждаемся что выбрана нужная БД
    $dbh->do("USE `$db`") if $db;
    for my $t (@$triggers) {
        # Устанавливаем оригинальный sql_mode
        $dbh->do("SET sql_mode = " . $dbh->quote($t->{SQL_MODE} // ''));
        my $definer = $t->{DEFINER} // 'root@%';
        $definer = join('@', map { "`$_`" } split /\@/, $definer, 2)
            if $definer =~ /\@/;
        my $sql = sprintf(
            "CREATE DEFINER=%s TRIGGER `%s` %s %s ON `%s` FOR EACH ROW %s",
            $definer,
            $t->{TRIGGER_NAME},
            $t->{ACTION_TIMING},
            $t->{EVENT_MANIPULATION},
            $t->{EVENT_OBJECT_TABLE},
            $t->{ACTION_STATEMENT}
        );
        $dbh->do($sql) or warn_("restore_triggers $t->{TRIGGER_NAME}: " . $dbh->errstr);
        $dbh->do("SET sql_mode = ''");
    }
}

# ═══════════════════════════════════════════════════════════════════════════════
# Верификация импорта
# ═══════════════════════════════════════════════════════════════════════════════

# Проверяет корректность импортированных данных по контрольным параметрам из JSON.
# Возвращает { passed => N, failed => N, skipped => N }
sub verify_import {
    my ($dbh, $tname, $use_db, $src_v, $src_rows, $src_cols) = @_;

    # Карта типов колонок из источника — для точной нормализации чисел
    my %col_type = map { $_->{name} => $_->{type} } @{ $src_cols // [] };

    unless (defined $src_v && !$src_v->{skipped}) {
        info(sprintf("  %-40s [VERIFY] пропуск — контрольные данные не были сохранены при экспорте", $tname));
        return { passed => 0, failed => 0, skipped => 1 };
    }

    my (@passed, @failed, @warnings);

    # ── 1. row_count ──────────────────────────────────────────────────────────
    my ($tgt_count) = $dbh->selectrow_array(
        "SELECT COUNT(*) FROM `$use_db`.`$tname`"
    );
    my $src_count = $src_v->{row_count} // 0;
    _check(\@passed, \@failed, 'row_count',
        $src_count, $tgt_count,
        sub { $_[0] == $_[1] },
        sprintf("%d ожидалось, %d в целевой", $src_count, $tgt_count // '?'));

    # ── 2. CHECKSUM TABLE ─────────────────────────────────────────────────────
    # CHECKSUM TABLE ненадёжен при кросс-серверном переносе: результат зависит
    # от версии MySQL (даже патч-версии), charset/collation таблицы, кол-ва
    # колонок в целевой схеме. Расхождение — предупреждение, не ошибка.
    if (defined $src_v->{checksum}) {
        my $src_ver = $src_v->{mysql_version} // '';
        my ($tgt_ver) = $dbh->selectrow_array("SELECT VERSION()");
        $tgt_ver //= '';

        my $src_full = $src_ver =~ /^(\d+\.\d+\.\d+)/ ? $1 : '';
        my $tgt_full = $tgt_ver =~ /^(\d+\.\d+\.\d+)/ ? $1 : '';

        if ($src_full && $tgt_full && $src_full eq $tgt_full) {
            # Точное совпадение версии — checksum должен совпадать
            my $chk = $dbh->selectrow_hashref("CHECKSUM TABLE `$use_db`.`$tname`");
            my $tgt_chk = defined $chk ? $chk->{Checksum}+0 : undef;
            _check(\@passed, \@failed, 'checksum',
                $src_v->{checksum}, $tgt_chk,
                sub { defined $_[1] && $_[0] == $_[1] },
                sprintf("src=%s tgt=%s", $src_v->{checksum}, $tgt_chk // 'N/A'));
        } else {
            # Разные версии — только информационная проверка
            my $chk = $dbh->selectrow_hashref("CHECKSUM TABLE `$use_db`.`$tname`");
            my $tgt_chk = defined $chk ? $chk->{Checksum}+0 : undef;
            push @passed, { check => 'checksum(advisory)',
                            detail => "src_ver=$src_ver tgt_ver=$tgt_ver" };
            if (!defined $tgt_chk || $src_v->{checksum} != $tgt_chk) {
                push @warnings, sprintf(
                    "checksum отличается (src=%s tgt=%s, src_ver=%s tgt_ver=%s) — " .
                    "норма при разных версиях MySQL или разном charset",
                    $src_v->{checksum}, $tgt_chk // 'N/A', $src_ver, $tgt_ver);
            }
        }
    }

    # ── 3. Первичный ключ: min / max / count_distinct ─────────────────────────
    if (my $pk = $src_v->{pk}) {
        my $cn = $pk->{column};
        my $ps = $dbh->selectrow_hashref(
            "SELECT MIN(`$cn`) AS mn, MAX(`$cn`) AS mx,
                    COUNT(DISTINCT `$cn`) AS cd FROM `$use_db`.`$tname`"
        );
        if ($ps) {
            my $pk_is_num = defined $pk->{min} && looks_like_number($pk->{min});
            my $mn = defined $ps->{mn}
                ? ($pk_is_num && looks_like_number($ps->{mn}) ? $ps->{mn}+0 : $ps->{mn})
                : undef;
            my $mx = defined $ps->{mx}
                ? ($pk_is_num && looks_like_number($ps->{mx}) ? $ps->{mx}+0 : $ps->{mx})
                : undef;
            _check(\@passed, \@failed, "pk($cn).min",
                $pk->{min}, $mn,
                sub {
                    return 1 unless defined $_[0];
                    return 0 unless defined $_[1];
                    no warnings 'numeric';
                    $pk_is_num ? $_[0] == $_[1] : $_[0] eq $_[1];
                },
                sprintf("src=%s tgt=%s", $pk->{min}//'NULL', $mn//'NULL'));
            _check(\@passed, \@failed, "pk($cn).max",
                $pk->{max}, $mx,
                sub {
                    return 1 unless defined $_[0];
                    return 0 unless defined $_[1];
                    no warnings 'numeric';
                    $pk_is_num ? $_[0] == $_[1] : $_[0] eq $_[1];
                },
                sprintf("src=%s tgt=%s", $pk->{max}//'NULL', $mx//'NULL'));
            _check(\@passed, \@failed, "pk($cn).distinct",
                $pk->{count_distinct}, $ps->{cd}+0,
                sub { $_[0] == $_[1] },
                sprintf("src=%d tgt=%d", $pk->{count_distinct}//0, $ps->{cd}//0));
        }
    }

    # ── 4. Числовые колонки: SUM / MIN / MAX ──────────────────────────────────
    if (my $cn_src = $src_v->{col_numeric}) {
        # Один агрегирующий запрос для всех числовых колонок
        my @cols = keys %$cn_src;
        # Проверяем что колонки существуют в целевой таблице
        my %tgt_col_set = map { $_ => 1 }
            @{ $dbh->selectcol_arrayref("DESCRIBE `$use_db`.`$tname`") // [] };
        my @existing = grep { $tgt_col_set{$_} } @cols;

        if (@existing) {
            my $agg_sql = "SELECT " .
                join(', ', map { "SUM(`$_`), MIN(`$_`), MAX(`$_`)" } @existing) .
                " FROM `$use_db`.`$tname`";
            my $agg = $dbh->selectrow_arrayref($agg_sql);
            if ($agg) {
                my $i = 0;
                for my $col (@existing) {
                    my ($tgt_sum, $tgt_min, $tgt_max) =
                        ($agg->[$i], $agg->[$i+1], $agg->[$i+2]);
                    $i += 3;
                    my $s = $cn_src->{$col};
                    for my $stat (qw(sum min max)) {
                        my $sv = $s->{$stat};
                        my $tv = { sum => $tgt_sum, min => $tgt_min, max => $tgt_max }->{$stat};
                        next unless defined $sv;
                        my $tv_num;
                        { no warnings 'numeric'; $tv_num = defined $tv ? $tv+0 : undef; }
                        _check(\@passed, \@failed, "$col.$stat",
                            $sv, $tv_num,
                            sub { defined $_[1] && abs($_[0] - $_[1]) < 0.001 },
                            sprintf("src=%s tgt=%s", $sv, $tv//'NULL'));
                    }
                }
            }
        }

        # Колонки только в источнике (удалены из целевой схемы)
        my @missing = grep { !$tgt_col_set{$_} } @cols;
        push @warnings, "числовые колонки отсутствуют в целевой таблице: " . join(', ', @missing)
            if @missing;
    }

    # ── 5. NULL-счётчики по колонкам ──────────────────────────────────────────
    if (my $cn_nulls = $src_v->{col_nulls}) {
        my %tgt_col_set = map { $_ => 1 }
            @{ $dbh->selectcol_arrayref("DESCRIBE `$use_db`.`$tname`") // [] };
        my @cols = grep { $tgt_col_set{$_} } keys %$cn_nulls;

        if (@cols) {
            my $null_sql = "SELECT " .
                join(', ', map { "SUM(`$_` IS NULL)" } @cols) .
                " FROM `$use_db`.`$tname`";
            my $null_res = $dbh->selectrow_arrayref($null_sql);
            if ($null_res) {
                for my $i (0..$#cols) {
                    my $col      = $cols[$i];
                    my $src_null = $cn_nulls->{$col} // 0;
                    my $tgt_null = $null_res->[$i] // 0;
                    _check(\@passed, \@failed, "$col.nulls",
                        $src_null, $tgt_null+0,
                        sub { $_[0] == $_[1] },
                        sprintf("src=%d tgt=%d", $src_null, $tgt_null));
                }
            }
        }
    }

    # ── 6. sample_hash — MD5 тех же строк, но прочитанных из целевой БД ───────
    # Берём PK-значения первой/средней/последней строки из источника,
    # запрашиваем их из целевой БД, считаем хэш — сравниваем с источником.
    # Это единственный способ проверить реальные данные в целевой БД.
    if (defined $src_v->{sample_hash} && $src_v->{pk} && $src_rows && @$src_rows) {
        my $pk_col = $src_v->{pk}{column};

        # Индексы строк-образцов (те же что при экспорте)
        my @sample_idxs = (0);
        push @sample_idxs, int(@$src_rows / 2) if @$src_rows > 2;
        push @sample_idxs, $#$src_rows          if @$src_rows > 1;

        # PK-значения образцов
        my @pk_vals = map { $src_rows->[$_]{$pk_col} } @sample_idxs;
        my $placeholders = join(', ', ('?') x @pk_vals);

        # Запрашиваем эти строки из целевой БД
        my $sth = $dbh->prepare(
            "SELECT * FROM `$use_db`.`$tname`
             WHERE `$pk_col` IN ($placeholders)
             ORDER BY FIELD(`$pk_col`, $placeholders)"
        );

        if ($sth && $sth->execute(@pk_vals, @pk_vals)) {
            my @tgt_samples;
            # Только колонки из источника — новые колонки целевой БД в хеш не включаем,
            # иначе хеши никогда не совпадут при расширенной схеме.
            my %src_col_set = map { $_->{name} => 1 } @{ $src_cols // [] };

            while (my $row = $sth->fetchrow_hashref()) {
                # Нормализуем точно как export_spz:
                #   - числовой тип И значение выглядит как число → +0
                #   - иначе → строка
                #   - NULL → undef
                # Колонки не из источника — пропускаем.
                my %norm;
                for my $k (keys %$row) {
                    next unless $src_col_set{$k};
                    my $v = $row->{$k};
                    if (!defined $v) {
                        $norm{$k} = undef;
                    } elsif (is_numeric_type($col_type{$k} // '')) {
                        # Идентично export_spz: $val + 0 без проверки значения.
                        # Подавляем предупреждение для случаев с нечисловыми данными
                        # в числовых колонках (лax SQL mode в источнике).
                        no warnings 'numeric';
                        $norm{$k} = $v + 0;
                    } else {
                        $norm{$k} = $v;
                    }
                }
                push @tgt_samples, \%norm;
            }
            $sth->finish();

            if (@tgt_samples == @pk_vals) {
                my $tgt_hash = md5_hex($json_canon->encode(\@tgt_samples));
                _check(\@passed, \@failed, 'sample_hash',
                    $src_v->{sample_hash}, $tgt_hash,
                    sub { $_[0] eq $_[1] },
                    sprintf("src=%s... tgt=%s...",
                        substr($src_v->{sample_hash}, 0, 12),
                        substr($tgt_hash,             0, 12)));
            } else {
                push @warnings, sprintf(
                    "sample_hash: найдено %d строк из %d по PK в целевой БД",
                    scalar @tgt_samples, scalar @pk_vals);
            }
        } else {
            push @warnings, "sample_hash: не удалось запросить образцы из целевой БД";
        }
    } elsif (defined $src_v->{sample_hash} && !$src_v->{pk}) {
        push @warnings, "sample_hash: пропущен — PK не определён в источнике";
    }

    # ── Итоговый вывод ────────────────────────────────────────────────────────
    my $pass_n = scalar @passed;
    my $fail_n = scalar @failed;
    my $status = $fail_n == 0 ? '[OK]' : '[FAIL]';

    if ($fail_n == 0 && !$opt_verbose) {
        info(sprintf("  %-40s [VERIFY] OK (%d проверок пройдено)", $tname, $pass_n));
    } else {
        info(sprintf("  %-40s [VERIFY] %s — пройдено:%d провалено:%d",
            $tname, $status, $pass_n, $fail_n));
        for my $p (@passed) {
            debug(sprintf("    ✓ %-35s %s", $p->{check}, $p->{detail}));
        }
        for my $f (@failed) {
            err_(sprintf("    ✗ %-35s РАСХОЖДЕНИЕ: %s", $f->{check}, $f->{detail}));
        }
    }
    for my $w (@warnings) {
        warn_("    ⚠ $tname: $w");
    }

    return { passed => $pass_n, failed => $fail_n };
}

# Хелпер: добавляет результат проверки в passed или failed
sub _check {
    my ($passed, $failed, $name, $src, $tgt, $cmp, $detail) = @_;
    if ($cmp->($src, $tgt)) {
        push @$passed, { check => $name, detail => $detail };
    } else {
        push @$failed, { check => $name, detail => $detail };
    }
}


# ═══════════════════════════════════════════════════════════════════════════════
# Интерактив
# ═══════════════════════════════════════════════════════════════════════════════

sub interactive_select {
    my ($tables, $index, $dbh) = @_;

    my %src_rows = map { $_->{table} => $_->{rows} } @{ $index->{tables} };

    print "\n";
    print "═" x 72 . "\n";
    printf "  %-3s %-38s %8s  %s\n", '#', 'Таблица', 'Строк', 'Целевая БД';
    print "─" x 72 . "\n";

    my $i = 0;
    for my $t (@$tables) {
        $i++;
        my $rows   = $src_rows{$t} // '?';
        my $use_db = $db_name || db_for_table($t, $index);
        my $status = '';
        if ($use_db && $target_dbs{$use_db}) {
            my $tbl_ok = $dbh->selectrow_array(
                "SELECT COUNT(*) FROM information_schema.TABLES
                 WHERE TABLE_SCHEMA=? AND TABLE_NAME=?", undef, $use_db, $t
            );
            $status = $tbl_ok ? $use_db : "! нет в $use_db";
        } else {
            $status = $use_db ? "! БД '$use_db' не найдена" : '?';
        }
        printf "  %-3d %-38s %8s  %s\n", $i, $t,
            ($rows =~ /^\d+$/ ? commify($rows) : $rows), $status;
    }

    print "═" x 72 . "\n\n";
    print "Выберите таблицы для импорта:\n";
    print "  all     — все таблицы\n";
    print "  1,3,5   — по номерам\n";
    print "  2-7     — диапазон\n";
    print "  имя     — по имени таблицы\n";
    print "  q/Enter — выход\n\n> ";

    my $input = <STDIN>; chomp $input; $input =~ s/^\s+|\s+$//g;
    return () if !$input || $input eq 'q';

    my %sel;
    if (lc($input) eq 'all') {
        %sel = map { $_ => 1 } 1 .. scalar @$tables;
    } else {
        for my $part (split /,/, $input) {
            $part =~ s/^\s+|\s+$//g;
            if    ($part =~ /^(\d+)-(\d+)$/) { $sel{$_}++ for $1..$2; }
            elsif ($part =~ /^\d+$/)          { $sel{$part}++; }
            elsif ($part) {
                my $found = 0;
                for my $j (0..$#$tables) {
                    if ($tables->[$j] eq $part) { $sel{$j+1}++; $found=1; }
                }
                warn_("Таблица '$part' не найдена") unless $found;
            }
        }
    }

    return map { $tables->[$_-1] }
           grep { $_ >= 1 && $_ <= scalar @$tables }
           sort { $a <=> $b } keys %sel;
}

sub ask_import_mode {
    print "\nРежим импорта:\n";
    print "  1) INSERT IGNORE          — пропускать дубликаты [умолчание]\n";
    print "  2) REPLACE INTO           — перезаписывать существующие строки\n";
    print "  3) TRUNCATE + INSERT      — очистить таблицу, затем вставить\n\n> ";
    my $in = <STDIN>; chomp $in; $in =~ s/^\s+|\s+$//g; $in ||= '1';

    return (0, 1) if $in eq '2';
    if ($in eq '3') {
        print "\nВНИМАНИЕ: TRUNCATE удалит все данные. Продолжить? [y/N] > ";
        my $c = <STDIN>; chomp $c;
        return (1, 0) if lc($c) eq 'y';
    }
    return (0, 0);
}


# ═══════════════════════════════════════════════════════════════════════════════
# Утилиты
# ═══════════════════════════════════════════════════════════════════════════════

sub is_numeric_type {
    my $type = lc(shift // '');
    return $type =~ /^(int|tinyint|smallint|mediumint|bigint|float|double|decimal|numeric|bit)\b/;
}

sub db_for_table {
    my ($tname, $index) = @_;
    return 'spz_bill' if $tname =~ /^bill/;
    return 'spz'      if $tname =~ /^contacts|^set_/;
    return $index->{database} || 'spz';
}

sub slurp {
    my $f = shift;
    open(my $fh, '<:utf8', $f) or die "Не удалось прочитать $f: $!\n";
    local $/; my $s = <$fh>; close $fh;
    return $s;
}

sub commify {
    my $n = reverse(shift // 0);
    $n =~ s/(\d{3})(?=\d)/$1,/g;
    return scalar reverse $n;
}

sub print_separator { print "─" x 72 . "\n" }

sub usage { print <<'END';
Использование: import_spz.pl [параметры]

Подключение:
  --host HOST        Хост целевой БД        (умолч: 127.0.0.1)
  --port PORT        Порт                   (умолч: 13306)
  --user USER        Пользователь           (умолч: root)
  --pass PASS        Пароль
  --database DB      Целевая БД (умолч — из JSON, multi-db)

Источник:
  --dir PATH         Папка с JSON-файлами   (умолч: ./backup/spzconfig)

Режим:
  --all              Все таблицы без интерактива
  -t, --table NAME   Только одна таблица
  --truncate         TRUNCATE перед импортом
  --replace          REPLACE INTO (вместо INSERT IGNORE)
  --no-triggers      Отключить триггеры на время импорта (DROP + CREATE после)
  --dry-run          Показать план, без изменений в БД
  --no-verify        Не проверять данные после импорта
  --verify-only      Только верификация, импорт не выполнять
  --batch N          Строк в одной транзакции (умолч: 200)
  -v, --verbose      Подробный вывод (в т.ч. пройденные проверки)
  --help             Эта справка

Верификация проверяет:
  row_count          — кол-во строк совпадает с источником
  checksum           — MySQL CHECKSUM TABLE (CRC32 всех данных)
  pk.min/max/distinct — диапазон и уникальность первичного ключа
  col.sum/min/max    — числовые колонки: сумма, минимум, максимум
  col.nulls          — кол-во NULL по каждой колонке
  sample_hash        — MD5 первой/средней/последней строки источника

Примеры:
  perl import_spz.pl
  perl import_spz.pl --all --truncate
  perl import_spz.pl -t spz_rules --replace -v
  perl import_spz.pl --verify-only --all
  perl import_spz.pl --host db2.local --port 3306 --pass secret
  perl import_spz.pl --all --dry-run

Лог: /tmp/import_spz.log

Примечание: по умолчанию скрипты читают/пишут в ./backup/ относительно
рабочего каталога. Запускайте из /data/spz/docker/spzgate_2.0.0:
  cd /data/spz/docker/spzgate_2.0.0
  perl tools/export_spz.pl -f -v --extra-db status
  perl tools/import_spz.pl --all --truncate --no-triggers
END
}