Дмитрий
f94552d452
92 файла одной пачкой. Исключены чужие зоны: CLAUDE.md, .claude/settings.json, docs/observer/.pii-counters.json. gitleaks staged: no leaks found. Не верифицировано тестами - сохранение труда в историю. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
447 lines
19 KiB
PHP
447 lines
19 KiB
PHP
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use App\Support\RussianRegions;
|
||
use Illuminate\Console\Command;
|
||
use Illuminate\Database\Connection;
|
||
use Illuminate\Support\Facades\DB;
|
||
use OpenSpout\Reader\XLSX\Reader as XlsxReader;
|
||
|
||
/**
|
||
* Импорт реестра нумерации Россвязи в `phone_ranges` (spec §6).
|
||
*
|
||
* php artisan phone-ranges:import --file=<csv|xlsx> [--force] [--dry-run]
|
||
* php artisan phone-ranges:import --dir=<dir с пакетом файлов> [...]
|
||
*
|
||
* Алгоритм:
|
||
* 1. Резолв входных файлов (--file | --dir; --url отложен — оператор качает пакет вручную).
|
||
* 2. Checksum-идемпотентность: совпал с предыдущим `completed` → status='rolled_back', выход.
|
||
* 3. Парсинг (CSV через str_getcsv ';', XLSX через openspout) → нормализованные строки.
|
||
* 4. Маппинг region → subject_code через RussianRegions::nameToCode(). Несматчившиеся → лог в error.
|
||
* 5. Сборка `phone_ranges_staging` (LIKE phone_ranges) + bulk INSERT.
|
||
* 6. --dry-run → staging остаётся для инспекции, swap НЕ делается, status='rolled_back'.
|
||
* Иначе → atomic RENAME swap + status='completed'.
|
||
*
|
||
* Запись идёт через `pgsql_supplier` (на проде crm_supplier_worker — член crm_migrator,
|
||
* INHERIT даёт CREATE; SET ROLE crm_migrator выравнивает ownership. На dev/test — postgres superuser).
|
||
*
|
||
* NB (swap — operator-validated): committing-swap (шаг 6 else) НЕ покрыт автотестом —
|
||
* RENAME коммитит и сломал бы общую тестовую БД. Свап проверяется первым реальным
|
||
* импортом оператора по runbook (Session 6). Тесты покрывают parse/map/dry-run/idempotency.
|
||
*/
|
||
class PhoneRangesImportCommand extends Command
|
||
{
|
||
/** @var string */
|
||
protected $signature = 'phone-ranges:import
|
||
{--file= : Путь к одному CSV/XLSX файлу реестра}
|
||
{--dir= : Каталог с пакетом файлов реестра (*.csv, *.xlsx)}
|
||
{--url= : (отложено) URL пакета — скачать вручную и использовать --dir}
|
||
{--force : Игнорировать checksum-идемпотентность}
|
||
{--dry-run : Распарсить и собрать staging, но не делать atomic swap}';
|
||
|
||
/** @var string */
|
||
protected $description = 'Импорт реестра нумерации Россвязи в phone_ranges (idempotent, atomic swap)';
|
||
|
||
/** Connection для DDL/записи (на проде crm_migrator-capable, на dev/test — superuser fallback). */
|
||
private const DDL_CONNECTION = 'pgsql_supplier';
|
||
|
||
/** Размер пачки для bulk INSERT в staging. */
|
||
private const INSERT_CHUNK = 1000;
|
||
|
||
public function handle(): int
|
||
{
|
||
$files = $this->resolveFiles();
|
||
if ($files === null) {
|
||
return self::FAILURE;
|
||
}
|
||
|
||
$checksum = $this->computeChecksum($files);
|
||
$dryRun = (bool) $this->option('dry-run');
|
||
$force = (bool) $this->option('force');
|
||
|
||
// 2. Идемпотентность по checksum (если не --force).
|
||
if (! $force) {
|
||
$prev = DB::table('phone_ranges_imports')
|
||
->where('checksum_sha256', $checksum)
|
||
->where('status', 'completed')
|
||
->orderByDesc('id')
|
||
->first();
|
||
|
||
if ($prev !== null) {
|
||
DB::table('phone_ranges_imports')->insert([
|
||
'source_url' => $this->sourceLabel($files),
|
||
'checksum_sha256' => $checksum,
|
||
'status' => 'rolled_back',
|
||
'rows_inserted' => 0,
|
||
'rows_updated' => 0,
|
||
'error' => "Идентично импорту #{$prev->id} (checksum совпал) — пропуск.",
|
||
'imported_at' => now(),
|
||
'completed_at' => now(),
|
||
]);
|
||
$this->info("Реестр идентичен импорту #{$prev->id} — пропуск (используйте --force для принудительного импорта).");
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
}
|
||
|
||
// 3. Журнал импорта (in_progress).
|
||
$importId = (int) DB::table('phone_ranges_imports')->insertGetId([
|
||
'source_url' => $this->sourceLabel($files),
|
||
'checksum_sha256' => $checksum,
|
||
'status' => 'in_progress',
|
||
'imported_at' => now(),
|
||
]);
|
||
|
||
try {
|
||
// 4. Парсинг + маппинг.
|
||
$unmatched = [];
|
||
$rows = [];
|
||
foreach ($files as $file) {
|
||
foreach ($this->parseFile($file) as $rec) {
|
||
$regionNormalized = RussianRegions::canonicalRegionName($rec['region']);
|
||
$subjectCode = $regionNormalized === null
|
||
? null
|
||
: (RussianRegions::nameToCode()[$regionNormalized] ?? null);
|
||
if ($subjectCode === null && trim($rec['region']) !== '') {
|
||
$unmatched[trim($rec['region'])] = true;
|
||
}
|
||
$rows[] = [
|
||
'def_code' => $rec['def_code'],
|
||
'from_num' => $rec['from_num'],
|
||
'to_num' => $rec['to_num'],
|
||
'operator' => $rec['operator'],
|
||
'region' => $rec['region'],
|
||
'region_normalized' => $regionNormalized,
|
||
'subject_code' => $subjectCode,
|
||
'imported_at' => now(),
|
||
'import_id' => $importId,
|
||
];
|
||
}
|
||
}
|
||
|
||
// 5. Сборка staging.
|
||
$this->buildStaging($rows, $importId);
|
||
|
||
$unmatchedNote = $unmatched === []
|
||
? ''
|
||
: 'Не сопоставлены регионы: '.implode(', ', array_keys($unmatched)).'.';
|
||
|
||
if ($dryRun) {
|
||
DB::table('phone_ranges_imports')->where('id', $importId)->update([
|
||
'status' => 'rolled_back',
|
||
'rows_inserted' => count($rows),
|
||
'error' => trim('dry-run (swap не выполнен). '.$unmatchedNote),
|
||
'completed_at' => now(),
|
||
]);
|
||
$this->info('dry-run: '.count($rows).' строк в phone_ranges_staging, swap не выполнен.');
|
||
if ($unmatchedNote !== '') {
|
||
$this->warn($unmatchedNote);
|
||
}
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
|
||
// 6. Atomic swap (operator-validated — см. docblock).
|
||
$this->atomicSwap();
|
||
|
||
DB::table('phone_ranges_imports')->where('id', $importId)->update([
|
||
'status' => 'completed',
|
||
'rows_inserted' => count($rows),
|
||
'error' => $unmatchedNote !== '' ? $unmatchedNote : null,
|
||
'completed_at' => now(),
|
||
]);
|
||
$this->info('Импортировано '.count($rows).' строк в phone_ranges (atomic swap выполнен).');
|
||
if ($unmatchedNote !== '') {
|
||
$this->warn($unmatchedNote);
|
||
}
|
||
|
||
return self::SUCCESS;
|
||
} catch (\Throwable $e) {
|
||
DB::table('phone_ranges_imports')->where('id', $importId)->update([
|
||
'status' => 'failed',
|
||
'error' => mb_substr($e->getMessage(), 0, 2000),
|
||
'completed_at' => now(),
|
||
]);
|
||
$this->error('Импорт упал: '.$e->getMessage());
|
||
|
||
return self::FAILURE;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @return list<string>|null Список файлов или null при ошибке валидации опций.
|
||
*/
|
||
private function resolveFiles(): ?array
|
||
{
|
||
$file = $this->option('file');
|
||
$dir = $this->option('dir');
|
||
$url = $this->option('url');
|
||
|
||
if ($url !== null) {
|
||
$this->error('--url отложен (пакет ~500-600 файлов). Скачайте вручную и используйте --dir.');
|
||
|
||
return null;
|
||
}
|
||
|
||
if ($file !== null) {
|
||
if (! is_file($file)) {
|
||
$this->error("Файл не найден: {$file}");
|
||
|
||
return null;
|
||
}
|
||
|
||
return [$file];
|
||
}
|
||
|
||
if ($dir !== null) {
|
||
if (! is_dir($dir)) {
|
||
$this->error("Каталог не найден: {$dir}");
|
||
|
||
return null;
|
||
}
|
||
$found = glob(rtrim($dir, '/\\').DIRECTORY_SEPARATOR.'*.{csv,xlsx}', GLOB_BRACE) ?: [];
|
||
if ($found === []) {
|
||
$this->error("В каталоге нет *.csv / *.xlsx: {$dir}");
|
||
|
||
return null;
|
||
}
|
||
sort($found);
|
||
|
||
return array_values($found);
|
||
}
|
||
|
||
$this->error('Укажите --file=<путь> или --dir=<каталог>.');
|
||
|
||
return null;
|
||
}
|
||
|
||
/**
|
||
* @param list<string> $files
|
||
*/
|
||
private function computeChecksum(array $files): string
|
||
{
|
||
if (count($files) === 1) {
|
||
return (string) hash_file('sha256', $files[0]);
|
||
}
|
||
|
||
$hashes = array_map(static fn (string $f): string => (string) hash_file('sha256', $f), $files);
|
||
sort($hashes);
|
||
|
||
return hash('sha256', implode('|', $hashes));
|
||
}
|
||
|
||
/**
|
||
* @param list<string> $files
|
||
*/
|
||
private function sourceLabel(array $files): string
|
||
{
|
||
return $this->option('url')
|
||
?? $this->option('dir')
|
||
?? ($files[0] ?? 'unknown');
|
||
}
|
||
|
||
/**
|
||
* Парсит один файл реестра в нормализованные строки.
|
||
*
|
||
* @return list<array{def_code:int, from_num:int, to_num:int, operator:string, region:string}>
|
||
*/
|
||
private function parseFile(string $path): array
|
||
{
|
||
$ext = strtolower(pathinfo($path, PATHINFO_EXTENSION));
|
||
|
||
return $ext === 'xlsx'
|
||
? $this->parseXlsx($path)
|
||
: $this->parseCsv($path);
|
||
}
|
||
|
||
/**
|
||
* @return list<array{def_code:int, from_num:int, to_num:int, operator:string, region:string}>
|
||
*/
|
||
private function parseCsv(string $path): array
|
||
{
|
||
$content = (string) file_get_contents($path);
|
||
// BOM strip + split строк (CRLF/CR/LF).
|
||
$content = preg_replace('/^\xEF\xBB\xBF/', '', $content) ?? $content;
|
||
$lines = preg_split('/\r\n|\r|\n/', rtrim($content)) ?: [];
|
||
if ($lines === []) {
|
||
return [];
|
||
}
|
||
|
||
$header = str_getcsv((string) array_shift($lines), ';');
|
||
$cols = $this->resolveColumns($header);
|
||
|
||
$out = [];
|
||
foreach ($lines as $line) {
|
||
if (trim($line) === '') {
|
||
continue;
|
||
}
|
||
$cells = str_getcsv($line, ';');
|
||
$rec = $this->mapCells($cells, $cols);
|
||
if ($rec !== null) {
|
||
$out[] = $rec;
|
||
}
|
||
}
|
||
|
||
return $out;
|
||
}
|
||
|
||
/**
|
||
* Парсинг XLSX через openspout (operator-real-files; CSV-ветка покрыта тестом).
|
||
*
|
||
* @return list<array{def_code:int, from_num:int, to_num:int, operator:string, region:string}>
|
||
*/
|
||
private function parseXlsx(string $path): array
|
||
{
|
||
$reader = new XlsxReader;
|
||
$reader->open($path);
|
||
|
||
$out = [];
|
||
$cols = null;
|
||
foreach ($reader->getSheetIterator() as $sheet) {
|
||
foreach ($sheet->getRowIterator() as $row) {
|
||
$cells = array_map(static fn ($c): string => (string) $c, $row->toArray());
|
||
if ($cols === null) {
|
||
$cols = $this->resolveColumns($cells);
|
||
|
||
continue;
|
||
}
|
||
$rec = $this->mapCells($cells, $cols);
|
||
if ($rec !== null) {
|
||
$out[] = $rec;
|
||
}
|
||
}
|
||
break; // только первый лист
|
||
}
|
||
$reader->close();
|
||
|
||
return $out;
|
||
}
|
||
|
||
/**
|
||
* Сопоставляет индексы колонок по заголовку (русские имена Россвязи) с позиционным fallback.
|
||
*
|
||
* @param list<string> $header
|
||
* @return array{def:int, from:int, to:int, operator:int, region:int}
|
||
*/
|
||
private function resolveColumns(array $header): array
|
||
{
|
||
$cols = ['def' => 0, 'from' => 1, 'to' => 2, 'operator' => 4, 'region' => 5];
|
||
|
||
foreach ($header as $i => $cell) {
|
||
$n = preg_replace('/[\s\/]+/u', '', mb_strtolower(trim((string) $cell))) ?? '';
|
||
if (str_contains($n, 'def') || str_contains($n, 'авс')) {
|
||
$cols['def'] = $i;
|
||
} elseif ($n === 'от') {
|
||
$cols['from'] = $i;
|
||
} elseif ($n === 'до') {
|
||
$cols['to'] = $i;
|
||
} elseif (str_contains($n, 'оператор')) {
|
||
$cols['operator'] = $i;
|
||
} elseif (str_contains($n, 'регион')) {
|
||
$cols['region'] = $i;
|
||
}
|
||
}
|
||
|
||
return $cols;
|
||
}
|
||
|
||
/**
|
||
* @param list<string> $cells
|
||
* @param array{def:int, from:int, to:int, operator:int, region:int} $cols
|
||
* @return array{def_code:int, from_num:int, to_num:int, operator:string, region:string}|null
|
||
*/
|
||
private function mapCells(array $cells, array $cols): ?array
|
||
{
|
||
$def = (int) preg_replace('/\D+/', '', $cells[$cols['def']] ?? '');
|
||
if ($def === 0) {
|
||
return null; // пустая/битая строка
|
||
}
|
||
|
||
return [
|
||
'def_code' => $def,
|
||
'from_num' => (int) preg_replace('/\D+/', '', $cells[$cols['from']] ?? '0'),
|
||
'to_num' => (int) preg_replace('/\D+/', '', $cells[$cols['to']] ?? '0'),
|
||
'operator' => trim((string) ($cells[$cols['operator']] ?? '')),
|
||
'region' => trim((string) ($cells[$cols['region']] ?? '')),
|
||
];
|
||
}
|
||
|
||
/**
|
||
* Собирает phone_ranges_staging (LIKE phone_ranges) и заливает строки.
|
||
*
|
||
* id: НЕ копируем серийный default через INCLUDING DEFAULTS — он ссылается на
|
||
* исходную последовательность phone_ranges, которую atomic-swap уничтожает
|
||
* (DROP phone_ranges_old CASCADE) после первого импорта, оставляя staging.id
|
||
* без default (NOT NULL violation на повторном импорте). Вместо этого даём
|
||
* staging собственную последовательность с уникальным по import_id именем,
|
||
* OWNED BY колонкой id → она переезжает при RENAME и дропается вместе со
|
||
* старой таблицей (без коллизий имён и без утечки последовательностей).
|
||
*
|
||
* @param list<array<string, mixed>> $rows
|
||
*/
|
||
private function buildStaging(array $rows, int $importId): void
|
||
{
|
||
$c = DB::connection(self::DDL_CONNECTION);
|
||
$this->elevate($c);
|
||
|
||
$seq = "phone_ranges_stg_seq_{$importId}";
|
||
$c->statement('DROP TABLE IF EXISTS phone_ranges_staging CASCADE');
|
||
$c->statement('CREATE TABLE phone_ranges_staging (LIKE phone_ranges INCLUDING CONSTRAINTS)');
|
||
$c->statement("CREATE SEQUENCE {$seq}");
|
||
$c->statement("ALTER TABLE phone_ranges_staging ALTER COLUMN id SET DEFAULT nextval('{$seq}')");
|
||
$c->statement("ALTER SEQUENCE {$seq} OWNED BY phone_ranges_staging.id");
|
||
$c->statement('CREATE INDEX IF NOT EXISTS idx_phone_ranges_staging_lookup ON phone_ranges_staging (def_code, from_num, to_num)');
|
||
|
||
foreach (array_chunk($rows, self::INSERT_CHUNK) as $chunk) {
|
||
$c->table('phone_ranges_staging')->insert($chunk);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Atomic swap живого phone_ranges на staging (spec §6.2 шаг 6).
|
||
*
|
||
* NB: НЕ покрыт автотестом (committing RENAME сломал бы общую тестовую БД).
|
||
* Проверяется первым реальным импортом оператора (Session 6 runbook).
|
||
* Сохраняет одну предыдущую версию (phone_ranges_old) для `phone-ranges:rollback`.
|
||
* GRANT'ы переустанавливаются (RENAME их не переносит); lookup-индекс на новой
|
||
* таблице носит имя idx_phone_ranges_staging_lookup (косметика — имя занято _old).
|
||
*/
|
||
private function atomicSwap(): void
|
||
{
|
||
$c = DB::connection(self::DDL_CONNECTION);
|
||
$this->elevate($c);
|
||
|
||
// Транзакция вокруг свапа (spec §6.2): PostgreSQL поддерживает транзакционный
|
||
// DDL, поэтому DROP+RENAME+RENAME+GRANT атомарны. Обрыв процесса между
|
||
// переименованиями не оставит phone_ranges несуществующей — откат вернёт
|
||
// живую таблицу (раньше 4 авто-коммит-statement'а оставляли окно, в котором
|
||
// Россвязь-lookup падал бы до ручного восстановления).
|
||
$c->transaction(function () use ($c) {
|
||
$c->statement('DROP TABLE IF EXISTS phone_ranges_old CASCADE');
|
||
$c->statement('ALTER TABLE phone_ranges RENAME TO phone_ranges_old');
|
||
$c->statement('ALTER TABLE phone_ranges_staging RENAME TO phone_ranges');
|
||
$c->statement('GRANT SELECT ON phone_ranges TO crm_app_user, crm_supplier_worker');
|
||
});
|
||
}
|
||
|
||
/**
|
||
* SET ROLE crm_migrator для корректного ownership на проде; на dev/test роль
|
||
* отсутствует → RESET и работаем как superuser (зеркало миграционного паттерна).
|
||
*/
|
||
private function elevate(Connection $c): void
|
||
{
|
||
try {
|
||
$c->statement('SET ROLE crm_migrator');
|
||
$canCreate = $c->selectOne("SELECT has_schema_privilege('crm_migrator', 'public', 'CREATE') AS ok");
|
||
if (! $canCreate || ! $canCreate->ok) {
|
||
$c->statement('RESET ROLE');
|
||
}
|
||
} catch (\Throwable) {
|
||
// окружение без роли — продолжаем как superuser
|
||
}
|
||
}
|
||
}
|