script-build-and-framework-tmp/visualization-old/backend/app/module_audit/router.py

from __future__ import annotations

import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse

from ..repositories import BuildNotFound
from ..sanitizer import sanitize
from .service import ModuleAuditService, ModuleNotFound


router = APIRouter(tags=["module-audit"])
service = ModuleAuditService()
_STATIC = Path(__file__).resolve().parent / "static"
_ASSETS = {
    "app.js": ("application/javascript; charset=utf-8", "app.js"),
    "styles.css": ("text/css; charset=utf-8", "styles.css"),
}


@router.get("/audit/{script_build_id}", response_class=HTMLResponse)
def audit_page(script_build_id: int):
    template = (_STATIC / "index.html").read_text(encoding="utf-8")
    return HTMLResponse(template.replace("__RUN_ID__", str(script_build_id)))


@router.get("/audit-assets/{asset_name}")
def audit_asset(asset_name: str):
    asset = _ASSETS.get(asset_name)
    if asset is None:
        raise HTTPException(status_code=404, detail="未找到审计页面资源")
    media_type, filename = asset
    return FileResponse(_STATIC / filename, media_type=media_type)


@router.get("/api/script-builds/{script_build_id}/module-audit")
def module_audit_index(script_build_id: int):
    try:
        return service.index(script_build_id)
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取模块数据对照失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc


@router.get("/api/script-builds/{script_build_id}/module-audit-stream")
def module_audit_stream(script_build_id: int):
    """Stream every module automatically; this is full loading, not click-to-load."""
    try:
        index = service.index(script_build_id)
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取模块数据对照失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc

    def rows():
        yield _ndjson({"type": "index", "data": index})
        modules = index.get("modules") or []
        with ThreadPoolExecutor(max_workers=4, thread_name_prefix="module-audit") as pool:
            futures = {
                pool.submit(service.detail, script_build_id, str(module["id"])): str(module["id"])
                for module in modules
            }
            for future in as_completed(futures):
                module_id = futures[future]
                try:
                    yield _ndjson(
                        {
                            "type": "detail",
                            "moduleId": module_id,
                            "data": future.result(),
                        }
                    )
                except Exception as exc:  # One bad module must not block the full audit.
                    yield _ndjson(
                        {
                            "type": "error",
                            "moduleId": module_id,
                            "message": f"{type(exc).__name__}: {sanitize(str(exc))}",
                        }
                    )
        yield _ndjson({"type": "complete", "moduleCount": len(modules)})

    return StreamingResponse(
        rows(),
        media_type="application/x-ndjson; charset=utf-8",
        headers={
            "Cache-Control": "no-store",
            "X-Content-Type-Options": "nosniff",
        },
    )


@router.get("/api/script-builds/{script_build_id}/module-audit/{module_id:path}")
def module_audit_detail(script_build_id: int, module_id: str):
    try:
        return service.detail(script_build_id, module_id)
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except ModuleNotFound as exc:
        raise HTTPException(status_code=404, detail=f"未找到模块 {module_id}") from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取模块完整数据失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc


def _ndjson(value: dict) -> bytes:
    return (json.dumps(value, ensure_ascii=False, separators=(",", ":")) + "\n").encode(
        "utf-8"
    )