8장: 아키텍처 분석과 시각화

from collections import defaultdict
from dataclasses import dataclass
 
 
@dataclass
class DependencyEdge:
    source: str
    target: str
    weight: int  # import 횟수
    edge_type: str  # direct, transitive
 
 
class DSMBuilder:
    """의존성 구조 매트릭스 생성기"""
 
    def __init__(self):
        self.modules: list[str] = []
        self.edges: list[DependencyEdge] = []
 
    def build_from_graph(self, graph: dict) -> dict:
        self.modules = sorted(graph["nodes"].keys())
        self.edges = [
            DependencyEdge(
                source=src, target=tgt, weight=1, edge_type="direct"
            )
            for src, tgt in graph["edges"]
        ]
 
        matrix = self._create_matrix()
        clusters = self._identify_clusters(matrix)
        cycles = self._detect_cycles()
 
        return {
            "modules": self.modules,
            "matrix": matrix,
            "clusters": clusters,
            "cycles": cycles,
        }
 
    def _create_matrix(self) -> list[list[int]]:
        size = len(self.modules)
        matrix = [[0] * size for _ in range(size)]
        module_index = {m: i for i, m in enumerate(self.modules)}
 
        for edge in self.edges:
            if edge.source in module_index and edge.target in module_index:
                i = module_index[edge.source]
                j = module_index[edge.target]
                matrix[i][j] = edge.weight
 
        return matrix
 
    def _identify_clusters(self, matrix: list[list[int]]) -> list[list[str]]:
        """강하게 연결된 모듈 그룹 식별"""
        n = len(self.modules)
        visited = [False] * n
        clusters = []
 
        def dfs(node: int, cluster: list[int]):
            visited[node] = True
            cluster.append(node)
            for j in range(n):
                if matrix[node][j] > 0 and not visited[j]:
                    dfs(j, cluster)
 
        for i in range(n):
            if not visited[i]:
                cluster: list[int] = []
                dfs(i, cluster)
                if len(cluster) > 1:
                    clusters.append([self.modules[idx] for idx in cluster])
 
        return clusters
 
    def _detect_cycles(self) -> list[list[str]]:
        """순환 의존성 탐지 (Tarjan 알고리즘 기반)"""
        graph = defaultdict(set)
        for edge in self.edges:
            graph[edge.source].add(edge.target)
 
        index_counter = [0]
        stack = []
        lowlink = {}
        index = {}
        on_stack = {}
        sccs = []
 
        def strongconnect(v: str):
            index[v] = index_counter[0]
            lowlink[v] = index_counter[0]
            index_counter[0] += 1
            stack.append(v)
            on_stack[v] = True
 
            for w in graph[v]:
                if w not in index:
                    strongconnect(w)
                    lowlink[v] = min(lowlink[v], lowlink[w])
                elif on_stack.get(w, False):
                    lowlink[v] = min(lowlink[v], index[w])
 
            if lowlink[v] == index[v]:
                scc = []
                while True:
                    w = stack.pop()
                    on_stack[w] = False
                    scc.append(w)
                    if w == v:
                        break
                if len(scc) > 1:
                    sccs.append(scc)
 
        for v in self.modules:
            if v not in index:
                strongconnect(v)
 
        return sccs

from dataclasses import dataclass
from pathlib import Path
 
 
@dataclass
class LayerViolation:
    source_module: str
    source_layer: str
    target_module: str
    target_layer: str
    violation_type: str  # upward, skip, cross
 
 
@dataclass
class ArchitectureRule:
    name: str
    layers: list[str]  # 상위에서 하위 순서
    layer_patterns: dict[str, list[str]]  # 레이어명 -> 경로 패턴
    allowed_skip: bool  # 레이어 건너뛰기 허용 여부
 
 
class LayerChecker:
    """레이어 위반을 탐지하는 아키텍처 검증기"""
 
    def __init__(self, rule: ArchitectureRule):
        self.rule = rule
        self.layer_order = {
            layer: idx for idx, layer in enumerate(rule.layers)
        }
 
    def check(self, dependencies: list[tuple[str, str]]) -> list[LayerViolation]:
        violations = []
 
        for source, target in dependencies:
            source_layer = self._identify_layer(source)
            target_layer = self._identify_layer(target)
 
            if source_layer is None or target_layer is None:
                continue
 
            source_idx = self.layer_order[source_layer]
            target_idx = self.layer_order[target_layer]
 
            # 상향 참조 위반
            if target_idx < source_idx:
                violations.append(LayerViolation(
                    source_module=source,
                    source_layer=source_layer,
                    target_module=target,
                    target_layer=target_layer,
                    violation_type="upward",
                ))
 
            # 레이어 건너뛰기 위반
            if (not self.rule.allowed_skip
                    and target_idx > source_idx + 1):
                violations.append(LayerViolation(
                    source_module=source,
                    source_layer=source_layer,
                    target_module=target,
                    target_layer=target_layer,
                    violation_type="skip",
                ))
 
        return violations
 
    def _identify_layer(self, module_name: str) -> str | None:
        for layer, patterns in self.rule.layer_patterns.items():
            if any(p in module_name for p in patterns):
                return layer
        return None
 
 
# 사용 예시
clean_architecture = ArchitectureRule(
    name="Clean Architecture",
    layers=["presentation", "application", "domain", "infrastructure"],
    layer_patterns={
        "presentation": ["routes", "controllers", "views", "pages"],
        "application": ["services", "use_cases", "handlers"],
        "domain": ["models", "entities", "value_objects"],
        "infrastructure": ["repositories", "database", "external"],
    },
    allowed_skip=False,
)

import pytest
from architecture_checker import LayerChecker, ArchitectureRule
 
 
@pytest.fixture
def checker():
    rule = ArchitectureRule(
        name="Clean Architecture",
        layers=["presentation", "application", "domain", "infrastructure"],
        layer_patterns={
            "presentation": ["routes", "controllers"],
            "application": ["services"],
            "domain": ["models", "entities"],
            "infrastructure": ["repositories", "database"],
        },
        allowed_skip=False,
    )
    return LayerChecker(rule)
 
 
def test_no_upward_dependencies(checker, project_dependencies):
    """하위 레이어가 상위 레이어를 참조하지 않는지 검증"""
    violations = checker.check(project_dependencies)
    upward = [v for v in violations if v.violation_type == "upward"]
 
    assert len(upward) == 0, (
        f"상향 참조 위반 {len(upward)}건 발견:\n"
        + "\n".join(
            f"  {v.source_module} ({v.source_layer}) -> "
            f"{v.target_module} ({v.target_layer})"
            for v in upward
        )
    )
 
 
def test_no_layer_skipping(checker, project_dependencies):
    """레이어를 건너뛰는 참조가 없는지 검증"""
    violations = checker.check(project_dependencies)
    skips = [v for v in violations if v.violation_type == "skip"]
 
    assert len(skips) == 0, (
        f"레이어 건너뛰기 위반 {len(skips)}건 발견"
    )

BOUNDARY_PROMPT = """다음은 모놀리식 애플리케이션의 모듈 분석 결과입니다.
 
모듈 목록과 메트릭:
{module_metrics}
 
의존성 그래프:
{dependency_edges}
 
변경 빈도 (최근 6개월):
{change_frequencies}
 
도메인 컨텍스트:
{domain_context}
 
마이크로서비스 분해를 위한 서비스 경계를 제안하세요.
 
각 서비스에 대해 다음을 포함하세요:
1. 서비스 이름과 역할
2. 포함될 모듈 목록
3. 분리 근거 (도메인 경계, 변경 독립성, 팀 구조)
4. 서비스 간 통신 방식 (동기/비동기)
5. 데이터 소유권
6. 예상되는 도전 과제"""
 
 
class BoundarySuggester:
    """마이크로서비스 경계 제안 도구"""
 
    def __init__(self, llm_client):
        self.llm_client = llm_client
 
    async def suggest_boundaries(
        self,
        dsm: dict,
        change_frequencies: dict[str, int],
        domain_context: str,
    ) -> dict:
        module_metrics = "\n".join(
            f"- {module}: 의존성 {self._count_deps(module, dsm)}개, "
            f"변경 빈도 {change_frequencies.get(module, 0)}회"
            for module in dsm["modules"]
        )
 
        dependency_edges = "\n".join(
            f"- {src} -> {tgt}"
            for src, tgt in zip(
                [m for m in dsm["modules"] for _ in dsm["modules"]],
                [t for _ in dsm["modules"] for t in dsm["modules"]],
            )
            if dsm["matrix"][dsm["modules"].index(src)][dsm["modules"].index(tgt)] > 0
        ) if dsm.get("matrix") else "없음"
 
        prompt = BOUNDARY_PROMPT.format(
            module_metrics=module_metrics,
            dependency_edges=dependency_edges,
            change_frequencies=str(change_frequencies),
            domain_context=domain_context,
        )
 
        response = await self.llm_client.generate(prompt)
        return self._parse_boundaries(response)
 
    def _count_deps(self, module: str, dsm: dict) -> int:
        if module not in dsm["modules"]:
            return 0
        idx = dsm["modules"].index(module)
        return sum(1 for v in dsm["matrix"][idx] if v > 0)
 
    def _parse_boundaries(self, response: str) -> dict:
        return {"raw_suggestion": response}

class MermaidDiagramGenerator:
    """분석 결과를 Mermaid 다이어그램으로 변환"""
 
    def generate_architecture_diagram(
        self,
        dsm: dict,
        layers: dict[str, list[str]],
    ) -> str:
        lines = ["graph TD"]
 
        # 레이어별 서브그래프 생성
        for layer_name, modules in layers.items():
            lines.append(f"    subgraph {layer_name}")
            for module in modules:
                node_id = module.replace(".", "_")
                short_name = module.split(".")[-1]
                lines.append(f"        {node_id}[{short_name}]")
            lines.append("    end")
 
        # 의존성 엣지 추가
        for i, src in enumerate(dsm["modules"]):
            for j, tgt in enumerate(dsm["modules"]):
                if dsm["matrix"][i][j] > 0:
                    src_id = src.replace(".", "_")
                    tgt_id = tgt.replace(".", "_")
                    lines.append(f"    {src_id} --> {tgt_id}")
 
        # 순환 의존성 강조
        for cycle in dsm.get("cycles", []):
            for k in range(len(cycle)):
                src_id = cycle[k].replace(".", "_")
                tgt_id = cycle[(k + 1) % len(cycle)].replace(".", "_")
                lines.append(
                    f"    {src_id} -. cycle .-> {tgt_id}"
                )
 
        return "\n".join(lines)
 
    def generate_c4_context(
        self,
        system_name: str,
        services: list[dict],
        external_systems: list[dict],
    ) -> str:
        """C4 모델 컨텍스트 다이어그램 생성"""
        lines = ["graph TD"]
 
        # 사용자
        lines.append("    User((사용자))")
 
        # 내부 서비스
        lines.append(f"    subgraph {system_name}")
        for service in services:
            sid = service["name"].replace(" ", "_")
            lines.append(f"        {sid}[{service['name']}]")
        lines.append("    end")
 
        # 외부 시스템
        for ext in external_systems:
            eid = ext["name"].replace(" ", "_")
            lines.append(f"    {eid}[/{ext['name']}/]")
 
        # 관계
        lines.append("    User --> " + services[0]["name"].replace(" ", "_"))
        for service in services:
            sid = service["name"].replace(" ", "_")
            for dep in service.get("depends_on", []):
                did = dep.replace(" ", "_")
                lines.append(f"    {sid} --> {did}")
 
        return "\n".join(lines)
 
    def generate_sequence_diagram(
        self,
        flow_name: str,
        steps: list[dict],
    ) -> str:
        """시퀀스 다이어그램 생성"""
        lines = ["sequenceDiagram"]
        lines.append(f"    Note over Client: {flow_name}")
 
        for step in steps:
            arrow = "->>" if step.get("async") else "->"
            lines.append(
                f"    {step['from']}{arrow}{step['to']}: "
                f"{step['action']}"
            )
            if step.get("response"):
                lines.append(
                    f"    {step['to']}-->>{step['from']}: "
                    f"{step['response']}"
                )
 
        return "\n".join(lines)

ARCHITECTURE_REVIEW_PROMPT = """아키텍트로서 다음 시스템 분석 결과를 리뷰하세요.
 
의존성 매트릭스:
{dsm_summary}
 
레이어 위반: {violations}건
순환 의존성: {cycles}건
 
핫스팟 (상위 5개):
{hotspots}
 
다음 관점에서 리뷰하세요:
1. 전체 아키텍처 건강도 평가 (1-10)
2. 가장 심각한 구조적 문제 3가지
3. 각 문제의 비즈니스 영향
4. 단기/중기/장기 개선 로드맵
5. 마이크로서비스 분해 시기 판단"""