""" Добавление записей в открытые таблицы из одноимённых TAB-файлов Скрипт открывает диалог выбора нескольких .tab-файлов, для каждого находит открытую таблицу с тем же именем файла (без учёта пути и регистра), проверяет совместимость схем и копирует записи порциями по 1000 шт. Копирование выполняется в фоновом потоке с прогресс-баром, счётчиком записей и возможностью отмены. """ import os from dataclasses import dataclass, field from pathlib import Path from typing import Callable, Dict, List, Optional, Tuple from PySide2.QtCore import Qt, Slot from PySide2.QtWidgets import (QCheckBox, QDialog, QHBoxLayout, QHeaderView, QLabel, QProgressBar, QPushButton, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget) import axipy # Размер пачки записей при вставке. # 1000 записей — компромисс между частотой вызовов insert() # и объёмом данных, теряемых при ошибке на одной пачке. BATCH_SIZE: int = 1000 # ------------------------------------------------------------------ # Исключения # ------------------------------------------------------------------ class SchemaMismatchError(Exception): """ Поле источника отсутствует в приёмнике или схемы несовместимы по структуре (последний рубеж проверки перед вставкой данных). """ pass # ------------------------------------------------------------------ # Совместимость типов атрибутов # ------------------------------------------------------------------ # Группы типов для сравнения. # Ключ — строка, возвращаемая Attribute.type_string. # Значение — кортеж (группа, разрядность_в_битах_для_целых). _TYPE_GROUPS: Dict[str, Tuple[str, Optional[int]]] = { 'bool': ('bool', None), 'int16': ('int', 16), 'int': ('int', 32), 'int64': ('int', 64), 'double': ('float', None), 'string': ('string', None), 'decimal': ('decimal', None), 'date': ('date', None), 'time': ('time', None), 'datetime': ('datetime', None), } # Максимальное количество символов в целой части ЦЕЛОГО типа со знаком # (худший случай — отрицательное число: знак «−» занимает 1 символ). # Используется для проверки «целое → десятичное», # т.к. знак числа заранее неизвестен. _INT_MAX_DIGITS_SIGNED: Dict[str, int] = { 'int16': 6, # −32767 → 5 цифр + знак 'int': 11, # −2147483647 → 10 цифр + знак 'int64': 20, # −9223372036854775807 → 19 цифр + знак } def _decimal_integer_digits(length: Optional[int], precision: Optional[int]) -> Optional[int]: """ Максимальное количество цифр в целой части для положительных чисел в Десятичном(L, P). Для положительных чисел знак «+» не хранится: - P > 0: L − P − 1 (минус запятая); - P == 0: L (нет ни запятой, ни знака). :param length: Общая длина поля (L). :param precision: Количество знаков после запятой (P). :return: Количество цифр в целой части или None, если length не задан. """ if length is None: return None if precision and precision > 0: return length - precision - 1 return length def _int_to_decimal_required(src_type: str, tgt_precision: Optional[int]) -> int: """ Сколько символов length нужно в Десятичном, чтобы вместить ЛЮБОЕ значение целого типа (и положительное, и отрицательное). :param src_type: Имя целого типа ('int16', 'int', 'int64'). :param tgt_precision: Точность целевого десятичного поля. :return: Минимально необходимая длина (length) десятичного поля. """ digits: int = _INT_MAX_DIGITS_SIGNED.get(src_type, 11) if tgt_precision and tgt_precision > 0: # целая часть со знаком + запятая + дробная часть return digits + 1 + tgt_precision # только целая часть со знаком return digits def check_attribute_compatibility(src_attr: axipy.Attribute, tgt_attr: axipy.Attribute) -> bool: """ Возвращает True, если атрибут из файла-источника совместим с атрибутом целевой таблицы (данные источника помещаются в приёмник). Правила: - каждая колонка источника должна присутствовать в приёмнике; - тип колонки источника должен быть не «шире» типа приёмника (короткие строки → длинные, короткие целые → длинные и т.д.); - позиции полей не сравниваются. :param src_attr: Атрибут из схемы файла-источника. :param tgt_attr: Атрибут из схемы таблицы-приёмника. :return: True, если данные можно скопировать без потери. """ src_type: str = src_attr.type_string tgt_type: str = tgt_attr.type_string src_len: Optional[int] = src_attr.length tgt_len: Optional[int] = tgt_attr.length src_prec: Optional[int] = src_attr.precision tgt_prec: Optional[int] = tgt_attr.precision src_info: Optional[Tuple[str, Optional[int]]] = _TYPE_GROUPS.get(src_type) tgt_info: Optional[Tuple[str, Optional[int]]] = _TYPE_GROUPS.get(tgt_type) if not src_info or not tgt_info: return False src_group, src_bits = src_info tgt_group, tgt_bits = tgt_info # --- string --- if src_group == 'string' and tgt_group == 'string': if src_len is None or tgt_len is None: return True return src_len <= tgt_len # --- decimal --- if src_group == 'decimal' and tgt_group == 'decimal': # Дробная часть источника должна помещаться. if src_prec is not None and tgt_prec is not None and src_prec > tgt_prec: return False # Целая часть источника должна помещаться. # Сравнение по положительным числам: если помещается максимум # положительных, то отрицательные тем более поместятся # (для них нужна на 1 цифру меньше). src_digits: Optional[int] = _decimal_integer_digits(src_len, src_prec) tgt_digits: Optional[int] = _decimal_integer_digits(tgt_len, tgt_prec) if src_digits is not None and tgt_digits is not None: return src_digits <= tgt_digits return True # --- целые --- if src_group == 'int' and tgt_group == 'int': if tgt_bits is None or src_bits is None: return True return src_bits <= tgt_bits # --- целое → float --- if src_group == 'int' and tgt_group == 'float': return True # --- целое → decimal (со знаком, т.к. содержимое заранее неизвестно) --- if src_group == 'int' and tgt_group == 'decimal' \ and src_type in _INT_MAX_DIGITS_SIGNED: required: int = _int_to_decimal_required(src_type, tgt_prec) if tgt_len is not None and tgt_len < required: return False return True # --- float --- if src_group == 'float' and tgt_group == 'float': return True # --- точное совпадение для остальных типов --- if src_group == tgt_group: return True return False # ------------------------------------------------------------------ # Проверка одного файла # ------------------------------------------------------------------ @dataclass class FileCheckResult: """ Результат проверки одного выбранного файла. :param file_path: Путь к выбранному файлу. :param status: Текущий статус обработки. :param message: Текстовое описание статуса. :param target_table: Ссылка на целевую таблицу-приёмник (если найдена). :param row: Индекс строки в таблице диалога. """ file_path: Path status: str = '' message: str = '' target_table: Optional[axipy.Table] = field(default=None, repr=False) row: int = -1 # Отображение статусов на иконки и текст для диалога. # Используется единообразно на всех этапах: проверка, копирование, завершение. _STATUS_DISPLAY: Dict[str, Tuple[str, str]] = { 'ready': ('✅', 'Готов'), 'skip': ('⏭️', 'Пропуск'), 'error': ('❌', 'Ошибка'), 'copy': ('⏳', 'Копируется…'), 'copied': ('✅', 'Скопировано'), 'copy_error': ('❌', 'Ошибка копирования'), 'canceled': ('⚠️', 'Не скопировано'), } def check_file(file_path: Path, open_tables: List[axipy.Table]) -> FileCheckResult: """ Все проверки одного выбранного файла. Шаги: 1. Файл не должен быть уже открыт в Аксиоме (по полному пути). 2. Поиск целевой таблицы по имени файла (без пути, без учёта регистра). 3. Скрытое открытие файла для чтения схемы. 4. Сравнение схем источника и приёмника. Скрытая таблица-источник закрывается сразу после сравнения схем. :param file_path: Путь к выбранному файлу. :param open_tables: Список открытых таблиц Аксиомы. :return: Результат проверки. """ result = FileCheckResult(file_path=file_path) # Имя файла сравниваем без учёта регистра на любой ОС (бизнес-требование). file_name_norm: str = file_path.name.lower() # Полный путь сравниваем через os.path.normcase: # - на Windows регистр игнорируется (файловая система нечувствительна); # - на Linux регистр учитывается (файловая система чувствительна). # Это обеспечивает корректное кроссплатформенное поведение. file_path_norm: str = os.path.normcase(str(file_path)) # --- Шаг 1: файл не должен быть уже открыт --- for tbl in open_tables: if tbl.file_name is not None \ and os.path.normcase(str(tbl.file_name)) == file_path_norm: result.status = 'skip' result.message = 'Файл уже открыт' return result # --- Шаг 2: поиск целевой таблицы по имени файла --- candidates: List[axipy.Table] = [ tbl for tbl in open_tables if tbl.file_name is not None and tbl.file_name.name.lower() == file_name_norm ] if len(candidates) == 0: result.status = 'error' result.message = 'Не найдена соответствующая открытая таблица' return result if len(candidates) > 1: result.status = 'error' result.message = f'Найдено несколько открытых таблиц ({len(candidates)})' return result target_table: axipy.Table = candidates[0] # --- Шаг 3: скрытое открытие файла --- try: source: axipy.Table = axipy.provider_manager.open_hidden( {'src': str(file_path)}) except Exception as e: result.status = 'error' result.message = f'Ошибка открытия файла: {e}' return result # --- Шаг 4: сравнение схем --- try: compatible: bool = True # Имена полей сравниваются без учёта регистра (семантика MapInfo). tgt_by_lower: Dict[str, axipy.Attribute] = { a.name.lower(): a for a in target_table.schema } for src_attr in source.schema: tgt_attr: Optional[axipy.Attribute] = \ tgt_by_lower.get(src_attr.name.lower()) if tgt_attr is None \ or not check_attribute_compatibility(src_attr, tgt_attr): compatible = False break finally: _close_quietly(source) if not compatible: result.status = 'error' result.message = 'Несовместимость схем' return result result.status = 'ready' result.message = 'Готов к копированию' result.target_table = target_table return result # ------------------------------------------------------------------ # Вспомогательные функции для копирования # ------------------------------------------------------------------ def _close_quietly(source: Optional[axipy.Table]) -> None: """ Закрывает таблицу, игнорируя любые ошибки закрытия. :param source: Таблица для закрытия (может быть None). """ if source is not None: try: source.close() except Exception: pass def _build_name_remap(src_schema: axipy.Schema, tgt_schema: axipy.Schema) -> Dict[str, str]: """ Строит сопоставление имён полей источника именам полей приёмника (без учёта регистра). Возвращает пустой словарь, если имена совпадают и пересборка записей не нужна. Выбрасывает SchemaMismatchError, если какое-либо поле источника отсутствует в приёмнике (последний рубеж защиты перед вставкой). :param src_schema: Схема таблицы-источника. :param tgt_schema: Схема таблицы-приёмника. :return: Словарь {имя_источника: имя_приёмника}, пустой {} если пересборка не нужна. :raises SchemaMismatchError: Если поле источника отсутствует в приёмнике. """ tgt_by_lower: Dict[str, str] = {a.name.lower(): a.name for a in tgt_schema} remap: Dict[str, str] = {} needed: bool = False for attr in src_schema: tgt_name: Optional[str] = tgt_by_lower.get(attr.name.lower()) if tgt_name is None: raise SchemaMismatchError( f'Поле "{attr.name}" отсутствует в таблице-приёмнике') remap[attr.name] = tgt_name if tgt_name != attr.name: needed = True return remap if needed else {} def _remap_feature(feature: axipy.Feature, remap: Optional[Dict[str, str]]) -> axipy.Feature: """ Пересобирает запись с именами полей приёмника (при необходимости). Если пересборка не нужна (remap пустой или None), возвращает исходную запись. :param feature: Исходная запись. :param remap: Словарь пересборки имён или None/пустой {}. :return: Запись с именами полей приёмника. """ if not remap: return feature props: Dict[str, object] = {} for key, value in feature.items(): new_key: Optional[str] = remap.get(key) if new_key is not None: props[new_key] = value return axipy.Feature(props, geometry=feature.geometry, style=feature.style) def _notify(func: Callable, *args) -> None: """ Безопасное обновление интерфейса из фонового потока. Если диалог уже закрыт/удалён, обращение к Qt-объекту вызывает RuntimeError («wrapped C/C++ object has been deleted»). В этом случае просто игнорируем обновление. :param func: Функция обновления интерфейса. :param args: Аргументы для функции. """ try: axipy.run_in_gui(func, *args) except RuntimeError: # Диалог закрыт — ожидаемая ситуация при отмене. pass except Exception: # Прочие ошибки обновления не должны прерывать копирование. pass def _fmt(n: int) -> str: """ Форматирует число с разделителями-пробелами для удобства чтения. :param n: Число для форматирования. :return: Строка с разделителями, например '3 000'. """ return f'{n:,}'.replace(',', ' ') # ------------------------------------------------------------------ # Фоновая задача копирования # ------------------------------------------------------------------ def _copy_task(task: axipy.Task, items: List[FileCheckResult], dialog: 'ResultsDialog') -> None: """ Фоновое копирование всех выбранных файлов. Записи читаются из source.items() и вставляются в целевую таблицу пачками по BATCH_SIZE. После каждой пачки проверяется отмена и обновляется прогресс. При несовместимости схем на этапе построения remap (SchemaMismatchError) или ошибке чтения/вставки фиксируется ошибка, закрывается источник, и обработка продолжается со следующим файлом. :param task: Объект фоновой задачи. :param items: Список файлов для копирования. :param dialog: Диалог для обновления интерфейса. """ task.range = task.Range(0, len(items)) task.message = 'Копирование данных…' total: int = len(items) for i, item in enumerate(items): task.raise_if_canceled() name: str = item.file_path.name _notify(dialog._ui_file_started, item.row, i, total, name) # --- Открытие источника и построение remap --- source: Optional[axipy.Table] = None remap: Dict[str, str] = {} try: source = axipy.provider_manager.open_hidden( {'src': str(item.file_path)}) remap = _build_name_remap(source.schema, item.target_table.schema) except SchemaMismatchError as e: _close_quietly(source) _notify(dialog._ui_file_error, item.row, f'Несовместимость схем: {e}') task.value = i + 1 continue except Exception as e: _close_quietly(source) _notify(dialog._ui_file_error, item.row, f'Ошибка открытия: {e}') task.value = i + 1 continue # --- Копирование пачками --- count: int = 0 error: Optional[Exception] = None try: batch: List[axipy.Feature] = [] for feature in source.items(): task.raise_if_canceled() batch.append(_remap_feature(feature, remap)) if len(batch) >= BATCH_SIZE: item.target_table.insert(batch) count += len(batch) _notify(dialog._ui_file_progress, item.row, i, total, name, count) batch = [] if batch: item.target_table.insert(batch) count += len(batch) except Exception as e: error = e finally: _close_quietly(source) # Если причиной ошибки была отмена — исключение пройдёт дальше # и задача завершится со статусом CANCELED. task.raise_if_canceled() if error is not None: msg: str = f'Ошибка копирования: {error}' if count > 0: msg += f' (скопировано записей до ошибки: {_fmt(count)})' _notify(dialog._ui_file_error, item.row, msg) else: _notify(dialog._ui_file_done, item.row, i, total, name, count) task.value = i + 1 # ------------------------------------------------------------------ # Диалог результатов # ------------------------------------------------------------------ class ResultsDialog(QDialog): """ Диалог отображения результатов проверки и процесса копирования. Содержит таблицу файлов с чекбоксами, статусами и описаниями, прогресс-бар и кнопки управления. После завершения копирования кнопка «Копировать» превращается в «Закрыть». """ def __init__(self, results: List[FileCheckResult], parent: Optional[QWidget] = None) -> None: """ Инициализация диалога. :param results: Результаты проверки файлов. :param parent: Родительский виджет. """ parent = parent or axipy.view_manager.global_parent super().__init__(parent) self.setWindowTitle('Добавление записей в открытые таблицы') self.setMinimumSize(780, 480) self.results: List[FileCheckResult] = results self.task: Optional[axipy.Task] = None self._copy_started: bool = False self._checkboxes: List[QCheckBox] = [] self._build_ui() self._populate_table() self._update_ok_state() # ---------------- построение интерфейса ---------------- def _build_ui(self) -> None: """Создаёт элементы интерфейса диалога.""" layout = QVBoxLayout(self) self.table = QTableWidget() self.table.setColumnCount(4) self.table.setHorizontalHeaderLabels(['', 'Файл', 'Статус', 'Описание']) self.table.verticalHeader().setVisible(False) self.table.setEditTriggers(QTableWidget.NoEditTriggers) self.table.setSelectionBehavior(QTableWidget.SelectRows) header = self.table.horizontalHeader() header.setSectionResizeMode(0, QHeaderView.ResizeToContents) header.setSectionResizeMode(1, QHeaderView.Interactive) header.setSectionResizeMode(2, QHeaderView.ResizeToContents) header.setSectionResizeMode(3, QHeaderView.Stretch) layout.addWidget(self.table) self.progress_label = QLabel('') self.progress_label.setVisible(False) layout.addWidget(self.progress_label) self.progress = QProgressBar() self.progress.setVisible(False) layout.addWidget(self.progress) btn_row = QHBoxLayout() btn_row.addStretch() self.btn_ok = QPushButton('Копировать') self.btn_cancel = QPushButton('Отмена') self.btn_ok.clicked.connect(self._on_ok_clicked) self.btn_cancel.clicked.connect(self._on_cancel_clicked) btn_row.addWidget(self.btn_ok) btn_row.addWidget(self.btn_cancel) layout.addLayout(btn_row) def _populate_table(self) -> None: """Заполняет таблицу файлов результатами проверки.""" self.table.setRowCount(len(self.results)) self._checkboxes.clear() for row, res in enumerate(self.results): res.row = row # Чекбокс в контейнере для корректного центрирования # и отсутствия артефакта выделения текста. container = QWidget() cell_layout = QHBoxLayout(container) cell_layout.setContentsMargins(0, 0, 0, 0) cell_layout.setAlignment(Qt.AlignCenter) chk = QCheckBox() chk.setChecked(res.status == 'ready') chk.setEnabled(res.status == 'ready') chk.stateChanged.connect(self._update_ok_state) cell_layout.addWidget(chk) self.table.setCellWidget(row, 0, container) self._checkboxes.append(chk) # Колонки: файл, статус, описание. self.table.setItem(row, 1, QTableWidgetItem(str(res.file_path))) icon, text = _STATUS_DISPLAY.get(res.status, ('?', res.status)) self.table.setItem(row, 2, QTableWidgetItem(f'{icon} {text}')) self.table.setItem(row, 3, QTableWidgetItem(res.message)) self.table.resizeColumnToContents(0) self.table.setColumnWidth(1, 320) def _get_checkbox(self, row: int) -> Optional[QCheckBox]: """ Возвращает чекбокс для указанной строки. :param row: Индекс строки в таблице. :return: Виджет чекбокса или None. """ if 0 <= row < len(self._checkboxes): return self._checkboxes[row] return None # ---------------- состояние кнопки ОК ---------------- @Slot() def _update_ok_state(self) -> None: """Обновляет доступность кнопки ОК в зависимости от чекбоксов.""" if self._copy_started: return has_checked: bool = False for row in range(self.table.rowCount()): res = self.results[row] if res.status == 'ready': chk = self._get_checkbox(row) if chk is not None and chk.isChecked(): has_checked = True break self.btn_ok.setEnabled(has_checked) # ---------------- запуск копирования ---------------- @Slot() def _on_ok_clicked(self) -> None: """Обработка нажатия кнопки ОК / Копировать / Закрыть.""" # После завершения копирования кнопка ОК работает как «Закрыть». if self._copy_started: self.accept() return items: List[FileCheckResult] = [] for res in self.results: if res.status == 'ready': chk = self._get_checkbox(res.row) if chk is not None and chk.isChecked(): items.append(res) if not items: return self._copy_started = True for res in items: res.status = 'copy' # Блокируем ВСЕ чекбоксы после старта копирования. for chk in self._checkboxes: chk.setEnabled(False) self.btn_ok.setEnabled(False) self.btn_ok.setText('Копирование…') self.progress.setRange(0, len(items)) self.progress.setValue(0) self.progress.setVisible(True) self.progress_label.setVisible(True) self.task = axipy.Task(_copy_task) self.task.value_changed.connect( lambda v: self.progress.setValue(int(float(v)))) self.task.finished.connect(self._on_task_finished) self.task.start(items, self) @Slot(axipy.Task) def _on_task_finished(self, task_instance: axipy.Task) -> None: """Обработка завершения фоновой задачи копирования.""" self.progress.setVisible(False) self.progress_label.setVisible(False) # Файлы, до которых задача не дошла (отмена/ошибка задачи). for res in self.results: if res.status == 'copy': res.status = 'canceled' self._set_status(res.row, 'canceled') self.btn_cancel.setEnabled(False) self.btn_ok.setText('Закрыть') self.btn_ok.setEnabled(True) # ---------------- отмена и закрытие ---------------- @Slot() def _on_cancel_clicked(self) -> None: """Обработка нажатия кнопки Отмена.""" if self.task is not None \ and self.task.status == axipy.Task.Status.RUNNING: self.task.cancel() else: self.reject() def closeEvent(self, event) -> None: """ Закрытие окна во время копирования отменяет задачу. :param event: Событие закрытия окна. """ if self.task is not None \ and self.task.status == axipy.Task.Status.RUNNING: self.task.cancel() super().closeEvent(event) # ---------------- обновление строк таблицы ---------------- def _set_status(self, row: int, status: str, message: Optional[str] = None) -> None: """ Устанавливает статус и описание для строки таблицы. :param row: Индекс строки. :param status: Ключ статуса из _STATUS_DISPLAY. :param message: Дополнительное описание (для столбца «Описание»). """ icon, text = _STATUS_DISPLAY.get(status, ('?', status)) self.table.setItem(row, 2, QTableWidgetItem(f'{icon} {text}')) if message is not None: self.table.setItem(row, 3, QTableWidgetItem(message)) def _ui_file_started(self, row: int, idx: int, total: int, name: str) -> None: """Обновление интерфейса при начале копирования файла.""" self._set_status(row, 'copy') self.progress_label.setText(f'Файл {idx + 1}/{total}: {name}…') def _ui_file_progress(self, row: int, idx: int, total: int, name: str, count: int) -> None: """Обновление интерфейса после вставки очередной пачки записей.""" self.progress_label.setText( f'Файл {idx + 1}/{total}: {name} — ' f'скопировано {_fmt(count)} записей…') def _ui_file_done(self, row: int, idx: int, total: int, name: str, count: int) -> None: """Обновление интерфейса при успешном завершении копирования файла.""" self.results[row].status = 'copied' self._set_status(row, 'copied', f'Скопировано записей: {_fmt(count)}') self.progress_label.setText( f'Файл {idx + 1}/{total}: {name} — готово ({_fmt(count)} записей)') def _ui_file_error(self, row: int, message: str) -> None: """Обновление интерфейса при ошибке копирования файла.""" self.results[row].status = 'copy_error' self._set_status(row, 'copy_error', message) # ------------------------------------------------------------------ # Точка входа # ------------------------------------------------------------------ def run() -> None: """ Главная функция скрипта. Открывает диалог выбора файлов, выполняет проверки и показывает диалог результатов. """ selected = axipy.open_files_dialog( filter_arg='MapInfo Tab (*.tab)', title='Выберите файлы для копирования') if not selected: return open_tables: List[axipy.Table] = list(axipy.data_manager.tables) results: List[FileCheckResult] = [] for path in selected: try: results.append(check_file(Path(path), open_tables)) except Exception as e: results.append(FileCheckResult( file_path=Path(path), status='error', message=f'Непредвиденная ошибка: {e}')) dialog = ResultsDialog(results) dialog.exec_() if __name__ == '__main__': run()