10장: 실전 프로젝트 -- LLM 코드 분석 파이프라인 구축

"""통합 AST 파싱 엔진"""
import ast
from dataclasses import dataclass, field
from pathlib import Path
 
 
@dataclass
class CodeEntity:
    """코드 엔티티 (함수, 클래스, 메서드)"""
    name: str
    entity_type: str  # function, class, method
    filepath: str
    line_start: int
    line_end: int
    source: str
    metrics: dict = field(default_factory=dict)
    children: list["CodeEntity"] = field(default_factory=list)
 
 
@dataclass
class FileAnalysis:
    """파일 분석 결과"""
    filepath: str
    language: str
    total_lines: int
    entities: list[CodeEntity]
    imports: list[str]
    global_metrics: dict
 
 
class UnifiedParser:
    """Python과 TypeScript를 지원하는 통합 파서"""
 
    def parse_file(self, filepath: str) -> FileAnalysis:
        path = Path(filepath)
        source = path.read_text()
 
        if path.suffix == ".py":
            return self._parse_python(filepath, source)
        elif path.suffix in (".ts", ".tsx"):
            return self._parse_typescript(filepath, source)
        else:
            raise ValueError(f"지원하지 않는 확장자: {path.suffix}")
 
    def parse_directory(self, dirpath: str) -> list[FileAnalysis]:
        results = []
        root = Path(dirpath)
 
        ignore_dirs = {"node_modules", ".git", "__pycache__", ".venv", "dist"}
 
        for path in root.rglob("*"):
            if any(p in path.parts for p in ignore_dirs):
                continue
            if path.suffix in (".py", ".ts", ".tsx"):
                try:
                    results.append(self.parse_file(str(path)))
                except (SyntaxError, UnicodeDecodeError) as e:
                    results.append(FileAnalysis(
                        filepath=str(path),
                        language="unknown",
                        total_lines=0,
                        entities=[],
                        imports=[],
                        global_metrics={"error": str(e)},
                    ))
 
        return results
 
    def _parse_python(self, filepath: str, source: str) -> FileAnalysis:
        tree = ast.parse(source)
        lines = source.splitlines()
        entities = []
        imports = []
 
        for node in ast.iter_child_nodes(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                entities.append(self._python_function_entity(
                    node, filepath, lines
                ))
            elif isinstance(node, ast.ClassDef):
                class_entity = self._python_class_entity(
                    node, filepath, lines
                )
                entities.append(class_entity)
            elif isinstance(node, ast.Import):
                for alias in node.names:
                    imports.append(alias.name)
            elif isinstance(node, ast.ImportFrom):
                if node.module:
                    imports.append(node.module)
 
        return FileAnalysis(
            filepath=filepath,
            language="python",
            total_lines=len(lines),
            entities=entities,
            imports=imports,
            global_metrics={
                "entity_count": len(entities),
                "import_count": len(imports),
                "avg_complexity": (
                    sum(
                        e.metrics.get("complexity", 0)
                        for e in entities
                    ) / max(len(entities), 1)
                ),
            },
        )
 
    def _python_function_entity(
        self,
        node: ast.FunctionDef,
        filepath: str,
        lines: list[str],
    ) -> CodeEntity:
        start = node.lineno - 1
        end = node.end_lineno or node.lineno
        source = "\n".join(lines[start:end])
 
        return CodeEntity(
            name=node.name,
            entity_type="function",
            filepath=filepath,
            line_start=node.lineno,
            line_end=end,
            source=source,
            metrics={
                "complexity": self._calc_complexity(node),
                "line_count": end - node.lineno + 1,
                "param_count": len(node.args.args),
                "has_docstring": ast.get_docstring(node) is not None,
                "return_count": sum(
                    1 for n in ast.walk(node)
                    if isinstance(n, ast.Return)
                ),
            },
        )
 
    def _python_class_entity(
        self,
        node: ast.ClassDef,
        filepath: str,
        lines: list[str],
    ) -> CodeEntity:
        start = node.lineno - 1
        end = node.end_lineno or node.lineno
        source = "\n".join(lines[start:end])
 
        children = []
        for child in node.body:
            if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
                children.append(self._python_function_entity(
                    child, filepath, lines
                ))
 
        return CodeEntity(
            name=node.name,
            entity_type="class",
            filepath=filepath,
            line_start=node.lineno,
            line_end=end,
            source=source,
            metrics={
                "method_count": len(children),
                "line_count": end - node.lineno + 1,
                "has_docstring": ast.get_docstring(node) is not None,
            },
            children=children,
        )
 
    def _parse_typescript(self, filepath: str, source: str) -> FileAnalysis:
        # ts-morph 기반 파싱 (Python에서는 subprocess로 호출)
        # 실제 구현에서는 Node.js 스크립트와 연동
        return FileAnalysis(
            filepath=filepath,
            language="typescript",
            total_lines=len(source.splitlines()),
            entities=[],
            imports=[],
            global_metrics={},
        )
 
    def _calc_complexity(self, node: ast.AST) -> int:
        complexity = 1
        for child in ast.walk(node):
            if isinstance(child, (ast.If, ast.While, ast.For)):
                complexity += 1
            elif isinstance(child, ast.BoolOp):
                complexity += len(child.values) - 1
            elif isinstance(child, ast.ExceptHandler):
                complexity += 1
        return complexity

"""통합 코드 스멜 탐지기"""
from dataclasses import dataclass
 
 
@dataclass
class DetectedSmell:
    smell_type: str
    severity: str
    entity_name: str
    filepath: str
    line_start: int
    description: str
    metrics: dict
    remediation_hours: float
 
 
class IntegratedSmellDetector:
    """AST 메트릭 + LLM 하이브리드 스멜 탐지"""
 
    # 메트릭 기반 탐지 임계값
    THRESHOLDS = {
        "complexity": {"warning": 10, "critical": 20},
        "line_count": {"warning": 40, "critical": 80},
        "param_count": {"warning": 4, "critical": 7},
        "method_count": {"warning": 10, "critical": 20},
    }
 
    def __init__(self, llm_client=None):
        self.llm_client = llm_client
 
    async def detect(
        self, file_analyses: list,
    ) -> list[DetectedSmell]:
        smells = []
 
        # 1단계: 메트릭 기반 탐지 (빠름, 비용 없음)
        for file_analysis in file_analyses:
            for entity in file_analysis.entities:
                smells.extend(self._check_metrics(entity))
 
        # 2단계: LLM 기반 심층 탐지 (느림, 비용 있음)
        if self.llm_client:
            high_priority = [
                s for s in smells if s.severity == "critical"
            ]
            for smell in high_priority:
                llm_smells = await self._llm_deep_analysis(smell)
                smells.extend(llm_smells)
 
        return self._deduplicate(smells)
 
    def _check_metrics(self, entity) -> list[DetectedSmell]:
        smells = []
        metrics = entity.metrics
 
        # 순환 복잡도 검사
        cx = metrics.get("complexity", 0)
        if cx >= self.THRESHOLDS["complexity"]["critical"]:
            smells.append(DetectedSmell(
                smell_type="high_complexity",
                severity="critical",
                entity_name=entity.name,
                filepath=entity.filepath,
                line_start=entity.line_start,
                description=(
                    f"순환 복잡도가 {cx}로 매우 높습니다. "
                    f"함수를 분해하세요."
                ),
                metrics={"complexity": cx},
                remediation_hours=cx * 0.3,
            ))
        elif cx >= self.THRESHOLDS["complexity"]["warning"]:
            smells.append(DetectedSmell(
                smell_type="moderate_complexity",
                severity="warning",
                entity_name=entity.name,
                filepath=entity.filepath,
                line_start=entity.line_start,
                description=f"순환 복잡도가 {cx}입니다. 개선을 검토하세요.",
                metrics={"complexity": cx},
                remediation_hours=cx * 0.2,
            ))
 
        # 함수 길이 검사
        line_count = metrics.get("line_count", 0)
        if line_count >= self.THRESHOLDS["line_count"]["critical"]:
            smells.append(DetectedSmell(
                smell_type="long_function",
                severity="critical",
                entity_name=entity.name,
                filepath=entity.filepath,
                line_start=entity.line_start,
                description=f"함수가 {line_count}줄로 너무 깁니다.",
                metrics={"line_count": line_count},
                remediation_hours=2.0,
            ))
 
        # 매개변수 수 검사
        param_count = metrics.get("param_count", 0)
        if param_count >= self.THRESHOLDS["param_count"]["critical"]:
            smells.append(DetectedSmell(
                smell_type="long_parameter_list",
                severity="warning",
                entity_name=entity.name,
                filepath=entity.filepath,
                line_start=entity.line_start,
                description=(
                    f"매개변수가 {param_count}개입니다. "
                    f"매개변수 객체 도입을 고려하세요."
                ),
                metrics={"param_count": param_count},
                remediation_hours=1.0,
            ))
 
        return smells
 
    async def _llm_deep_analysis(self, smell: DetectedSmell) -> list[DetectedSmell]:
        # LLM을 통한 추가 스멜 탐지 (간략화)
        return []
 
    def _deduplicate(self, smells: list[DetectedSmell]) -> list[DetectedSmell]:
        seen = set()
        unique = []
        for smell in smells:
            key = (smell.smell_type, smell.entity_name, smell.filepath)
            if key not in seen:
                seen.add(key)
                unique.append(smell)
        return unique

"""리팩터링 계획 수립 및 실행"""
from dataclasses import dataclass
 
 
@dataclass
class RefactorAction:
    target_file: str
    target_entity: str
    action_type: str  # extract_function, rename, move, inline
    description: str
    priority: int
    estimated_hours: float
 
 
@dataclass
class RefactorPlan:
    actions: list[RefactorAction]
    total_hours: float
    expected_improvements: dict
 
 
class RefactorPlanner:
    """우선순위 기반 리팩터링 계획 수립"""
 
    def __init__(self, llm_client):
        self.llm_client = llm_client
 
    async def create_plan(
        self,
        smells: list,
        budget_hours: float = 40,
    ) -> RefactorPlan:
        # 우선순위 정렬: 심각도 x 수정 용이성
        prioritized = sorted(
            smells,
            key=lambda s: (
                {"critical": 4, "warning": 2, "info": 1}.get(
                    s.severity, 1
                ) / max(s.remediation_hours, 0.5)
            ),
            reverse=True,
        )
 
        actions = []
        total_hours = 0
 
        for smell in prioritized:
            if total_hours + smell.remediation_hours > budget_hours:
                continue
 
            action = await self._plan_action(smell)
            if action:
                actions.append(action)
                total_hours += action.estimated_hours
 
        return RefactorPlan(
            actions=actions,
            total_hours=total_hours,
            expected_improvements={
                "complexity_reduction": self._estimate_complexity_reduction(
                    actions
                ),
                "smell_reduction": len(actions),
            },
        )
 
    async def _plan_action(self, smell) -> RefactorAction | None:
        action_map = {
            "high_complexity": "extract_function",
            "long_function": "extract_function",
            "long_parameter_list": "introduce_parameter_object",
            "duplicate_code": "extract_common_function",
            "feature_envy": "move_method",
        }
 
        action_type = action_map.get(smell.smell_type)
        if not action_type:
            return None
 
        # LLM에게 구체적인 리팩터링 지시 요청
        prompt = f"""다음 코드 스멜에 대한 구체적인 리팩터링 지시를 생성하세요.
 
스멜 유형: {smell.smell_type}
대상: {smell.entity_name} ({smell.filepath}:{smell.line_start})
설명: {smell.description}
권장 액션: {action_type}
 
다음을 포함하세요:
1. 구체적인 리팩터링 단계
2. 추출할 함수의 이름과 매개변수
3. 주의사항"""
 
        response = await self.llm_client.generate(prompt)
 
        return RefactorAction(
            target_file=smell.filepath,
            target_entity=smell.entity_name,
            action_type=action_type,
            description=response,
            priority=1,
            estimated_hours=smell.remediation_hours,
        )
 
    def _estimate_complexity_reduction(
        self, actions: list[RefactorAction]
    ) -> float:
        return sum(
            3.0 if a.action_type == "extract_function" else 1.0
            for a in actions
        )

"""전체 파이프라인 오케스트레이터"""
import json
import asyncio
from dataclasses import dataclass, asdict
from datetime import datetime
from pathlib import Path
 
 
@dataclass
class PipelineResult:
    project_path: str
    analyzed_at: str
    files_analyzed: int
    total_entities: int
    smells_detected: int
    critical_smells: int
    refactor_actions: int
    estimated_hours: float
    report_path: str
 
 
class CodeAnalysisPipeline:
    """LLM 코드 분석 통합 파이프라인"""
 
    def __init__(
        self,
        llm_client=None,
        output_dir: str = "./reports",
    ):
        self.llm_client = llm_client
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
 
    async def run(
        self,
        project_path: str,
        budget_hours: float = 40,
    ) -> PipelineResult:
        print(f"프로젝트 분석 시작: {project_path}")
 
        # 1단계: AST 파싱
        print("  [1/7] AST 파싱 중...")
        parser = UnifiedParser()
        file_analyses = parser.parse_directory(project_path)
        print(f"  -> {len(file_analyses)}개 파일 분석 완료")
 
        # 2단계: 코드 스멜 탐지
        print("  [2/7] 코드 스멜 탐지 중...")
        detector = IntegratedSmellDetector(self.llm_client)
        smells = await detector.detect(file_analyses)
        print(f"  -> {len(smells)}개 스멜 감지")
 
        # 3단계: 기술 부채 정량화
        print("  [3/7] 기술 부채 계산 중...")
        total_debt_hours = sum(s.remediation_hours for s in smells)
        critical_count = sum(
            1 for s in smells if s.severity == "critical"
        )
 
        # 4단계: 보안 분석
        print("  [4/7] 보안 분석 중...")
        # security_scanner.scan(file_analyses)
 
        # 5단계: 리팩터링 계획
        print("  [5/7] 리팩터링 계획 수립 중...")
        planner = RefactorPlanner(self.llm_client)
        plan = await planner.create_plan(smells, budget_hours)
 
        # 6단계: 아키텍처 분석
        print("  [6/7] 아키텍처 분석 중...")
        # architecture_analyzer.analyze(file_analyses)
 
        # 7단계: 리포트 생성
        print("  [7/7] 리포트 생성 중...")
        report = self._generate_report(
            file_analyses, smells, plan
        )
        report_path = self._save_report(report)
 
        return PipelineResult(
            project_path=project_path,
            analyzed_at=datetime.now().isoformat(),
            files_analyzed=len(file_analyses),
            total_entities=sum(
                len(f.entities) for f in file_analyses
            ),
            smells_detected=len(smells),
            critical_smells=critical_count,
            refactor_actions=len(plan.actions),
            estimated_hours=plan.total_hours,
            report_path=str(report_path),
        )
 
    def _generate_report(
        self, file_analyses, smells, plan
    ) -> dict:
        return {
            "summary": {
                "files": len(file_analyses),
                "entities": sum(
                    len(f.entities) for f in file_analyses
                ),
                "smells": len(smells),
                "critical": sum(
                    1 for s in smells if s.severity == "critical"
                ),
                "total_debt_hours": round(
                    sum(s.remediation_hours for s in smells), 1
                ),
            },
            "smells": [
                {
                    "type": s.smell_type,
                    "severity": s.severity,
                    "entity": s.entity_name,
                    "file": s.filepath,
                    "line": s.line_start,
                    "description": s.description,
                }
                for s in smells
            ],
            "plan": {
                "actions": [
                    {
                        "file": a.target_file,
                        "entity": a.target_entity,
                        "action": a.action_type,
                        "hours": a.estimated_hours,
                    }
                    for a in plan.actions
                ],
                "total_hours": plan.total_hours,
            },
        }
 
    def _save_report(self, report: dict) -> Path:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        path = self.output_dir / f"analysis_{timestamp}.json"
        path.write_text(
            json.dumps(report, indent=2, ensure_ascii=False)
        )
        return path