report_generator.py

"""
report_generator.py

Conformance / Governance report generator for AI-Sovereign-Sentinel.
این ماژول یک گزارش استاندارد برای مناقصات دولتی (G-Cloud / NCSC / DSIT)
تولید می‌کند که نشان می‌دهد Sovereign چه کارهایی انجام می‌دهد و
آمار رفتاری مدل‌ها در عمل چه بوده است.
"""

import json
import time
from typing import Dict, Any, List

from sovereign_core import (
    ENGINE_NAME,
    SOVEREIGN_VERSION,
    AUTHORITY_NAME,
    AUDIT_LOG_FILE,
    export_audit_log_json,
)

CONFORMANCE_VERSION = "1.0-gov-ready"


def _summarise_audit_entries(entries: List[Dict[str, Any]]) -> Dict[str, Any]:
    """
    Build simple aggregate stats over the centralised audit log.
    این فقط یک summary سبک است که بعداً می‌توانی پیچیده‌ترش کنی.
    """
    total_events = len(entries)

    by_event_type: Dict[str, int] = {}
    by_risk_level: Dict[str, int] = {}
    environments = set()
    policy_violations = 0
    max_risk_score = 0

    for e in entries:
        et = e.get("event_type", "unknown")
        rl = (e.get("risk_level") or "UNSPECIFIED").upper()
        env = e.get("deployment_env") or "unspecified"

        by_event_type[et] = by_event_type.get(et, 0) + 1
        by_risk_level[rl] = by_risk_level.get(rl, 0) + 1
        environments.add(env)

        if e.get("policy_violation"):
            policy_violations += 1

        rs = e.get("risk_score")
        if isinstance(rs, int) and rs > max_risk_score:
            max_risk_score = rs

    return {
        "total_events": total_events,
        "events_by_type": by_event_type,
        "events_by_risk_level": by_risk_level,
        "distinct_environments": sorted(list(environments)),
        "policy_violations": policy_violations,
        "max_observed_risk_score": max_risk_score,
    }


def generate_conformance_report(
    output_path: str = "sovereign_conformance_report.json",
    audit_log_path: str = AUDIT_LOG_FILE,
) -> Dict[str, Any]:
    """
    Generate a conformance report that can be attached to tenders /
    security reviews / G-Cloud submissions.

    خروجی یک dict است و همچنین در یک فایل JSON ذخیره می‌شود.
    """

    # 1) Load audit entries (centralised audit journal)
    entries = export_audit_log_json(path=audit_log_path)
    audit_summary = _summarise_audit_entries(entries)

    # 2) Static capability map – ربط دادن Sovereign به نیازهای دولت
    capability_mapping: Dict[str, Any] = {
        "centralised_audit_log": {
            "description": "JSONL-based audit journal with CSV/JSON export for G-Cloud / NCSC evidence.",
            "meets_requirements": [
                "standardised audit trails",
                "reusable modular components",
                "exportable evidence for oversight and investigation",
            ],
        },
        "api_first_architecture": {
            "description": "Sovereign exposes REST endpoints for monitoring, governance, policy checks and risk evaluation.",
            "endpoints": [
                "/monitor",
                "/govern",
                "/evaluate-risk",
                "/behavior-score",
                "/policy-check",
            ],
            "meets_requirements": [
                "API-based interoperability across government systems",
                "ability to plug into existing digital infrastructure",
            ],
        },
        "policy_engine": {
            "description": "Configurable policy library for LLM safety, restrictions and ethical guardrails.",
            "meets_requirements": [
                "standardised safety rules",
                "reusable policy sets",
                "future alignment with UK AI Safety regulation",
            ],
        },
        "multi_model_support": {
            "description": "Supports multiple model families (frontier and narrow) via generic adapter pattern.",
            "meets_requirements": [
                "scalable tech stack for narrow and large language models",
                "vendor-agnostic control layer",
            ],
        },
        "semantic_risk_scoring": {
            "description": "Semantic analysis of prompts/outputs with 0-100 risk score and categorical levels.",
            "meets_requirements": [
                "behavioural oversight",
                "risk scoring for AI systems",
                "supporting safe deployment in financial, defence and public-sector settings",
            ],
        },
        "deployment_environment_binding": {
            "description": "Every certificate and audit event is bound to an explicit environment (dev, staging, prod, restricted, etc.).",
            "meets_requirements": [
                "clear separation of environments",
                "reduced risk of cross-environment contamination",
            ],
        },
        "enterprise_access_control": {
            "description": "Enterprise key / token gating to prevent shadow-IT and unauthorised operation.",
            "meets_requirements": [
                "access control for critical infrastructure",
                "alignment with zero-trust security principles",
            ],
        },
    }

    # 3) High-level metadata about the Sovereign engine
    report: Dict[str, Any] = {
        "report_type": "Sovereign_AI_Conformance_Report",
        "report_version": CONFORMANCE_VERSION,
        "generated_at": int(time.time()),
        "engine": {
            "name": ENGINE_NAME,
            "sovereign_version": SOVEREIGN_VERSION,
            "authority": AUTHORITY_NAME,
        },
        "governance_and_security": {
            "description": (
                "Sovereign provides a cryptographic integrity and behaviour-governance "
                "layer for AI models used in banking, defence, government and critical "
                "national infrastructure."
            ),
            "capabilities": capability_mapping,
        },
        "audit_summary": audit_summary,
        "provenance": {
            "audit_log_path": audit_log_path,
            "entry_count": audit_summary["total_events"],
        },
    }

    # 4) Save to file
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2)

    return report


# برای سازگاری با نسخه‌ی قبلی که از این اسم استفاده می‌کرد
def export_conformance_report_json(
    path: str = "sovereign_conformance_report.json",
) -> str:
    """
    Compatibility wrapper so app.py می‌تواند همان
    export_conformance_report_json را صدا بزند.
    """
    generate_conformance_report(output_path=path)
    return path


if __name__ == "__main__":
    r = generate_conformance_report()
    print("Conformance report generated:", "sovereign_conformance_report.json")