"""E05 안정성평가 보고서 API (요구 347-5 · 351 · 요구서 4-3 · PLAN 534-3). - `GET /api/projects/{id}/stab/report` — 작업본(없고 2D 결과가 있으면 처음 만듦 + 초기값) · 판 · 해석 지문이 바뀌었는지(`stale`) · 고칠 수 있는지 - `PUT …/stab/report` — [저장] `{rev, book}` · 판이 다르면 409 · 서버 재계산(`with_server_values`) - `POST …/stab/report/refresh` — [보고서 갱신] `{rev}` · 프로그램 몫 시트 · 그림만 다시 · 덮인 시트에 사용자 고침이 있었으면 `overwritten` + 옛 판 `_지난/` - `POST …/stab/report/reset` — [초기화] 초기값을 작업본에 덮어씀 - `POST …/stab/report/xlsx` · `…/pdf` `{book}` — 화면 지금 책을 서버 재계산 → xlsx · PDF(저장 안 함) - `GET …/stab/report/xlsx` · `…/pdf` — 작업본으로(보기만 허용된 회사도) - `GET …/stab/report/fig?key=` — 그림 PNG 계산 자리 = ③ 서버 단독. 권한 = main.py `protected_with_company` (지정받은 회사 = GET 만 · 그 밖 403). """ from __future__ import annotations import asyncio from pathlib import Path from typing import Any from urllib.parse import quote from uuid import UUID import aiomysql from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import FileResponse, Response from pydantic import BaseModel from common_util.common_util_auth import verify_session from common_util.common_util_print_pdf import PrintError from common_util.common_util_spreadsheet import RecalcError, with_server_values from common_util.common_util_storage import resolve_stored_project_path from common_util.common_util_workbook_xlsx import workbook_to_xlsx from config.config_db import get_db_pool from E05_StabReport import E05_StabReport_Build as build from E05_StabReport import E05_StabReport_Print as printer from E05_StabReport import E05_StabReport_Store as store router = APIRouter(prefix="/api/projects", tags=["E05 Stab Report"]) XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" TITLE = "안정성평가_보고서" _LOCKS: dict[str, asyncio.Lock] = {} class SaveRequest(BaseModel): rev: str | None = None book: dict[str, Any] class RevRequest(BaseModel): rev: str | None = None class BookRequest(BaseModel): book: dict[str, Any] def _lock(project_id: UUID) -> asyncio.Lock: return _LOCKS.setdefault(str(project_id), asyncio.Lock()) async def _project(project_id: UUID) -> tuple[Path, int | None]: async with ( get_db_pool().acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor, ): await cursor.execute( "SELECT storage_path, company_id FROM projects" " WHERE id = %s AND project_kind = 'STABILITY' AND deleted_at IS NULL", (str(project_id),), ) row = await cursor.fetchone() if not row or not row["storage_path"]: raise HTTPException(status_code=404, detail="안정성평가 프로젝트가 아닙니다.") return Path(resolve_stored_project_path(row["storage_path"])), row["company_id"] def _editable(session: dict[str, Any], company_id: int | None) -> bool: return session["role"] == "SYSTEM_ADMIN" or session.get("company_id") == company_id def _check_book(book: dict[str, Any]) -> None: sheets = book.get("시트") if book.get("종류") != "통합문서" or not isinstance(sheets, list) or not sheets: raise HTTPException(status_code=400, detail="통합문서 꼴이 아닙니다.") def _state(root: Path) -> dict[str, Any]: book, meta = store.load(root) return { "book": book, "rev": store.revision(root), "meta": meta, "stale": bool(meta) and meta.get("fingerprint") != build.analysis_fingerprint(root), "has_initial": (build.initial_dir(root) / build.REPORT).is_file(), } def _ensure(root: Path) -> dict[str, Any]: """작업본이 없으면 처음 만듦(2D 결과가 없으면 빈 상태 · 까닭).""" if store.load(root)[0] is None: try: store.create(root) except build.MissingInput as error: return {"book": None, "reason": str(error)} return _state(root) def _recalc_errors(error: Exception) -> HTTPException: if isinstance(error, PrintError): return HTTPException(status_code=503, detail=str(error)) return HTTPException(status_code=422, detail=str(error)) @router.get("/{project_id}/stab/report", response_model=None) async def get_report( project_id: UUID, session: dict[str, Any] = Depends(verify_session) ) -> dict[str, Any]: root, company_id = await _project(project_id) async with _lock(project_id): try: state = await asyncio.to_thread(_ensure, root) except (RecalcError, PrintError) as error: raise _recalc_errors(error) from error return {"status": "success", "editable": _editable(session, company_id), **state} @router.put("/{project_id}/stab/report", response_model=None) async def put_report(project_id: UUID, payload: SaveRequest) -> dict[str, Any]: root, _ = await _project(project_id) _check_book(payload.book) async with _lock(project_id): if store.load(root)[0] is None: raise HTTPException(status_code=409, detail="보고서가 아직 없습니다.") if payload.rev != store.revision(root): raise HTTPException( status_code=409, detail="다른 곳에서 보고서가 바뀌었습니다 — 다시 여세요." ) try: await asyncio.to_thread(store.save, root, payload.book) except RecalcError as error: raise _recalc_errors(error) from error return {"status": "success", **await asyncio.to_thread(_state, root)} @router.post("/{project_id}/stab/report/refresh", response_model=None) async def refresh_report(project_id: UUID, payload: RevRequest) -> dict[str, Any]: root, _ = await _project(project_id) async with _lock(project_id): current = store.revision(root) if current is not None and payload.rev != current: raise HTTPException( status_code=409, detail="다른 곳에서 보고서가 바뀌었습니다 — 다시 여세요." ) try: result = await asyncio.to_thread(store.refresh, root) except build.MissingInput as error: raise HTTPException(status_code=409, detail=str(error)) from error except (RecalcError, PrintError) as error: raise _recalc_errors(error) from error return { "status": "success", "overwritten": result["overwritten"], **await asyncio.to_thread(_state, root), } @router.post("/{project_id}/stab/report/reset", response_model=None) async def reset_report(project_id: UUID) -> dict[str, Any]: root, _ = await _project(project_id) async with _lock(project_id): if not await asyncio.to_thread(store.reset, root): raise HTTPException(status_code=404, detail="초기값이 없습니다.") return {"status": "success", **await asyncio.to_thread(_state, root)} def _file(data: bytes, media: str, ext: str, inline: bool = False) -> Response: kind = "inline" if inline else "attachment" name = quote(f"{TITLE}.{ext}") return Response( content=data, media_type=media, headers={"Content-Disposition": f"{kind}; filename*=UTF-8''{name}"}, ) def _export(root: Path, book: dict[str, Any] | None, kind: str) -> bytes: if book is None: book = store.load(root)[0] if book is None: raise HTTPException(status_code=404, detail="보고서가 아직 없습니다.") figures = store.figures_of(root, book) if kind == "xlsx": return workbook_to_xlsx(with_server_values(book), figures) return printer.report_pdf(book, figures) # 인쇄 HTML 이 다시 풂(브라우저 값 안 씀) async def _export_response(project_id: UUID, book: dict[str, Any] | None, kind: str) -> Response: root, _ = await _project(project_id) if book is not None: _check_book(book) try: data = await asyncio.to_thread(_export, root, book, kind) except (RecalcError, PrintError) as error: raise _recalc_errors(error) from error if kind == "xlsx": return _file(data, XLSX, "xlsx") return _file(data, "application/pdf", "pdf", inline=True) @router.post("/{project_id}/stab/report/xlsx") async def post_xlsx(project_id: UUID, payload: BookRequest) -> Response: return await _export_response(project_id, payload.book, "xlsx") @router.get("/{project_id}/stab/report/xlsx") async def get_xlsx(project_id: UUID) -> Response: return await _export_response(project_id, None, "xlsx") @router.post("/{project_id}/stab/report/pdf") async def post_pdf(project_id: UUID, payload: BookRequest) -> Response: return await _export_response(project_id, payload.book, "pdf") @router.get("/{project_id}/stab/report/pdf") async def get_pdf(project_id: UUID) -> Response: return await _export_response(project_id, None, "pdf") @router.get("/{project_id}/stab/report/fig") async def get_figure(project_id: UUID, key: str) -> FileResponse: root, _ = await _project(project_id) if not key.startswith("fig/") or ".." in key: raise HTTPException(status_code=400, detail="그림 열쇠가 틀렸습니다.") path = build.fig_path(build.stage_dir(root), key) if not path.is_file(): raise HTTPException(status_code=404, detail="그림이 없습니다.") return FileResponse(path, media_type="image/png", headers={"Cache-Control": "no-cache"})