위험 관리 시스템

# EU AI Act Article 9 준수 - AI 위험 관리 시스템 구현
# ISO/IEC 23894 및 NIST AI RMF 기반

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
from datetime import datetime
import json

class RiskSeverity(Enum):
    CATASTROPHIC = 5   # 생명 위협, 심각한 인권 침해
    CRITICAL = 4       # 중대한 피해
    SERIOUS = 3        # 상당한 피해
    MINOR = 2          # 경미한 피해
    NEGLIGIBLE = 1     # 무시 가능

class RiskLikelihood(Enum):
    ALMOST_CERTAIN = 5  # 거의 확실
    LIKELY = 4          # 가능성 높음
    POSSIBLE = 3        # 가능
    UNLIKELY = 2        # 가능성 낮음
    RARE = 1            # 희박

@dataclass
class Risk:
    """식별된 위험"""
    risk_id: str
    name: str
    description: str
    category: str  # safety, rights, discrimination, privacy, security
    scenario: str  # 발생 시나리오
    affected_groups: List[str]
    severity: RiskSeverity
    likelihood: RiskLikelihood
    risk_score: int = 0
    mitigations: List[str] = field(default_factory=list)
    residual_risk: Optional[int] = None
    status: str = "identified"

    def __post_init__(self):
        self.risk_score = self.severity.value * self.likelihood.value

@dataclass
class AIRiskManagementSystem:
    """EU AI Act Article 9 준수 위험 관리 시스템"""

    system_name: str
    system_version: str
    intended_purpose: str
    high_risk_category: str  # Annex III 분류
    risks: List[Risk] = field(default_factory=list)
    created_date: datetime = field(default_factory=datetime.now)
    last_updated: datetime = field(default_factory=datetime.now)

    # 위험 수용 기준 (조직 정책)
    RISK_ACCEPTANCE_THRESHOLD = 9  # 3x3 이하 수용

    def identify_risks(self):
        """Step 1: 위험 식별 (Article 9(2)(a))"""

        # 예시: 채용 AI 시스템의 위험 식별
        identified_risks = [
            Risk(
                risk_id="R001",
                name="성별 차별",
                description="학습 데이터의 역사적 편향으로 인한 성별 기반 차별적 예측",
                category="discrimination",
                scenario="여성 지원자가 동일 조건의 남성보다 낮은 점수를 받음",
                affected_groups=["여성 지원자", "소수 성별"],
                severity=RiskSeverity.CRITICAL,
                likelihood=RiskLikelihood.POSSIBLE
            ),
            Risk(
                risk_id="R002",
                name="불투명한 거절",
                description="지원자가 거절 이유를 이해할 수 없음",
                category="rights",
                scenario="AI가 거절 결정을 내렸으나 설명 제공 불가",
                affected_groups=["모든 지원자"],
                severity=RiskSeverity.SERIOUS,
                likelihood=RiskLikelihood.LIKELY
            ),
            Risk(
                risk_id="R003",
                name="개인정보 과다 수집",
                description="채용에 불필요한 SNS 데이터 분석",
                category="privacy",
                scenario="지원자의 정치 성향, 종교 정보가 분석에 사용됨",
                affected_groups=["모든 지원자"],
                severity=RiskSeverity.CRITICAL,
                likelihood=RiskLikelihood.UNLIKELY
            ),
        ]

        self.risks.extend(identified_risks)
        print(f"📋 {len(identified_risks)}개 위험 식별 완료")
        return identified_risks

    def analyze_and_evaluate_risks(self):
        """Step 2 & 3: 위험 분석 및 평가 (Article 9(2)(b))"""

        print("\n📊 위험 분석 및 평가 결과:")
        print("-" * 70)
        print(f"{'ID':<6} {'위험명':<20} {'심각도':<12} {'가능성':<12} {'점수':<6} {'판정'}")
        print("-" * 70)

        for risk in self.risks:
            acceptable = risk.risk_score <= self.RISK_ACCEPTANCE_THRESHOLD
            status = "✅ 수용" if acceptable else "❌ 완화필요"
            risk.status = "accepted" if acceptable else "needs_mitigation"

            print(f"{risk.risk_id:<6} {risk.name:<20} "
                  f"{risk.severity.name:<12} {risk.likelihood.name:<12} "
                  f"{risk.risk_score:<6} {status}")

        return [r for r in self.risks if r.status == "needs_mitigation"]

    def apply_mitigations(self, risk: Risk, mitigations: List[Dict]):
        """Step 4: 위험 완화 (Article 9(2)(c)) - 우선순위 적용"""

        print(f"\n🛡️ {risk.name} 위험 완화 조치:")

        for i, mitigation in enumerate(mitigations, 1):
            priority = mitigation["priority"]
            measure = mitigation["measure"]
            effect = mitigation["risk_reduction"]

            priority_labels = {
                1: "설계 단계 제거 (Design out)",
                2: "기술적 보호 조치",
                3: "사용자 정보 제공/교육"
            }

            print(f"   {i}. [{priority_labels[priority]}] {measure}")
            risk.mitigations.append(measure)

        # 잔존 위험 계산
        total_reduction = sum(m["risk_reduction"] for m in mitigations)
        risk.residual_risk = max(1, risk.risk_score - total_reduction)
        risk.status = "mitigated"

        acceptable = risk.residual_risk <= self.RISK_ACCEPTANCE_THRESHOLD
        print(f"   → 잔존 위험 점수: {risk.residual_risk} "
              f"({'✅ 수용 가능' if acceptable else '⚠️ 추가 조치 필요'})")

        return risk.residual_risk

    def generate_report(self) -> str:
        """위험 관리 문서화 (Article 9(3))"""

        report = {
            "시스템 정보": {
                "시스템명": self.system_name,
                "버전": self.system_version,
                "의도된 용도": self.intended_purpose,
                "고위험 분류": self.high_risk_category,
                "평가일": self.last_updated.isoformat()
            },
            "위험 요약": {
                "총 식별 위험": len(self.risks),
                "완화 필요": len([r for r in self.risks if r.status == "needs_mitigation"]),
                "완화 완료": len([r for r in self.risks if r.status == "mitigated"]),
                "수용": len([r for r in self.risks if r.status == "accepted"])
            },
            "위험 상세": [
                {
                    "ID": r.risk_id,
                    "명칭": r.name,
                    "설명": r.description,
                    "위험 점수": r.risk_score,
                    "완화 조치": r.mitigations,
                    "잔존 위험": r.residual_risk,
                    "상태": r.status
                }
                for r in self.risks
            ]
        }

        return json.dumps(report, ensure_ascii=False, indent=2)

# 실행 예시
rms = AIRiskManagementSystem(
    system_name="AI 채용 심사 시스템",
    system_version="2.0",
    intended_purpose="이력서 스크리닝 및 적합도 점수 산출",
    high_risk_category="Annex III(4)(a) - 고용 분야 AI"
)

# 1. 위험 식별
rms.identify_risks()

# 2-3. 위험 분석 및 평가
high_risks = rms.analyze_and_evaluate_risks()

# 4. 위험 완화 (R001 성별 차별 예시)
if high_risks:
    rms.apply_mitigations(
        high_risks[0],  # 성별 차별 위험
        [
            {"priority": 1, "measure": "성별 관련 피처 제거 및 대리변수 탐지", "risk_reduction": 5},
            {"priority": 2, "measure": "공정성 제약 손실함수 적용 (Equalized Odds)", "risk_reduction": 4},
            {"priority": 3, "measure": "분기별 공정성 감사 및 결과 공개", "risk_reduction": 2}
        ]
    )

# 보고서 생성
print("\n" + "="*60)
print("📄 위험 관리 보고서 (EU AI Act Article 9 준수)")
print("="*60)
print(rms.generate_report())