# 例1: AIコード生成におけるバイアス
# AIに「ユーザープロフィール」の実装を依頼すると…
 
# BAD: AIが生成しがちなバイアスのあるコード
class UserProfile:
    def __init__(self):
        self.gender = ""         # 二項対立を前提
        self.title = "Mr./Mrs."  # 性別に基づく敬称
        self.maiden_name = ""    # ジェンダーバイアス
        self.age = 0             # 年齢差別の温床
 
    def calculate_insurance_rate(self):
        if self.gender == "female":
            return self.base_rate * 0.9  # 性別に基づく料率差
        return self.base_rate
 
# GOOD: バイアスを考慮した設計
class UserProfile:
    def __init__(self):
        self.display_name = ""
        self.pronouns = ""           # 自己申告の代名詞
        self.honorific = ""          # 任意の敬称
        self.previous_names = []     # 性別を前提としない
        self.date_of_birth = None    # 必要な場合のみ
 
    def calculate_insurance_rate(self):
        # 性別ではなく、リスク要因に基づく
        risk_factors = self.assess_risk_factors()
        return self.base_rate * risk_factors.multiplier

# 例2: AI推薦システムのバイアス
# 採用AIが学習データの偏りを反映する問題
 
# BAD: 過去の採用データをそのまま学習
class BiasedRecruitmentAI:
    def rank_candidates(self, candidates):
        """過去の採用実績に基づくランキング
        問題: 過去に男性が多く採用された部署では
        男性候補者が不当に高いスコアを得る"""
        return self.model.predict(candidates)
 
# GOOD: バイアス監査と緩和を組み込む
class FairRecruitmentAI:
    def rank_candidates(self, candidates):
        raw_scores = self.model.predict(candidates)
 
        # バイアス監査
        audit = self.bias_auditor.check(
            scores=raw_scores,
            protected_attributes=["gender", "ethnicity", "age"],
            fairness_metric="demographic_parity",
        )
 
        if audit.has_bias:
            # バイアス緩和を適用
            adjusted_scores = self.mitigator.adjust(
                scores=raw_scores,
                bias_report=audit,
            )
            self.log_bias_mitigation(audit, adjusted_scores)
            return adjusted_scores
 
        return raw_scores

# バイアス検出フレームワーク
from dataclasses import dataclass
from typing import Any
 
@dataclass
class BiasMetric:
    name: str
    value: float
    threshold: float
    is_biased: bool
    affected_groups: list[str]
    recommendation: str
 
class BiasDetector:
    """AIシステムのバイアスを検出するフレームワーク"""
 
    def demographic_parity(
        self,
        predictions: list[int],
        protected_attribute: list[str],
    ) -> BiasMetric:
        """人口統計的均等性: 各グループの陽性率が等しいか"""
        groups = set(protected_attribute)
        positive_rates = {}
 
        for group in groups:
            group_mask = [
                a == group for a in protected_attribute
            ]
            group_preds = [
                p for p, m in zip(predictions, group_mask) if m
            ]
            positive_rates[group] = (
                sum(group_preds) / len(group_preds)
                if group_preds else 0
            )
 
        max_rate = max(positive_rates.values())
        min_rate = min(positive_rates.values())
        disparity = max_rate - min_rate
 
        return BiasMetric(
            name="demographic_parity",
            value=disparity,
            threshold=0.1,  # 10%以上の差は要注意
            is_biased=disparity > 0.1,
            affected_groups=[
                g for g, r in positive_rates.items()
                if r == min_rate
            ],
            recommendation=(
                "陽性率に大きな差があります。"
                "学習データの分布を確認してください。"
            ),
        )
 
    def equalized_odds(
        self,
        predictions: list[int],
        labels: list[int],
        protected_attribute: list[str],
    ) -> BiasMetric:
        """均等化オッズ: 各グループのTPR/FPRが等しいか"""
        groups = set(protected_attribute)
        tpr_by_group = {}
 
        for group in groups:
            group_indices = [
                i for i, a in enumerate(protected_attribute)
                if a == group
            ]
            true_positives = sum(
                1 for i in group_indices
                if predictions[i] == 1 and labels[i] == 1
            )
            actual_positives = sum(
                1 for i in group_indices if labels[i] == 1
            )
            tpr_by_group[group] = (
                true_positives / actual_positives
                if actual_positives > 0 else 0
            )
 
        max_tpr = max(tpr_by_group.values())
        min_tpr = min(tpr_by_group.values())
        disparity = max_tpr - min_tpr
 
        return BiasMetric(
            name="equalized_odds",
            value=disparity,
            threshold=0.1,
            is_biased=disparity > 0.1,
            affected_groups=[
                g for g, r in tpr_by_group.items()
                if r == min_tpr
            ],
            recommendation=(
                "真陽性率にグループ間で差があります。"
                "モデルの学習パラメータを調整してください。"
            ),
        )

# AI生成コードのメタデータを記録するデコレータ
import functools
import json
from datetime import datetime, timezone
from pathlib import Path
 
def ai_generated(
    model: str,
    prompt_summary: str,
    human_reviewed: bool = False,
    reviewer: str = "",
    modifications: str = "",
):
    """AI生成コードに透明性メタデータを付与するデコレータ"""
 
    def decorator(func):
        func._ai_metadata = {
            "generated_by": model,
            "prompt_summary": prompt_summary,
            "generation_date": datetime.now(timezone.utc).isoformat(),
            "human_reviewed": human_reviewed,
            "reviewer": reviewer,
            "modifications": modifications,
            "transparency_level": (
                "L3" if human_reviewed else "L2"
            ),
        }
 
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
 
        return wrapper
    return decorator
 
# 使用例
@ai_generated(
    model="Claude Opus 4",
    prompt_summary="年齢に基づかない保険料率計算の実装",
    human_reviewed=True,
    reviewer="tanaka@example.com",
    modifications="リスク要因の重み付けを手動調整",
)
def calculate_insurance_rate(risk_factors: dict) -> float:
    """リスク要因に基づく保険料率計算（年齢差別排除版）"""
    base_rate = 10000
    multiplier = 1.0
 
    for factor, value in risk_factors.items():
        if factor in APPROVED_RISK_FACTORS:
            multiplier *= RISK_WEIGHTS[factor][value]
 
    return base_rate * multiplier

// AI意思決定の監査ログシステム
interface AIDecisionLog {
  timestamp: string;
  component: string;
  decision: string;
  alternatives: {
    option: string;
    score: number;
    reason: string;
  }[];
  selectedOption: string;
  selectionReason: string;
  humanOverride: boolean;
  overrideReason?: string;
}
 
class AIAuditTrail {
  private logs: AIDecisionLog[] = [];
  private storePath: string;
 
  constructor(storePath: string) {
    this.storePath = storePath;
  }
 
  logDecision(log: AIDecisionLog): void {
    this.logs.push(log);
    this.persistLog(log);
  }
 
  // 規制当局向けの監査レポート生成
  generateAuditReport(
    startDate: string,
    endDate: string
  ): AuditReport {
    const filteredLogs = this.logs.filter(
      (log) => log.timestamp >= startDate && log.timestamp <= endDate
    );
 
    return {
      totalDecisions: filteredLogs.length,
      humanOverrides: filteredLogs.filter((l) => l.humanOverride).length,
      overrideRate:
        filteredLogs.filter((l) => l.humanOverride).length /
        filteredLogs.length,
      decisionBreakdown: this.categorizeDecisions(filteredLogs),
      recommendations: this.generateRecommendations(filteredLogs),
    };
  }
 
  private persistLog(log: AIDecisionLog): void {
    // 改ざん防止のためハッシュチェーンで保存
    const previousHash = this.getLastHash();
    const entry = { ...log, previousHash, hash: "" };
    entry.hash = this.computeHash(JSON.stringify(entry));
    // ストレージに書き込み
  }
}

# コード送信前のPII（個人識別情報）除去
import re
from typing import NamedTuple
 
class PIIPattern(NamedTuple):
    name: str
    pattern: str
    replacement: str
 
class CodeSanitizer:
    """AIに送信する前にコードからPIIを除去"""
 
    PII_PATTERNS = [
        PIIPattern(
            "email",
            r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
            "user@example.com",
        ),
        PIIPattern(
            "phone_jp",
            r'0\d{1,4}-\d{1,4}-\d{4}',
            "000-0000-0000",
        ),
        PIIPattern(
            "ip_address",
            r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b',
            "192.0.2.1",
        ),
        PIIPattern(
            "api_key",
            r'(?:api[_-]?key|token|secret)["\s]*[:=]["\s]*[a-zA-Z0-9_\-]{20,}',
            'API_KEY="REDACTED"',
        ),
        PIIPattern(
            "my_number",
            r'\b\d{4}\s?\d{4}\s?\d{4}\b',
            "0000 0000 0000",
        ),
    ]
 
    def sanitize(self, code: str) -> tuple[str, list[str]]:
        """コードからPIIを除去し、除去した項目を返す"""
        removed = []
        sanitized = code
 
        for pii in self.PII_PATTERNS:
            matches = re.findall(pii.pattern, sanitized)
            if matches:
                removed.append(
                    f"{pii.name}: {len(matches)}件を除去"
                )
                sanitized = re.sub(
                    pii.pattern, pii.replacement, sanitized
                )
 
        return sanitized, removed
 
# 使用例
sanitizer = CodeSanitizer()
clean_code, report = sanitizer.sanitize(original_code)
print(f"除去レポート: {report}")
# AI APIに clean_code を送信

# .github/workflows/ethics-check.yml
# AI倫理チェックをCIパイプラインに統合
 
name: AI Ethics Check
on:
  pull_request:
    branches: [main]
 
jobs:
  bias-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Check for biased variable names
        run: |
          # 偏りのある変数名・関数名を検出
          python scripts/check_inclusive_naming.py \
            --config .ethics/naming-rules.yaml \
            --path src/
 
      - name: Check for hardcoded cultural assumptions
        run: |
          # ハードコードされた文化的前提を検出
          python scripts/check_cultural_assumptions.py \
            --rules .ethics/cultural-rules.yaml \
            --path src/
 
      - name: PII leak detection
        run: |
          # コード内のPII（個人情報）漏洩を検出
          python scripts/detect_pii.py \
            --patterns .ethics/pii-patterns.yaml \
            --path src/ --path tests/
 
      - name: AI transparency check
        run: |
          # AI生成コードに適切なメタデータがあるか確認
          python scripts/check_ai_transparency.py \
            --require-metadata \
            --min-level L2 \
            --path src/
 
      - name: License compliance
        run: |
          # AI生成コードのライセンス互換性を確認
          python scripts/check_license_compliance.py \
            --policy .ethics/license-policy.yaml
 
  fairness-test:
    runs-on: ubuntu-latest
    if: contains(github.event.pull_request.labels.*.name, 'ai-model')
    steps:
      - uses: actions/checkout@v4
 
      - name: Run fairness tests
        run: |
          python -m pytest tests/fairness/ \
            --fairness-threshold 0.1 \
            --protected-attributes gender,age,ethnicity

# AI生成コードの著作権リスクを理解する
 
# リスク1: 学習データの著作物混入
# AIモデルはオープンソースコードを含む大量のデータで学習されている
# 生成されたコードが既存のGPLコードと酷似する可能性がある
 
# リスク2: AI生成物の著作権帰属
# 多くの法域で「AI生成物に著作権は発生しない」とされる傾向
# → 自社のコアIPをAI生成のみに依存するリスク
 
# リスク3: ライセンス汚染
# AI生成コードがGPL等のコピーレフトライセンスの
# コードを含む場合、プロジェクト全体に影響する可能性
 
# 対策の例
class LicenseComplianceChecker:
    """AI生成コードのライセンス互換性をチェック"""
 
    def check_similarity(
        self,
        generated_code: str,
        threshold: float = 0.85,
    ) -> list[dict]:
        """生成コードと既知のOSSコードの類似性を検査"""
        results = []
        for oss_project in self.oss_database:
            similarity = self.compute_similarity(
                generated_code, oss_project.code
            )
            if similarity > threshold:
                results.append({
                    "project": oss_project.name,
                    "license": oss_project.license,
                    "similarity": similarity,
                    "risk": self.assess_risk(oss_project.license),
                    "recommendation": self.recommend_action(
                        oss_project.license
                    ),
                })
        return results
 
    def assess_risk(self, license_type: str) -> str:
        risk_map = {
            "MIT": "低 - 帰属表示のみ",
            "Apache-2.0": "低 - 帰属表示+特許条項",
            "BSD-2-Clause": "低 - 帰属表示のみ",
            "GPL-3.0": "高 - コピーレフト感染の可能性",
            "AGPL-3.0": "最高 - SaaSにも適用",
            "SSPL": "最高 - サービス提供にも制約",
        }
        return risk_map.get(license_type, "不明 - 法務確認必須")

# BAD: リリース後にバイアスが発覚して慌てて対処
class AfterThoughtApproach:
    def develop(self):
        model = self.train_model(data)
        self.deploy_to_production(model)
        # ... 数ヶ月後 ...
        # ニュース「御社のAIは差別的だ！」
        self.panic_fix(model)  # ← 手遅れ
 
# GOOD: 設計段階から公平性を組み込む（Fairness by Design）
class FairnessByDesign:
    def develop(self):
        # Phase 1: データ監査
        data = self.collect_data()
        bias_report = self.audit_data_bias(data)
        data = self.mitigate_data_bias(data, bias_report)
 
        # Phase 2: モデル学習（公平性制約付き）
        model = self.train_model(
            data,
            fairness_constraints={
                "demographic_parity_gap": 0.05,
                "equalized_odds_gap": 0.05,
            },
        )
 
        # Phase 3: 公平性テスト
        fairness_result = self.test_fairness(model)
        if not fairness_result.passes_all_criteria:
            raise FairnessViolation(fairness_result)
 
        # Phase 4: 段階的デプロイ
        self.canary_deploy(model, monitor_fairness=True)

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()

# デバッグ用ユーティリティ
import logging
import traceback
from functools import wraps
 
# ロガーの設定
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)
 
def debug_decorator(func):
    """関数の入出力をログ出力するデコレータ"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.debug(f"呼び出し: {func.__name__}(args={args}, kwargs={kwargs})")
        try:
            result = func(*args, **kwargs)
            logger.debug(f"戻り値: {func.__name__} -> {result}")
            return result
        except Exception as e:
            logger.error(f"例外発生: {func.__name__}: {e}")
            logger.error(traceback.format_exc())
            raise
    return wrapper
 
@debug_decorator
def process_data(items):
    """データ処理（デバッグ対象）"""
    if not items:
        raise ValueError("空のデータ")
    return [item * 2 for item in items]

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md

# チーム間のAPI契約定義
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
 
class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class APIContract:
    """チーム間のAPI契約"""
    endpoint: str
    method: str
    owner_team: str
    consumers: List[str]
    sla_ms: int  # レスポンスタイムSLA
    priority: Priority
 
    def validate_sla(self, actual_ms: int) -> bool:
        """SLA準拠の確認"""
        return actual_ms <= self.sla_ms
 
    def to_openapi(self) -> dict:
        """OpenAPI形式で出力"""
        return {
            'path': self.endpoint,
            'method': self.method,
            'x-owner': self.owner_team,
            'x-consumers': self.consumers,
            'x-sla-ms': self.sla_ms
        }
 
# 使用例
contracts = [
    APIContract(
        endpoint="/api/v1/users",
        method="GET",
        owner_team="user-team",
        consumers=["order-team", "notification-team"],
        sla_ms=200,
        priority=Priority.HIGH
    ),
    APIContract(
        endpoint="/api/v1/orders",
        method="POST",
        owner_team="order-team",
        consumers=["payment-team", "inventory-team"],
        sla_ms=500,
        priority=Priority.CRITICAL
    )
]

# セキュアコーディング例
import hashlib
import secrets
import hmac
from typing import Optional
 
class SecurityUtils:
    """セキュリティユーティリティ"""
 
    @staticmethod
    def generate_token(length: int = 32) -> str:
        """暗号学的に安全なトークン生成"""
        return secrets.token_urlsafe(length)
 
    @staticmethod
    def hash_password(password: str, salt: Optional[str] = None) -> tuple:
        """パスワードのハッシュ化"""
        if salt is None:
            salt = secrets.token_hex(16)
        hashed = hashlib.pbkdf2_hmac(
            'sha256',
            password.encode('utf-8'),
            salt.encode('utf-8'),
            iterations=100000
        )
        return hashed.hex(), salt
 
    @staticmethod
    def verify_password(password: str, hashed: str, salt: str) -> bool:
        """パスワードの検証"""
        new_hash, _ = SecurityUtils.hash_password(password, salt)
        return hmac.compare_digest(new_hash, hashed)
 
    @staticmethod
    def sanitize_input(value: str) -> str:
        """入力値のサニタイズ"""
        dangerous_chars = ['<', '>', '"', "'", '&', '\\']
        result = value
        for char in dangerous_chars:
            result = result.replace(char, '')
        return result.strip()
 
# 使用例
token = SecurityUtils.generate_token()
hashed, salt = SecurityUtils.hash_password("my_password")
is_valid = SecurityUtils.verify_password("my_password", hashed, salt)