79 lines
3.4 KiB
Python
79 lines
3.4 KiB
Python
import json
|
||
import os
|
||
from typing import Dict, Any
|
||
from abc import abstractmethod
|
||
|
||
class BaseMigrator:
|
||
"""Базовый класс для миграторов с общими методами работы с файлами маппинга."""
|
||
|
||
def _load_mapping(self, filename: str) -> Dict[int, int]:
|
||
"""Загружает маппинг из JSON файла."""
|
||
if os.path.exists(filename):
|
||
with open(filename, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
return {int(k): int(v) for k, v in data.items()}
|
||
return {}
|
||
|
||
def _load_nested_mapping(self, filename: str) -> Dict[Any, Any]:
|
||
"""Загрузка вложенного маппинга из JSON-файла (для колонок)."""
|
||
if os.path.exists(filename):
|
||
with open(filename, 'r') as f:
|
||
data = json.load(f)
|
||
|
||
# Для вложенного маппинга колонок: {project_id: {src_col_id: target_col_id}}
|
||
result = {}
|
||
for project_id_str, column_map in data.items():
|
||
project_id = int(project_id_str)
|
||
result[project_id] = {}
|
||
for src_col_str, target_col_id in column_map.items():
|
||
result[project_id][int(src_col_str)] = int(target_col_id)
|
||
return result
|
||
return {}
|
||
|
||
def _save_mapping(self, filename: str, mapping: Dict[int, int]):
|
||
"""Сохраняет маппинг в JSON файл."""
|
||
with open(filename, "w", encoding="utf-8") as f:
|
||
json.dump(mapping, f, ensure_ascii=False, indent=2)
|
||
|
||
def _save_nested_mapping(self, filename: str, mapping: Dict[Any, Any]):
|
||
"""Сохранение вложенного маппинга в JSON-файл."""
|
||
# Преобразуем в формат, который можно сериализовать в JSON
|
||
serializable_mapping = {}
|
||
for project_id, column_map in mapping.items():
|
||
serializable_mapping[str(project_id)] = {str(k): v for k, v in column_map.items()}
|
||
|
||
with open(filename, 'w') as f:
|
||
json.dump(serializable_mapping, f, indent=2)
|
||
|
||
def _delete_mapping_file(self, filename: str) -> bool:
|
||
"""
|
||
Удаляет файл маппинга, если он существует.
|
||
|
||
Args:
|
||
filename: Путь к файлу маппинга.
|
||
Returns:
|
||
bool: True, если файл удален или не существует, False в случае ошибки.
|
||
"""
|
||
try:
|
||
if filename and os.path.exists(filename):
|
||
os.remove(filename)
|
||
print(f"Файл маппинга {filename} удалён")
|
||
return True
|
||
else:
|
||
print("Файл маппинга не найден или не задан")
|
||
return True
|
||
except Exception as e:
|
||
print(f"[{type(e).__name__}] Ошибка при удалении файла маппинга: {e}")
|
||
return False
|
||
|
||
|
||
@abstractmethod
|
||
def migrate(self):
|
||
"""Абстрактный метод для миграции данных."""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def delete_all(self):
|
||
"""Абстрактный метод для удаления всех данных."""
|
||
pass
|