src/services/code_change_tracker.py - master - Watashicuvu/agentic-tools

src/services/code_change_tracker.py
288 строк12 KB
Watashicuvu
library indexer and new cli util
26 апр 2026, 23:01
26 апр 2026, 23:012cd1ea2
"""
Code Change Tracker — отслеживание изменений кода для проверки документации.

Сравнивает текущее состояние с предыдущим (через file_hashes) и определяет:
- added — новые файлы/функции
- removed — удалённые файлы/функции
- modified — изменённые сигнатуры функций
- moved — переименованные/перемещённые файлы (эвристика)
"""

import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Set

from src.services.repository_context import RepositoryContext

logger = logging.getLogger(__name__)


@dataclass
class CodeChange:
    """Изменение в коде."""
    
    type: str  # "added", "removed", "modified", "moved"
    file_path: str
    function_name: Optional[str] = None
    old_signature: Optional[str] = None
    new_signature: Optional[str] = None
    architectural_zone: Optional[str] = None
    line_number: Optional[int] = None
    
    def to_dict(self) -> Dict:
        return {
            "type": self.type,
            "file_path": self.file_path,
            "function_name": self.function_name,
            "old_signature": self.old_signature,
            "new_signature": self.new_signature,
            "architectural_zone": self.architectural_zone,
            "line_number": self.line_number
        }


class CodeChangeTracker:
    """
    Отслеживает изменения кода для проверки актуальности документации.
    
    Используется вместе с DocConsistencyChecker для поиска противоречий
    между кодом и документацией.
    """
    
    def __init__(self, repo_ctx: RepositoryContext):
        self.repo_ctx = repo_ctx
        self.repo_root = repo_ctx.repo_root
        self.heuristics = repo_ctx.heuristics
    
    def detect_changes(self) -> List[CodeChange]:
        """
        Сравнивает file_hashes с текущим состоянием.
        
        Returns:
            Список изменений
        """
        changes = []
        
        # Получаем текущие файлы
        from src.services.repository_context import SafeFileWalker
        walker = SafeFileWalker(self.repo_root, self.heuristics)
        current_files = {
            str(f.relative_to(self.repo_root))
            for f in walker.walk_source_files()
        }
        
        # Получаем проиндексированные файлы
        indexed_files = set(self.repo_ctx.file_hashes.keys())
        
        # 1. Удалённые файлы
        removed_files = indexed_files - current_files
        for file_path in removed_files:
            changes.append(CodeChange(
                type="removed",
                file_path=file_path,
                architectural_zone=self._get_file_zone(file_path)
            ))
        
        # 2. Новые и изменённые файлы
        for file_path in current_files:
            if file_path not in indexed_files:
                # Новый файл
                changes.append(CodeChange(
                    type="added",
                    file_path=file_path,
                    architectural_zone=self._get_file_zone(file_path)
                ))
            else:
                # Проверяем изменения
                current_hash = self._compute_file_hash(file_path)
                if self.repo_ctx.file_hashes.get(file_path) != current_hash:
                    # Файл изменён — проверяем функции
                    func_changes = self._detect_function_changes(file_path)
                    changes.extend(func_changes)
                    
                    # Если нет изменений функций, но файл изменён
                    if not func_changes:
                        changes.append(CodeChange(
                            type="modified",
                            file_path=file_path,
                            architectural_zone=self._get_file_zone(file_path)
                        ))
        
        # 3. Эвристика для переименованных файлов
        renamed = self._detect_renamed_files(changes)
        changes.extend(renamed)
        
        logger.debug(
            f"📊 Detected {len(changes)} code changes: "
            f"{sum(1 for c in changes if c.type == 'added')} added, "
            f"{sum(1 for c in changes if c.type == 'removed')} removed, "
            f"{sum(1 for c in changes if c.type == 'modified')} modified"
        )
        
        return changes
    
    def _detect_function_changes(self, file_path: str) -> List[CodeChange]:
        """
        Определяет изменения функций в файле.
        
        Args:
            file_path: Относительный путь к файлу
            
        Returns:
            Список изменений функций
        """
        changes = []
        analysis = self.repo_ctx.static_analyzer.analyze_file(self.repo_root / file_path)
        
        if not analysis.get("success"):
            return changes
        
        # Получаем предыдущий анализ (если есть)
        old_analysis = self.repo_ctx.file_analyses.get(file_path, {})
        old_functions = {
            f["name"]: f for f in old_analysis.get("functions", [])
        }
        
        # Проверяем каждую функцию
        for func in analysis.get("functions", []):
            func_name = func["name"]
            new_signature = self._extract_signature(func)
            
            if func_name not in old_functions:
                # Новая функция
                changes.append(CodeChange(
                    type="added",
                    file_path=file_path,
                    function_name=func_name,
                    new_signature=new_signature,
                    architectural_zone=analysis.get("architectural_zone"),
                    line_number=func.get("line_start")
                ))
            else:
                # Проверяем изменения
                old_func = old_functions[func_name]
                old_signature = self._extract_signature(old_func)
                
                if new_signature != old_signature:
                    # Сигнатура изменилась
                    changes.append(CodeChange(
                        type="modified",
                        file_path=file_path,
                        function_name=func_name,
                        old_signature=old_signature,
                        new_signature=new_signature,
                        architectural_zone=analysis.get("architectural_zone"),
                        line_number=func.get("line_start")
                    ))
        
        # Проверяем удалённые функции
        current_functions = {f["name"] for f in analysis.get("functions", [])}
        for func_name in old_functions:
            if func_name not in current_functions:
                old_signature = self._extract_signature(old_functions[func_name])
                changes.append(CodeChange(
                    type="removed",
                    file_path=file_path,
                    function_name=func_name,
                    old_signature=old_signature,
                    architectural_zone=analysis.get("architectural_zone")
                ))
        
        return changes
    
    def _extract_signature(self, func_analysis: Dict) -> str:
        """
        Извлекает сигнатуру функции из анализа.
        
        Args:
            func_analysis: Результат анализа функции
            
        Returns:
            Сигнатура функции
        """
        ctx = func_analysis.get("context", {})
        name = func_analysis.get("name", "unknown")
        
        # Параметры
        params = []
        for param in ctx.get("parameters", []):
            param_str = param["name"]
            if param.get("type_annotation"):
                param_str += f": {param['type_annotation']}"
            if param.get("default_value"):
                param_str += f" = {param['default_value']}"
            params.append(param_str)
        
        # Возвращаемый тип
        return_ann = ctx.get("type_semantics", {}).get("return", "Any")
        
        return f"def {name}({', '.join(params)}) -> {return_ann}"
    
    def _get_file_zone(self, file_path: str) -> Optional[str]:
        """Получает архитектурную зону файла."""
        analysis = self.repo_ctx.file_analyses.get(file_path, {})
        return analysis.get("architectural_zone")
    
    def _compute_file_hash(self, file_path: str) -> str:
        """Вычисляет хэш файла."""
        from src.services.static_analyzer import StaticAnalyzer
        return StaticAnalyzer._compute_hash(self.repo_root / file_path)
    
    def _detect_renamed_files(self, changes: List[CodeChange]) -> List[CodeChange]:
        """
        Эвристика для обнаружения переименованных файлов.
        
        Если файл удалён и добавлен файл с похожим именем/содержанием —
        предполагаем переименование.
        """
        renamed = []
        
        removed = [c for c in changes if c.type == "removed"]
        added = [c for c in changes if c.type == "added"]
        
        for rem in removed:
            for add in added:
                # Сравниваем имена файлов
                rem_name = Path(rem.file_path).stem
                add_name = Path(add.file_path).stem
                
                if rem_name == add_name:
                    # Одинаковые имена — возможно переименование пути
                    if rem.file_path != add.file_path:
                        renamed.append(CodeChange(
                            type="moved",
                            file_path=add.file_path,
                            old_signature=rem.file_path,  # Старый путь
                            new_signature=add.file_path,  # Новый путь
                            architectural_zone=add.architectural_zone
                        ))
                        # Помечаем как обработанные
                        rem.type = "processed"
                        add.type = "processed"
        
        # Удаляем обработанные
        changes = [c for c in changes if c.type != "processed"]
        
        return renamed
    
    def get_changed_functions(self) -> List[CodeChange]:
        """
        Возвращает изменения функций (добавленные/изменённые/удалённые).
        
        Returns:
            Список изменений функций
        """
        changes = self.detect_changes()
        return [c for c in changes if c.function_name]
    
    def get_changed_zones(self) -> Set[str]:
        """
        Возвращает изменённые архитектурные зоны.
        
        Returns:
            Множество названий зон
        """
        changes = self.detect_changes()
        zones = {c.architectural_zone for c in changes if c.architectural_zone}
        return zones - {None}