【系统架构设计百科】技术债务：量化、可视化与偿还策略

# 技术债利息计算示例
principal = 3  # 本金：节省的天数
interest_rate = 1.5 / 5  # 利息率：额外时间 / 正常修改时间
modifications = 28  # 一年内修改次数

# 简单利息
simple_interest = principal + (interest_rate * principal * modifications)
print(f"简单利息模型：{simple_interest:.1f} 天")  # 28.2 天

# 复合利息（更接近真实情况，因为债务会累积）
compound_cost = principal * (1 + interest_rate) ** modifications
print(f"复合利息模型：{compound_cost:.1f} 天")  # 远超简单利息

# CodeScene 代码健康度评分维度
code_health_dimensions:
  - name: "函数长度"
    description: "超过 30 行的函数开始扣分"
    weight: 0.20
    thresholds:
      healthy: "<= 15 行"
      warning: "16-30 行"
      alert: "> 30 行"

  - name: "嵌套深度"
    description: "超过 3 层嵌套开始扣分"
    weight: 0.15
    thresholds:
      healthy: "<= 2 层"
      warning: "3 层"
      alert: "> 3 层"

  - name: "函数参数数量"
    description: "超过 4 个参数开始扣分"
    weight: 0.10
    thresholds:
      healthy: "<= 3 个"
      warning: "4 个"
      alert: "> 4 个"

  - name: "代码重复"
    description: "重复代码块检测"
    weight: 0.15
    thresholds:
      healthy: "无重复"
      warning: "1-2 处重复"
      alert: "> 2 处重复"

  - name: "模块耦合度"
    description: "文件间的依赖关系复杂度"
    weight: 0.20
    thresholds:
      healthy: "<= 5 个依赖"
      warning: "6-10 个依赖"
      alert: "> 10 个依赖"

  - name: "认知复杂度"
    description: "基于 SonarSource 认知复杂度算法"
    weight: 0.20
    thresholds:
      healthy: "<= 10"
      warning: "11-20"
      alert: "> 20"

# 使用 git log 分析时间耦合（CodeScene 内部原理的简化版）
# 找出总是一起修改的文件对

git log --pretty=format:"%H" --since="2025-01-01" | while read commit; do
    git diff-tree --no-commit-id --name-only -r "$commit" 2>/dev/null
    echo "---"
done | python3 -c "
import sys
from collections import defaultdict
from itertools import combinations

commits = []
current = []
for line in sys.stdin:
    line = line.strip()
    if line == '---':
        if current:
            commits.append(current)
        current = []
    elif line:
        current.append(line)

coupling = defaultdict(int)
file_changes = defaultdict(int)

for files in commits:
    for f in files:
        file_changes[f] += 1
    for a, b in combinations(sorted(files), 2):
        coupling[(a, b)] += 1

print('文件A | 文件B | 共同修改次数 | 耦合度')
print('-' * 60)
for (a, b), count in sorted(coupling.items(), key=lambda x: -x[1])[:20]:
    total = min(file_changes[a], file_changes[b])
    degree = count / total if total > 0 else 0
    if degree > 0.5 and count >= 3:
        print(f'{a} | {b} | {count} | {degree:.2f}')
"

// 规则：S1192 - 字符串字面量不应重复
// 修复时间：2 分钟/处
// 技术债类型：可维护性

// 违反示例（产生技术债）
public class OrderService {
    public void createOrder(Order order) {
        if (order.getStatus().equals("PENDING")) {  // 字符串重复
            log.info("Order status: PENDING");       // 字符串重复
            notifyService.send("PENDING");           // 字符串重复
        }
    }

    public void cancelOrder(Order order) {
        if (order.getStatus().equals("PENDING")) {  // 第 4 次重复
            order.setStatus("CANCELLED");
        }
    }
}

// 修复后（偿还技术债）
public class OrderService {
    private static final String STATUS_PENDING = "PENDING";
    private static final String STATUS_CANCELLED = "CANCELLED";

    public void createOrder(Order order) {
        if (order.getStatus().equals(STATUS_PENDING)) {
            log.info("Order status: " + STATUS_PENDING);
            notifyService.send(STATUS_PENDING);
        }
    }

    public void cancelOrder(Order order) {
        if (order.getStatus().equals(STATUS_PENDING)) {
            order.setStatus(STATUS_CANCELLED);
        }
    }
}

// 规则：S3776 - 认知复杂度不应过高
// 修复时间：基础 5 分钟 + 每超出 1 点增加 1 分钟
// 阈值：默认 15

// 违反示例：认知复杂度 = 23
public String processPayment(Payment payment) {       // +0
    if (payment == null) {                             // +1
        return "INVALID";
    }
    if (payment.getAmount() <= 0) {                    // +1
        return "INVALID_AMOUNT";
    }
    for (PaymentRule rule : rules) {                   // +1
        if (rule.isApplicable(payment)) {              // +2（嵌套）
            if (rule.getType().equals("BLOCK")) {      // +3（嵌套）
                return "BLOCKED";
            } else if (rule.getType().equals("WARN")) {// +3（嵌套+else if）
                log.warn("Warning rule triggered");
                if (payment.isRetry()) {               // +4（嵌套）
                    continue;                          // +1
                }
            }
        }
    }
    try {                                              // +0
        gateway.charge(payment);
    } catch (GatewayException e) {                     // +1
        if (e.isRetryable()) {                         // +2（嵌套）
            return processPayment(payment.retry());    // +1（递归）
        }
        throw e;
    }
    return "SUCCESS";
}

{
  "qualityGate": {
    "name": "TechDebt-Strict",
    "conditions": [
      {
        "metric": "new_technical_debt_ratio",
        "operator": "GREATER_THAN",
        "value": "5",
        "period": "1",
        "description": "新代码的技术债比率不超过 5%"
      },
      {
        "metric": "new_coverage",
        "operator": "LESS_THAN",
        "value": "80",
        "period": "1",
        "description": "新代码的测试覆盖率不低于 80%"
      },
      {
        "metric": "new_duplicated_lines_density",
        "operator": "GREATER_THAN",
        "value": "3",
        "period": "1",
        "description": "新代码的重复率不超过 3%"
      },
      {
        "metric": "new_bugs",
        "operator": "GREATER_THAN",
        "value": "0",
        "period": "1",
        "description": "新代码不允许引入 Bug"
      },
      {
        "metric": "new_vulnerabilities",
        "operator": "GREATER_THAN",
        "value": "0",
        "period": "1",
        "description": "新代码不允许引入安全漏洞"
      }
    ]
  }
}

# GitHub Actions 集成 SonarQube 示例
name: SonarQube Analysis
on:
  pull_request:
    branches: [main, develop]
  push:
    branches: [main]

jobs:
  sonarqube:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0  # 需要完整历史记录来计算新代码

      - name: Set up JDK 17
        uses: actions/setup-java@v4
        with:
          java-version: '17'
          distribution: 'temurin'

      - name: Cache SonarQube packages
        uses: actions/cache@v4
        with:
          path: ~/.sonar/cache
          key: ${{ runner.os }}-sonar

      - name: Build and analyze
        env:
          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
          SONAR_HOST_URL: ${{ secrets.SONAR_HOST_URL }}
        run: |
          mvn clean verify sonar:sonar \
            -Dsonar.projectKey=order-service \
            -Dsonar.qualitygate.wait=true \
            -Dsonar.newCode.referenceBranch=main

{
  "dashboard": {
    "title": "技术债务监控仪表板",
    "panels": [
      {
        "title": "技术债总量趋势",
        "type": "timeseries",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "sonarqube_technical_debt_minutes{project='order-service'}",
            "legendFormat": "技术债（分钟）"
          }
        ]
      },
      {
        "title": "各模块技术债分布",
        "type": "piechart",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "sonarqube_technical_debt_minutes{project='order-service'} by (module)",
            "legendFormat": "{{module}}"
          }
        ]
      },
      {
        "title": "技术债比率（Quality Gate）",
        "type": "gauge",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "sonarqube_technical_debt_ratio{project='order-service'}",
            "legendFormat": "TDR"
          }
        ],
        "thresholds": [
          { "value": 5, "color": "green" },
          { "value": 10, "color": "yellow" },
          { "value": 20, "color": "red" }
        ]
      },
      {
        "title": "热点文件 Top 10",
        "type": "table",
        "datasource": "PostgreSQL",
        "targets": [
          {
            "rawSql": "SELECT file_path, complexity, change_frequency, debt_minutes FROM code_metrics ORDER BY complexity * change_frequency DESC LIMIT 10"
          }
        ]
      },
      {
        "title": "每周新增 vs 偿还",
        "type": "barchart",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "increase(sonarqube_new_debt_minutes[7d])",
            "legendFormat": "新增技术债"
          },
          {
            "expr": "increase(sonarqube_resolved_debt_minutes[7d])",
            "legendFormat": "偿还技术债"
          }
        ]
      }
    ]
  }
}

"""SonarQube Metrics Exporter for Prometheus"""
import time
import requests
from prometheus_client import start_http_server, Gauge

SONAR_URL = "https://sonar.example.com"
SONAR_TOKEN = "squ_xxxxxxxxxxxx"
PROJECT_KEY = "order-service"

# 定义 Prometheus 指标
tech_debt_minutes = Gauge(
    'sonarqube_technical_debt_minutes',
    'Technical debt in minutes',
    ['project', 'module']
)

tech_debt_ratio = Gauge(
    'sonarqube_technical_debt_ratio',
    'Technical debt ratio percentage',
    ['project']
)

code_smells = Gauge(
    'sonarqube_code_smells',
    'Number of code smells',
    ['project', 'severity']
)

coverage = Gauge(
    'sonarqube_coverage',
    'Code coverage percentage',
    ['project']
)


def fetch_metrics():
    """从 SonarQube API 拉取指标"""
    metrics = "sqale_index,sqale_debt_ratio,code_smells,coverage"
    url = f"{SONAR_URL}/api/measures/component"
    params = {
        "component": PROJECT_KEY,
        "metricKeys": metrics
    }
    response = requests.get(url, params=params, auth=(SONAR_TOKEN, ""))
    data = response.json()

    for measure in data["component"]["measures"]:
        metric = measure["metric"]
        value = float(measure["value"])

        if metric == "sqale_index":
            tech_debt_minutes.labels(project=PROJECT_KEY, module="total").set(value)
        elif metric == "sqale_debt_ratio":
            tech_debt_ratio.labels(project=PROJECT_KEY).set(value)
        elif metric == "coverage":
            coverage.labels(project=PROJECT_KEY).set(value)


def fetch_code_smells_by_severity():
    """按严重程度拉取代码异味数量"""
    for severity in ["BLOCKER", "CRITICAL", "MAJOR", "MINOR", "INFO"]:
        url = f"{SONAR_URL}/api/issues/search"
        params = {
            "componentKeys": PROJECT_KEY,
            "types": "CODE_SMELL",
            "severities": severity,
            "ps": 1
        }
        response = requests.get(url, params=params, auth=(SONAR_TOKEN, ""))
        total = response.json()["total"]
        code_smells.labels(project=PROJECT_KEY, severity=severity).set(total)


if __name__ == "__main__":
    start_http_server(9090)
    print("SonarQube Exporter started on :9090")
    while True:
        fetch_metrics()
        fetch_code_smells_by_severity()
        time.sleep(300)  # 每 5 分钟刷新一次

// 旧 API（待废弃）
import com.google.common.logging.GoogleLogger;

public class UserService {
    private static final GoogleLogger logger = GoogleLogger.forEnclosingClass();

    public void createUser(User user) {
        logger.info("Creating user: %s", user.getName());
        // ...
        logger.warning("User creation took %d ms", elapsed);
    }
}

// 新 API（迁移目标）
import com.google.flogger.FluentLogger;

public class UserService {
    private static final FluentLogger logger = FluentLogger.forEnclosingClass();

    public void createUser(User user) {
        logger.atInfo().log("Creating user: %s", user.getName());
        // ...
        logger.atWarning().log("User creation took %d ms", elapsed);
    }
}

"""简化的代码迁移变换规则（模拟 Google 的 Rosie 工作流）"""
import ast
import re
from pathlib import Path
from dataclasses import dataclass


@dataclass
class MigrationRule:
    """代码迁移规则定义"""
    name: str
    old_pattern: str
    new_pattern: str
    description: str


# 定义迁移规则
RULES = [
    MigrationRule(
        name="logger_info",
        old_pattern=r'logger\.info\("([^"]*)"(.*?)\)',
        new_pattern=r'logger.atInfo().log("\1"\2)',
        description="迁移 info 级别日志调用"
    ),
    MigrationRule(
        name="logger_warning",
        old_pattern=r'logger\.warning\("([^"]*)"(.*?)\)',
        new_pattern=r'logger.atWarning().log("\1"\2)',
        description="迁移 warning 级别日志调用"
    ),
    MigrationRule(
        name="logger_severe",
        old_pattern=r'logger\.severe\("([^"]*)"(.*?)\)',
        new_pattern=r'logger.atSevere().log("\1"\2)',
        description="迁移 severe 级别日志调用"
    ),
    MigrationRule(
        name="import_statement",
        old_pattern=r'import com\.google\.common\.logging\.GoogleLogger;',
        new_pattern='import com.google.flogger.FluentLogger;',
        description="迁移 import 语句"
    ),
    MigrationRule(
        name="logger_declaration",
        old_pattern=r'GoogleLogger\.forEnclosingClass\(\)',
        new_pattern='FluentLogger.forEnclosingClass()',
        description="迁移 Logger 声明"
    ),
    MigrationRule(
        name="logger_type",
        old_pattern=r'GoogleLogger',
        new_pattern='FluentLogger',
        description="迁移 Logger 类型引用"
    ),
]


def apply_migration(file_path: Path, rules: list[MigrationRule]) -> dict:
    """对单个文件应用迁移规则"""
    content = file_path.read_text()
    original = content
    applied_rules = []

    for rule in rules:
        new_content = re.sub(rule.old_pattern, rule.new_pattern, content)
        if new_content != content:
            applied_rules.append(rule.name)
            content = new_content

    if content != original:
        return {
            "file": str(file_path),
            "rules_applied": applied_rules,
            "original": original,
            "migrated": content,
            "changed": True
        }
    return {"file": str(file_path), "changed": False}


def split_into_changelists(results: list[dict], owners_map: dict) -> list[dict]:
    """将变更按 OWNERS 分组为独立的 CL"""
    cl_groups = {}

    for result in results:
        if not result["changed"]:
            continue
        file_path = result["file"]
        owner = find_owner(file_path, owners_map)

        if owner not in cl_groups:
            cl_groups[owner] = {
                "owner": owner,
                "files": [],
                "description": "自动化代码迁移：GoogleLogger → FluentLogger"
            }
        cl_groups[owner]["files"].append(result)

    return list(cl_groups.values())


def find_owner(file_path: str, owners_map: dict) -> str:
    """根据 OWNERS 文件查找代码所有者"""
    path = Path(file_path)
    for parent in path.parents:
        if str(parent) in owners_map:
            return owners_map[str(parent)]
    return "unowned"

// 童子军法则检查清单 —— 代码审查时使用

interface BoyScoutChecklist {
  // 必做项：接触到的代码必须改善
  required: {
    renameUnclearVariables: boolean;      // 重命名含义不清的变量
    extractMagicNumbers: boolean;          // 提取魔法数字为常量
    addMissingTypeAnnotations: boolean;    // 补充缺失的类型注解
    removeDeadCode: boolean;               // 删除死代码
  };

  // 建议项：如果时间允许
  recommended: {
    extractLongMethods: boolean;           // 拆分过长的方法
    reduceParameterCount: boolean;         // 减少参数数量
    addUnitTests: boolean;                 // 补充单元测试
    updateOutdatedComments: boolean;       // 更新过时的注释
  };

  // 禁止项：不在本次提交中做
  prohibited: {
    largeScaleRefactoring: boolean;        // 大规模重构
    architectureChanges: boolean;          // 架构变更
    dependencyUpgrades: boolean;           // 依赖升级
  };
}

"""技术债预算管理器"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum


class DebtSeverity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


@dataclass
class DebtItem:
    """单条技术债记录"""
    id: str
    title: str
    severity: DebtSeverity
    estimated_hours: float
    interest_rate: float  # 每月产生的额外成本（小时）
    module: str
    created_at: datetime = field(default_factory=datetime.now)
    resolved_at: datetime | None = None


@dataclass
class DebtBudget:
    """技术债预算"""
    max_total_hours: float        # 最大允许的技术债总量（小时）
    max_critical_items: int       # 最大允许的关键级别债务数
    monthly_paydown_target: float # 每月偿还目标（小时）
    items: list[DebtItem] = field(default_factory=list)

    @property
    def total_debt_hours(self) -> float:
        """当前技术债总量"""
        return sum(
            item.estimated_hours
            for item in self.items
            if item.resolved_at is None
        )

    @property
    def monthly_interest(self) -> float:
        """每月利息总量"""
        return sum(
            item.interest_rate
            for item in self.items
            if item.resolved_at is None
        )

    @property
    def critical_count(self) -> int:
        """关键级别债务数量"""
        return sum(
            1 for item in self.items
            if item.resolved_at is None
            and item.severity == DebtSeverity.CRITICAL
        )

    def is_over_budget(self) -> bool:
        """是否超出预算"""
        return (
            self.total_debt_hours > self.max_total_hours
            or self.critical_count > self.max_critical_items
        )

    def get_paydown_priority(self) -> list[DebtItem]:
        """获取偿还优先级列表（按 ROI 排序）"""
        active_items = [
            item for item in self.items
            if item.resolved_at is None
        ]
        # ROI = 利息率 / 修复成本
        return sorted(
            active_items,
            key=lambda item: item.interest_rate / item.estimated_hours,
            reverse=True
        )

    def generate_report(self) -> str:
        """生成预算报告"""
        status = "超出预算" if self.is_over_budget() else "预算内"
        utilization = self.total_debt_hours / self.max_total_hours * 100

        report = f"""
技术债预算报告
==============
状态：{status}
预算利用率：{utilization:.1f}%
技术债总量：{self.total_debt_hours:.0f} / {self.max_total_hours:.0f} 小时
关键级别：{self.critical_count} / {self.max_critical_items}
每月利息：{self.monthly_interest:.1f} 小时
每月偿还目标：{self.monthly_paydown_target:.0f} 小时

优先偿还项：
"""
        for i, item in enumerate(self.get_paydown_priority()[:5], 1):
            roi = item.interest_rate / item.estimated_hours
            report += (
                f"  {i}. [{item.severity.value}] {item.title}\n"
                f"     模块：{item.module}\n"
                f"     修复成本：{item.estimated_hours}h，"
                f"月利息：{item.interest_rate}h，"
                f"ROI：{roi:.2f}\n"
            )
        return report

// 使用 ArchUnit 定义架构适应度函数
import com.tngtech.archunit.core.domain.JavaClasses;
import com.tngtech.archunit.core.importer.ClassFileImporter;
import com.tngtech.archunit.lang.ArchRule;
import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.*;
import static com.tngtech.archunit.library.Architectures.layeredArchitecture;

public class ArchitectureFitnessTest {

    private final JavaClasses classes = new ClassFileImporter()
        .importPackages("com.example.orderservice");

    // 分层架构约束：控制器层不能直接访问数据访问层
    @Test
    void layerDependenciesShouldBeRespected() {
        layeredArchitecture()
            .consideringAllDependencies()
            .layer("Controller").definedBy("..controller..")
            .layer("Service").definedBy("..service..")
            .layer("Repository").definedBy("..repository..")
            .layer("Domain").definedBy("..domain..")
            .whereLayer("Controller").mayOnlyAccessLayers("Service", "Domain")
            .whereLayer("Service").mayOnlyAccessLayers("Repository", "Domain")
            .whereLayer("Repository").mayOnlyAccessLayers("Domain")
            .check(classes);
    }

    // 依赖方向约束：domain 包不应依赖任何基础设施包
    @Test
    void domainShouldNotDependOnInfrastructure() {
        noClasses()
            .that().resideInAPackage("..domain..")
            .should().dependOnClassesThat()
            .resideInAnyPackage(
                "..controller..",
                "..repository..",
                "..config..",
                "..infrastructure.."
            )
            .check(classes);
    }

    // 命名约束：Repository 类必须以 Repository 结尾
    @Test
    void repositoriesShouldFollowNamingConvention() {
        classes()
            .that().resideInAPackage("..repository..")
            .and().areAnnotatedWith(Repository.class)
            .should().haveSimpleNameEndingWith("Repository")
            .check(classes);
    }

    // 循环依赖检测
    @Test
    void noCircularDependencies() {
        slices()
            .matching("com.example.orderservice.(*)..")
            .should().beFreeOfCycles()
            .check(classes);
    }
}

# 设计评审技术债检查清单
design_review_checklist:
  architecture_impact:
    - question: "本次变更是否引入了新的模块间依赖？"
      red_flag: "引入了循环依赖或跨层调用"
      action: "重新设计接口，使用事件驱动解耦"

    - question: "是否复用了已有的抽象，还是创建了重复的实现？"
      red_flag: "与现有模块功能重叠超过 50%"
      action: "提取公共抽象或合并模块"

    - question: "是否有明确的废弃计划？"
      red_flag: "引入了临时方案但没有设定清理时间"
      action: "创建技术债票据，设定截止日期"

  code_quality:
    - question: "是否有充分的测试覆盖？"
      red_flag: "核心逻辑没有单元测试"
      action: "测试覆盖率必须达到 80% 以上"

    - question: "是否遵循了现有的编码规范？"
      red_flag: "引入了与项目风格不一致的模式"
      action: "按照项目规范调整"

  operational:
    - question: "是否添加了必要的监控指标？"
      red_flag: "关键路径没有度量"
      action: "添加延迟、错误率、吞吐量指标"

    - question: "是否有降级和熔断策略？"
      red_flag: "外部依赖没有容错机制"
      action: "实现降级方案"

## 技术债票据模板

**标题**：[模块名] 简要描述
**严重程度**：Critical / High / Medium / Low
**创建日期**：YYYY-MM-DD
**截止日期**：YYYY-MM-DD（必须设定）
**创建人**：@engineer
**所属模块**：module-name

### 背景
为什么产生这笔技术债？当时的约束条件是什么？

### 影响范围
哪些模块、功能或团队会受到影响？

### 利息估算
- 每次涉及该区域的修改额外耗时：X 小时
- 预计未来 6 个月的修改频率：Y 次/月
- 累计利息估算：X × Y × 6 = Z 小时

### 偿还方案
描述如何修复这笔技术债。

### 偿还成本
估计需要多少人天来修复。

### 验收标准
如何验证技术债已被偿还？

"""代码健康度指标采集与聚合"""
from dataclasses import dataclass
from typing import Optional


@dataclass
class CodeHealthMetrics:
    """代码健康度指标集"""
    # 可维护性指标
    cyclomatic_complexity: float        # 平均圈复杂度
    cognitive_complexity: float         # 平均认知复杂度
    avg_function_length: float          # 平均函数长度（行）
    duplication_percentage: float       # 代码重复率（%）
    tech_debt_ratio: float              # 技术债比率（%）

    # 可靠性指标
    bug_density: float                  # Bug 密度（Bug/千行代码）
    defect_escape_rate: float           # 缺陷逃逸率（%）
    mttr_hours: float                   # 平均修复时间（小时）

    # 安全性指标
    vulnerability_count: int            # 漏洞数量
    security_hotspot_count: int         # 安全热点数量
    dependency_vulnerabilities: int     # 依赖漏洞数量

    # 可测试性指标
    line_coverage: float                # 行覆盖率（%）
    branch_coverage: float              # 分支覆盖率（%）
    mutation_score: Optional[float]     # 变异测试分数（%）

    # 演进性指标
    change_failure_rate: float          # 变更失败率（%）
    merge_conflict_rate: float          # 合并冲突率（%）
    avg_build_time_seconds: float       # 平均构建时间（秒）

    def calculate_health_score(self) -> float:
        """计算综合健康评分（0-100）"""
        scores = {
            "maintainability": self._maintainability_score(),
            "reliability": self._reliability_score(),
            "security": self._security_score(),
            "testability": self._testability_score(),
            "evolvability": self._evolvability_score(),
        }

        weights = {
            "maintainability": 0.25,
            "reliability": 0.25,
            "security": 0.20,
            "testability": 0.15,
            "evolvability": 0.15,
        }

        total = sum(
            scores[key] * weights[key]
            for key in scores
        )
        return round(total, 1)

    def _maintainability_score(self) -> float:
        """可维护性评分"""
        score = 100.0
        if self.cyclomatic_complexity > 10:
            score -= (self.cyclomatic_complexity - 10) * 3
        if self.cognitive_complexity > 15:
            score -= (self.cognitive_complexity - 15) * 2
        if self.duplication_percentage > 3:
            score -= (self.duplication_percentage - 3) * 5
        if self.tech_debt_ratio > 5:
            score -= (self.tech_debt_ratio - 5) * 2
        return max(0, min(100, score))

    def _reliability_score(self) -> float:
        """可靠性评分"""
        score = 100.0
        score -= self.bug_density * 10
        score -= self.defect_escape_rate * 2
        if self.mttr_hours > 4:
            score -= (self.mttr_hours - 4) * 3
        return max(0, min(100, score))

    def _security_score(self) -> float:
        """安全性评分"""
        score = 100.0
        score -= self.vulnerability_count * 15
        score -= self.security_hotspot_count * 5
        score -= self.dependency_vulnerabilities * 8
        return max(0, min(100, score))

    def _testability_score(self) -> float:
        """可测试性评分"""
        score = 0.0
        score += self.line_coverage * 0.4
        score += self.branch_coverage * 0.4
        if self.mutation_score is not None:
            score += self.mutation_score * 0.2
        else:
            score += self.line_coverage * 0.2
        return max(0, min(100, score))

    def _evolvability_score(self) -> float:
        """演进性评分"""
        score = 100.0
        score -= self.change_failure_rate * 2
        score -= self.merge_conflict_rate * 3
        if self.avg_build_time_seconds > 300:
            score -= (self.avg_build_time_seconds - 300) / 60 * 5
        return max(0, min(100, score))


# 使用示例
metrics = CodeHealthMetrics(
    cyclomatic_complexity=8.5,
    cognitive_complexity=12.3,
    avg_function_length=22.0,
    duplication_percentage=4.2,
    tech_debt_ratio=8.7,
    bug_density=0.3,
    defect_escape_rate=5.0,
    mttr_hours=3.5,
    vulnerability_count=0,
    security_hotspot_count=2,
    dependency_vulnerabilities=1,
    line_coverage=78.5,
    branch_coverage=65.2,
    mutation_score=58.0,
    change_failure_rate=8.0,
    merge_conflict_rate=5.0,
    avg_build_time_seconds=240,
)

health_score = metrics.calculate_health_score()
print(f"综合健康评分：{health_score}/100")

# 技术债分析流水线（GitHub Actions）
name: Tech Debt Analysis Pipeline
on:
  push:
    branches: [main]
  schedule:
    - cron: '0 2 * * 1'  # 每周一凌晨 2 点运行完整分析

jobs:
  static-analysis:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: SonarQube 静态分析
        uses: sonarsource/sonarqube-scan-action@v2
        env:
          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
          SONAR_HOST_URL: ${{ secrets.SONAR_HOST_URL }}

      - name: 依赖漏洞扫描
        uses: aquasecurity/trivy-action@master
        with:
          scan-type: 'fs'
          scan-ref: '.'
          format: 'json'
          output: 'trivy-results.json'

  behavioral-analysis:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: 热点分析（简化版）
        run: |
          echo "=== 变更频率 Top 20 ==="
          git log --since="6 months ago" --name-only --pretty=format: \
            | grep -v '^$' \
            | sort | uniq -c | sort -rn | head -20

          echo ""
          echo "=== 大文件检测（可能的复杂度热点）==="
          find src -name "*.java" -o -name "*.ts" -o -name "*.py" \
            | xargs wc -l 2>/dev/null \
            | sort -rn | head -20

      - name: 时间耦合分析
        run: |
          python3 scripts/temporal_coupling.py \
            --since "6 months ago" \
            --threshold 0.7 \
            --min-commits 5

  report:
    needs: [static-analysis, behavioral-analysis]
    runs-on: ubuntu-latest
    steps:
      - name: 聚合分析结果
        run: |
          python3 scripts/aggregate_debt_report.py \
            --sonar-project order-service \
            --output debt-report.html

      - name: 发送报告
        uses: actions/upload-artifact@v4
        with:
          name: tech-debt-report
          path: debt-report.html

#!/bin/bash
# .git/hooks/pre-commit —— 提交前技术债检查

set -e

echo "正在检查技术债..."

# 检查函数长度（超过 50 行的函数）
LONG_FUNCTIONS=$(git diff --cached --diff-filter=ACM \
    -U0 -- '*.java' '*.ts' '*.py' \
    | grep -c "^+.*function\|^+.*def \|^+.*public.*(" || true)

# 检查 TODO/FIXME/HACK 标记
DEBT_MARKERS=$(git diff --cached --diff-filter=ACM \
    | grep -c "^+.*TODO\|^+.*FIXME\|^+.*HACK\|^+.*XXX" || true)

if [ "$DEBT_MARKERS" -gt 0 ]; then
    echo "警告：发现 $DEBT_MARKERS 处技术债标记（TODO/FIXME/HACK/XXX）"
    echo "请确保已为每处标记创建了对应的技术债票据"
    # 非阻塞，仅警告
fi

# 检查是否引入了新的循环导入（Python）
PYTHON_FILES=$(git diff --cached --name-only --diff-filter=ACM -- '*.py')
if [ -n "$PYTHON_FILES" ]; then
    echo "检查循环导入..."
    python3 -c "
import ast
import sys
from pathlib import Path

files = sys.argv[1:]
for f in files:
    try:
        tree = ast.parse(Path(f).read_text())
        imports = [
            node for node in ast.walk(tree)
            if isinstance(node, (ast.Import, ast.ImportFrom))
        ]
        if len(imports) > 20:
            print(f'警告：{f} 有 {len(imports)} 条导入语句，可能存在过度耦合')
    except Exception:
        pass
" $PYTHON_FILES
fi

echo "技术债检查完成"

"""技术债「痛苦指数」投票系统"""
from collections import defaultdict


class PainIndexSurvey:
    """收集团队对各模块维护痛苦程度的投票"""

    def __init__(self, modules: list[str]):
        self.modules = modules
        self.votes: dict[str, list[int]] = defaultdict(list)

    def submit_vote(self, engineer: str, module: str, pain_score: int):
        """
        提交投票
        pain_score: 1-5
          1 = 维护轻松
          2 = 偶尔不便
          3 = 经常困扰
          4 = 严重阻碍
          5 = 痛不欲生
        """
        if module not in self.modules:
            raise ValueError(f"未知模块：{module}")
        if not 1 <= pain_score <= 5:
            raise ValueError("评分范围：1-5")
        self.votes[module].append(pain_score)

    def get_results(self) -> list[dict]:
        """获取投票结果，按痛苦指数排序"""
        results = []
        for module in self.modules:
            scores = self.votes.get(module, [])
            if scores:
                avg = sum(scores) / len(scores)
                results.append({
                    "module": module,
                    "avg_pain": round(avg, 1),
                    "votes": len(scores),
                    "max_pain": max(scores),
                    "urgency": "紧急" if avg >= 4 else "关注" if avg >= 3 else "正常"
                })
        return sorted(results, key=lambda x: x["avg_pain"], reverse=True)