Разработка AI-системы управления техническим долгом (Tech Debt AI)

AI-система управления техническим долгом

Технический долг — это накопленные компромиссы: "сделаем быстро сейчас, перепишем потом". Проблема в том, что "потом" не наступает, потому что сложно измерить долг и обосновать время на рефакторинг менеджменту. AI-система делает технический долг измеримым, приоритизированным и управляемым как обычный backlog.

Обнаружение и измерение долга

from anthropic import Anthropic
from pathlib import Path
import ast
import subprocess
import json
from dataclasses import dataclass, field
from typing import Literal

client = Anthropic()

@dataclass
class DebtItem:
    id: str
    file: str
    category: Literal["code_smell", "architecture", "security", "test_coverage", "documentation", "dependency"]
    severity: Literal["critical", "high", "medium", "low"]
    title: str
    description: str
    estimated_hours: float
    business_impact: str
    quick_fix: bool = False

class TechDebtScanner:

    def scan_project(self, project_root: str) -> list[DebtItem]:
        """Полное сканирование технического долга проекта"""
        all_items = []

        # 1. Зависимости с уязвимостями
        all_items.extend(self._scan_dependencies(project_root))

        # 2. Сложность кода
        all_items.extend(self._scan_complexity(project_root))

        # 3. AI-анализ архитектурных проблем
        all_items.extend(self._ai_scan_architecture(project_root))

        # 4. TODO/FIXME/HACK комментарии
        all_items.extend(self._scan_comments(project_root))

        return all_items

    def _scan_dependencies(self, project_root: str) -> list[DebtItem]:
        """Сканирует зависимости на уязвимости и устаревшие версии"""
        items = []

        # Safety для Python зависимостей
        result = subprocess.run(
            ["safety", "check", "--json", "--full-report"],
            capture_output=True, text=True, cwd=project_root
        )

        if result.returncode != 0 and result.stdout:
            try:
                vulns = json.loads(result.stdout)
                for vuln in vulns:
                    items.append(DebtItem(
                        id=f"dep_{vuln.get('package_name', 'unknown')}",
                        file="requirements.txt",
                        category="security",
                        severity="critical" if "critical" in str(vuln).lower() else "high",
                        title=f"Уязвимость в {vuln.get('package_name')} {vuln.get('affected_versions')}",
                        description=vuln.get("vulnerability", ""),
                        estimated_hours=0.5,
                        business_impact="Потенциальная уязвимость безопасности",
                        quick_fix=True,
                    ))
            except json.JSONDecodeError:
                pass

        # pip-audit как альтернатива
        result = subprocess.run(
            ["pip-audit", "--format=json"],
            capture_output=True, text=True, cwd=project_root
        )
        # Обработка аналогично...

        return items

    def _scan_complexity(self, project_root: str) -> list[DebtItem]:
        """Находит функции с высокой цикломатической сложностью"""
        items = []

        result = subprocess.run(
            ["radon", "cc", "-j", "-n", "C", project_root],  # Только оценка C и выше
            capture_output=True, text=True
        )

        if result.stdout:
            data = json.loads(result.stdout)
            for file_path, functions in data.items():
                for func in functions:
                    complexity = func.get("complexity", 0)
                    if complexity >= 10:
                        hours = complexity * 0.5  # Грубая оценка
                        items.append(DebtItem(
                            id=f"cc_{file_path}_{func['name']}",
                            file=file_path,
                            category="code_smell",
                            severity="critical" if complexity >= 20 else "high" if complexity >= 15 else "medium",
                            title=f"Высокая сложность: {func['name']} (CC={complexity})",
                            description=f"Цикломатическая сложность {complexity} при пороге 10. Функция трудно тестируема и поддерживаема.",
                            estimated_hours=hours,
                            business_impact="Увеличивает время изменений, риск багов при модификации",
                        ))

        return items

    def _scan_comments(self, project_root: str) -> list[DebtItem]:
        """Находит TODO/FIXME/HACK маркеры"""
        items = []

        result = subprocess.run(
            ["grep", "-rn", "--include=*.py", r"#\s*\(TODO\|FIXME\|HACK\|XXX\|BUG\)", project_root],
            capture_output=True, text=True
        )

        for line in result.stdout.splitlines():
            if ":" in line:
                parts = line.split(":", 2)
                if len(parts) >= 3:
                    file_path, line_num, comment = parts
                    severity = "high" if "HACK" in comment or "FIXME" in comment else "low"
                    items.append(DebtItem(
                        id=f"todo_{hash(line)}",
                        file=file_path,
                        category="code_smell",
                        severity=severity,
                        title=f"Технический маркер в коде",
                        description=comment.strip(),
                        estimated_hours=2.0,
                        business_impact="Задокументированный технический долг",
                        quick_fix=False,
                    ))

        return items

    def _ai_scan_architecture(self, project_root: str) -> list[DebtItem]:
        """AI-анализ архитектурных проблем"""
        items = []

        # Читаем структуру проекта
        structure = []
        for root, dirs, files in Path(project_root).walk():
            dirs[:] = [d for d in dirs if d not in {".git", "__pycache__", ".venv"}]
            for f in files:
                if f.endswith(".py"):
                    structure.append(str(Path(root) / f))

        # Анализируем файлы с размером > 500 строк (потенциальные God Objects)
        large_files = []
        for fp in structure[:50]:
            try:
                lines = Path(fp).read_text().splitlines()
                if len(lines) > 500:
                    large_files.append((fp, len(lines)))
            except Exception:
                pass

        if not large_files:
            return items

        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""Проанализируй список больших файлов на архитектурные проблемы.

Файлы (путь, кол-во строк):
{json.dumps(large_files, ensure_ascii=False)}

Верни JSON:
[{{
  "file": "...",
  "issue": "...",
  "severity": "high|medium",
  "estimated_hours": <число>,
  "recommendation": "..."
}}]"""
            }]
        )

        text = response.content[0].text
        start = text.find("[")
        end = text.rfind("]") + 1
        arch_issues = json.loads(text[start:end])

        for issue in arch_issues:
            items.append(DebtItem(
                id=f"arch_{hash(issue['file'])}",
                file=issue["file"],
                category="architecture",
                severity=issue.get("severity", "medium"),
                title=issue.get("issue", "Архитектурная проблема"),
                description=issue.get("recommendation", ""),
                estimated_hours=issue.get("estimated_hours", 8.0),
                business_impact="Замедляет разработку новых фич",
            ))

        return items

Приоритизация и планирование

class TechDebtPlanner:

    def prioritize(
        self,
        items: list[DebtItem],
        available_hours: float,
        team_velocity: float = 0.7,
    ) -> dict:
        """Приоритизирует долг с учётом доступного времени"""

        # Скор = impact * urgency / effort
        severity_weights = {"critical": 100, "high": 40, "medium": 10, "low": 2}

        scored = []
        for item in items:
            base_score = severity_weights[item.severity]
            effort = max(item.estimated_hours, 0.5)
            # Quick fixes в приоритете
            if item.quick_fix:
                base_score *= 2
            score = base_score / effort
            scored.append((score, item))

        scored.sort(key=lambda x: x[0], reverse=True)

        # Формируем план
        selected = []
        total_hours = 0.0
        effective_hours = available_hours * team_velocity

        for _, item in scored:
            if total_hours + item.estimated_hours <= effective_hours:
                selected.append(item)
                total_hours += item.estimated_hours

        return {
            "selected_items": selected,
            "total_hours": total_hours,
            "debt_reduced_hours": total_hours,
            "remaining_items": [item for _, item in scored if item not in selected],
            "sprint_capacity": available_hours,
        }

    def generate_jira_tickets(self, items: list[DebtItem]) -> list[dict]:
        """Генерирует Jira-задачи для технического долга"""

        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=4096,
            messages=[{
                "role": "user",
                "content": f"""Сформируй Jira-задачи для технического долга.

Элементы долга:
{json.dumps([{
    "title": i.title,
    "description": i.description,
    "severity": i.severity,
    "hours": i.estimated_hours,
    "business_impact": i.business_impact,
} for i in items], ensure_ascii=False, indent=2)}

Для каждого создай Jira-задачу:
{{
  "summary": "...",
  "description": "...",
  "story_points": <1-13>,
  "priority": "Highest|High|Medium|Low",
  "labels": ["tech-debt", "<category>"],
  "acceptance_criteria": ["..."]
}}"""
            }]
        )

        text = response.content[0].text
        start = text.find("[")
        end = text.rfind("]") + 1
        return json.loads(text[start:end])

Дашборд технического долга

def generate_debt_dashboard(project_root: str) -> dict:
    """Генерирует полный отчёт по техническому долгу"""
    scanner = TechDebtScanner()
    items = scanner.scan_project(project_root)

    by_severity = {}
    for item in items:
        by_severity.setdefault(item.severity, []).append(item)

    total_hours = sum(i.estimated_hours for i in items)

    return {
        "total_items": len(items),
        "total_hours": total_hours,
        "debt_index": round(total_hours / max(len(list(Path(project_root).rglob("*.py"))), 1), 2),
        "by_severity": {k: len(v) for k, v in by_severity.items()},
        "by_category": {},
        "top_10_critical": sorted(
            [i for i in items if i.severity in ("critical", "high")],
            key=lambda x: x.estimated_hours,
            reverse=True
        )[:10],
    }

Практический кейс: SaaS с 4-летним долгом

Контекст: SaaS для HR, 4 года разработки, 3 смены команды. Жалобы: любая новая фича занимает в 3–4 раза дольше ожидаемого.

Результаты сканирования:

• 847 элементов технического долга
• Критических: 23 (12 уязвимостей в зависимостях, 11 God Objects)
• Общая оценка: 1340 часов для полного погашения
• Debt Index: 8.7 (высокий — норма < 3.0)

Plan of attack:

• Sprint 1 (20ч): все уязвимости зависимостей — обновление пакетов, 0.5ч каждый
• Sprint 2–4 (60ч): 4 главных God Object → декомпозиция
• Sprint 5–8 (80ч): test coverage с 28% → 70%

Результаты через 4 месяца:

• Debt Index: 8.7 → 3.2
• Время реализации типовой фичи: -41%
• Bug rate в production: -38%

ROI расчёт: 160 часов на погашение долга сохранили ~320 часов замедления за квартал — окупаемость 2:1 за первый квартал.

Сроки

• Базовый сканер (complexity + TODO + зависимости): 3–5 дней
• AI-анализ архитектурных проблем: 1 неделя
• Приоритизация + генерация Jira-задач: 1 неделя
• Dashboard с историческими трендами: 2 недели