Aislo/common_util/common_util_help_router.py

"""[?] 사용자 가이드(54-2 · 538) — `A05_EduDetail/guide/<페이지>/*.md` 를 부를 때마다 읽어 줌.

사용자가 메모장으로 파일을 고치면 화면 고침 없이 다음에 열 때 바뀐 글이 보임.
파일 이름 순서 = 모달 왼쪽 차례 순서 · 제목 = 첫 `# ` 줄(없으면 파일 이름).
`_` 로 시작하는 파일 · 그림은 항목 아님.
판(version) = 파일 sha256 앞 16자 · 페이지 제목 = `00_설명서` 의 제목.
그림 = 페이지 폴더의 `이름.webp|png|jpg` · 판 = 고친 시각 · 내려주기는 로그인만 · `no-cache`.
바로 고침(538-3) = 시스템 관리자만 · 판이 같을 때만 그 줄 범위를 갈아 원자 쓰기 · 다르면 409.
그림 바꾸기(538-4) = 시스템 관리자만 · png · jpg · webp · 5 MB · Pillow 확인 · 폭 1800 px 로 줄임 →
WebP 로 그 이름에 덮음 · `_manual.json`(손으로 바꾼 그림 이름 — 자동 캡처가 안 덮음)에 더함.
"""

from __future__ import annotations

import codecs
import hashlib
import io
import json
import re
import threading
from pathlib import Path
from typing import Any

from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import FileResponse
from PIL import Image, ImageOps
from pydantic import BaseModel, Field

from common_util.common_util_atomic import atomic_write_bytes
from common_util.common_util_auth import require_system_admin, verify_session

router = APIRouter(prefix="/api/help", tags=["help"])

HELP_ROOT = Path(__file__).resolve().parent.parent / "A05_EduDetail" / "guide"
PAGE_RE = re.compile(r"^[BDE]\d{2}$")
IMG_RE = re.compile(r"^[A-Za-z0-9_-]+\.(webp|png|jpg)$")


def file_version(blob: bytes) -> str:
    return hashlib.sha256(blob).hexdigest()[:16]


def image_version(path: Path) -> str:
    return format(path.stat().st_mtime_ns, "x")


def page_folder(page: str, root: Path = HELP_ROOT) -> Path:
    if not PAGE_RE.match(page):
        raise HTTPException(status_code=404, detail="없는 페이지")
    return root / page


def image_path(page: str, name: str, root: Path = HELP_ROOT) -> Path:
    if not IMG_RE.match(name):
        raise HTTPException(status_code=404, detail="없는 그림")
    return page_folder(page, root) / name


def read_images(folder: Path) -> dict[str, str]:
    return {p.name: image_version(p) for p in sorted(folder.iterdir()) if IMG_RE.match(p.name)}


def read_help(page: str, root: Path = HELP_ROOT) -> list[dict[str, str]]:
    folder = page_folder(page, root)
    items = []
    for path in sorted(folder.glob("*.md")) if folder.is_dir() else []:
        if path.name.startswith("_"):
            continue
        blob = path.read_bytes()
        text = blob.decode("utf-8-sig")
        head = next((line[2:].strip() for line in text.splitlines() if line.startswith("# ")), "")
        items.append(
            {
                "key": path.stem,
                "title": head or path.stem,
                "text": text,
                "version": file_version(blob),
            }
        )
    return items


def read_guide(root: Path = HELP_ROOT) -> list[dict[str, Any]]:
    """모든 페이지 — 폴더 이름 순 · 글 없는 폴더는 뺌."""
    pages = []
    for folder in sorted(p for p in root.iterdir() if p.is_dir() and PAGE_RE.match(p.name)):
        items = read_help(folder.name, root)
        if not items:
            continue
        manual = next((i for i in items if i["key"] == "00_설명서"), None)
        pages.append(
            {
                "page": folder.name,
                "title": manual["title"] if manual else folder.name,
                "items": items,
                "images": read_images(folder),
            }
        )
    return pages


@router.get("")
async def get_guide(session: dict[str, Any] = Depends(verify_session)) -> dict[str, Any]:
    return {"pages": read_guide(HELP_ROOT), "admin": session.get("role") == "SYSTEM_ADMIN"}


MEDIA = {b"RIFF": "image/webp", b"\x89PNG": "image/png"}


@router.get("/{page}/img/{name}")
def get_image(page: str, name: str) -> FileResponse:
    path = image_path(page, name, HELP_ROOT)
    if not path.is_file():
        raise HTTPException(status_code=404, detail="없는 그림")
    with path.open("rb") as f:  # 올린 그림은 이름과 달리 WebP — 머리 글자로 가름
        media = MEDIA.get(f.read(4), "image/jpeg")
    return FileResponse(path, media_type=media, headers={"Cache-Control": "no-cache"})


class HelpEdit(BaseModel):
    version: str = Field(max_length=64)
    start: int = Field(ge=0)
    end: int = Field(ge=0)
    text: str = Field(max_length=200_000)


# ponytail: 서버 한 벌 안 잠금 — 판 확인 ~ 쓰기 사이 경합만 막음 · 서버 여러 벌이면 파일 잠금
_write_lock = threading.Lock()


def item_path(page: str, key: str, root: Path = HELP_ROOT) -> Path:
    folder = page_folder(page, root)
    bad = not key or key.startswith((".", "_")) or any(c in key for c in "/\\")
    path = folder / f"{key}.md"
    if bad or not path.is_file():
        raise HTTPException(status_code=404, detail="없는 항목")
    return path


def replace_lines(page: str, key: str, edit: HelpEdit, root: Path = HELP_ROOT) -> dict[str, str]:
    """판이 같을 때만 start~end 줄(0부터 · 끝 포함)을 글로 갈아 끼움 — 빈 글 = 그 줄 지움."""
    path = item_path(page, key, root)
    with _write_lock:
        blob = path.read_bytes()
        if file_version(blob) != edit.version:
            raise HTTPException(status_code=409, detail="다른 관리자가 먼저 고침")
        text = blob.decode("utf-8-sig")
        newline = "\r\n" if "\r\n" in text else "\n"
        lines = text.split(newline)
        if edit.start > edit.end or edit.end >= len(lines):
            raise HTTPException(status_code=400, detail="줄 범위 틀림")
        new = edit.text.replace("\r\n", "\n").split("\n") if edit.text else []
        lines[edit.start : edit.end + 1] = new
        out = newline.join(lines).encode("utf-8")
        atomic_write_bytes(path, codecs.BOM_UTF8 + out if blob.startswith(codecs.BOM_UTF8) else out)
    return next(item for item in read_help(page, root) if item["key"] == key)


@router.put("/{page}/{key}", dependencies=[Depends(require_system_admin)])
def put_item(page: str, key: str, edit: HelpEdit) -> dict[str, dict[str, str]]:
    return {"item": replace_lines(page, key, edit, HELP_ROOT)}


MAX_IMG_BYTES = 5 * 1024 * 1024
MAX_IMG_WIDTH = 1800
UPLOAD_SUFFIXES = {".png", ".jpg", ".jpeg", ".webp"}


def to_webp(blob: bytes) -> bytes:
    if not blob or len(blob) > MAX_IMG_BYTES:
        raise HTTPException(status_code=400, detail="그림은 5 MB 이하")
    try:
        with Image.open(io.BytesIO(blob)) as raw:
            if raw.format not in {"PNG", "JPEG", "MPO", "WEBP"}:  # MPO = 사진기 JPEG
                raise HTTPException(status_code=400, detail="png · jpg · webp 그림만")
            img = ImageOps.exif_transpose(raw)
            if img.width > MAX_IMG_WIDTH:
                img = img.resize((MAX_IMG_WIDTH, round(img.height * MAX_IMG_WIDTH / img.width)))
            if img.mode not in ("RGB", "RGBA"):
                img = img.convert(
                    "RGBA" if "transparency" in img.info or "A" in img.mode else "RGB"
                )
            out = io.BytesIO()
            img.save(out, "WEBP", quality=90)
    except (OSError, ValueError, Image.DecompressionBombError) as error:
        raise HTTPException(status_code=400, detail="그림 파일 아님") from error
    return out.getvalue()


def save_image(page: str, name: str, blob: bytes, root: Path = HELP_ROOT) -> str:
    path = image_path(page, name, root)
    if not path.parent.is_dir():
        raise HTTPException(status_code=404, detail="없는 페이지")
    webp = to_webp(blob)
    manual = path.parent / "_manual.json"
    with _write_lock:
        atomic_write_bytes(path, webp)
        names = set(json.loads(manual.read_text(encoding="utf-8"))) if manual.is_file() else set()
        names.add(name)
        text = json.dumps(sorted(names), ensure_ascii=False, indent=2) + "\n"
        atomic_write_bytes(manual, text.encode("utf-8"))
    return image_version(path)


@router.post("/{page}/img", dependencies=[Depends(require_system_admin)])
async def post_image(
    page: str, name: str = Form(...), file: UploadFile = File(...)
) -> dict[str, str]:
    if Path(file.filename or "").suffix.lower() not in UPLOAD_SUFFIXES:
        raise HTTPException(status_code=400, detail="png · jpg · webp 그림만")
    blob = await file.read(MAX_IMG_BYTES + 1)
    version = await run_in_threadpool(save_image, page, name, blob, HELP_ROOT)
    return {"name": name, "version": version}