cert_study_app/services/question_concept_service.py

import json
import re
from dataclasses import dataclass
from typing import Optional

from cert_study_app.models import Question


@dataclass(frozen=True)
class ConceptRule:
    category: str
    subcategory: str
    tags: tuple[str, ...]
    keywords: tuple[str, ...]


AZ104_CONCEPT_RULES: tuple[ConceptRule, ...] = (
    ConceptRule("az104_identity_governance", "entra_id", ("Microsoft Entra ID", "tenant", "user", "group"), ("entra", "azure ad", "tenant", "user account", "group", "사용자", "그룹", "테넌트")),
    ConceptRule("az104_identity_governance", "rbac", ("RBAC", "role assignment", "custom role"), ("rbac", "role assignment", "custom role", "role definition", "assignablescopes", "role1", "역할", "권한")),
    ConceptRule("az104_identity_governance", "policy", ("Azure Policy", "initiative", "compliance"), ("policy", "initiative", "compliance", "deny", "audit", "정책", "규정")),
    ConceptRule("az104_identity_governance", "resource_group", ("resource group", "subscription", "management group"), ("resource group", "subscription", "management group", "tag", "lock", "리소스 그룹", "구독", "태그", "잠금")),
    ConceptRule("az104_storage", "blob", ("Blob Storage", "container", "access tier"), ("blob", "container", "access tier", "hot tier", "cool tier", "archive", "컨테이너")),
    ConceptRule("az104_storage", "files", ("Azure Files", "file share", "sync"), ("azure files", "file share", "file sync", "파일 공유")),
    ConceptRule("az104_storage", "storage_account", ("storage account", "replication", "SAS"), ("storage account", "sas", "shared access signature", "replication", "grs", "lrs", "zrs", "스토리지")),
    ConceptRule("az104_compute", "vm", ("virtual machine", "availability set", "disk"), ("virtual machine", "vm", "availability set", "disk", "managed disk", "가상 머신", "디스크")),
    ConceptRule("az104_compute", "vmss", ("VM scale set", "autoscale"), ("scale set", "vmss", "autoscale", "확장 집합", "자동 크기 조정")),
    ConceptRule("az104_compute", "app_service", ("App Service", "web app", "deployment slot"), ("app service", "web app", "deployment slot", "app plan", "웹앱")),
    ConceptRule("az104_compute", "aks_acr", ("AKS", "ACR", "container"), ("aks", "kubernetes", "acr", "container registry", "container instance", "컨테이너")),
    ConceptRule("az104_compute", "migration", ("Azure Migrate", "replication", "assessment"), ("azure migrate", "migration", "replication appliance", "assessment", "마이그레이션")),
    ConceptRule("az104_networking", "vnet_subnet", ("VNet", "subnet", "peering"), ("vnet", "virtual network", "subnet", "peering", "address space", "가상 네트워크", "서브넷")),
    ConceptRule("az104_networking", "nsg", ("NSG", "security rule", "RDP", "SSH"), ("nsg", "network security group", "security rule", "inbound", "outbound", "rdp", "ssh", "네트워크 보안 그룹")),
    ConceptRule("az104_networking", "load_balancer", ("Load Balancer", "backend pool", "health probe"), ("load balancer", "backend pool", "health probe", "frontend ip", "부하 분산")),
    ConceptRule("az104_networking", "app_gateway", ("Application Gateway", "WAF", "listener"), ("application gateway", "waf", "listener", "routing rule", "애플리케이션 게이트웨이")),
    ConceptRule("az104_networking", "dns", ("DNS", "private DNS", "zone"), ("dns", "private dns", "zone", "record set", "name resolution", "이름 확인")),
    ConceptRule("az104_networking", "vpn_expressroute", ("VPN Gateway", "ExpressRoute", "connection"), ("vpn gateway", "expressroute", "local network gateway", "connection", "gateway subnet")),
    ConceptRule("az104_monitor_recovery", "monitor", ("Azure Monitor", "Log Analytics", "alert"), ("azure monitor", "log analytics", "alert", "metric", "activity log", "모니터", "경고")),
    ConceptRule("az104_monitor_recovery", "backup", ("Azure Backup", "recovery vault", "restore"), ("backup", "recovery services vault", "restore", "snapshot", "백업", "복원")),
)


CATEGORY_LABELS = {
    "az104_identity_governance": "AZ-104 · Identity & Governance",
    "az104_storage": "AZ-104 · Storage",
    "az104_compute": "AZ-104 · Compute",
    "az104_networking": "AZ-104 · Networking",
    "az104_monitor_recovery": "AZ-104 · Monitor & Recovery",
    "identity": "ID 및 액세스",
    "governance": "거버넌스",
    "storage": "스토리지",
    "compute_vm": "VM/컴퓨팅",
    "network": "네트워크",
    "monitoring": "모니터링",
    "backup_recovery": "백업/복구",
    "containers": "컨테이너",
    "migration": "마이그레이션",
    "uncategorized": "미분류",
}


SUBCATEGORY_LABELS = {
    "entra_id": "Entra ID",
    "rbac": "RBAC/역할",
    "policy": "Policy",
    "resource_group": "구독/리소스 그룹",
    "blob": "Blob",
    "files": "Azure Files",
    "storage_account": "Storage Account",
    "vm": "VM",
    "vmss": "VM Scale Set",
    "app_service": "App Service",
    "vnet_subnet": "VNet/Subnet",
    "nsg": "NSG",
    "load_balancer": "Load Balancer",
    "app_gateway": "Application Gateway",
    "dns": "DNS",
    "vpn_expressroute": "VPN/ExpressRoute",
    "monitor": "Azure Monitor",
    "backup": "Backup",
    "aks_acr": "AKS/ACR",
    "migration": "Migration",
}


def _visual_text(question: Question) -> str:
    try:
        visual = json.loads(question.visual_analysis_json) if question.visual_analysis_json else {}
    except Exception:
        return ""
    if not isinstance(visual, dict):
        return ""
    parts = [
        visual.get("stem"),
        visual.get("source_content"),
        visual.get("notes"),
        json.dumps(visual.get("answer_areas") or [], ensure_ascii=False),
        json.dumps(visual.get("statements") or [], ensure_ascii=False),
    ]
    return "\n".join(str(part) for part in parts if part)


def question_concept_text(question: Question) -> str:
    parts = [
        question.stem,
        question.raw_text,
        question.explanation,
        question.parent_stem,
        " ".join(str(option) for option in question.get_options() or []),
        _visual_text(question),
    ]
    return "\n".join(str(part) for part in parts if part).lower()


def classify_question_concept(question: Question) -> dict:
    text = question_concept_text(question)
    scores: list[tuple[int, ConceptRule]] = []
    for rule in AZ104_CONCEPT_RULES:
        score = 0
        for keyword in rule.keywords:
            pattern = re.escape(keyword.lower())
            matches = len(re.findall(pattern, text))
            score += matches * (3 if " " in keyword else 1)
        if score:
            scores.append((score, rule))

    if not scores:
        return {
            "category": "uncategorized",
            "subcategory": None,
            "concept_tags": [],
            "confidence": 0,
        }

    scores.sort(key=lambda item: item[0], reverse=True)
    top_score, top_rule = scores[0]
    second_score = scores[1][0] if len(scores) > 1 else 0
    confidence = min(95, 45 + top_score * 8 + max(0, top_score - second_score) * 5)
    related_tags = list(top_rule.tags)
    return {
        "category": top_rule.category,
        "subcategory": top_rule.subcategory,
        "concept_tags": related_tags[:8],
        "confidence": confidence,
    }


def apply_question_concept(question: Question, overwrite: bool = False) -> dict:
    result = classify_question_concept(question)
    if overwrite or not question.category:
        question.category = result["category"]
    if overwrite or not question.subcategory:
        question.subcategory = result["subcategory"]
    if overwrite or not question.get_concept_tags():
        question.set_concept_tags(result["concept_tags"])
    return result


def concept_label(category: Optional[str], subcategory: Optional[str] = None) -> str:
    category_label = CATEGORY_LABELS.get(category or "uncategorized", category or "미분류")
    if subcategory:
        return f"{category_label} · {SUBCATEGORY_LABELS.get(subcategory, subcategory)}"
    return category_label


def classify_question_batch(db, source: Optional[str] = None, limit: int = 1000, overwrite: bool = False) -> dict:
    query = db.query(Question)
    if source:
        query = query.filter(Question.source == source)
    if not overwrite:
        query = query.filter(Question.category.is_(None))
    questions = query.order_by(Question.id.asc()).limit(limit).all()
    summary = {"checked": 0, "classified": 0, "uncategorized": 0}
    for question in questions:
        result = apply_question_concept(question, overwrite=overwrite)
        summary["checked"] += 1
        if result["category"] == "uncategorized":
            summary["uncategorized"] += 1
        else:
            summary["classified"] += 1
    db.commit()
    return summary