den.veremeytsev/dags_scripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
499d7f4399f0992e06c855daa89735bcfc8ba557
dags_scripts/migration_reporter.py
T

3235 lines
148 KiB
Python
Raw Normal View History

Add DAG and migration_reporter module
2026-10-29 16:14:13 +03:00
#!/usr/bin/env python3
"""
migration_reporter — единый инструмент анализа Kotlin-миграций PostgreSQL.
Пайплайн: [git pull] → анализ .kt файлов → кэш JSON → HTML-отчет (+ Excel c --excel).
HTML строится напрямую из данных анализа (без обратного парсинга Excel)
и содержит вкладку "Батчи" — группировку по дате загрузки файлов в репозиторий.
Команды:
run [git pull] → анализ → кэш → HTML (+ Excel с --excel)
analyze только анализ + кэш JSON (шаг DAG)
excel только Excel + кэш (ручной запуск)
html HTML из кэша (ручной запуск)
Примеры:
python migration_reporter.py run --dir pgsql --skip-pull
python migration_reporter.py run --dir pgsql # c git pull
python migration_reporter.py run --dir pgsql --excel # с Excel-отчетом
python migration_reporter.py analyze --dir pgsql --output c.json # шаг DAG
python migration_reporter.py excel --dir pgsql --output r.xlsx # вручную Excel
python migration_reporter.py html --cache c.json --output r.html # вручную HTML
Выходные артефакты (в папке --output-dir):
migration_report_YYYYMMDD_HHMMSS.changes.json / .html / [.xlsx]
latest.html / [latest.xlsx] — фиксированные имена последнего отчета
"""
import os
import re
import sys
import json
import shutil
import argparse
import subprocess
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from collections import defaultdict, OrderedDict
from enum import Enum
try:
from openpyxl import Workbook
from openpyxl.styles import (
Font, PatternFill, Alignment, Border, Side,
NamedStyle, numbers
)
from openpyxl.utils import get_column_letter
except ImportError:
print("❌ Требуется openpyxl. Установите: pip install openpyxl")
sys.exit(1)
# UTF-8 вывод в консоль (Windows cp1251 без эмодзи, контейнеры без локали)
for _stream in (sys.stdout, sys.stderr):
if _stream and hasattr(_stream, "reconfigure"):
_stream.reconfigure(encoding="utf-8", errors="replace")
class ChangeCategory(Enum):
"""Категории изменений для группировки"""
NEW_TABLE = "🆕 Новая таблица"
DROPPED_TABLE = "🗑️ Удаленная таблица"
NEW_COLUMN = "➕ Новая колонка"
DROPPED_COLUMN = "➖ Удаленная колонка"
MODIFIED_COLUMN = "✏️ Измененная колонка"
NEW_INDEX = "📑 Новый индекс"
DROPPED_INDEX = "📑 Удаленный индекс"
NEW_FOREIGN_KEY = "🔗 Новый внешний ключ"
NEW_CONSTRAINT = "🔒 Новый constraint"
RAW_SQL = "💻 Сырой SQL"
COMMENT = "💬 Комментарий"
OTHER = "📌 Прочее"
@dataclass
class MigrationChange:
"""Одно изменение в миграции"""
sequence: int
category: ChangeCategory
file_name: str
file_path: str
line_number: int
table_name: str
description: str
details: str
code_snippet: str
impact_level: str
migration_date: str = ""
file_created_date: str = ""
file_modified_date: str = ""
author: str = ""
git_last_commit_date: str = ""
git_last_author: str = ""
git_commit_hash: str = ""
git_commit_message: str = ""
programmer_comment: str = ""
db_comment: str = ""
related_tables: List[str] = field(default_factory=list)
columns_info: Dict[str, str] = field(default_factory=dict)
class MigrationAnalyzer:
"""Анализатор миграций с полной поддержкой COMMENT ON и комментариев"""
def __init__(self, directory: Path, debug: bool = False, use_git: bool = True):
self.directory = directory
self.debug = debug
self.use_git = use_git
self.changes: List[MigrationChange] = []
self.all_tables: Dict[str, Dict] = {}
self.sequence_counter = 0
self.files_with_content = 0
self.files_without_content = 0
self.found_patterns = defaultdict(int)
self.git_info_cache = {}
self.db_comments = defaultdict(list)
# Проверяем git
self.git_available = self._check_git_available()
if debug and use_git and not self.git_available:
print("[DEBUG] Git не найден, используется информация о файлах")
# Паттерны
self.patterns = {
# Комментарии
'jsdoc_comment': re.compile(r'/\*\*\s*(.*?)\s*\*/', re.MULTILINE | re.DOTALL),
'line_comment': re.compile(r'//\s*(.*?)$', re.MULTILINE),
'block_comment': re.compile(r'/\*\s*(.*?)\s*\*/', re.MULTILINE | re.DOTALL),
'commit_hash': re.compile(r'\b([0-9a-f]{7,40})\b', re.IGNORECASE),
# COMMENT ON
'comment_on_table': re.compile(
r'COMMENT\s+ON\s+TABLE\s+(\w+(?:\.\w+)?)\s+IS\s+(?:\'([^\']*)\'|"([^"]*)")',
re.IGNORECASE | re.MULTILINE
),
'comment_on_column': re.compile(
r'COMMENT\s+ON\s+COLUMN\s+(\w+(?:\.\w+)?)\.(\w+)\s+IS\s+(?:\'([^\']*)\'|"([^"]*)")',
re.IGNORECASE | re.MULTILINE
),
'comment_on_view': re.compile(
r'COMMENT\s+ON\s+VIEW\s+(\w+(?:\.\w+)?)\s+IS\s+(?:\'([^\']*)\'|"([^"]*)")',
re.IGNORECASE | re.MULTILINE
),
'comment_on_index': re.compile(
r'COMMENT\s+ON\s+INDEX\s+(\w+(?:\.\w+)?)\s+IS\s+(?:\'([^\']*)\'|"([^"]*)")',
re.IGNORECASE | re.MULTILINE
),
# SQL
'raw_sql_exec': re.compile(r'exec\s*\(\s*"""\s*(.*?)\s*"""', re.MULTILINE | re.DOTALL),
'raw_sql_transaction': re.compile(r'transaction\s*\(\s*"""\s*(.*?)\s*"""', re.MULTILINE | re.DOTALL),
'raw_sql_exec_single': re.compile(r'exec\s*\(\s*"((?:[^"\\]|\\.)*)"', re.MULTILINE),
'sql_variable': re.compile( r'(?:@Language\s*\(\s*"SQL"\s*\)\s*)?(?:val|var)\s+\w+\s*=\s*"""\s*(.*?)\s*"""\s*\.trimMargin\s*\(\s*\)', re.MULTILINE | re.DOTALL),
# Exposed DSL
'exposed_table': re.compile(
r'(?:object|class)\s+(\w+)\s*:\s*Table\s*\(\s*"([^"]+)"\s*\)',
re.MULTILINE
),
'exposed_column': re.compile(
r'val\s+(\w+)\s*=\s*(\w+)\s*\(\s*"([^"]+)"',
re.MULTILINE
),
'exposed_reference': re.compile(
r'val\s+(\w+)\s*=\s*reference\s*\(\s*"([^"]+)"\s*,\s*(\w+)',
re.MULTILINE
),
'exposed_index': re.compile(
r'(?:val\s+(\w+)\s*=\s*)?index\s*\(\s*"([^"]+)"',
re.MULTILINE
),
'exposed_pk': re.compile(
r'(?:override\s+)?val\s+primaryKey\s*=\s*PrimaryKey\s*\(([^)]+)\)',
re.MULTILINE
),
# SchemaUtils
'schema_utils': re.compile(
r'SchemaUtils\.(create|drop|createTable|dropTable|createMissingTablesAndColumns)\s*\(([^)]+)\)',
re.MULTILINE
),
# SQL паттерны
'sql_create_table': re.compile(
r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?(\w+(?:\.\w+)?)',
re.IGNORECASE | re.MULTILINE
),
'sql_alter_table_add': re.compile(
r'ALTER\s+TABLE\s+(?:ONLY\s+)?(\w+(?:\.\w+)?)\s+ADD\s+(?:COLUMN\s+)?(?:IF\s+NOT\s+EXISTS\s+)?(\w+)\s+(.+?)(?:,|\n|;|$)',
re.IGNORECASE | re.MULTILINE
),
'sql_alter_table_drop': re.compile(
r'ALTER\s+TABLE\s+(\w+(?:\.\w+)?)\s+DROP\s+(?:COLUMN\s+)?(?:IF\s+EXISTS\s+)?(\w+)',
re.IGNORECASE | re.MULTILINE
),
'sql_alter_column': re.compile(
r'ALTER\s+TABLE\s+(\w+(?:\.\w+)?)\s+ALTER\s+COLUMN\s+(\w+)\s+(.+)',
re.IGNORECASE | re.MULTILINE
),
'sql_create_index': re.compile(
r'CREATE\s+(?:UNIQUE\s+)?INDEX\s+(?:IF\s+NOT\s+EXISTS\s+)?(?:CONCURRENTLY\s+)?(\w+)\s+ON\s+(\w+(?:\.\w+)?)',
re.IGNORECASE | re.MULTILINE
),
'sql_drop_table': re.compile(
r'DROP\s+TABLE\s+(?:IF\s+EXISTS\s+)?(\w+(?:\.\w+)?)',
re.IGNORECASE | re.MULTILINE
),
'sql_foreign_key': re.compile(
r'FOREIGN\s+KEY\s*\(([^)]+)\)\s*REFERENCES\s+(\w+)\s*\(([^)]+)\)',
re.IGNORECASE | re.MULTILINE
),
'sql_insert': re.compile(
r'INSERT\s+INTO\s+(\w+(?:\.\w+)?)',
re.IGNORECASE | re.MULTILINE
),
'sql_update': re.compile(
r'UPDATE\s+(\w+(?:\.\w+)?)\s+SET',
re.IGNORECASE | re.MULTILINE
),
'any_sql_like': re.compile(
r'(?:CREATE|ALTER|DROP|INSERT|UPDATE|DELETE|TRUNCATE|GRANT|REVOKE|COMMENT)\s+',
re.IGNORECASE | re.MULTILINE
),
# Метаинформация
'migration_version': re.compile(r'(?:V|v)(\d+[._]\d+[._]\d+)', re.MULTILINE),
'author_info': re.compile(r'@author\s+(\w+)', re.MULTILINE),
}
def _check_git_available(self) -> bool:
"""Проверяет доступность git"""
try:
result = subprocess.run(
['git', 'rev-parse', '--git-dir'],
cwd=str(self.directory),
capture_output=True,
text=True,
timeout=5
)
return result.returncode == 0
except:
return False
def _get_git_info(self, file_path: Path) -> Dict[str, str]:
"""Получает информацию из git о файле"""
if not self.git_available or not self.use_git:
return {}
cache_key = str(file_path)
if cache_key in self.git_info_cache:
return self.git_info_cache[cache_key]
info = {
'last_commit_date': '',
'last_author': '',
'created_date': '',
'commit_hash': '',
'commit_message': ''
}
try:
# Полная информация о последнем коммите
result = subprocess.run(
['git', 'log', '-1', '--format=%H|%ai|%an|%s', '--', str(file_path.name)],
cwd=str(self.directory),
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0 and result.stdout.strip():
parts = result.stdout.strip().split('|')
if len(parts) >= 4:
info['commit_hash'] = parts[0][:8]
info['last_commit_date'] = parts[1][:10]
info['last_author'] = parts[2]
info['commit_message'] = parts[3][:200]
# Дата первого коммита
result = subprocess.run(
['git', 'log', '--diff-filter=A', '--format=%ai', '--', str(file_path.name)],
cwd=str(self.directory),
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0 and result.stdout.strip():
info['created_date'] = result.stdout.strip().split('\n')[-1][:10]
except Exception as e:
if self.debug:
print(f" [DEBUG] Git ошибка для {file_path.name}: {e}")
self.git_info_cache[cache_key] = info
return info
Update migration_reporter.py
2026-11-02 14:39:36 +03:00
def _get_file_dates(self, file_path: Path, git_info: Optional[Dict[str, str]] = None) -> Dict[str, str]:
"""Даты файла по данным git (источник правды):
created_date — когда файл добавлен в репозиторий (commit add),
modified_date — дата последнего commit'а файла.
Даты файловой системы (mtime/ctime) — только fallback (--no-git, git недоступен)."""
git_info = git_info or {}
fs_dates = self._get_fs_file_dates(file_path)
return {
'created_date': git_info.get('created_date') or fs_dates['created_date'],
'modified_date': git_info.get('last_commit_date') or fs_dates['modified_date'],
}
def _get_fs_file_dates(self, file_path: Path) -> Dict[str, str]:
"""Fallback: даты файлов из файловой системы"""
Add DAG and migration_reporter module
2026-10-29 16:14:13 +03:00
try:
stat = file_path.stat()
return {
'created_date': datetime.fromtimestamp(stat.st_ctime).strftime('%Y-%m-%d'),
'modified_date': datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d'),
}
except:
return {
Update migration_reporter.py
2026-11-02 14:39:36 +03:00
'created_date': '',
'modified_date': '',
Add DAG and migration_reporter module
2026-10-29 16:14:13 +03:00
}
def _extract_programmer_comments(self, content: str) -> Dict[str, Any]:
"""Извлекает комментарии программистов"""
comments = {
'jsdoc': [],
'line_comments': [],
'block_comments': [],
'commit_hashes': []
}
# JSDoc комментарии
for match in self.patterns['jsdoc_comment'].finditer(content):
comment_text = match.group(1).strip()
comment_line = content[:match.start()].count('\n') + 1
comments['jsdoc'].append({'text': comment_text, 'line': comment_line})
# Однострочные комментарии
for match in self.patterns['line_comment'].finditer(content):
comment_text = match.group(1).strip()
comment_line = content[:match.start()].count('\n') + 1
if comment_text and not comment_text.startswith('import'):
comments['line_comments'].append({'text': comment_text, 'line': comment_line})
# Блочные комментарии
for match in self.patterns['block_comment'].finditer(content):
comment_text = match.group(1).strip()
comment_line = content[:match.start()].count('\n') + 1
comments['block_comments'].append({'text': comment_text, 'line': comment_line})
# Commit-хэши из комментариев
all_text = ' '.join(
[c['text'] for c in comments['jsdoc'] + comments['line_comments'] + comments['block_comments']])
for match in self.patterns['commit_hash'].finditer(all_text):
if match.group(1) not in comments['commit_hashes']:
comments['commit_hashes'].append(match.group(1))
return comments
def _extract_db_comments(self, content: str) -> Dict[str, List[Dict[str, str]]]:
"""Извлекает COMMENT ON комментарии"""
comments = {'tables': [], 'columns': [], 'views': [], 'indexes': []}
# COMMENT ON TABLE
for match in self.patterns['comment_on_table'].finditer(content):
table_name = match.group(1)
comment_text = match.group(2) or match.group(3) or ''
line_num = content[:match.start()].count('\n') + 1
cleaned = self._clean_comment_text(comment_text)
comments['tables'].append({'table': table_name, 'comment': cleaned, 'line': line_num})
self.db_comments[table_name].append({'type': 'table', 'comment': cleaned, 'line': line_num})
# COMMENT ON COLUMN
for match in self.patterns['comment_on_column'].finditer(content):
table_name = match.group(1)
column_name = match.group(2)
comment_text = match.group(3) or match.group(4) or ''
line_num = content[:match.start()].count('\n') + 1
cleaned = self._clean_comment_text(comment_text)
comments['columns'].append({
'table': table_name, 'column': column_name,
'comment': cleaned, 'line': line_num
})
self.db_comments[f"{table_name}.{column_name}"].append({
'type': 'column', 'table': table_name, 'column': column_name,
'comment': cleaned, 'line': line_num
})
# COMMENT ON VIEW
for match in self.patterns['comment_on_view'].finditer(content):
view_name = match.group(1)
comment_text = match.group(2) or match.group(3) or ''
line_num = content[:match.start()].count('\n') + 1
comments['views'].append({
'view': view_name,
'comment': self._clean_comment_text(comment_text),
'line': line_num
})
# COMMENT ON INDEX
for match in self.patterns['comment_on_index'].finditer(content):
index_name = match.group(1)
comment_text = match.group(2) or match.group(3) or ''
line_num = content[:match.start()].count('\n') + 1
comments['indexes'].append({
'index': index_name,
'comment': self._clean_comment_text(comment_text),
'line': line_num
})
return comments
def _clean_comment_text(self, text: str) -> str:
"""Очищает текст комментария"""
if not text:
return ""
return text.replace("''", "'").strip()
def _format_comments(self, comments: Dict[str, Any]) -> str:
"""Форматирует комментарии в читаемый вид"""
parts = []
for comment in comments['jsdoc']:
text = comment['text'].strip()
text = re.sub(r'^\s*\*\s?', '', text, flags=re.MULTILINE).strip()
if text:
parts.append(f"/** {text} */")
for comment in comments['line_comments'][:5]:
text = comment['text'].strip()
if text and len(text) > 3:
parts.append(f"// {text}")
for commit_hash in comments['commit_hashes'][:3]:
parts.append(f"Commit: {commit_hash}")
return "\n".join(parts) if parts else ""
def _combine_comments(self, programmer_comment: str, db_comments: Dict[str, List[Dict[str, str]]]) -> str:
"""Объединяет комментарии из кода и COMMENT ON"""
parts = []
if programmer_comment:
parts.append(programmer_comment)
if db_comments['tables']:
for tc in db_comments['tables'][:3]:
parts.append(f"COMMENT ON TABLE {tc['table']}: {tc['comment']}")
if db_comments['columns']:
for cc in db_comments['columns'][:5]:
parts.append(f"COMMENT ON COLUMN {cc['table']}.{cc['column']}: {cc['comment']}")
if db_comments['views']:
for vc in db_comments['views'][:3]:
parts.append(f"COMMENT ON VIEW {vc['view']}: {vc['comment']}")
if db_comments['indexes']:
for ic in db_comments['indexes'][:3]:
parts.append(f"COMMENT ON INDEX {ic['index']}: {ic['comment']}")
return "\n\n".join(parts) if parts else ""
def analyze(self):
"""Основной метод анализа"""
kt_files = sorted(self.directory.rglob("*.kt"))
print(f"[*] Анализирую {len(kt_files)} Kotlin-файлов...")
if self.git_available:
print("[*] Git информация доступна")
for idx, kt_file in enumerate(kt_files, 1):
if self.debug and idx % 100 == 0:
print(f" Прогресс: {idx}/{len(kt_files)}")
try:
self._analyze_file(kt_file)
except Exception as e:
print(f" ⚠️ Ошибка в файле {kt_file.name}: {e}")
if self.debug:
import traceback
traceback.print_exc()
print(f"\n[✓] Файлов с содержимым: {self.files_with_content}")
print(f"[✓] Файлов без содержимого: {self.files_without_content}")
print(f"[✓] Найдено изменений: {len(self.changes)}")
return self.changes
def _analyze_direct_sql(self, content: str, file_path: Path, file_name: str,
migration_date: str, file_dates: Dict, git_info: Dict,
author: str, programmer_comment: str,
commit_hash: str, commit_message: str):
"""Находит CREATE TABLE напрямую в контенте файла"""
# Очищаем контент от trimMargin символов
content_clean = self._clean_trim_margin(content)
# Ищем все CREATE TABLE в контенте
create_pattern = re.compile(
r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[\w.]+?\s*\((.*?)\)\s*;',
re.IGNORECASE | re.DOTALL
)
for match in create_pattern.finditer(content_clean):
sql = match.group(0)
line_num = content_clean[:match.start()].count('\n') + 1
table_match = re.search(r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?([\w.]+)',
sql, re.IGNORECASE)
if table_match:
table_name = table_match.group(1).split('.')[-1]
if self.debug:
print(f" [DEBUG] Найдена CREATE TABLE напрямую: {table_name}")
columns_info = self._extract_columns_from_create_table(sql)
related_tables = self._extract_related_tables(sql, table_name)
db_comment = self._get_table_comment(table_name)
details = self._format_table_details(table_name, columns_info, related_tables)
if db_comment:
details += f"\nКомментарий: {db_comment}"
self._create_change(
ChangeCategory.NEW_TABLE, file_path, file_name, line_num,
table_name, f"Создание таблицы {table_name}",
details, sql[:200], "HIGH",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message,
related_tables, columns_info, db_comment
)
# Также ищем CREATE INDEX
index_pattern = re.compile(
r'CREATE\s+(?:UNIQUE\s+)?INDEX\s+(\w+)\s+ON\s+(\w+)',
re.IGNORECASE
)
for match in index_pattern.finditer(content_clean):
idx_name = match.group(1)
table_name = match.group(2)
line_num = content_clean[:match.start()].count('\n') + 1
self._create_change(
ChangeCategory.NEW_INDEX, file_path, file_name, line_num,
table_name, f"Создание индекса {idx_name}",
f"CREATE INDEX {idx_name} ON {table_name}",
match.group(0), "MEDIUM",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
def _analyze_file(self, file_path: Path):
"""Анализирует один .kt файл"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
except Exception as e:
if self.debug:
print(f" ⚠️ Ошибка чтения {file_path.name}: {e}")
return
file_name = file_path.name
git_info = self._get_git_info(file_path)
Update migration_reporter.py
2026-11-02 14:39:36 +03:00
file_dates = self._get_file_dates(file_path, git_info)
Add DAG and migration_reporter module
2026-10-29 16:14:13 +03:00
migration_date = git_info.get('created_date') or file_dates['created_date']
author = git_info.get('last_author') or self._extract_author(content) or 'неизвестен'
# Извлекаем комментарии
all_comments = self._extract_programmer_comments(content)
programmer_comment = self._format_comments(all_comments)
db_comments = self._extract_db_comments(content)
combined_comment = self._combine_comments(programmer_comment, db_comments)
commit_hash = git_info.get('commit_hash', '')
commit_message = git_info.get('commit_message', '')
content_found = False
# Анализируем COMMENT ON
if db_comments and any(db_comments.values()):
content_found = True
self.found_patterns['comment_on'] += 1
self._analyze_db_comments(db_comments, file_path, file_name, migration_date,
file_dates, git_info, author, combined_comment,
commit_hash, commit_message)
# Анализируем SQL в exec/transaction
if self._has_raw_sql(content):
content_found = True
self.found_patterns['raw_sql'] += 1
self._analyze_raw_sql_all(content, file_path, file_name, migration_date,
file_dates, git_info, author, combined_comment,
commit_hash, commit_message)
# ВАЖНО: Анализируем SQL в переменных (query, sql и т.д.)
if self.patterns['sql_variable'].search(content):
content_found = True
self.found_patterns['sql_variable'] += 1
self._analyze_sql_variables(content, file_path, file_name, migration_date,
file_dates, git_info, author, combined_comment,
commit_hash, commit_message)
# ДОПОЛНИТЕЛЬНО: Ищем SQL в up { } блоках
if 'CREATE TABLE' in content.upper() or 'ALTER TABLE' in content.upper():
content_found = True
self.found_patterns['direct_sql'] += 1
self._analyze_direct_sql(content, file_path, file_name, migration_date,
file_dates, git_info, author, combined_comment,
commit_hash, commit_message)
# Exposed
if self.patterns['exposed_table'].search(content):
content_found = True
self.found_patterns['exposed_table'] += 1
self._analyze_exposed_tables(content, file_path, file_name, migration_date,
file_dates, git_info, author, combined_comment,
commit_hash, commit_message)
# SchemaUtils
if self.patterns['schema_utils'].search(content):
content_found = True
self.found_patterns['schema_utils'] += 1
self._analyze_schema_utils(content, file_path, file_name, migration_date,
file_dates, git_info, author, combined_comment,
commit_hash, commit_message)
if content_found:
self.files_with_content += 1
else:
self.files_without_content += 1
def _create_change(self, category: ChangeCategory, file_path: Path,
file_name: str, line_num: int, table_name: str,
description: str, details: str, code_snippet: str,
impact_level: str, migration_date: str,
file_dates: Dict[str, str], git_info: Dict[str, str],
author: str, programmer_comment: str = "",
commit_hash: str = "", commit_message: str = "",
related_tables: List[str] = None,
columns_info: Dict[str, str] = None,
db_comment: str = ""):
"""Создает объект изменения"""
self.sequence_counter += 1
if related_tables is None:
related_tables = []
if columns_info is None:
columns_info = {}
self.changes.append(MigrationChange(
sequence=self.sequence_counter,
category=category,
file_name=file_name,
file_path=str(file_path),
line_number=line_num,
table_name=table_name,
description=description,
details=details,
code_snippet=code_snippet,
impact_level=impact_level,
migration_date=migration_date,
file_created_date=file_dates.get('created_date', ''),
file_modified_date=file_dates.get('modified_date', ''),
author=author,
git_last_commit_date=git_info.get('last_commit_date', ''),
git_last_author=git_info.get('last_author', ''),
git_commit_hash=commit_hash or git_info.get('commit_hash', ''),
git_commit_message=commit_message or git_info.get('commit_message', ''),
programmer_comment=programmer_comment,
db_comment=db_comment,
related_tables=related_tables,
columns_info=columns_info
))
def _has_raw_sql(self, content: str) -> bool:
return bool(
self.patterns['raw_sql_exec'].search(content) or
self.patterns['raw_sql_transaction'].search(content) or
self.patterns['raw_sql_exec_single'].search(content)
)
def _analyze_db_comments(self, db_comments: Dict, file_path: Path, file_name: str,
migration_date: str, file_dates: Dict, git_info: Dict,
author: str, combined_comment: str,
commit_hash: str, commit_message: str):
"""Создает записи для COMMENT ON"""
for tc in db_comments['tables']:
self._create_change(
ChangeCategory.COMMENT, file_path, file_name, tc['line'],
tc['table'], f"Комментарий к таблице {tc['table']}",
f"COMMENT ON TABLE {tc['table']} IS '{tc['comment']}'",
f"COMMENT ON TABLE {tc['table']}", "LOW",
migration_date, file_dates, git_info, author,
combined_comment, commit_hash, commit_message,
db_comment=tc['comment']
)
for cc in db_comments['columns']:
self._create_change(
ChangeCategory.COMMENT, file_path, file_name, cc['line'],
f"{cc['table']}.{cc['column']}",
f"Комментарий к колонке {cc['table']}.{cc['column']}",
f"COMMENT ON COLUMN {cc['table']}.{cc['column']} IS '{cc['comment']}'",
f"COMMENT ON COLUMN {cc['table']}.{cc['column']}", "LOW",
migration_date, file_dates, git_info, author,
combined_comment, commit_hash, commit_message,
db_comment=cc['comment']
)
for vc in db_comments['views']:
self._create_change(
ChangeCategory.COMMENT, file_path, file_name, vc['line'],
vc['view'], f"Комментарий к представлению {vc['view']}",
f"COMMENT ON VIEW {vc['view']} IS '{vc['comment']}'",
f"COMMENT ON VIEW {vc['view']}", "LOW",
migration_date, file_dates, git_info, author,
combined_comment, commit_hash, commit_message,
db_comment=vc['comment']
)
for ic in db_comments['indexes']:
self._create_change(
ChangeCategory.COMMENT, file_path, file_name, ic['line'],
ic['index'], f"Комментарий к индексу {ic['index']}",
f"COMMENT ON INDEX {ic['index']} IS '{ic['comment']}'",
f"COMMENT ON INDEX {ic['index']}", "LOW",
migration_date, file_dates, git_info, author,
combined_comment, commit_hash, commit_message,
db_comment=ic['comment']
)
def _analyze_raw_sql_all(self, content: str, file_path: Path, file_name: str,
migration_date: str, file_dates: Dict, git_info: Dict,
author: str, programmer_comment: str,
commit_hash: str, commit_message: str):
"""Извлекает SQL из разных источников с диагностикой"""
# Проверяем exec с многострочным SQL
exec_matches = list(self.patterns['raw_sql_exec'].finditer(content))
if self.debug and exec_matches:
print(f" [DEBUG] {file_name}: найдено {len(exec_matches)} exec с многострочным SQL")
for match in exec_matches:
sql = match.group(1).strip()
line_num = content[:match.start()].count('\n') + 1
if self.debug and 'CREATE TABLE' in sql.upper():
print(f" [DEBUG] Найден CREATE TABLE в exec, строка {line_num}")
self._parse_sql_block(sql, file_path, file_name, line_num, migration_date,
file_dates, git_info, author, programmer_comment,
commit_hash, commit_message)
# Проверяем transaction
transaction_matches = list(self.patterns['raw_sql_transaction'].finditer(content))
if self.debug and transaction_matches:
print(f" [DEBUG] {file_name}: найдено {len(transaction_matches)} transaction")
for match in transaction_matches:
sql = match.group(1).strip()
line_num = content[:match.start()].count('\n') + 1
if self.debug and 'CREATE TABLE' in sql.upper():
print(f" [DEBUG] Найден CREATE TABLE в transaction, строка {line_num}")
self._parse_sql_block(sql, file_path, file_name, line_num, migration_date,
file_dates, git_info, author, programmer_comment,
commit_hash, commit_message)
# Проверяем однострочный SQL
single_matches = list(self.patterns['raw_sql_exec_single'].finditer(content))
if self.debug and single_matches:
print(f" [DEBUG] {file_name}: найдено {len(single_matches)} однострочных exec")
for match in single_matches:
sql = match.group(1).strip()
line_num = content[:match.start()].count('\n') + 1
if self.debug and 'CREATE TABLE' in sql.upper():
print(f" [DEBUG] Найден CREATE TABLE в однострочном SQL, строка {line_num}")
self._parse_sql_statement(sql, file_path, file_name, line_num, migration_date,
file_dates, git_info, author, programmer_comment,
commit_hash, commit_message)
# Дополнительно: ищем CREATE TABLE напрямую в контенте
if self.debug:
create_matches = list(re.finditer(r'CREATE\s+TABLE', content, re.IGNORECASE))
if create_matches:
print(f" [DEBUG] {file_name}: найдено {len(create_matches)} CREATE TABLE в файле")
for cm in create_matches[:3]:
line = content[:cm.start()].count('\n') + 1
print(f" Строка {line}: {content[cm.start():cm.start() + 100].strip()}")
def _analyze_sql_variables(self, content: str, file_path: Path, file_name: str,
migration_date: str, file_dates: Dict, git_info: Dict,
author: str, programmer_comment: str,
commit_hash: str, commit_message: str):
"""Анализирует SQL в переменных с поддержкой trimMargin"""
for match in self.patterns['sql_variable'].finditer(content):
sql = match.group(1).strip()
line_num = content[:match.start()].count('\n') + 1
# Убираем символы | из trimMargin
sql_clean = self._clean_trim_margin(sql)
if self.debug:
print(f" [DEBUG] Найден SQL в переменной, строка {line_num}")
print(f" [DEBUG] SQL после очистки: {sql_clean[:200]}")
self._parse_sql_block(sql_clean, file_path, file_name, line_num, migration_date,
file_dates, git_info, author, programmer_comment,
commit_hash, commit_message)
def _clean_trim_margin(self, sql: str) -> str:
"""Очищает SQL от символов trimMargin"""
lines = sql.split('\n')
cleaned_lines = []
for line in lines:
# Убираем ведущие пробелы и символ |
cleaned = re.sub(r'^\s*\|', '', line)
cleaned_lines.append(cleaned)
return '\n'.join(cleaned_lines)
def _parse_sql_block(self, sql: str, file_path: Path, file_name: str,
line_num: int, migration_date: str, file_dates: Dict,
git_info: Dict, author: str, programmer_comment: str,
commit_hash: str, commit_message: str):
"""Парсит блок SQL"""
statements = [s.strip() for s in sql.split(';') if s.strip()]
for statement in statements:
self._parse_sql_statement(statement, file_path, file_name, line_num,
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message)
def _parse_sql_statement(self, sql: str, file_path: Path, file_name: str,
line_num: int, migration_date: str, file_dates: Dict,
git_info: Dict, author: str, programmer_comment: str,
commit_hash: str, commit_message: str):
"""Парсит один SQL запрос с расширенной диагностикой"""
if not sql.strip():
return
sql_clean = re.sub(r'--.*$', '', sql, flags=re.MULTILINE).strip()
sql_upper = sql_clean.upper()
# РАСШИРЕННАЯ ПРОВЕРКА CREATE TABLE
# Проверяем разные варианты написания
if ('CREATE' in sql_upper and 'TABLE' in sql_upper and
not sql_upper.startswith('CREATE INDEX') and
not sql_upper.startswith('CREATE UNIQUE INDEX') and
not sql_upper.startswith('CREATE OR REPLACE')):
# Ищем название таблицы разными способами
table_name = None
# Стандартный паттерн
match = self.patterns['sql_create_table'].search(sql_clean)
if match:
table_name = match.group(1)
# Если не нашли, пробуем другие варианты
if not table_name:
# CREATE TABLE IF NOT EXISTS schema.table_name
match = re.search(r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?(?:[\w]+\.)?(\w+)',
sql_clean, re.IGNORECASE)
if match:
table_name = match.group(1)
# Если нашли имя таблицы
if table_name:
if self.debug:
print(f" [DEBUG] Найдена CREATE TABLE: {table_name}")
print(f" [DEBUG] SQL: {sql_clean[:200]}")
related_tables = self._extract_related_tables(sql_clean, table_name)
columns_info = self._extract_columns_from_create_table(sql_clean)
details = self._format_table_details(table_name, columns_info, related_tables)
db_comment = self._get_table_comment(table_name)
if db_comment:
details += f"\nКомментарий: {db_comment}"
self._create_change(
ChangeCategory.NEW_TABLE, file_path, file_name, line_num,
table_name, f"Создание таблицы {table_name}",
details, sql_clean[:200], "HIGH",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message,
related_tables, columns_info, db_comment
)
return
# ALTER TABLE ADD
elif 'ALTER TABLE' in sql_upper and 'ADD' in sql_upper:
match = self.patterns['sql_alter_table_add'].search(sql_clean)
if match:
table_name = match.group(1)
col_name = match.group(2)
col_def = match.group(3).rstrip(';').strip()[:100]
self._create_change(
ChangeCategory.NEW_COLUMN, file_path, file_name, line_num,
table_name, f"Добавление колонки {col_name} в {table_name}",
f"Тип: {col_def}\nSQL: {sql_clean[:300]}", sql_clean[:200], "MEDIUM",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
# ALTER TABLE DROP
elif 'ALTER TABLE' in sql_upper and 'DROP' in sql_upper:
match = self.patterns['sql_alter_table_drop'].search(sql_clean)
if match:
table_name = match.group(1)
col_name = match.group(2)
self._create_change(
ChangeCategory.DROPPED_COLUMN, file_path, file_name, line_num,
table_name, f"Удаление колонки {col_name} из {table_name}",
f"SQL: {sql_clean[:300]}", sql_clean[:200], "HIGH",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
# ALTER COLUMN
elif 'ALTER TABLE' in sql_upper and 'ALTER COLUMN' in sql_upper:
match = self.patterns['sql_alter_column'].search(sql_clean)
if match:
table_name = match.group(1)
col_name = match.group(2)
alteration = match.group(3).strip()[:100]
self._create_change(
ChangeCategory.MODIFIED_COLUMN, file_path, file_name, line_num,
table_name, f"Изменение колонки {table_name}.{col_name}",
f"Изменение: {alteration}", sql_clean[:200], "MEDIUM",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
# CREATE INDEX
elif sql_upper.startswith('CREATE INDEX') or sql_upper.startswith('CREATE UNIQUE INDEX'):
match = self.patterns['sql_create_index'].search(sql_clean)
if match:
idx_name = match.group(1)
table_name = match.group(2)
is_concurrent = 'CONCURRENTLY' in sql_upper
concurrent_note = " (CONCURRENTLY)" if is_concurrent else " (⚠️ без CONCURRENTLY)"
self._create_change(
ChangeCategory.NEW_INDEX, file_path, file_name, line_num,
table_name, f"Создание индекса {idx_name}{concurrent_note}",
f"SQL: {sql_clean[:300]}", sql_clean[:200],
"HIGH" if not is_concurrent else "MEDIUM",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
# DROP TABLE
elif sql_upper.startswith('DROP TABLE'):
match = self.patterns['sql_drop_table'].search(sql_clean)
if match:
table_name = match.group(1)
self._create_change(
ChangeCategory.DROPPED_TABLE, file_path, file_name, line_num,
table_name, f"Удаление таблицы {table_name}",
f"SQL: {sql_clean[:300]}", sql_clean[:200], "HIGH",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
# INSERT
elif sql_upper.startswith('INSERT INTO'):
match = self.patterns['sql_insert'].search(sql_clean)
if match:
table_name = match.group(1)
self._create_change(
ChangeCategory.RAW_SQL, file_path, file_name, line_num,
table_name, f"Вставка данных в {table_name}",
f"SQL: {sql_clean[:300]}", sql_clean[:200], "LOW",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
# UPDATE
elif sql_upper.startswith('UPDATE'):
match = self.patterns['sql_update'].search(sql_clean)
if match:
table_name = match.group(1)
self._create_change(
ChangeCategory.RAW_SQL, file_path, file_name, line_num,
table_name, f"Обновление данных в {table_name}",
f"SQL: {sql_clean[:300]}", sql_clean[:200], "LOW",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
# FOREIGN KEY
elif 'FOREIGN KEY' in sql_upper:
match = self.patterns['sql_foreign_key'].search(sql_clean)
if match:
fk_cols = match.group(1).strip()
ref_table = match.group(2).strip()
ref_cols = match.group(3).strip()
table_match = re.search(r'ALTER\s+TABLE\s+(\w+)', sql_clean, re.IGNORECASE)
table_name = table_match.group(1) if table_match else "неизвестно"
self._create_change(
ChangeCategory.NEW_FOREIGN_KEY, file_path, file_name, line_num,
table_name, f"Внешний ключ {fk_cols} → {ref_table}({ref_cols})",
f"SQL: {sql_clean[:300]}", sql_clean[:200], "MEDIUM",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message,
[ref_table]
)
# Другие SQL
elif re.match(r'(?:CREATE|ALTER|DROP|INSERT|UPDATE|DELETE|TRUNCATE|GRANT|REVOKE)\s+',
sql_upper, re.IGNORECASE):
self._create_change(
ChangeCategory.RAW_SQL, file_path, file_name, line_num,
"разные", "Выполнение SQL",
f"SQL: {sql_clean[:300]}", sql_clean[:200], "LOW",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
def _extract_related_tables(self, content: str, current_table: str) -> List[str]:
"""Извлекает связанные таблицы"""
related = []
for match in re.finditer(r'REFERENCES\s+(\w+)', content, re.IGNORECASE):
ref_table = match.group(1)
if ref_table != current_table and ref_table not in related:
related.append(ref_table)
return related
def _extract_columns_from_create_table(self, sql: str) -> Dict[str, str]:
"""Извлекает колонки из CREATE TABLE с расширенной поддержкой"""
columns = {}
# Ищем тело CREATE TABLE
body_match = re.search(r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[\w.]+?\s*\((.*?)\)\s*;',
sql, re.IGNORECASE | re.DOTALL)
if not body_match:
body_match = re.search(r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[\w.]+?\s*\((.*?)\)',
sql, re.IGNORECASE | re.DOTALL)
if body_match:
body = body_match.group(1)
definitions = self._split_definitions(body)
for definition in definitions:
definition = definition.strip()
if not definition:
continue
# Пропускаем constraints
if any(definition.upper().startswith(kw) for kw in
['PRIMARY KEY', 'FOREIGN KEY', 'UNIQUE', 'CHECK', 'CONSTRAINT', 'INDEX']):
continue
# Расширенный паттерн для колонок
# Поддерживает: name TYPE, "name" TYPE, name TYPE GENERATED ALWAYS AS IDENTITY
col_match = re.match(
r'"?(\w+)"?\s+([A-Za-z0-9_]+(?:\s+GENERATED\s+ALWAYS\s+AS\s+IDENTITY)?)',
definition, re.IGNORECASE)
if col_match:
col_name = col_match.group(1)
col_type = col_match.group(2).strip()
# Добавляем модификаторы
modifiers = []
if 'NOT NULL' in definition.upper():
modifiers.append("NOT NULL")
if 'PRIMARY KEY' in definition.upper():
modifiers.append("PK")
if 'DEFAULT' in definition.upper():
modifiers.append("DEFAULT")
if 'REFERENCES' in definition.upper():
# Извлекаем ссылку
ref_match = re.search(r'REFERENCES\s+(\w+)', definition, re.IGNORECASE)
if ref_match:
modifiers.append(f"FK→{ref_match.group(1)}")
if 'GENERATED' in definition.upper():
modifiers.append("IDENTITY")
if 'NULL' in definition.upper() and 'NOT NULL' not in definition.upper():
modifiers.append("NULL")
if modifiers:
col_type += " " + " ".join(modifiers)
columns[col_name] = col_type
return columns
def _split_definitions(self, body: str) -> List[str]:
"""Разбивает тело CREATE TABLE на отдельные определения"""
definitions = []
current = []
paren_depth = 0
in_quotes = False
quote_char = None
for char in body:
# Обработка кавычек
if char in ["'", '"']:
if not in_quotes:
in_quotes = True
quote_char = char
elif char == quote_char:
in_quotes = False
quote_char = None
# Обработка скобок (только вне кавычек)
if not in_quotes:
if char == '(':
paren_depth += 1
elif char == ')':
paren_depth -= 1
elif char == ',' and paren_depth == 0:
definitions.append(''.join(current))
current = []
continue
current.append(char)
if current:
definitions.append(''.join(current))
return definitions
def _format_table_details(self, table_name: str, columns: Dict[str, str],
related_tables: List[str]) -> str:
"""Форматирует детали таблицы"""
details = f"Таблица: {table_name}\n"
if columns:
details += "\nКолонки:\n"
for col_name, col_type in columns.items():
details += f" • {col_name}: {col_type}\n"
if related_tables:
details += f"\nСвязанные таблицы: {', '.join(related_tables)}"
return details
def _analyze_exposed_tables(self, content: str, file_path: Path, file_name: str,
migration_date: str, file_dates: Dict, git_info: Dict,
author: str, programmer_comment: str,
commit_hash: str, commit_message: str):
"""Анализирует Exposed таблицы"""
for match in self.patterns['exposed_table'].finditer(content):
class_name = match.group(1)
table_name = match.group(2)
line_num = content[:match.start()].count('\n') + 1
body_start = content.find('{', match.end())
if body_start == -1:
continue
body_end = self._find_matching_brace(content, body_start)
if body_end == -1:
continue
body = content[body_start:body_end]
columns = OrderedDict()
related_tables = []
for col_match in self.patterns['exposed_column'].finditer(body):
col_type = col_match.group(2)
col_db_name = col_match.group(3)
remaining = body[col_match.end():col_match.end() + 200]
modifiers = []
if 'nullable()' in remaining.split('\n')[0]:
modifiers.append("nullable")
if 'autoIncrement()' in remaining.split('\n')[0]:
modifiers.append("auto-increment")
modifier_str = f" ({', '.join(modifiers)})" if modifiers else ""
columns[col_db_name] = f"{col_type}{modifier_str}"
for ref_match in self.patterns['exposed_reference'].finditer(body):
ref_table = ref_match.group(3)
if ref_table not in related_tables:
related_tables.append(ref_table)
details = f"Класс: {class_name}\nТаблица: {table_name}"
if columns:
details += f"\nКолонок: {len(columns)}"
for col_name, col_type in list(columns.items())[:10]:
details += f"\n • {col_name}: {col_type}"
if related_tables:
details += f"\nСвязанные таблицы: {', '.join(related_tables)}"
# Получаем комментарий из БД
db_comment = self._get_table_comment(table_name)
if db_comment:
details += f"\nКомментарий: {db_comment}"
self._create_change(
ChangeCategory.NEW_TABLE, file_path, file_name, line_num,
table_name, f"Создание таблицы {table_name} (Exposed)",
details, f"object {class_name} : Table(\"{table_name}\")", "HIGH",
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message,
related_tables, dict(columns), db_comment
)
def _analyze_schema_utils(self, content: str, file_path: Path, file_name: str,
migration_date: str, file_dates: Dict, git_info: Dict,
author: str, programmer_comment: str,
commit_hash: str, commit_message: str):
"""Анализирует SchemaUtils вызовы"""
for match in self.patterns['schema_utils'].finditer(content):
operation = match.group(1).lower()
tables = match.group(2)
line_num = content[:match.start()].count('\n') + 1
if 'create' in operation:
category = ChangeCategory.NEW_TABLE
desc = f"Создание через SchemaUtils.{operation}"
impact = "HIGH"
elif 'drop' in operation:
category = ChangeCategory.DROPPED_TABLE
desc = f"Удаление через SchemaUtils.{operation}"
impact = "HIGH"
else:
category = ChangeCategory.OTHER
desc = f"SchemaUtils.{operation}"
impact = "MEDIUM"
self._create_change(
category, file_path, file_name, line_num,
tables, desc, f"SchemaUtils.{operation}({tables})",
match.group(0), impact,
migration_date, file_dates, git_info, author,
programmer_comment, commit_hash, commit_message
)
def _find_matching_brace(self, text: str, start: int) -> int:
depth = 0
for i in range(start, len(text)):
if text[i] == '{':
depth += 1
elif text[i] == '}':
depth -= 1
if depth == 0:
return i
return -1
def _extract_author(self, content: str) -> str:
match = self.patterns['author_info'].search(content)
return match.group(1) if match else ""
def _get_table_comment(self, table_name: str) -> str:
"""Получает комментарий к таблице из db_comments"""
if table_name in self.db_comments:
for comment_info in self.db_comments[table_name]:
if comment_info['type'] == 'table':
return comment_info['comment']
return ""
class ExcelReportGenerator:
"""Генератор Excel-отчета"""
COLORS = {
'header_bg': PatternFill(start_color="2F5496", end_color="2F5496", fill_type="solid"),
'header_font': Font(color="FFFFFF", bold=True, size=11),
'high_impact': PatternFill(start_color="FFE0E0", end_color="FFE0E0", fill_type="solid"),
'medium_impact': PatternFill(start_color="FFF3E0", end_color="FFF3E0", fill_type="solid"),
'low_impact': PatternFill(start_color="E8F5E9", end_color="E8F5E9", fill_type="solid"),
'recent_change': PatternFill(start_color="FFF9C4", end_color="FFF9C4", fill_type="solid"),
'comment_bg': PatternFill(start_color="F3E5F5", end_color="F3E5F5", fill_type="solid"),
'db_comment_bg': PatternFill(start_color="E0F7FA", end_color="E0F7FA", fill_type="solid"),
'category_colors': {
ChangeCategory.NEW_TABLE: "1B5E20",
ChangeCategory.DROPPED_TABLE: "B71C1C",
ChangeCategory.NEW_COLUMN: "0D47A1",
ChangeCategory.DROPPED_COLUMN: "D32F2F",
ChangeCategory.NEW_INDEX: "4A148C",
ChangeCategory.NEW_FOREIGN_KEY: "E65100",
ChangeCategory.RAW_SQL: "616161",
ChangeCategory.COMMENT: "00838F",
}
}
def __init__(self, changes: List[MigrationChange], output_path: str):
self.changes = changes
self.output_path = output_path
self.wb = Workbook()
self.recent_threshold = datetime.now().strftime('%Y-%m-%d')
def generate(self):
"""Генерирует полный Excel-отчет"""
print("\n[*] Генерирую Excel-отчет...")
self._create_table_structure_sheet() # НОВЫЙ ЛИСТ
self._create_summary_sheet()
self._create_changes_sheet()
self._create_comments_sheet()
self._create_db_comments_sheet()
self._create_by_table_sheet()
self._create_by_file_sheet()
self._create_impact_analysis_sheet()
self._create_timeline_sheet()
self._create_recent_changes_sheet()
self.wb.save(self.output_path)
print(f"[✓] Отчет сохранен: {self.output_path}")
print(f" Листов: {len(self.wb.sheetnames)}")
print(f" Изменений: {len(self.changes)}")
def _create_table_structure_sheet(self):
"""Создает лист со структурой созданных таблиц"""
ws = self.wb.create_sheet("📖 Структура таблиц", 0) # Первый лист
# Заголовок
ws.merge_cells('A1:H1')
ws['A1'] = "СТРУКТУРА СОЗДАННЫХ ТАБЛИЦ"
ws['A1'].font = Font(bold=True, size=16, color="1B5E20")
ws['A1'].alignment = Alignment(horizontal='center', vertical='center')
ws.merge_cells('A2:H2')
ws['A2'] = f"Сгенерировано: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
ws['A2'].font = Font(color="757575", size=10)
ws['A2'].alignment = Alignment(horizontal='center')
# Собираем все созданные таблицы
created_tables = {}
for change in self.changes:
if change.category == ChangeCategory.NEW_TABLE:
table_name = change.table_name
if table_name not in created_tables:
created_tables[table_name] = {
'columns': change.columns_info,
'comment': change.db_comment or change.programmer_comment,
'related_tables': change.related_tables,
'created_date': change.file_created_date,
'modified_date': change.file_modified_date,
'file': change.file_name,
'author': change.author or change.git_last_author,
'commit': change.git_commit_hash
}
else:
# Обновляем существующую запись
if change.columns_info:
created_tables[table_name]['columns'].update(change.columns_info)
if change.db_comment and not created_tables[table_name]['comment']:
created_tables[table_name]['comment'] = change.db_comment
if change.related_tables:
created_tables[table_name]['related_tables'].extend(
[t for t in change.related_tables
if t not in created_tables[table_name]['related_tables']]
)
row = 4
for table_name in sorted(created_tables.keys()):
info = created_tables[table_name]
# Заголовок таблицы
ws.merge_cells(f'A{row}:H{row}')
cell = ws.cell(row=row, column=1, value=f"📋 ТАБЛИЦА: {table_name}")
cell.font = Font(bold=True, size=13, color="FFFFFF")
cell.fill = PatternFill(start_color="1B5E20", end_color="1B5E20", fill_type="solid")
cell.alignment = Alignment(horizontal='left', vertical='center')
# Применяем стиль ко всей строке
for col in range(1, 9):
ws.cell(row=row, column=col).fill = PatternFill(
start_color="1B5E20", end_color="1B5E20", fill_type="solid"
)
ws.cell(row=row, column=col).border = Border(
left=Side(style='thin', color='1B5E20'),
right=Side(style='thin', color='1B5E20'),
top=Side(style='thin', color='1B5E20'),
bottom=Side(style='thin', color='1B5E20')
)
row += 1
# Комментарий
if info['comment']:
ws.merge_cells(f'A{row}:H{row}')
comment_cell = ws.cell(row=row, column=1, value=f"💬 Комментарий: {info['comment']}")
comment_cell.font = Font(italic=True, color="4A148C")
comment_cell.fill = PatternFill(start_color="F3E5F5", end_color="F3E5F5", fill_type="solid")
comment_cell.alignment = Alignment(wrap_text=True, vertical='top')
for col in range(1, 9):
ws.cell(row=row, column=col).fill = PatternFill(
start_color="F3E5F5", end_color="F3E5F5", fill_type="solid"
)
row += 1
# Информация о создании
info_parts = []
if info['created_date']:
info_parts.append(f"📅 Создана: {info['created_date']}")
if info['modified_date']:
info_parts.append(f"✏️ Изменена: {info['modified_date']}")
if info['file']:
info_parts.append(f"📁 Файл: {info['file']}")
if info['author']:
info_parts.append(f"👤 Автор: {info['author']}")
if info['commit']:
info_parts.append(f"🔗 Commit: {info['commit']}")
if info_parts:
ws.merge_cells(f'A{row}:H{row}')
info_cell = ws.cell(row=row, column=1, value=" | ".join(info_parts))
info_cell.font = Font(size=9, color="616161")
info_cell.alignment = Alignment(horizontal='left')
row += 1
# Заголовки колонок
col_headers = ["Колонка", "Тип данных", "COMMENT ON", "Связи", "PK", "FK", "NOT NULL", "Default"]
for col, header in enumerate(col_headers, 1):
cell = ws.cell(row=row, column=col, value=header)
cell.font = Font(bold=True, color="1B5E20")
cell.fill = PatternFill(start_color="E8F5E9", end_color="E8F5E9", fill_type="solid")
cell.alignment = Alignment(horizontal='center')
cell.border = Border(
left=Side(style='thin', color='dee2e6'),
right=Side(style='thin', color='dee2e6'),
top=Side(style='thin', color='dee2e6'),
bottom=Side(style='thin', color='dee2e6')
)
row += 1
# Данные колонок
if info['columns']:
for col_name, col_type in sorted(info['columns'].items()):
# Определяем атрибуты колонки
is_pk = 'PK' in col_type.upper() or 'PRIMARY' in col_type.upper()
is_fk = 'REFERENCES' in col_type.upper() or 'FK' in col_type.upper()
is_not_null = 'NOT NULL' in col_type.upper()
has_default = 'DEFAULT' in col_type.upper()
# Извлекаем чистый тип
clean_type = col_type
for attr in ['NOT NULL', 'NULL', 'PRIMARY KEY', 'PK', 'REFERENCES', 'DEFAULT', 'UNIQUE', 'CHECK']:
clean_type = re.sub(rf'\s*{attr}.*$', '', clean_type, flags=re.IGNORECASE)
# Ищем комментарий к колонке
col_comment = ""
for change in self.changes:
if (change.category == ChangeCategory.COMMENT and
change.table_name == f"{table_name}.{col_name}"):
col_comment = change.db_comment
break
# Ищем связи
fk_ref = ""
fk_match = re.search(r'REFERENCES\s+(\w+)', col_type, re.IGNORECASE)
if fk_match:
fk_ref = f"→ {fk_match.group(1)}"
ws.cell(row=row, column=1, value=col_name).font = Font(bold=True)
ws.cell(row=row, column=2, value=clean_type.strip())
comment_cell = ws.cell(row=row, column=3, value=col_comment)
if col_comment:
comment_cell.fill = PatternFill(start_color="E0F7FA", end_color="E0F7FA", fill_type="solid")
ws.cell(row=row, column=4, value=fk_ref)
# Отметки
ws.cell(row=row, column=5, value="✓" if is_pk else "")
ws.cell(row=row, column=6, value="✓" if is_fk else "")
ws.cell(row=row, column=7, value="✓" if is_not_null else "")
ws.cell(row=row, column=8, value="✓" if has_default else "")
# Выравнивание отметок
for col in range(5, 9):
ws.cell(row=row, column=col).alignment = Alignment(horizontal='center')
# Границы
for col in range(1, 9):
ws.cell(row=row, column=col).border = Border(
left=Side(style='thin', color='dee2e6'),
right=Side(style='thin', color='dee2e6'),
top=Side(style='thin', color='dee2e6'),
bottom=Side(style='thin', color='dee2e6')
)
row += 1
else:
ws.merge_cells(f'A{row}:H{row}')
ws.cell(row=row, column=1, value="Колонки не определены")
ws.cell(row=row, column=1).font = Font(color="999999", italic=True)
row += 1
# Связанные таблицы
if info['related_tables']:
ws.merge_cells(f'A{row}:H{row}')
fk_header = ws.cell(row=row, column=1, value="🔗 СВЯЗАННЫЕ ТАБЛИЦЫ:")
fk_header.font = Font(bold=True, color="E65100")
fk_header.fill = PatternFill(start_color="FFF3E0", end_color="FFF3E0", fill_type="solid")
for col in range(1, 9):
ws.cell(row=row, column=col).fill = PatternFill(
start_color="FFF3E0", end_color="FFF3E0", fill_type="solid"
)
row += 1
ws.merge_cells(f'A{row}:H{row}')
ws.cell(row=row, column=1, value=", ".join(info['related_tables']))
ws.cell(row=row, column=1).font = Font(color="E65100")
row += 1
# Разделитель
row += 1
ws.merge_cells(f'A{row}:H{row}')
ws.cell(row=row, column=1, value="")
row += 1
# Ширина колонок
widths = [25, 30, 40, 20, 8, 8, 10, 10]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
# Закрепление
ws.freeze_panes = 'A5'
print(f"[✓] Лист 'Структура таблиц' создан: {len(created_tables)} таблиц")
def _create_summary_sheet(self):
"""Создает лист с общей сводкой"""
ws = self.wb.create_sheet("📊 Общая сводка") # ИСПРАВЛЕНО: создаем новый лист
ws.merge_cells('A1:H1')
ws['A1'] = "ОТЧЕТ ПО МИГРАЦИЯМ БАЗЫ ДАННЫХ"
ws['A1'].font = Font(bold=True, size=16, color="2F5496")
ws['A1'].alignment = Alignment(horizontal='center', vertical='center')
ws.merge_cells('A2:H2')
ws['A2'] = f"Сгенерирован: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
ws['A2'].alignment = Alignment(horizontal='center')
ws['A2'].font = Font(color="757575", size=10)
if self.changes:
dates = set(c.file_modified_date for c in self.changes if c.file_modified_date)
comments_count = sum(1 for c in self.changes if c.programmer_comment)
db_comments_count = sum(1 for c in self.changes if c.db_comment)
commits_count = sum(1 for c in self.changes if c.git_commit_hash)
row = 3
ws.merge_cells(f'A{row}:H{row}')
ws[f'A{row}'] = (f"Период: {min(dates)} — {max(dates)} | "
f"Комментариев: {comments_count} | "
f"COMMENT ON: {db_comments_count} | "
f"Commit'ов: {commits_count}")
ws[f'A{row}'].font = Font(color="424242", size=10)
ws[f'A{row}'].alignment = Alignment(horizontal='center')
row = 5
ws.merge_cells(f'A{row}:H{row}')
ws[f'A{row}'] = "СТАТИСТИКА ПО КАТЕГОРИЯМ"
ws[f'A{row}'].font = Font(bold=True, size=12, color="2F5496")
row = 6
headers = ["Категория", "Количество", "%", "Влияние", "Комментарии", "COMMENT ON", "Commit'ы", "Описание"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=row, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = self.COLORS['header_bg']
cell.alignment = Alignment(horizontal='center')
category_counts = defaultdict(int)
category_comments = defaultdict(int)
category_db_comments = defaultdict(int)
category_commits = defaultdict(int)
for change in self.changes:
category_counts[change.category] += 1
if change.programmer_comment:
category_comments[change.category] += 1
if change.db_comment:
category_db_comments[change.category] += 1
if change.git_commit_hash:
category_commits[change.category] += 1
total = len(self.changes)
row = 7
for category, count in sorted(category_counts.items(), key=lambda x: x[1], reverse=True):
impact = self._get_category_impact(category)
ws.cell(row=row, column=1, value=category.value).font = Font(
color=self.COLORS['category_colors'].get(category, "000000"), bold=True)
ws.cell(row=row, column=2, value=count).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=3,
value=f"{(count / total) * 100:.1f}%" if total > 0 else "0%").alignment = Alignment(
horizontal='center')
impact_cell = ws.cell(row=row, column=4, value=impact)
impact_cell.alignment = Alignment(horizontal='center')
if impact == "HIGH":
impact_cell.fill = self.COLORS['high_impact']
elif impact == "MEDIUM":
impact_cell.fill = self.COLORS['medium_impact']
else:
impact_cell.fill = self.COLORS['low_impact']
ws.cell(row=row, column=5, value=category_comments[category]).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=6, value=category_db_comments[category]).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=7, value=category_commits[category]).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=8, value=self._get_category_description(category))
row += 1
row += 1
ws.cell(row=row, column=1, value="ВСЕГО").font = Font(bold=True)
ws.cell(row=row, column=2, value=total).font = Font(bold=True)
widths = [30, 15, 10, 18, 15, 15, 15, 50]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
def _create_changes_sheet(self):
"""Создает лист со всеми изменениями"""
ws = self.wb.create_sheet("📋 Все изменения")
headers = [
"№", "Категория", "Файл", "Строка", "Таблица",
"Описание", "Детали", "Код", "Влияние",
"Дата миграции", "Файл создан", "Файл изменен",
"Git commit", "Commit message", "Git автор",
"Комментарий программиста", "COMMENT ON", "Связанные таблицы", "Автор"
]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF", size=11)
cell.fill = self.COLORS['header_bg']
cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
for row, change in enumerate(self.changes, 2):
ws.cell(row=row, column=1, value=change.sequence).alignment = Alignment(horizontal='center')
cat_cell = ws.cell(row=row, column=2, value=change.category.value)
cat_color = self.COLORS['category_colors'].get(change.category, "000000")
cat_cell.font = Font(color=cat_color, bold=True)
ws.cell(row=row, column=3, value=change.file_name)
ws.cell(row=row, column=4, value=change.line_number).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=5, value=change.table_name).font = Font(bold=True)
ws.cell(row=row, column=6, value=change.description)
details_cell = ws.cell(row=row, column=7, value=change.details)
details_cell.alignment = Alignment(wrap_text=True, vertical='top')
ws.cell(row=row, column=8, value=change.code_snippet)
ws.cell(row=row, column=8).font = Font(name="Courier New", size=9)
impact_cell = ws.cell(row=row, column=9, value=change.impact_level)
impact_cell.alignment = Alignment(horizontal='center')
if change.impact_level == "HIGH":
impact_cell.fill = self.COLORS['high_impact']
impact_cell.font = Font(bold=True, color="C62828")
elif change.impact_level == "MEDIUM":
impact_cell.fill = self.COLORS['medium_impact']
else:
impact_cell.fill = self.COLORS['low_impact']
ws.cell(row=row, column=10, value=change.migration_date).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=11, value=change.file_created_date).alignment = Alignment(horizontal='center')
mod_cell = ws.cell(row=row, column=12, value=change.file_modified_date)
mod_cell.alignment = Alignment(horizontal='center')
if change.file_modified_date >= self.recent_threshold:
mod_cell.fill = self.COLORS['recent_change']
ws.cell(row=row, column=13, value=change.git_commit_hash).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=14, value=change.git_commit_message)
ws.cell(row=row, column=15, value=change.git_last_author)
comment_cell = ws.cell(row=row, column=16, value=change.programmer_comment)
comment_cell.alignment = Alignment(wrap_text=True)
if change.programmer_comment:
comment_cell.fill = self.COLORS['comment_bg']
db_comment_cell = ws.cell(row=row, column=17, value=change.db_comment)
db_comment_cell.alignment = Alignment(wrap_text=True)
if change.db_comment:
db_comment_cell.fill = self.COLORS['db_comment_bg']
related_str = ", ".join(change.related_tables) if change.related_tables else ""
ws.cell(row=row, column=18, value=related_str)
ws.cell(row=row, column=19, value=change.author)
widths = [5, 20, 25, 8, 18, 35, 50, 40, 10, 12, 12, 12, 12, 35, 15, 40, 35, 25, 12]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
if len(self.changes) > 0:
ws.auto_filter.ref = f"A1:{get_column_letter(len(headers))}{len(self.changes) + 1}"
ws.freeze_panes = 'A2'
def _create_comments_sheet(self):
"""Создает лист с комментариями программистов"""
ws = self.wb.create_sheet("💬 Комментарии")
changes_with_comments = [c for c in self.changes if c.programmer_comment]
ws.merge_cells('A1:H1')
ws['A1'] = "КОММЕНТАРИИ ПРОГРАММИСТОВ К МИГРАЦИЯМ"
ws['A1'].font = Font(bold=True, size=14, color="6A1B9A")
ws['A1'].alignment = Alignment(horizontal='center')
ws.merge_cells('A2:H2')
ws['A2'] = f"Найдено комментариев: {len(changes_with_comments)}"
ws['A2'].font = Font(color="757575")
ws['A2'].alignment = Alignment(horizontal='center')
if not changes_with_comments:
ws.merge_cells('A4:H4')
ws['A4'] = "Комментарии не найдены"
ws['A4'].font = Font(color="757575", italic=True)
ws['A4'].alignment = Alignment(horizontal='center')
return
headers = ["№", "Таблица", "Описание", "Комментарий",
"Commit", "Commit message", "Файл", "Дата"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=4, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = PatternFill(start_color="6A1B9A", end_color="6A1B9A", fill_type="solid")
row = 5
for change in changes_with_comments:
ws.cell(row=row, column=1, value=change.sequence).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=2, value=change.table_name).font = Font(bold=True)
ws.cell(row=row, column=3, value=change.description)
comment_cell = ws.cell(row=row, column=4, value=change.programmer_comment)
comment_cell.alignment = Alignment(wrap_text=True, vertical='top')
comment_cell.fill = self.COLORS['comment_bg']
ws.cell(row=row, column=5, value=change.git_commit_hash)
ws.cell(row=row, column=6, value=change.git_commit_message)
ws.cell(row=row, column=7, value=change.file_name)
ws.cell(row=row, column=8, value=change.git_last_commit_date)
row += 1
widths = [5, 20, 35, 50, 12, 40, 25, 12]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
ws.freeze_panes = 'A5'
def _create_db_comments_sheet(self):
"""Создает лист с COMMENT ON комментариями"""
ws = self.wb.create_sheet("💬 COMMENT ON")
comment_changes = [c for c in self.changes if c.db_comment]
ws.merge_cells('A1:F1')
ws['A1'] = "КОММЕНТАРИИ ИЗ БАЗЫ ДАННЫХ (COMMENT ON)"
ws['A1'].font = Font(bold=True, size=14, color="00838F")
ws['A1'].alignment = Alignment(horizontal='center')
ws.merge_cells('A2:F2')
ws['A2'] = f"Найдено COMMENT ON комментариев: {len(comment_changes)}"
ws['A2'].font = Font(color="757575")
ws['A2'].alignment = Alignment(horizontal='center')
if not comment_changes:
ws.merge_cells('A4:F4')
ws['A4'] = "COMMENT ON комментарии не найдены"
ws['A4'].font = Font(color="757575", italic=True)
ws['A4'].alignment = Alignment(horizontal='center')
return
headers = ["№", "Объект", "Тип", "Комментарий", "Файл", "Дата"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=4, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = PatternFill(start_color="00838F", end_color="00838F", fill_type="solid")
row = 5
for change in comment_changes:
ws.cell(row=row, column=1, value=change.sequence).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=2, value=change.table_name).font = Font(bold=True)
if 'COLUMN' in change.description:
obj_type = "Колонка"
elif 'VIEW' in change.description:
obj_type = "Представление"
elif 'INDEX' in change.description:
obj_type = "Индекс"
else:
obj_type = "Таблица"
ws.cell(row=row, column=3, value=obj_type)
comment_cell = ws.cell(row=row, column=4, value=change.db_comment)
comment_cell.alignment = Alignment(wrap_text=True)
comment_cell.fill = self.COLORS['db_comment_bg']
ws.cell(row=row, column=5, value=change.file_name)
ws.cell(row=row, column=6, value=change.file_modified_date)
row += 1
widths = [5, 30, 15, 60, 25, 12]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
ws.freeze_panes = 'A5'
def _create_by_table_sheet(self):
"""Создает лист с группировкой по таблицам"""
ws = self.wb.create_sheet("📊 По таблицам")
changes_by_table = defaultdict(list)
for change in self.changes:
changes_by_table[change.table_name].append(change)
headers = ["Таблица", "Изменений", "Первое появление", "Последнее изменение",
"Дата создания", "Дата изменения", "Связанные таблицы",
"Комментарии", "COMMENT ON", "Commit", "Категории", "Файлы"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = self.COLORS['header_bg']
row = 2
for table_name, changes in sorted(changes_by_table.items()):
categories = set(c.category.value for c in changes)
files = set(c.file_name for c in changes)
all_related = set()
all_comments = []
all_db_comments = []
all_commits = set()
for change in changes:
all_related.update(change.related_tables)
if change.programmer_comment:
all_comments.append(change.programmer_comment)
if change.db_comment:
all_db_comments.append(change.db_comment)
if change.git_commit_hash:
all_commits.add(change.git_commit_hash)
first_change = min(changes, key=lambda c: c.sequence)
last_change = max(changes, key=lambda c: c.sequence)
ws.cell(row=row, column=1, value=table_name).font = Font(bold=True)
ws.cell(row=row, column=2, value=len(changes)).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=3, value=f"#{first_change.sequence}")
ws.cell(row=row, column=4, value=f"#{last_change.sequence}")
ws.cell(row=row, column=5, value=first_change.file_created_date).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=6, value=last_change.file_modified_date).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=7, value=", ".join(sorted(all_related)) if all_related else "")
comments_str = "\n".join(all_comments[:2]) if all_comments else ""
ws.cell(row=row, column=8, value=comments_str)
ws.cell(row=row, column=8).alignment = Alignment(wrap_text=True)
db_comments_str = "\n".join(all_db_comments[:2]) if all_db_comments else ""
ws.cell(row=row, column=9, value=db_comments_str)
ws.cell(row=row, column=9).alignment = Alignment(wrap_text=True)
ws.cell(row=row, column=10, value=", ".join(sorted(all_commits)[:3]) if all_commits else "")
ws.cell(row=row, column=11, value=", ".join(categories))
ws.cell(row=row, column=12, value=", ".join(files))
row += 1
widths = [25, 12, 12, 12, 15, 15, 25, 40, 40, 20, 40, 40]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
if row > 2:
ws.auto_filter.ref = f"A1:{get_column_letter(len(headers))}{row - 1}"
ws.freeze_panes = 'A2'
def _create_by_file_sheet(self):
"""Создает лист с группировкой по файлам"""
ws = self.wb.create_sheet("📁 По файлам")
changes_by_file = defaultdict(list)
for change in self.changes:
changes_by_file[change.file_name].append(change)
headers = ["Файл", "Изменений", "Таблиц", "Категории",
"HIGH", "Commit", "Комментарии", "COMMENT ON",
"Создан", "Изменен", "Git автор"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = self.COLORS['header_bg']
row = 2
for file_name, changes in sorted(changes_by_file.items()):
tables = set(c.table_name for c in changes)
categories = set(c.category.value for c in changes)
high_count = sum(1 for c in changes if c.impact_level == "HIGH")
comments_count = sum(1 for c in changes if c.programmer_comment)
db_comments_count = sum(1 for c in changes if c.db_comment)
commits = set(c.git_commit_hash for c in changes if c.git_commit_hash)
first_change = changes[0]
ws.cell(row=row, column=1, value=file_name)
ws.cell(row=row, column=2, value=len(changes)).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=3, value=len(tables)).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=4, value=", ".join(categories))
high_cell = ws.cell(row=row, column=5, value=high_count)
high_cell.alignment = Alignment(horizontal='center')
if high_count > 0:
high_cell.fill = self.COLORS['high_impact']
high_cell.font = Font(bold=True, color="C62828")
ws.cell(row=row, column=6, value=", ".join(sorted(commits)[:3]) if commits else "")
ws.cell(row=row, column=7, value=comments_count).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=8, value=db_comments_count).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=9, value=first_change.file_created_date).alignment = Alignment(horizontal='center')
mod_cell = ws.cell(row=row, column=10, value=first_change.file_modified_date)
mod_cell.alignment = Alignment(horizontal='center')
if first_change.file_modified_date >= self.recent_threshold:
mod_cell.fill = self.COLORS['recent_change']
ws.cell(row=row, column=11, value=first_change.git_last_author)
row += 1
widths = [30, 12, 15, 40, 12, 25, 12, 12, 15, 15, 15]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
if row > 2:
ws.auto_filter.ref = f"A1:{get_column_letter(len(headers))}{row - 1}"
ws.freeze_panes = 'A2'
def _create_impact_analysis_sheet(self):
"""Создает лист с анализом влияния"""
ws = self.wb.create_sheet("⚠️ Анализ влияния")
high_impact_changes = [c for c in self.changes if c.impact_level in ("HIGH", "MEDIUM")]
high_impact_changes.sort(key=lambda c: (c.impact_level, c.file_modified_date), reverse=True)
ws.merge_cells('A1:K1')
ws['A1'] = "АНАЛИЗ ПОТЕНЦИАЛЬНО ОПАСНЫХ ИЗМЕНЕНИЙ"
ws['A1'].font = Font(bold=True, size=14, color="C62828")
ws['A1'].alignment = Alignment(horizontal='center')
ws.merge_cells('A2:K2')
ws['A2'] = f"Изменений с высоким/средним влиянием: {len(high_impact_changes)}"
ws['A2'].font = Font(color="757575")
ws['A2'].alignment = Alignment(horizontal='center')
headers = ["№", "Влияние", "Категория", "Таблица", "Описание",
"Файл", "Дата", "Commit", "Комментарий", "Git автор", "Рекомендации"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=4, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = PatternFill(start_color="C62828", end_color="C62828", fill_type="solid")
row = 5
for change in high_impact_changes:
ws.cell(row=row, column=1, value=change.sequence).alignment = Alignment(horizontal='center')
impact_cell = ws.cell(row=row, column=2, value=change.impact_level)
impact_cell.alignment = Alignment(horizontal='center')
if change.impact_level == "HIGH":
impact_cell.fill = self.COLORS['high_impact']
impact_cell.font = Font(bold=True, color="C62828")
else:
impact_cell.fill = self.COLORS['medium_impact']
ws.cell(row=row, column=3, value=change.category.value)
ws.cell(row=row, column=4, value=change.table_name).font = Font(bold=True)
ws.cell(row=row, column=5, value=change.description)
ws.cell(row=row, column=6, value=f"{change.file_name}:{change.line_number}")
date_cell = ws.cell(row=row, column=7, value=change.file_modified_date)
date_cell.alignment = Alignment(horizontal='center')
if change.file_modified_date >= self.recent_threshold:
date_cell.fill = self.COLORS['recent_change']
ws.cell(row=row, column=8, value=change.git_commit_hash)
comment_cell = ws.cell(row=row, column=9,
value=change.programmer_comment[:150] if change.programmer_comment else "")
comment_cell.alignment = Alignment(wrap_text=True)
if change.programmer_comment:
comment_cell.fill = self.COLORS['comment_bg']
ws.cell(row=row, column=10, value=change.git_last_author or change.author)
recommendations = self._get_recommendations(change)
rec_cell = ws.cell(row=row, column=11, value=recommendations)
rec_cell.alignment = Alignment(wrap_text=True)
row += 1
widths = [5, 10, 20, 18, 35, 25, 15, 12, 35, 15, 50]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
ws.freeze_panes = 'A5'
def _create_timeline_sheet(self):
"""Создает лист с хронологией"""
ws = self.wb.create_sheet("📅 Хронология")
timeline_changes = sorted(self.changes,
key=lambda c: (c.file_modified_date, c.sequence),
reverse=True)
headers = ["Дата изменения", "Дата создания", "№", "Категория", "Таблица",
"Описание", "Файл", "Commit", "Комментарий", "Автор"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = PatternFill(start_color="1B5E20", end_color="1B5E20", fill_type="solid")
current_date = ""
row = 2
for change in timeline_changes:
if change.file_modified_date != current_date:
current_date = change.file_modified_date
ws.merge_cells(f'A{row}:J{row}')
date_cell = ws.cell(row=row, column=1, value=f"📅 {current_date}")
date_cell.font = Font(bold=True, size=11, color="1B5E20")
date_cell.fill = PatternFill(start_color="E8F5E9", end_color="E8F5E9", fill_type="solid")
if current_date >= self.recent_threshold:
date_cell.fill = self.COLORS['recent_change']
date_cell.font = Font(bold=True, size=11, color="F57F17")
row += 1
ws.cell(row=row, column=1, value="")
ws.cell(row=row, column=2, value=change.file_created_date).alignment = Alignment(horizontal='center')
ws.cell(row=row, column=3, value=f"#{change.sequence}").alignment = Alignment(horizontal='center')
ws.cell(row=row, column=4, value=change.category.value)
ws.cell(row=row, column=5, value=change.table_name).font = Font(bold=True)
ws.cell(row=row, column=6, value=change.description)
ws.cell(row=row, column=7, value=change.file_name)
ws.cell(row=row, column=8, value=change.git_commit_hash).alignment = Alignment(horizontal='center')
comment_text = change.programmer_comment or change.db_comment
comment_cell = ws.cell(row=row, column=9, value=comment_text[:100] if comment_text else "")
comment_cell.alignment = Alignment(wrap_text=True)
if comment_text:
comment_cell.fill = self.COLORS['comment_bg']
ws.cell(row=row, column=10, value=change.author)
row += 1
widths = [15, 15, 8, 20, 18, 35, 25, 12, 30, 12]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
ws.freeze_panes = 'A2'
def _create_recent_changes_sheet(self):
"""Создает лист с недавними изменениями"""
ws = self.wb.create_sheet("🕐 Недавние изменения")
recent_changes = [c for c in self.changes
if c.file_modified_date >= self.recent_threshold]
ws.merge_cells('A1:J1')
ws['A1'] = f"НЕДАВНИЕ ИЗМЕНЕНИЯ (после {self.recent_threshold})"
ws['A1'].font = Font(bold=True, size=14, color="F57F17")
ws['A1'].alignment = Alignment(horizontal='center')
ws.merge_cells('A2:J2')
ws['A2'] = f"Найдено изменений: {len(recent_changes)}"
ws['A2'].font = Font(color="757575")
ws['A2'].alignment = Alignment(horizontal='center')
if not recent_changes:
ws.merge_cells('A4:J4')
ws['A4'] = "Недавних изменений не найдено"
ws['A4'].font = Font(color="757575", italic=True)
ws['A4'].alignment = Alignment(horizontal='center')
return
headers = ["№", "Дата", "Категория", "Таблица",
"Описание", "Файл", "Commit", "Комментарий",
"Git автор", "Влияние"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=4, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = PatternFill(start_color="F57F17", end_color="F57F17", fill_type="solid")
recent_changes.sort(key=lambda c: c.file_modified_date, reverse=True)
row = 5
for change in recent_changes:
ws.cell(row=row, column=1, value=change.sequence).alignment = Alignment(horizontal='center')
date_cell = ws.cell(row=row, column=2, value=change.file_modified_date)
date_cell.alignment = Alignment(horizontal='center')
date_cell.fill = self.COLORS['recent_change']
ws.cell(row=row, column=3, value=change.category.value)
ws.cell(row=row, column=4, value=change.table_name).font = Font(bold=True)
ws.cell(row=row, column=5, value=change.description)
ws.cell(row=row, column=6, value=f"{change.file_name}:{change.line_number}")
ws.cell(row=row, column=7, value=change.git_commit_hash)
comment_text = change.programmer_comment or change.db_comment
comment_cell = ws.cell(row=row, column=8, value=comment_text[:150] if comment_text else "")
comment_cell.alignment = Alignment(wrap_text=True)
if comment_text:
comment_cell.fill = self.COLORS['comment_bg']
ws.cell(row=row, column=9, value=change.git_last_author or change.author)
impact_cell = ws.cell(row=row, column=10, value=change.impact_level)
impact_cell.alignment = Alignment(horizontal='center')
if change.impact_level == "HIGH":
impact_cell.fill = self.COLORS['high_impact']
impact_cell.font = Font(bold=True, color="C62828")
row += 1
widths = [5, 15, 20, 18, 35, 25, 12, 35, 15, 10]
for col, width in enumerate(widths, 1):
ws.column_dimensions[get_column_letter(col)].width = width
ws.freeze_panes = 'A5'
def _get_category_impact(self, category: ChangeCategory) -> str:
high_impact = {ChangeCategory.DROPPED_TABLE, ChangeCategory.DROPPED_COLUMN}
medium_impact = {ChangeCategory.NEW_TABLE, ChangeCategory.NEW_INDEX,
ChangeCategory.NEW_FOREIGN_KEY, ChangeCategory.MODIFIED_COLUMN}
if category in high_impact:
return "HIGH"
elif category in medium_impact:
return "MEDIUM"
return "LOW"
def _get_category_description(self, category: ChangeCategory) -> str:
descriptions = {
ChangeCategory.NEW_TABLE: "Создание новой таблицы",
ChangeCategory.DROPPED_TABLE: "Удаление таблицы",
ChangeCategory.NEW_COLUMN: "Добавление колонки",
ChangeCategory.DROPPED_COLUMN: "Удаление колонки",
ChangeCategory.MODIFIED_COLUMN: "Изменение колонки",
ChangeCategory.NEW_INDEX: "Создание индекса",
ChangeCategory.NEW_FOREIGN_KEY: "Добавление внешнего ключа",
ChangeCategory.RAW_SQL: "Выполнение SQL",
ChangeCategory.COMMENT: "Добавление комментария",
}
return descriptions.get(category, "Прочее изменение")
def _get_recommendations(self, change: MigrationChange) -> str:
recommendations = []
if change.category == ChangeCategory.DROPPED_TABLE:
recommendations.append("⚠️ Проверьте зависимости (FK, VIEW)")
recommendations.append("⚠️ Создайте резервную копию")
elif change.category == ChangeCategory.DROPPED_COLUMN:
recommendations.append("⚠️ Проверьте использование в индексах и FK")
recommendations.append("⚠️ Убедитесь, что ORM не маппит колонку")
elif change.category == ChangeCategory.NEW_INDEX:
if "без CONCURRENTLY" in change.description:
recommendations.append("❌ КРИТИЧНО: Используйте CREATE INDEX CONCURRENTLY")
return "\n".join(recommendations) if recommendations else "Особых рекомендаций нет"
# ===================================================================
# HTML-ОТЧЕТ (из данных в памяти, без обратного парсинга Excel)
# ===================================================================
class HTMLReportGenerator:
"""Генератор HTML-отчета напрямую из списка изменений (change_to_dict).
Вкладки:
📖 Структура таблиц — карточки таблиц (колонки, FK, история изменений)
📋 Все изменения — единый список с поиском
🗂 Батчи — группировка по дате загрузки файлов в репозиторий
Фильтры: батч (дата загрузки), период, поиск. Модальное окно с деталями.
"""
NO_DATE = "без даты"
def __init__(self, output_path: str = "report.html"):
self.output_path = output_path
self.changes: List[Dict[str, Any]] = []
self.tables: Dict[str, Dict[str, Any]] = {}
self.batches: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
self.all_dates: set = set()
self.changes_by_id: Dict[int, Dict[str, Any]] = {}
def load_from_dicts(self, change_dicts: List[Dict[str, Any]]) -> "HTMLReportGenerator":
"""Строит таблицы и батчи из списка change_to_dict"""
self.changes = list(change_dicts)
tables_tmp: Dict[str, Dict[str, Any]] = {}
for ch in self.changes:
self.changes_by_id[ch["sequence"]] = ch
self.batches[ch.get("batch") or self.NO_DATE].append(ch)
for key in ("date", "migration_date", "created_date", "modified_date", "batch"):
if ch.get(key):
self.all_dates.add(ch[key])
table_name = ch.get("table") or "разные"
table = tables_tmp.setdefault(table_name, {
"name": table_name,
"comment": "",
"columns": OrderedDict(),
"related": set(),
"files": set(),
"batches": set(),
"dates": set(),
"changes": [],
})
table["changes"].append(ch)
if ch.get("file"):
table["files"].add(ch["file"])
if ch.get("batch"):
table["batches"].add(ch["batch"])
for d in (ch.get("date"), ch.get("created_date"), ch.get("modified_date")):
if d:
table["dates"].add(d)
if ch.get("related_tables"):
for rt in re.split(r',\s*', ch["related_tables"]):
rt = rt.strip()
if rt and rt != table_name:
table["related"].add(rt)
# Структура колонок — только из изменений NEW_TABLE
if ch.get("category") == ChangeCategory.NEW_TABLE.value:
for col_name, col_type in (ch.get("columns_info") or {}).items():
table["columns"].setdefault(col_name, col_type)
if not table["comment"]:
table["comment"] = ch.get("db_comment") or ch.get("programmer_comment") or ""
self.tables = tables_tmp
print(f"[✓] Данные загружены: таблиц {len(self.tables)}, "
f"изменений {len(self.changes)}, батчей {len(self.batches)}")
return self
def generate(self):
"""Записывает HTML-файл"""
print("\n[*] Генерирую HTML-отчет...")
output = Path(self.output_path)
output.parent.mkdir(parents=True, exist_ok=True)
html = self._build_html()
with open(output, "w", encoding="utf-8") as f:
f.write(html)
print(f"[✓] HTML сохранен: {output}")
print(f" Размер: {os.path.getsize(output) / 1024:.1f} KB")
# ---------- вспомогательные визуальные хелперы ----------
@staticmethod
def _impact_badge(impact: str) -> str:
if impact == "HIGH":
return '<span class="badge badge-danger">HIGH</span>'
if impact == "MEDIUM":
return '<span class="badge badge-warning">MEDIUM</span>'
return ""
@staticmethod
def _category_icon(category: str) -> str:
cat = category.lower()
if "таблиц" in cat:
return "🆕" if "новая" in cat else "🗑"
if "колонк" in cat:
return "➕" if "новая" in cat else ("➖" if "удаленная" in cat else "✏️")
if "индекс" in cat:
return "📑"
if "внешн" in cat:
return "🔗"
if "коммент" in cat:
return "💬"
if "sql" in cat:
return "💻"
return "📌"
def _sorted_batches(self) -> List[str]:
"""Батчи по дате загрузки: свежие сверху, 'без даты' в конце"""
batches = sorted((b for b in self.batches if b != self.NO_DATE), reverse=True)
if self.NO_DATE in self.batches:
batches.append(self.NO_DATE)
return batches
def _change_entry_html(self, ch: Dict[str, Any]) -> str:
dates = " ".join(sorted({d for d in (
ch.get("date"), ch.get("created_date"), ch.get("modified_date")
) if d}))
icon = self._category_icon(ch.get("category", ""))
return (
f'<div class="change-entry" data-dates="{dates}" data-batch="{ch.get("batch") or ""}" '
f'onclick="openChangeDetail({ch.get("sequence", 0)})">'
f'<div>{icon} <strong>{ch.get("description", "")}</strong></div>'
f'<div style="color:#666;">Таблица: {ch.get("table", "")} | {ch.get("date", "")} | '
f'{ch.get("file", "")} {self._impact_badge(ch.get("impact", ""))}</div>'
f'<div style="font-size:0.8em;color:#999;">👆 Кликните для деталей</div>'
f'</div>'
)
# ---------- вкладка «Структура таблиц» ----------
def _generate_table_cards_html(self) -> str:
if not self.tables:
return "<p>Нет данных о таблицах</p>"
cards = [self._generate_single_card(self.tables[name]) for name in sorted(self.tables)]
return "\n".join(cards)
def _generate_single_card(self, table: Dict[str, Any]) -> str:
# Колонки
if table["columns"]:
cols = []
for col_name in sorted(table["columns"]):
col_type = table["columns"][col_name]
upper = col_type.upper()
badges = []
if "PK" in upper or "PRIMARY" in upper:
badges.append('<span class="badge badge-pk">PK</span>')
clean_type = col_type
fk_match = re.search(r'REFERENCES\s+(\w+)', col_type, re.IGNORECASE)
if fk_match:
badges.append(f'<span class="badge badge-fk">FK→{fk_match.group(1)}</span>')
for attr in ["NOT NULL", "NULL", "PRIMARY KEY", "PK", "REFERENCES",
"FK", "DEFAULT", "UNIQUE", "CHECK"]:
clean_type = re.sub(rf'\s*{attr}.*$', '', clean_type, flags=re.IGNORECASE)
badges_html = " ".join(badges)
cols.append(
f'<tr><td><strong>{col_name}</strong></td>'
f'<td>{clean_type.strip()}</td><td>{badges_html}</td></tr>')
columns_html = ('<table class="mini-table"><thead><tr><th>Колонка</th><th>Тип</th>'
'<th>Связи</th></tr></thead><tbody>' + "".join(cols) + '</tbody></table>')
else:
columns_html = "<p style='color: #999;'>Колонки не определены</p>"
# История изменений (последние 10)
changes_html = ""
recent = sorted(table["changes"], key=lambda c: c.get("sequence", 0), reverse=True)[:10]
if recent:
entries = "\n".join(self._change_entry_html(ch) for ch in recent)
changes_html = (f'<div class="changes-history">'
f'<strong>📜 История (кликните для деталей):</strong>{entries}</div>')
fk_html = ""
if table["related"]:
links = "".join(f'<span class="fk-link">→ {rt}</span>' for rt in sorted(table["related"]))
fk_html = f'<div class="fk-links">{links}</div>'
comment_html = f'<div class="card-comment">💬 {table["comment"]}</div>' if table["comment"] else ""
dates = sorted(table["dates"])
info_parts = []
if dates:
info_parts.append(f"📅 {min(dates)}")
info_parts.append(f"✏️ {max(dates)}")
info_html = f'<div class="card-info">{" | ".join(info_parts)}</div>' if info_parts else ""
dates_attr = " ".join(sorted(table["dates"]))
batches_attr = " ".join(sorted(table["batches"]))
return f"""
<div class="table-card" data-dates="{dates_attr}" data-batch="{batches_attr}">
<div class="card-header" onclick="toggleCard(this)">
<h3><span>📋 {table["name"]}</span><span class="toggle-icon">▼</span></h3>
</div>
<div class="card-body">
{comment_html}
{info_html}
{columns_html}
{fk_html}
{changes_html}
</div>
</div>
"""
# ---------- вкладка «Батчи» ----------
def _generate_batches_html(self) -> str:
parts = []
for batch in self._sorted_batches():
batch_changes = self.batches[batch]
files = sorted(set(ch.get("file", "") for ch in batch_changes if ch.get("file")))
Update migration_reporter.py
2026-11-02 14:39:36 +03:00
commits = sorted({ch.get("git_commit") for ch in batch_changes if ch.get("git_commit")})
Add DAG and migration_reporter module
2026-10-29 16:14:13 +03:00
entries = "\n".join(self._change_entry_html(ch)
for ch in sorted(batch_changes, key=lambda c: c["sequence"]))
files_html = "".join(f'<span class="batch-file">📄 {fname}</span>' for fname in files)
parts.append(f"""
<div class="batch-section" data-batch="{batch}">
<div class="batch-header">
<span>🗂 {batch}</span>
Update migration_reporter.py
2026-11-02 14:39:36 +03:00
<span class="batch-meta">{len(commits)} коммитов • {len(files)} файлов • {len(batch_changes)} изменений</span>
Add DAG and migration_reporter module
2026-10-29 16:14:13 +03:00
</div>
<div class="batch-files">{files_html}</div>
<div class="changes-history">{entries}</div>
</div>""")
return "\n".join(parts) if parts else "<p>Нет данных о батчах</p>"
# ---------- главный шаблон ----------
def _generate_date_options(self) -> str:
return "\n".join(f'<option value="{d}">{d}</option>'
for d in sorted(self.all_dates, reverse=True))
def _generate_batch_options(self) -> str:
return "\n".join(f'<option value="{b}">{b} ({len(self.batches[b])} изм.)</option>'
for b in self._sorted_batches())
def _get_total_columns(self) -> int:
return sum(len(t["columns"]) for t in self.tables.values())
def _build_html(self) -> str:
tables_cards_html = self._generate_table_cards_html()
changes_html = "\n".join(self._change_entry_html(ch)
for ch in sorted(self.changes, key=lambda c: c.get("sequence", 0),
reverse=True)) or "<p>Нет изменений</p>"
batches_html = self._generate_batches_html()
date_options = self._generate_date_options()
batch_options = self._generate_batch_options()
return f"""<!DOCTYPE html>
<html lang="ru">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Отчет Миграций</title>
<style>
:root {{
--primary: #2F5496;
--success: #1B5E20;
--warning: #F57F17;
--danger: #C62828;
--purple: #6A1B9A;
--bg-light: #f8f9fa;
--border: #dee2e6;
}}
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
body {{
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: #f0f2f5;
padding: 20px;
}}
.container {{ max-width: 1400px; margin: 0 auto; }}
.header {{
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 30px;
border-radius: 12px;
text-align: center;
margin-bottom: 20px;
}}
.header h1 {{ font-size: 2em; margin-bottom: 10px; }}
.stats-bar {{
display: flex;
justify-content: center;
gap: 20px;
flex-wrap: wrap;
margin-bottom: 20px;
}}
.stat {{
background: white;
border-radius: 8px;
padding: 15px 25px;
text-align: center;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}}
.stat-number {{ font-size: 1.8em; font-weight: bold; color: var(--primary); }}
.stat-label {{ font-size: 0.85em; color: #666; }}
.filters-panel {{
background: white;
border-radius: 8px;
padding: 20px;
margin-bottom: 20px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}}
.filters-title {{ font-weight: bold; margin-bottom: 15px; color: var(--primary); }}
.filter-group {{
display: flex;
flex-wrap: wrap;
gap: 10px;
align-items: center;
margin-bottom: 10px;
}}
.filter-label {{ font-size: 0.9em; color: #666; min-width: 80px; }}
.filter-select {{
padding: 8px 12px;
border: 1px solid var(--border);
border-radius: 4px;
font-size: 0.9em;
outline: none;
cursor: pointer;
}}
.filter-btn {{
padding: 8px 15px;
border: 1px solid var(--border);
border-radius: 4px;
background: white;
cursor: pointer;
font-size: 0.9em;
transition: all 0.3s;
}}
.filter-btn:hover {{ background: var(--bg-light); }}
.filter-btn.active {{ background: var(--primary); color: white; border-color: var(--primary); }}
.filter-btn.reset {{ background: #f0f0f0; color: #666; }}
.tabs {{
display: flex;
background: white;
border-radius: 8px;
margin-bottom: 20px;
overflow: hidden;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}}
.tab {{
flex: 1;
padding: 15px;
cursor: pointer;
border: none;
background: none;
font-size: 0.95em;
color: #666;
transition: all 0.3s;
}}
.tab:hover {{ background: var(--bg-light); }}
.tab.active {{ background: var(--primary); color: white; font-weight: 600; }}
.tab-content {{ display: none; }}
.tab-content.active {{ display: block; }}
.search-box {{ margin-bottom: 20px; }}
.search-input {{
width: 100%;
padding: 15px 20px;
font-size: 1em;
border: 2px solid var(--border);
border-radius: 25px;
outline: none;
background: white;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}}
.tables-grid {{
display: grid;
grid-template-columns: repeat(auto-fill, minmax(400px, 1fr));
gap: 20px;
}}
.table-card {{
background: white;
border-radius: 12px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
overflow: hidden;
transition: transform 0.3s;
}}
.table-card:hover {{ transform: translateY(-3px); }}
.table-card.hidden, .batch-section.hidden {{ display: none; }}
.card-header {{
background: linear-gradient(135deg, #1B5E20 0%, #2E7D32 100%);
color: white;
padding: 15px 20px;
cursor: pointer;
}}
.card-header h3 {{
display: flex;
justify-content: space-between;
align-items: center;
}}
.card-body {{ padding: 15px; }}
.card-comment {{
background: #F3E5F5;
border-left: 3px solid var(--purple);
padding: 10px;
margin-bottom: 10px;
border-radius: 4px;
font-size: 0.9em;
}}
.card-info {{ font-size: 0.8em; color: #666; margin-bottom: 10px; }}
.mini-table {{ width: 100%; border-collapse: collapse; font-size: 0.85em; margin-bottom: 10px; }}
.mini-table th {{
background: #E8F5E9;
color: #1B5E20;
padding: 6px 8px;
text-align: left;
border: 1px solid var(--border);
}}
.mini-table td {{ padding: 6px 8px; border: 1px solid var(--border); }}
.badge {{
display: inline-block;
padding: 2px 6px;
border-radius: 3px;
font-size: 0.75em;
font-weight: 600;
}}
.badge-pk {{ background: #C8E6C9; color: #1B5E20; }}
.badge-fk {{ background: #FFF9C4; color: #F57F17; }}
.badge-danger {{ background: #FFCDD2; color: #C62828; }}
.badge-warning {{ background: #FFF9C4; color: #F57F17; }}
.changes-history {{
max-height: 260px;
overflow-y: auto;
background: var(--bg-light);
border-radius: 4px;
padding: 8px;
font-size: 0.8em;
}}
.change-entry {{
padding: 10px;
border-bottom: 1px solid #e0e0e0;
cursor: pointer;
transition: background 0.2s;
}}
.change-entry:hover {{ background: #e8e8e8; }}
.change-entry:last-child {{ border-bottom: none; }}
.fk-links {{ display: flex; flex-wrap: wrap; gap: 5px; margin-top: 5px; }}
.fk-link {{
background: #FFF3E0;
border: 1px solid #F57F17;
border-radius: 4px;
padding: 3px 8px;
font-size: 0.8em;
color: #E65100;
}}
.batch-section {{
background: white;
border-radius: 12px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
margin-bottom: 20px;
overflow: hidden;
}}
.batch-header {{
background: linear-gradient(135deg, #2F5496 0%, #667eea 100%);
color: white;
padding: 12px 20px;
display: flex;
justify-content: space-between;
align-items: center;
font-size: 1.05em;
}}
.batch-meta {{ font-size: 0.8em; opacity: 0.85; }}
.batch-files {{
padding: 10px 15px 0 15px;
display: flex;
flex-wrap: wrap;
gap: 6px;
}}
.batch-file {{
background: #EDF2F7;
border: 1px solid var(--border);
border-radius: 4px;
padding: 2px 8px;
font-size: 0.78em;
color: #2D3748;
}}
.batch-section .changes-history {{
max-height: 400px;
margin: 10px 15px 15px 15px;
background: white;
}}
.modal {{
display: none;
position: fixed;
top: 0;
left: 0;
width: 100%;
height: 100%;
background: rgba(0,0,0,0.5);
z-index: 1000;
justify-content: center;
align-items: center;
}}
.modal.active {{ display: flex; }}
.modal-content {{
background: white;
border-radius: 12px;
max-width: 800px;
width: 90%;
max-height: 80vh;
overflow-y: auto;
padding: 30px;
box-shadow: 0 10px 40px rgba(0,0,0,0.3);
animation: slideIn 0.3s ease-out;
}}
@keyframes slideIn {{
from {{ transform: translateY(-50px); opacity: 0; }}
to {{ transform: translateY(0); opacity: 1; }}
}}
.modal-header {{
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 20px;
padding-bottom: 15px;
border-bottom: 2px solid var(--border);
}}
.modal-title {{
font-size: 1.3em;
font-weight: bold;
color: var(--primary);
}}
.modal-close {{
cursor: pointer;
font-size: 1.5em;
color: #999;
background: none;
border: none;
padding: 5px;
}}
.modal-close:hover {{ color: #333; }}
.detail-section {{
margin-bottom: 20px;
padding: 15px;
background: var(--bg-light);
border-radius: 8px;
}}
.detail-label {{
font-weight: bold;
color: var(--primary);
margin-bottom: 5px;
}}
.detail-value {{
color: #333;
line-height: 1.5;
}}
.code-block {{
background: #1e1e1e;
color: #d4d4d4;
padding: 15px;
border-radius: 8px;
font-family: 'Courier New', monospace;
font-size: 0.85em;
overflow-x: auto;
white-space: pre-wrap;
word-break: break-all;
}}
@media (max-width: 768px) {{
.tables-grid {{ grid-template-columns: 1fr; }}
.modal-content {{ width: 95%; padding: 20px; }}
}}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>📊 Отчет о всех миграциях</h1>
<p>Сгенерировано: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
</div>
<div class="stats-bar">
<div class="stat"><div class="stat-number" id="statTables">{len(self.tables)}</div><div class="stat-label">Таблиц</div></div>
<div class="stat"><div class="stat-number">{self._get_total_columns()}</div><div class="stat-label">Колонок</div></div>
<div class="stat"><div class="stat-number" id="statChanges">{len(self.changes)}</div><div class="stat-label">Изменений</div></div>
<div class="stat"><div class="stat-number">{len(self.batches)}</div><div class="stat-label">Батчей загрузки</div></div>
</div>
<div class="filters-panel">
<div class="filters-title">🔍 Фильтры</div>
<div class="filter-group">
<span class="filter-label">Батч:</span>
<select class="filter-select" id="batchFilter" onchange="applyAllFilters()">
<option value="all">Все батчи (по дате загрузки файлов)</option>
{batch_options}
</select>
</div>
<div class="filter-group">
<span class="filter-label">Период:</span>
<select class="filter-select" id="dateFilter" onchange="applyAllFilters()">
<option value="all">Все даты</option>
{date_options}
</select>
<button class="filter-btn reset" onclick="resetFilters()">Сбросить</button>
</div>
<div class="filter-group">
<span class="filter-label">Быстрый выбор:</span>
<button class="filter-btn" onclick="setQuickDate('today')">Сегодня</button>
<button class="filter-btn" onclick="setQuickDate('week')">7 дней</button>
<button class="filter-btn" onclick="setQuickDate('month')">30 дней</button>
</div>
</div>
<div class="tabs">
<button class="tab" onclick="showTab('batches', this)">🗂 Батчи загрузки</button>
<button class="tab active" onclick="showTab('structure', this)">📖 Структура таблиц</button>
<button class="tab" onclick="showTab('changes', this)">📋 Все изменения</button>
</div>
<div id="batches-tab" class="tab-content">
<div class="search-box">
<input type="text" class="search-input" id="batchSearch" placeholder="🔍 Поиск по батчам..." onkeyup="applyAllFilters()">
</div>
<div id="batchesList">
{batches_html}
</div>
</div>
<div id="structure-tab" class="tab-content active">
<div class="search-box">
<input type="text" class="search-input" id="tableSearch" placeholder="🔍 Поиск по таблицам..." onkeyup="applyAllFilters()">
</div>
<div class="tables-grid" id="tablesGrid">
{tables_cards_html}
</div>
</div>
<div id="changes-tab" class="tab-content">
<div class="search-box">
<input type="text" class="search-input" id="changeSearch" placeholder="🔍 Поиск по изменениям..." onkeyup="applyAllFilters()">
</div>
<div id="changesList">
{changes_html}
</div>
</div>
</div>
<!-- Модальное окно для деталей -->
<div class="modal" id="detailModal" onclick="if(event.target === this) closeModal()">
<div class="modal-content">
<div class="modal-header">
<div class="modal-title" id="modalTitle">Детали изменения</div>
<button class="modal-close" onclick="closeModal()">✕</button>
</div>
<div id="modalBody"></div>
</div>
</div>
<script>
const changesData = {json.dumps(self.changes_by_id, ensure_ascii=False)};
const allDates = {json.dumps(sorted(self.all_dates))};
function showTab(tabName, btn) {{
document.querySelectorAll('.tab-content').forEach(t => t.classList.remove('active'));
document.getElementById(tabName + '-tab').classList.add('active');
document.querySelectorAll('.tab').forEach(t => t.classList.remove('active'));
btn.classList.add('active');
}}
function toggleCard(header) {{
const body = header.nextElementSibling;
const icon = header.querySelector('.toggle-icon');
const hidden = body.style.display === 'none';
body.style.display = hidden ? '' : 'none';
if (icon) icon.textContent = hidden ? '▼' : '▶';
}}
function setQuickDate(period) {{
const select = document.getElementById('dateFilter');
const today = new Date();
let threshold;
if (period === 'today') threshold = new Date();
else if (period === 'week') threshold = new Date(today.getTime() - 7 * 24 * 3600 * 1000);
else threshold = new Date(today.getTime() - 30 * 24 * 3600 * 1000);
const thresholdStr = threshold.toISOString().slice(0, 10);
const filtered = allDates.filter(d => d >= thresholdStr);
if (filtered.length > 0) select.value = filtered[0];
applyAllFilters();
}}
function resetFilters() {{
document.getElementById('batchFilter').value = 'all';
document.getElementById('dateFilter').value = 'all';
['tableSearch', 'changeSearch', 'batchSearch'].forEach(id => {{
const el = document.getElementById(id);
if (el) el.value = '';
}});
applyAllFilters();
}}
function applyAllFilters() {{
const tableSearch = document.getElementById('tableSearch')?.value.toLowerCase() || '';
const changeSearch = document.getElementById('changeSearch')?.value.toLowerCase() || '';
const batchSearch = document.getElementById('batchSearch')?.value.toLowerCase() || '';
const batchFilter = document.getElementById('batchFilter').value;
const dateFilter = document.getElementById('dateFilter').value;
// Карточки таблиц
const cards = document.querySelectorAll('#tablesGrid .table-card');
let visibleCount = 0;
cards.forEach(card => {{
const cardDates = card.getAttribute('data-dates') || '';
const cardBatches = card.getAttribute('data-batch') || '';
const cardText = card.textContent.toLowerCase();
const dateMatch = dateFilter === 'all' || cardDates.includes(dateFilter);
const batchMatch = batchFilter === 'all' || cardBatches.includes(batchFilter);
const searchMatch = !tableSearch || cardText.includes(tableSearch);
if (dateMatch && batchMatch && searchMatch) {{
card.classList.remove('hidden');
visibleCount++;
}} else {{
card.classList.add('hidden');
}}
}});
document.getElementById('statTables').textContent = visibleCount;
// Список всех изменений
const changeItems = document.querySelectorAll('#changesList .change-entry');
let visibleChanges = 0;
changeItems.forEach(item => {{
const itemDates = item.getAttribute('data-dates') || '';
const itemBatch = item.getAttribute('data-batch') || '';
const itemText = item.textContent.toLowerCase();
const dateMatch = dateFilter === 'all' || itemDates.includes(dateFilter);
const batchMatch = batchFilter === 'all' || itemBatch.includes(batchFilter);
const searchMatch = !changeSearch || itemText.includes(changeSearch);
if (dateMatch && batchMatch && searchMatch) {{
item.style.display = '';
visibleChanges++;
}} else {{
item.style.display = 'none';
}}
}});
document.getElementById('statChanges').textContent = visibleChanges;
// Вкладка батчей: скрываем секции без видимых записей
document.querySelectorAll('#batchesList .batch-section').forEach(section => {{
let sectionVisible = 0;
section.querySelectorAll('.change-entry').forEach(item => {{
const itemDates = item.getAttribute('data-dates') || '';
const itemText = item.textContent.toLowerCase();
const dateMatch = dateFilter === 'all' || itemDates.includes(dateFilter);
const batchMatch = batchFilter === 'all' ||
section.getAttribute('data-batch') === batchFilter;
const searchMatch = !batchSearch || itemText.includes(batchSearch);
if (dateMatch && batchMatch && searchMatch) {{
item.style.display = '';
sectionVisible++;
}} else {{
item.style.display = 'none';
}}
}});
section.classList.toggle('hidden', sectionVisible === 0);
}});
}}
function openChangeDetail(sequence) {{
const change = changesData[sequence];
if (!change) return;
document.getElementById('modalTitle').textContent = 'Изменение #' + sequence;
let html = '';
html += '<div class="detail-section">';
html += '<div class="detail-label">📋 Описание</div>';
html += '<div class="detail-value">' + (change.description || 'Нет описания') + '</div>';
html += '</div>';
html += '<div class="detail-section">';
html += '<div class="detail-label">📌 Категория</div>';
html += '<div class="detail-value">' + (change.category || 'Не указана') + '</div>';
html += '<div class="detail-label" style="margin-top:10px;">⚡ Влияние</div>';
html += '<div class="detail-value">' + (change.impact || 'Не указано') + '</div>';
html += '</div>';
html += '<div class="detail-section">';
html += '<div class="detail-label">📊 Таблица</div>';
html += '<div class="detail-value">' + (change.table || 'Не указана') + '</div>';
html += '</div>';
html += '<div class="detail-section">';
html += '<div class="detail-label">🗂 Батч (дата загрузки файла)</div>';
html += '<div class="detail-value">' + (change.batch || 'Не указана') + '</div>';
html += '</div>';
if (change.created_date || change.modified_date) {{
html += '<div class="detail-section">';
html += '<div class="detail-label">📅 Даты</div>';
if (change.created_date) html += '<div class="detail-value">Создан файл: ' + change.created_date + '</div>';
if (change.modified_date) html += '<div class="detail-value">Изменен файл: ' + change.modified_date + '</div>';
html += '</div>';
}}
if (change.file) {{
html += '<div class="detail-section">';
html += '<div class="detail-label">📁 Файл</div>';
html += '<div class="detail-value">' + change.file + (change.line ? ' (строка ' + change.line + ')' : '') + '</div>';
html += '</div>';
}}
if (change.git_commit || change.git_author) {{
html += '<div class="detail-section">';
html += '<div class="detail-label">🔗 Git</div>';
if (change.git_commit) html += '<div class="detail-value">Commit: ' + change.git_commit + '</div>';
if (change.commit_message) html += '<div class="detail-value">Message: ' + change.commit_message + '</div>';
if (change.git_author) html += '<div class="detail-value">Автор: ' + change.git_author + '</div>';
html += '</div>';
}}
if (change.programmer_comment || change.db_comment) {{
html += '<div class="detail-section">';
html += '<div class="detail-label">💬 Комментарии</div>';
if (change.programmer_comment) html += '<div class="detail-value">Программист: ' + change.programmer_comment + '</div>';
if (change.db_comment) html += '<div class="detail-value">COMMENT ON: ' + change.db_comment + '</div>';
html += '</div>';
}}
if (change.related_tables) {{
html += '<div class="detail-section">';
html += '<div class="detail-label">🔗 Связанные таблицы</div>';
html += '<div class="detail-value">' + change.related_tables + '</div>';
html += '</div>';
}}
if (change.code_snippet) {{
html += '<div class="detail-section">';
html += '<div class="detail-label">💻 Код</div>';
html += '<div class="code-block">' + change.code_snippet.replace(/</g, '&lt;').replace(/>/g, '&gt;') + '</div>';
html += '</div>';
}}
document.getElementById('modalBody').innerHTML = html;
document.getElementById('detailModal').classList.add('active');
}}
function closeModal() {{
document.getElementById('detailModal').classList.remove('active');
}}
document.addEventListener('keydown', function(e) {{
if (e.key === 'Escape') closeModal();
}});
applyAllFilters();
</script>
</body>
</html>"""
# ===================================================================
# ПАЙПЛАЙН: анализ → Excel + кэш → HTML
# ===================================================================
def change_to_dict(c: "MigrationChange") -> Dict[str, Any]:
"""Сериализует изменение для HTML-генератора и кэша JSON"""
created = c.file_created_date or ""
modified = c.file_modified_date or ""
migration = c.migration_date or ""
return {
"sequence": c.sequence,
"category": c.category.value,
"table": c.table_name,
"description": c.description,
"details": c.details,
"code_snippet": c.code_snippet,
"impact": c.impact_level,
"file": c.file_name,
"line": c.line_number,
"migration_date": migration,
"created_date": created,
"modified_date": modified,
"date": modified or created or migration,
"batch": created or migration or modified or HTMLReportGenerator.NO_DATE,
"git_commit": c.git_commit_hash,
"commit_message": c.git_commit_message,
"git_author": c.git_last_author,
"author": c.author,
"programmer_comment": c.programmer_comment,
"db_comment": c.db_comment,
"related_tables": ", ".join(c.related_tables) if c.related_tables else "",
"columns_info": dict(c.columns_info) if c.columns_info else {},
}
def _git_pull(repo_dir: Path):
"""Обновляет локальную копию репозитория миграций"""
print(f"[*] Обновляю репозиторий: {repo_dir}")
result = subprocess.run(
["git", "-C", str(repo_dir), "pull", "--ff-only"],
capture_output=True, text=True, timeout=300
)
if result.returncode != 0:
print(f"❌ git pull не выполнен:\n{result.stderr}")
sys.exit(3)
print(result.stdout.strip() or "[✓] Репозиторий обновлен")
def analyze_directory(migrations_dir: Path, debug: bool = False,
use_git: bool = True) -> List["MigrationChange"]:
"""Анализирует директорию с .kt миграциями"""
if not migrations_dir.exists():
print(f"❌ Директория не найдена: {migrations_dir}", file=sys.stderr)
sys.exit(2)
print("=" * 60)
print("🔍 АНАЛИЗ МИГРАЦИЙ")
print("=" * 60)
analyzer = MigrationAnalyzer(migrations_dir, debug=debug, use_git=use_git)
changes = analyzer.analyze()
if not changes:
print("\n❌ Изменений не найдено!")
sys.exit(1)
return changes
def generate_excel_file(changes: List["MigrationChange"], output_path: Path) -> Path:
"""Генерирует Excel-отчет + копию latest.xlsx"""
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
generator = ExcelReportGenerator(changes, str(output_path))
generator.generate()
shutil.copy2(output_path, output_path.parent / "latest.xlsx")
print(f"[✓] Копия последнего отчета: {output_path.parent / 'latest.xlsx'}")
return output_path
def read_cache(cache_path: Path) -> List[Dict[str, Any]]:
"""Читает кэш изменений (JSON)"""
with open(cache_path, "r", encoding="utf-8") as f:
return json.load(f)
def write_cache(change_dicts: List[Dict[str, Any]], cache_path: Path) -> Path:
"""Сохраняет сериализованные изменения в JSON-кэш (прослойка между шагами DAG и HTML)"""
cache_path = Path(cache_path)
cache_path.parent.mkdir(parents=True, exist_ok=True)
with open(cache_path, "w", encoding="utf-8") as f:
json.dump(change_dicts, f, ensure_ascii=False, indent=1)
print(f"[✓] Кэш изменений сохранен: {cache_path}")
return cache_path
def generate_html_file(change_dicts: List[Dict[str, Any]], output_path: Path) -> Path:
"""Генерирует HTML-отчет + копию latest.html"""
output_path = Path(output_path)
HTMLReportGenerator(str(output_path)).load_from_dicts(change_dicts).generate()
shutil.copy2(output_path, output_path.parent / "latest.html")
print(f"[✓] Копия последнего отчета: {output_path.parent / 'latest.html'}")
return output_path
def cmd_run(args):
"""Полный цикл: [git pull] → анализ → кэш → HTML (+ Excel с --excel)"""
migrations_dir = Path(args.dir)
repo_dir = Path(args.repo_dir) if args.repo_dir else Path(args.dir).resolve().parent
if not args.skip_pull:
_git_pull(repo_dir)
changes = analyze_directory(migrations_dir, debug=args.debug, use_git=not args.no_git)
change_dicts = [change_to_dict(c) for c in changes]
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_dir = Path(args.output_dir)
cache_file = output_dir / f"migration_report_{timestamp}.changes.json"
html_file = output_dir / f"migration_report_{timestamp}.html"
report_file = output_dir / f"migration_report_{timestamp}.xlsx"
write_cache(change_dicts, cache_file)
if args.excel:
generate_excel_file(changes, report_file)
generate_html_file(change_dicts, html_file)
print(f"\nCACHE_FILE: {cache_file.resolve()}")
if args.excel:
print(f"REPORT_FILE: {report_file.resolve()}")
print(f"HTML_FILE: {html_file.resolve()}")
def cmd_analyze(args):
"""Только анализ + кэш изменений (без Excel и HTML)"""
changes = analyze_directory(Path(args.dir), debug=args.debug, use_git=not args.no_git)
change_dicts = [change_to_dict(c) for c in changes]
cache_file = write_cache(change_dicts, Path(args.output))
print(f"\nCACHE_FILE: {cache_file.resolve()}")
def cmd_excel(args):
"""Анализ + Excel-отчет + кэш изменений (для последующей генерации HTML)"""
migrations_dir = Path(args.dir)
changes = analyze_directory(migrations_dir, debug=args.debug, use_git=not args.no_git)
change_dicts = [change_to_dict(c) for c in changes]
report_file = generate_excel_file(changes, Path(args.output))
cache_file = write_cache(change_dicts, Path(args.output).with_suffix(".changes.json"))
print(f"\nREPORT_FILE: {report_file.resolve()}")
print(f"CACHE_FILE: {cache_file.resolve()}")
def cmd_html(args):
"""HTML-отчет из кэша изменений"""
cache_path = Path(args.cache)
if not cache_path.exists():
print(f"❌ Файл кэша не найден: {cache_path}", file=sys.stderr)
sys.exit(2)
change_dicts = read_cache(cache_path)
html_file = generate_html_file(change_dicts, Path(args.output))
print(f"\nHTML_FILE: {html_file.resolve()}")
def main():
parser = argparse.ArgumentParser(
prog="migration_reporter",
description="Единый инструмент анализа Kotlin-миграций: Excel + HTML отчеты",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Примеры:
python migration_reporter.py run --dir pgsql --skip-pull
python migration_reporter.py run --dir pgsql
python migration_reporter.py excel --dir pgsql --output reports/report.xlsx
python migration_reporter.py html --cache reports/report.changes.json --output reports/report.html
"""
)
sub = parser.add_subparsers(dest="command", required=True)
p_run = sub.add_parser("run", help="Полный цикл: [git pull] → анализ → кэш → HTML (+ Excel с --excel)")
p_run.add_argument("--dir", default="pgsql", help="Директория с .kt миграциями (default: pgsql)")
p_run.add_argument("--repo-dir", default=None,
help="Корень git-репозитория для pull (default: родитель --dir)")
p_run.add_argument("--output-dir", "-o", default="reports", help="Папка для отчетов (default: reports)")
p_run.add_argument("--skip-pull", action="store_true", help="Не выполнять git pull")
p_run.add_argument("--excel", action="store_true", help="Генерировать Excel-отчет (по умолчанию выключен)")
p_run.add_argument("--no-git", action="store_true", help="Не использовать git в анализе")
p_run.add_argument("--debug", action="store_true", help="Детальная диагностика")
p_analyze = sub.add_parser("analyze", help="Только анализ + кэш изменений (шаг DAG)")
p_analyze.add_argument("--dir", default="pgsql", help="Директория с .kt миграциями")
p_analyze.add_argument("--output", required=True, help="Путь к файлу кэша (.changes.json)")
p_analyze.add_argument("--no-git", action="store_true", help="Не использовать git в анализе")
p_analyze.add_argument("--debug", action="store_true", help="Детальная диагностика")
p_excel = sub.add_parser("excel", help="Анализ + Excel-отчет + кэш изменений")
p_excel.add_argument("--dir", default="pgsql", help="Директория с .kt миграциями")
p_excel.add_argument("--output", required=True, help="Путь к выходному Excel-файлу")
p_excel.add_argument("--no-git", action="store_true", help="Не использовать git в анализе")
p_excel.add_argument("--debug", action="store_true", help="Детальная диагностика")
p_html = sub.add_parser("html", help="HTML-отчет из кэша изменений")
p_html.add_argument("--cache", required=True, help="Путь к файлу кэша (.changes.json)")
p_html.add_argument("--output", required=True, help="Путь к выходному HTML-файлу")
args = parser.parse_args()
if args.command == "run":
cmd_run(args)
elif args.command == "analyze":
cmd_analyze(args)
elif args.command == "excel":
cmd_excel(args)
elif args.command == "html":
cmd_html(args)
if __name__ == "__main__":
main()
Reference in New Issue Copy Permalink