#!/usr/bin/env python3
"""
レガシーコードの依存関係を可視化するスクリプト。
各モジュールの依存数・被依存数を分析し、
リファクタリングの優先度を判断する。
"""
import ast
import sys
from pathlib import Path
from dataclasses import dataclass, field
from collections import defaultdict
 
 
@dataclass
class ModuleDependency:
    """モジュールの依存関係情報"""
    module: str
    imports: list[str] = field(default_factory=list)
    imported_by: list[str] = field(default_factory=list)
 
    @property
    def afferent_coupling(self) -> int:
        """求心性結合度: このモジュールに依存しているモジュール数"""
        return len(self.imported_by)
 
    @property
    def efferent_coupling(self) -> int:
        """遠心性結合度: このモジュールが依存しているモジュール数"""
        return len(self.imports)
 
    @property
    def instability(self) -> float:
        """不安定度: 0.0 (安定) ~ 1.0 (不安定)"""
        total = self.afferent_coupling + self.efferent_coupling
        if total == 0:
            return 0.0
        return self.efferent_coupling / total
 
 
def analyze_dependencies(src_path: str) -> dict[str, ModuleDependency]:
    """ソースコードの依存関係を分析"""
    modules: dict[str, ModuleDependency] = {}
 
    for py_file in Path(src_path).rglob("*.py"):
        module_name = str(py_file.relative_to(src_path)).replace("/", ".").rstrip(".py")
        tree = ast.parse(py_file.read_text())
 
        imports = []
        for node in ast.walk(tree):
            if 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)
 
        modules[module_name] = ModuleDependency(
            module=module_name, imports=imports
        )
 
    # 被依存関係を逆引き
    for mod_name, mod_dep in modules.items():
        for imp in mod_dep.imports:
            if imp in modules:
                modules[imp].imported_by.append(mod_name)
 
    return modules
 
 
def print_dependency_report(modules: dict[str, ModuleDependency]) -> None:
    """依存関係レポートを出力"""
    print("=" * 70)
    print("  依存関係分析レポート")
    print("=" * 70)
    print(f"{'モジュール':<30} {'依存数':>6} {'被依存':>6} {'不安定度':>8}")
    print("-" * 70)
 
    sorted_modules = sorted(
        modules.values(),
        key=lambda m: m.efferent_coupling,
        reverse=True
    )
    for mod in sorted_modules[:20]:
        stability = "安定" if mod.instability < 0.3 else (
            "中間" if mod.instability < 0.7 else "不安定"
        )
        print(f"{mod.module:<30} {mod.efferent_coupling:>6} "
              f"{mod.afferent_coupling:>6} {mod.instability:>7.2f} ({stability})")
 
    # 循環依存の検出
    print("\n--- 循環依存の検出 ---")
    for mod_name, mod_dep in modules.items():
        for imp in mod_dep.imports:
            if imp in modules and mod_name in modules[imp].imports:
                print(f"  ⚠ {mod_name} <-> {imp}")

# ────────────────────────────────────────
# Seam がない状態: テスト不可能
# ────────────────────────────────────────
class OrderProcessor:
    """依存を直接生成 → テスト時にDBやメールサーバが必要"""
 
    def process(self, order):
        # 静的メソッド → テスト時に差し替え不可
        db = DatabaseHelper.get_connection()
        result = db.execute("SELECT stock FROM products ...", order.product_id)
 
        if result.stock < order.quantity:
            raise InsufficientStockError()
 
        # 直接生成 → テスト時に差し替え不可
        inventory = InventoryChecker()
        inventory.reserve(order.product_id, order.quantity)
 
        # 静的メソッド → テスト時にメール送信が実行される
        EmailSender.send(
            to=order.customer_email,
            subject="注文確定",
            body=f"注文 {order.id} が確定しました"
        )
 
 
# ────────────────────────────────────────
# Object Seam: コンストラクタインジェクション
# ────────────────────────────────────────
class OrderProcessor:
    """依存を注入 → テスト時にテストダブルに差し替え可能"""
 
    def __init__(self, db_connection, inventory_checker, email_sender):
        self._db = db_connection              # 注入 → Stub に差し替え可能
        self._inventory = inventory_checker   # 注入 → Mock に差し替え可能
        self._email = email_sender            # 注入 → Fake に差し替え可能
 
    def process(self, order):
        result = self._db.execute("SELECT stock FROM products ...", order.product_id)
 
        if result.stock < order.quantity:
            raise InsufficientStockError()
 
        self._inventory.reserve(order.product_id, order.quantity)
        self._email.send(
            to=order.customer_email,
            subject="注文確定",
            body=f"注文 {order.id} が確定しました"
        )
 
 
# テスト: テストダブルを注入
def test_order_process_sends_email():
    fake_db = FakeDatabase(stock=10)
    mock_inventory = Mock()
    mock_email = Mock()
 
    processor = OrderProcessor(fake_db, mock_inventory, mock_email)
    processor.process(Order(product_id="P001", quantity=2, customer_email="a@b.com"))
 
    mock_email.send.assert_called_once()
    mock_inventory.reserve.assert_called_once_with("P001", 2)

# ────────────────────────────────────────
# Step 1: テスト困難な部分をメソッドに抽出
# ────────────────────────────────────────
class OrderProcessor:
    def process(self, order):
        price = self._calculate_price(order)
        self._save_to_database(order, price)    # 抽出したメソッド
        self._send_notification(order)          # 抽出したメソッド
        return price
 
    def _calculate_price(self, order):
        """純粋な計算ロジック ── テスト可能"""
        base_price = order.unit_price * order.quantity
        if order.quantity >= 10:
            return int(base_price * 0.9)  # 10個以上で10%割引
        return base_price
 
    def _save_to_database(self, order, price):
        """DB保存 ── テスト困難な外部依存"""
        db = DatabaseHelper.get_connection()
        db.execute("INSERT INTO orders VALUES (%s, %s)", order.id, price)
 
    def _send_notification(self, order):
        """メール送信 ── テスト困難な外部依存"""
        EmailSender.send(order.customer_email, "注文確定",
                         f"合計: {order.total}円")
 
 
# ────────────────────────────────────────
# Step 2: テスト用サブクラスでオーバーライド
# ────────────────────────────────────────
class TestableOrderProcessor(OrderProcessor):
    """テスト用: DB とメールを差し替え"""
 
    def __init__(self):
        self.saved_orders: list[tuple] = []
        self.sent_emails: list[str] = []
 
    def _save_to_database(self, order, price):
        """DB を使わず、保存内容を記録"""
        self.saved_orders.append((order.id, price))
 
    def _send_notification(self, order):
        """メール送信せず、送信先を記録"""
        self.sent_emails.append(order.customer_email)
 
 
# ────────────────────────────────────────
# Step 3: テスト
# ────────────────────────────────────────
import pytest
 
class TestOrderProcessor:
    def test_process_calculates_correct_price(self):
        """ビジネスロジック（価格計算）のテスト"""
        processor = TestableOrderProcessor()
        order = Order(id="O001", unit_price=1000, quantity=2,
                      customer_email="test@example.com")
 
        result = processor.process(order)
 
        assert result == 2000
 
    def test_process_applies_bulk_discount(self):
        """10個以上で10%割引のテスト"""
        processor = TestableOrderProcessor()
        order = Order(id="O002", unit_price=1000, quantity=10,
                      customer_email="test@example.com")
 
        result = processor.process(order)
 
        assert result == 9000  # 10000 * 0.9
 
    def test_process_saves_to_database(self):
        """DB保存が呼ばれることを確認"""
        processor = TestableOrderProcessor()
        order = Order(id="O003", unit_price=500, quantity=3,
                      customer_email="test@example.com")
 
        processor.process(order)
 
        assert len(processor.saved_orders) == 1
        assert processor.saved_orders[0] == ("O003", 1500)
 
    def test_process_sends_notification(self):
        """メール送信が呼ばれることを確認"""
        processor = TestableOrderProcessor()
        order = Order(id="O004", unit_price=500, quantity=1,
                      customer_email="customer@example.com")
 
        processor.process(order)
 
        assert processor.sent_emails == ["customer@example.com"]

# ────────────────────────────────────────
# 特性テスト: 現在の振る舞いを「記録」する
# ────────────────────────────────────────
class TestLegacyPriceCalculatorCharacterization:
    """
    特性テスト: LegacyPriceCalculator の現在の振る舞いを記録。
    このテストが通る限り、リファクタリングは安全。
    """
 
    def setup_method(self):
        self.calculator = LegacyPriceCalculator()
 
    def test_single_item_basic_price(self):
        """単品の基本価格"""
        result = self.calculator.calculate(
            items=[{"price": 100, "qty": 1}]
        )
        assert result == 100
 
    def test_multiple_items_basic_price(self):
        """複数品の基本価格（割引なし）"""
        result = self.calculator.calculate(
            items=[{"price": 100, "qty": 5}]
        )
        assert result == 500
 
    def test_bulk_discount_at_10(self):
        """10個以上で割引が適用される"""
        result = self.calculator.calculate(
            items=[{"price": 100, "qty": 10}]
        )
        assert result == 900  # 10%割引? 仕様不明だが記録
 
    def test_empty_items(self):
        """空リストの場合"""
        result = self.calculator.calculate(items=[])
        assert result == 0
 
    def test_zero_price(self):
        """価格0の商品"""
        result = self.calculator.calculate(
            items=[{"price": 0, "qty": 5}]
        )
        assert result == 0
 
    def test_negative_price_boundary(self):
        """負の価格 ── バグの可能性あるが現状の振る舞いを記録"""
        result = self.calculator.calculate(
            items=[{"price": -100, "qty": 1}]
        )
        assert result == -100  # ★ バグかもしれないが、現状の記録が目的
 
    def test_large_quantity(self):
        """大量注文"""
        result = self.calculator.calculate(
            items=[{"price": 100, "qty": 1000}]
        )
        assert result == 80000  # 20%割引? 仕様不明
 
 
# ────────────────────────────────────────
# 特性テストの自動生成（大量ケース）
# ────────────────────────────────────────
import json
import itertools
 
def generate_characterization_tests(output_path: str = "characterization_tests.json"):
    """
    特性テストを自動生成して JSON に保存。
    レガシーコードの振る舞いを網羅的に記録する。
    """
    calculator = LegacyPriceCalculator()
    test_cases = []
 
    # 境界値を含む入力の組み合わせを生成
    prices = [0, 1, 50, 100, 500, 999, 1000, 5000, 9999, 10000]
    quantities = [0, 1, 5, 9, 10, 11, 50, 100, 500, 1000]
 
    for price, qty in itertools.product(prices, quantities):
        try:
            result = calculator.calculate(
                items=[{"price": price, "qty": qty}]
            )
            test_cases.append({
                "input": {"price": price, "qty": qty},
                "expected": result,
                "error": None
            })
        except Exception as e:
            test_cases.append({
                "input": {"price": price, "qty": qty},
                "expected": None,
                "error": str(e)
            })
 
    with open(output_path, "w") as f:
        json.dump(test_cases, f, indent=2, ensure_ascii=False)
 
    print(f"生成されたテストケース: {len(test_cases)}件")
    print(f"保存先: {output_path}")
    return test_cases
 
 
# ────────────────────────────────────────
# 保存したケースを使ったパラメタライズテスト
# ────────────────────────────────────────
def load_test_cases(path: str) -> list[dict]:
    with open(path) as f:
        return json.load(f)
 
 
@pytest.mark.parametrize("case", load_test_cases("characterization_tests.json"))
def test_refactored_matches_legacy(case):
    """リファクタリング後のコードがレガシーと同じ結果を返すことを確認"""
    calculator = RefactoredPriceCalculator()  # 新しい実装
 
    if case["error"]:
        with pytest.raises(Exception):
            calculator.calculate(items=[case["input"]])
    else:
        result = calculator.calculate(items=[case["input"]])
        assert result == case["expected"], (
            f"入力: {case['input']}, "
            f"期待: {case['expected']}, 実際: {result}"
        )

import hashlib
from pathlib import Path
 
class GoldenMasterTest:
    """
    Golden Master テスト:
    レガシーシステムの出力をスナップショットとして保存し、
    リファクタリング後も同じ出力が得られることを検証する。
    """
    GOLDEN_DIR = Path("tests/golden_masters")
 
    def __init__(self, system_under_test):
        self.sut = system_under_test
        self.GOLDEN_DIR.mkdir(parents=True, exist_ok=True)
 
    def capture(self, test_name: str, inputs: list[dict]) -> None:
        """ゴールデンマスターを生成・保存"""
        outputs = []
        for inp in inputs:
            try:
                result = self.sut.process(**inp)
                outputs.append({"input": inp, "output": result, "error": None})
            except Exception as e:
                outputs.append({"input": inp, "output": None, "error": str(e)})
 
        golden_path = self.GOLDEN_DIR / f"{test_name}.json"
        with open(golden_path, "w") as f:
            json.dump(outputs, f, indent=2, ensure_ascii=False, default=str)
 
        # チェックサムも保存
        checksum = hashlib.sha256(
            json.dumps(outputs, sort_keys=True).encode()
        ).hexdigest()
        (self.GOLDEN_DIR / f"{test_name}.sha256").write_text(checksum)
 
        print(f"ゴールデンマスター保存: {golden_path} ({len(outputs)} ケース)")
 
    def verify(self, test_name: str) -> bool:
        """現在の出力がゴールデンマスターと一致するか検証"""
        golden_path = self.GOLDEN_DIR / f"{test_name}.json"
        with open(golden_path) as f:
            golden = json.load(f)
 
        mismatches = []
        for case in golden:
            try:
                actual = self.sut.process(**case["input"])
                if actual != case["output"]:
                    mismatches.append({
                        "input": case["input"],
                        "expected": case["output"],
                        "actual": actual,
                    })
            except Exception as e:
                if case["error"] is None or str(e) != case["error"]:
                    mismatches.append({
                        "input": case["input"],
                        "expected_error": case["error"],
                        "actual_error": str(e),
                    })
 
        if mismatches:
            print(f"不一致: {len(mismatches)} 件")
            for m in mismatches[:5]:
                print(f"  {m}")
            return False
 
        print(f"OK: {len(golden)} ケース全て一致")
        return True

# ────────────────────────────────────────
# 状況: 巨大な process() メソッドに「ロイヤルティ割引」を追加したい
# しかし、process() にはテストがなく、構造が複雑で変更が怖い
# ────────────────────────────────────────
 
# BEFORE: 変更したい巨大メソッド (テストなし)
class OrderProcessor:
    def process(self, order):
        # ... 200行の複雑な処理 (理解が難しい) ...
        total = self._legacy_calculate(order)
        # ... さらに100行の複雑な処理 ...
        return total
 
 
# AFTER: 新機能は新しいテスト済みメソッドとして「芽生え」させる
class OrderProcessor:
    def process(self, order):
        # ... 200行の複雑な処理 (変更なし) ...
        total = self._legacy_calculate(order)
 
        # ★ Sprout: 新機能を独立したメソッドとして追加
        discount = self._calculate_loyalty_discount(order, total)
        total = total - discount
 
        # ... さらに100行の複雑な処理 (変更なし) ...
        return total
 
    def _calculate_loyalty_discount(self, order, total: int) -> int:
        """
        新機能: ロイヤルティ割引（テスト付き）
 
        3年以上の顧客は5%割引、5年以上は10%割引。
        ★ この新メソッドにはテストがある → 安全
        """
        years = order.customer.loyalty_years
        if years >= 5:
            return int(total * 0.10)
        elif years >= 3:
            return int(total * 0.05)
        return 0
 
 
# ────────────────────────────────────────
# テスト: 新しい Sprout メソッドのみテスト
# ────────────────────────────────────────
class TestLoyaltyDiscount:
    def setup_method(self):
        self.processor = OrderProcessor()
 
    def test_no_discount_for_new_customer(self):
        order = create_order(loyalty_years=1)
        assert self.processor._calculate_loyalty_discount(order, 10000) == 0
 
    def test_5_percent_for_3_year_customer(self):
        order = create_order(loyalty_years=3)
        assert self.processor._calculate_loyalty_discount(order, 10000) == 500
 
    def test_10_percent_for_5_year_customer(self):
        order = create_order(loyalty_years=5)
        assert self.processor._calculate_loyalty_discount(order, 10000) == 1000
 
    def test_10_percent_for_veteran_customer(self):
        order = create_order(loyalty_years=10)
        assert self.processor._calculate_loyalty_discount(order, 10000) == 1000

# 新しい機能を独立したクラスとして追加
class LoyaltyDiscountCalculator:
    """
    ロイヤルティ割引計算 ── Sprout Class
    完全にテスト済みの独立したクラス。
    """
    TIERS = [
        (10, Decimal("0.15")),  # 10年以上: 15%
        (5, Decimal("0.10")),   # 5年以上: 10%
        (3, Decimal("0.05")),   # 3年以上: 5%
    ]
 
    def calculate(self, customer: Customer, amount: Decimal) -> Decimal:
        """ロイヤルティ割引額を計算"""
        rate = self._get_discount_rate(customer.loyalty_years)
        return (amount * rate).quantize(Decimal("1"))
 
    def _get_discount_rate(self, years: int) -> Decimal:
        """顧客の忠誠年数に基づく割引率"""
        for min_years, rate in self.TIERS:
            if years >= min_years:
                return rate
        return Decimal("0")
 
 
# 既存コードは最小限の変更で統合
class OrderProcessor:
    def __init__(self):
        self._loyalty_calculator = LoyaltyDiscountCalculator()  # Sprout
 
    def process(self, order):
        # ... 既存の複雑な処理 (変更なし) ...
        total = self._legacy_calculate(order)
 
        # Sprout Class の呼び出しを1行追加
        discount = self._loyalty_calculator.calculate(order.customer, total)
        total -= discount
 
        # ... 既存の複雑な処理 (変更なし) ...
        return total

# BEFORE: 複雑なレポート生成ロジック (変更したくない)
class ReportGenerator:
    def generate(self, data):
        # ... 複雑なレポート生成ロジック (200行) ...
        return report
 
 
# AFTER: 既存メソッドをラップ
class ReportGenerator:
    def generate(self, data):
        """ラッパー: 前後処理を追加"""
        self._log_generation_start(data)           # ← 新: ラップ (前処理)
        self._validate_input(data)                 # ← 新: ラップ (前処理)
        report = self._generate_legacy(data)       # ← 旧: リネーム
        self._record_metrics(report)               # ← 新: ラップ (後処理)
        self._log_generation_complete(report)       # ← 新: ラップ (後処理)
        return report
 
    def _generate_legacy(self, data):
        """元の複雑なロジック ── 変更なし"""
        # ... 200行のレガシーコード ...
        return report
 
    def _log_generation_start(self, data):
        """新: レポート生成開始のログ"""
        logger.info(f"レポート生成開始: {data.get('report_type')}")
 
    def _validate_input(self, data):
        """新: 入力データの事前バリデーション"""
        if not data:
            raise ValueError("データが空です")
        if 'report_type' not in data:
            raise ValueError("report_type が指定されていません")
 
    def _record_metrics(self, report):
        """新: メトリクス記録"""
        metrics.increment('reports_generated')
        metrics.histogram('report_size', len(str(report)))
 
    def _log_generation_complete(self, report):
        """新: レポート生成完了のログ"""
        logger.info(f"レポート生成完了: {len(str(report))} 文字")

from enum import Enum
from typing import Protocol
 
 
class FeatureFlag(Enum):
    """Feature Flag の一元管理"""
    NEW_ORDER_CREATION = "new_order_creation"
    NEW_ORDER_RETRIEVAL = "new_order_retrieval"
    NEW_PAYMENT_PROCESSING = "new_payment_processing"
    NEW_NOTIFICATION = "new_notification"
 
 
class FeatureFlagService:
    """Feature Flag サービス ── 環境変数・DB・設定ファイルから取得"""
 
    def __init__(self, config: dict[str, bool]):
        self._config = config
 
    def is_enabled(self, flag: FeatureFlag) -> bool:
        """指定されたフラグが有効かどうかを返す"""
        return self._config.get(flag.value, False)
 
    @classmethod
    def from_env(cls) -> "FeatureFlagService":
        """環境変数からフラグを読み込む"""
        import os
        config = {}
        for flag in FeatureFlag:
            config[flag.value] = os.getenv(
                f"FF_{flag.value.upper()}", "false"
            ).lower() == "true"
        return cls(config)
 
 
class OrderService(Protocol):
    """注文サービスのインターフェース"""
    def create_order(self, order_data: dict) -> Order: ...
    def get_order(self, order_id: str) -> Order: ...
 
 
class OrderFacade:
    """
    Strangler Fig Facade:
    Feature Flag に基づいてレガシーと新システムをルーティング。
    """
 
    def __init__(self, legacy: OrderService, new: OrderService,
                 flags: FeatureFlagService):
        self._legacy = legacy
        self._new = new
        self._flags = flags
 
    def create_order(self, order_data: dict) -> Order:
        if self._flags.is_enabled(FeatureFlag.NEW_ORDER_CREATION):
            return self._new.create_order(order_data)
        return self._legacy.create_order(order_data)
 
    def get_order(self, order_id: str) -> Order:
        if self._flags.is_enabled(FeatureFlag.NEW_ORDER_RETRIEVAL):
            return self._new.get_order(order_id)
        return self._legacy.get_order(order_id)
 
 
# ────────────────────────────────────────
# 使用例: 段階的な移行
# ────────────────────────────────────────
 
# Phase 2: 新機能のみ新システムで処理
flags = FeatureFlagService({
    "new_order_creation": True,     # 新システム
    "new_order_retrieval": False,   # まだレガシー
    "new_payment_processing": False,
    "new_notification": False,
})
 
facade = OrderFacade(
    legacy=LegacyOrderService(),
    new=NewOrderService(),
    flags=flags
)
 
# Phase 3: 既存機能も段階的に移行
flags = FeatureFlagService({
    "new_order_creation": True,     # 移行済み
    "new_order_retrieval": True,    # 移行済み ← 新たに有効化
    "new_payment_processing": False,
    "new_notification": False,
})

# ロールバック可能な Facade
class SafeOrderFacade:
    """
    安全な Strangler Fig Facade:
    新システムでエラーが発生した場合、自動的にレガシーにフォールバック。
    """
 
    def __init__(self, legacy, new, flags, metrics):
        self._legacy = legacy
        self._new = new
        self._flags = flags
        self._metrics = metrics
 
    def create_order(self, order_data: dict) -> Order:
        if self._flags.is_enabled(FeatureFlag.NEW_ORDER_CREATION):
            try:
                result = self._new.create_order(order_data)
                self._metrics.increment("new_system.success")
                return result
            except Exception as e:
                self._metrics.increment("new_system.fallback")
                logger.warning(
                    f"新システムでエラー、レガシーにフォールバック: {e}"
                )
                # 自動フォールバック
                return self._legacy.create_order(order_data)
        return self._legacy.create_order(order_data)

class ParallelVerificationFacade:
    """
    並行実行 Facade:
    新旧両方のシステムで処理を実行し、結果を比較。
    レスポンスはレガシーを返しつつ、不一致をログに記録。
    """
 
    def __init__(self, legacy, new, comparator, metrics):
        self._legacy = legacy
        self._new = new
        self._comparator = comparator
        self._metrics = metrics
 
    def create_order(self, order_data: dict) -> Order:
        # レガシーの結果を取得（これが正式なレスポンス）
        legacy_result = self._legacy.create_order(order_data)
 
        # 新システムの結果を非同期で取得（Shadow Mode）
        try:
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor() as executor:
                future = executor.submit(self._new.create_order, order_data)
                new_result = future.result(timeout=5)
 
            # 結果を比較
            if self._comparator.are_equivalent(legacy_result, new_result):
                self._metrics.increment("parallel.match")
            else:
                self._metrics.increment("parallel.mismatch")
                logger.warning(
                    f"新旧の結果が不一致: "
                    f"legacy={legacy_result}, new={new_result}"
                )
        except Exception as e:
            self._metrics.increment("parallel.new_system_error")
            logger.error(f"新システムでエラー (Shadow Mode): {e}")
 
        # ★ 常にレガシーの結果を返す
        return legacy_result

#!/bin/bash
# レガシーコードの考古学: Git 履歴から改善の優先度を分析
 
echo "=== レガシーコード考古学レポート ==="
 
# 1. 変更頻度が最も高いファイル（過去6ヶ月）
echo ""
echo "--- 変更頻度 Top 20 (過去6ヶ月) ---"
git log --format=format: --name-only --since="6 months ago" \
  | sort | uniq -c | sort -rn | head -20
 
# 2. 最も多くの開発者が触ったファイル（知識が分散）
echo ""
echo "--- 最多開発者ファイル Top 10 ---"
git log --format='%aN' --name-only --since="1 year ago" \
  | awk '/^$/{next} /^[^\/]/{author=$0; next} {print author, $0}' \
  | sort -u | awk '{print $NF}' | sort | uniq -c | sort -rn | head -10
 
# 3. 最近変更されていないが大きなファイル（忘れられたレガシー）
echo ""
echo "--- 大きいが最近変更されていないファイル ---"
find src/ -name "*.py" -exec wc -l {} \; 2>/dev/null \
  | sort -rn | head -10
 
# 4. TODO / FIXME / HACK の分布
echo ""
echo "--- TODO/FIXME/HACK の数 ---"
grep -rn "TODO\|FIXME\|HACK\|XXX" src/ 2>/dev/null | wc -l
 
# 5. バグ修正コミットが多いファイル
echo ""
echo "--- バグ修正が多いファイル Top 10 ---"
git log --oneline --grep="fix\|bug\|hotfix" --name-only --since="1 year ago" \
  | grep -v "^[a-f0-9]" | sort | uniq -c | sort -rn | head -10

class NotificationService:
    def send_alert(self, user_id: str, message: str) -> bool:
        # DB から直接ユーザー情報を取得
        import sqlite3
        conn = sqlite3.connect("/var/db/production.db")
        cursor = conn.execute(
            "SELECT email, phone FROM users WHERE id = ?", (user_id,)
        )
        user = cursor.fetchone()
        if not user:
            return False
 
        # メール送信
        import smtplib
        smtp = smtplib.SMTP("mail.production.com", 587)
        smtp.send_message(create_email(user[0], message))
 
        # SMS送信
        import requests
        requests.post("https://api.sms-provider.com/send",
                      json={"phone": user[1], "text": message})
 
        return True

def calculate_shipping(weight, destination, is_member, order_total):
    """配送料金計算（レガシー: 仕様書なし）"""
    base = weight * 100
    if destination == "overseas":
        base *= 3
    if is_member:
        base *= 0.8
    if order_total > 10000:
        base = 0
    if weight > 30:
        base += 2000
    return int(base)