server.py

"""
server.py — 얼굴 일괄 교체 (HF Space ↔ Comfy Cloud)

  POST /api/swap/jobs                 이미지 1장 + 성별 → 작업 생성 (즉시 {jobId} 반환)
  GET  /api/swap/jobs/{id}            상태 조회
  POST /api/swap/cancel               작업 취소 (대기·진행 중인 작업을 Comfy Cloud 에서도 멈춤)
  GET  /api/swap/jobs/{id}/thumb      결과 미리보기(JPEG)
  GET  /api/swap/jobs/{id}/view       결과 원본 크기(미리보기용, 파일명·확장자 유지 형식)
  GET  /api/swap/jobs/{id}/download   결과 다운로드 — 업로드한 파일명·확장자 그대로
  POST /api/swap/zip                  선택한 작업 ZIP — 폴더 구조·파일명 그대로
  POST /api/swap/regen                지난 결과의 원본으로 새 결과 하나 더 (원본 재업로드 없이)
  GET  /api/whoami                    로그인한 사용자 (HF OAuth, 선택)
  GET  /api/dashboard                 생성 기록 대시보드 집계 (기간: ?from=YYYY-MM-DD&to=YYYY-MM-DD)
  GET  /api/swap/faces                삽입할 얼굴(남자/여자) 기본 정보
  POST /api/swap/faces/{gender}       얼굴 교체 업로드
  GET  /api/swap/defaults             워크플로우 기본값(고급 설정)
  GET  /api/history                   지난 결과 목록 (영구 저장분 포함)
  POST /api/history/hide              목록에서 지우기 (저장소 파일은 남김)
  GET  /api/history/status            영구 저장 상태
  GET  /api/health                    워크플로우·API 키 상태
"""
from __future__ import annotations

import hashlib
import io
import json
import os
import re
import tempfile
import threading
import time
import traceback
import unicodedata
import uuid
import zipfile
from typing import Any, Dict, List, Optional
from urllib.parse import quote

from fastapi import Body, Depends, FastAPI, File, Form, HTTPException, Query, Request, UploadFile
from fastapi.responses import FileResponse, HTMLResponse, Response, StreamingResponse
from PIL import Image

import comfy_client as cc
import history

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
FACE_DIR = os.path.join(BASE_DIR, "assets", "faces")

MAX_CONCURRENT = int(os.environ.get("COMFY_MAX_CONCURRENT", "4"))
JOB_TTL_SEC = int(os.environ.get("JOB_TTL_SEC", str(6 * 3600)))
GENDERS = ("male", "female")

_OPEN_API = ("/api/health", "/api/whoami")          # 로그인 전에도 열려 있어야 하는 것(상태 확인, 로그인 여부 확인)


def _guard(request: Request) -> None:
    """로그인이 필요한 환경에서는 /api/* 전체를 로그인한 사람만 쓰게 한다(화면 '/' 은 index_page 가 로그인 안내를 보여 준다)."""
    if not _login_required():
        return
    path = request.url.path
    if not path.startswith("/api/") or path in _OPEN_API:
        return
    if not get_current_user(request)["is_logged_in"]:
        raise HTTPException(401, "로그인이 필요해요. 화면을 새로고침한 뒤 Hugging Face 로 로그인해 주세요.")


app = FastAPI(title="Face swap batch", dependencies=[Depends(_guard)], docs_url=None, redoc_url=None, openapi_url=None)   # 로그인 전에는 경로 목록(문서)도 노출하지 않는다

history.init()

# ── 로그인(HF OAuth) — 사용 통계에 '누가 썼는지'를 남기기 위함 ────────────────────
# Space 의 README.md 에 `hf_oauth: true` 가 있으면 HF 가 OAUTH_CLIENT_ID 등을 넣어 준다. 그때만 로그인 기능을 붙인다.
try:
    from huggingface_hub import attach_huggingface_oauth, parse_huggingface_oauth
except Exception:  # noqa: BLE001
    attach_huggingface_oauth = parse_huggingface_oauth = None
_OAUTH_ATTACHED = False
if attach_huggingface_oauth is not None and os.environ.get("OAUTH_CLIENT_ID"):
    try:
        attach_huggingface_oauth(app)
        _OAUTH_ATTACHED = True
        print("[OAUTH] HF OAuth 연결됨")
    except Exception as _e:  # noqa: BLE001
        print(f"[OAUTH] 연결 실패: {_e}")
# 테스트·로컬 개발용(Space 에서는 쓰지 않는다): DEV_USER="아이디|이름" → 그 사람으로 로그인한 것처럼 동작, DEV_OAUTH=1 → 로그인 필수 화면 확인용
_DEV_USER = os.environ.get("DEV_USER", "")
if os.environ.get("DEV_OAUTH") == "1":
    _OAUTH_ATTACHED = True
# README 에 hf_oauth: true 가 있어 로그인이 붙으면, 로그인한 사람만 쓸 수 있다(SPAO V2 와 같은 방식). README 에 없으면 예전처럼 로그인 없이 동작한다.
# 로그인이 붙어 있어도 로그인 없이 쓰게 하려면 REQUIRE_LOGIN=false.
REQUIRE_LOGIN = os.environ.get("REQUIRE_LOGIN", "true").lower() in ("1", "true", "yes", "on")


def get_current_user(request: Request) -> Dict[str, Any]:
    anon = {"username": "anonymous", "name": "", "email": "", "is_logged_in": False}
    if _DEV_USER:
        uid, _, nm = _DEV_USER.partition("|")
        return {"username": uid, "name": nm or uid, "email": "", "is_logged_in": True}
    if not _OAUTH_ATTACHED or parse_huggingface_oauth is None:
        return anon
    try:
        info = parse_huggingface_oauth(request)
        ui = getattr(info, "user_info", None) if info else None
        if not ui:
            return anon
        return {"username": getattr(ui, "preferred_username", "") or "anonymous", "name": getattr(ui, "name", "") or "",
                "email": getattr(ui, "email", "") or "", "is_logged_in": True}
    except Exception as e:  # noqa: BLE001
        print(f"[OAUTH] 사용자 확인 실패: {e}")
        return anon


def _login_required() -> bool:
    return REQUIRE_LOGIN and _OAUTH_ATTACHED


def _require_login(user: Dict[str, Any]) -> None:
    if _login_required() and not user["is_logged_in"]:
        raise HTTPException(401, "로그인이 필요해요. 화면 위쪽의 '로그인'을 눌러 Hugging Face 로 로그인한 뒤 다시 시도해 주세요.")


_LOGIN_GATE_HTML = """<!DOCTYPE html><html lang="ko"><head><meta charset="utf-8"/>
<meta name="viewport" content="width=device-width, initial-scale=1"/><title>AI 얼굴 교체 · 로그인</title>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/gh/orioncactus/pretendard@v1.3.9/dist/web/static/pretendard.min.css"/>
<style>*{font-family:'Pretendard',-apple-system,sans-serif;box-sizing:border-box}body{margin:0;background:#1a1a1a;color:#f9fafb;min-height:100vh;display:flex;align-items:center;justify-content:center}
.card{width:min(420px,92vw);background:#1e1e1e;border:1px solid #2a2a2a;border-radius:20px;padding:36px 32px;text-align:center}
.logo{width:44px;height:44px;border-radius:12px;background:#3B82F6;display:inline-flex;align-items:center;justify-content:center;margin-bottom:18px}
h1{font-size:18px;margin:0 0 6px}p{font-size:13px;color:#9ca3af;line-height:1.6;margin:0 0 24px}
a.btn{display:block;background:#3B82F6;color:#fff;text-decoration:none;font-weight:600;font-size:14px;padding:13px;border-radius:12px}a.btn:hover{background:#2563EB}
.note{font-size:11.5px;color:#6b7280;margin-top:16px}</style></head><body><div class="card">
<div class="logo"><svg width="22" height="22" viewBox="0 0 24 24" fill="none" stroke="#fff" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M9.937 15.5A2 2 0 0 0 8.5 14.063l-6.135-1.582a.5.5 0 0 1 0-.962L8.5 9.936A2 2 0 0 0 9.937 8.5l1.582-6.135a.5.5 0 0 1 .963 0L14.063 8.5A2 2 0 0 0 15.5 9.937l6.135 1.581a.5.5 0 0 1 0 .964L15.5 14.063a2 2 0 0 0-1.437 1.437l-1.582 6.135a.5.5 0 0 1-.963 0z"/><path d="M20 3v4"/><path d="M22 5h-4"/><path d="M4 17v2"/><path d="M5 18H3"/></svg></div>
<h1>AI 얼굴 교체</h1><p>이랜드이노플 조직 멤버만 사용할 수 있습니다.<br/>Hugging Face 계정으로 로그인해 주세요.</p>
<a class="btn" id="login" href="/oauth/huggingface/login">🤗 Hugging Face로 로그인</a>
<div class="note" id="note">로그인 후 자동으로 돌아옵니다. 조직 멤버가 아니면 접근이 거부됩니다.</div></div>
<script>if(window.top!==window.self){var a=document.getElementById("login");a.target="_blank";a.rel="noopener";document.getElementById("note").textContent="새 탭에서 로그인한 뒤, 이 화면을 새로고침해 주세요. 조직 멤버가 아니면 접근이 거부됩니다.";}</script>
</body></html>"""

_SEM = threading.Semaphore(MAX_CONCURRENT)
_JOBS: Dict[str, Dict[str, Any]] = {}
_LOCK = threading.Lock()
_FACES: Dict[str, bytes] = {}     # faceId → 원본 바이트 (기본 얼굴 + 사용자가 교체한 얼굴)

STAGE_LABEL = {
    "queued_local": "대기 중",
    "uploading": "이미지 업로드 중",
    "submitting": "워크플로우 제출 중",
    "queued_waiting": "Comfy 대기열",
    "queued_limited": "Comfy 대기열 (동시 실행 제한)",
    "pending": "Comfy 대기열",
    "preparing": "GPU 준비 중",
    "executing": "생성 중",
    "running": "생성 중",
    "downloading": "결과 받는 중",
    "done": "완료",
    "error": "실패",
    "cancelled": "취소됨",
}


# ── 얼굴 ──────────────────────────────────────────────────────────────────────
def _load_default_faces() -> None:
    for g in GENDERS:
        path = os.path.join(FACE_DIR, f"{g}.jpg")
        if os.path.exists(path):
            with open(path, "rb") as f:
                _FACES[f"default-{g}"] = f.read()
        else:
            print(f"[FACE] 기본 얼굴이 없습니다: {path}")


_load_default_faces()


def _jpeg_thumb(data: bytes, max_side: int, quality: int = 85) -> bytes:
    img = Image.open(io.BytesIO(data))
    img.load()
    from PIL import ImageOps
    img = ImageOps.exif_transpose(img).convert("RGB")
    img.thumbnail((max_side, max_side), Image.LANCZOS)
    out = io.BytesIO()
    img.save(out, "JPEG", quality=quality)
    return out.getvalue()


@app.get("/api/swap/faces")
def faces_info():
    return {g: {"id": f"default-{g}", "available": f"default-{g}" in _FACES} for g in GENDERS}


@app.post("/api/swap/faces/{gender}")
async def upload_face(gender: str, file: UploadFile = File(...)):
    if gender not in GENDERS:
        raise HTTPException(400, "gender 는 male 또는 female 이어야 합니다.")
    data = await file.read()
    try:
        Image.open(io.BytesIO(data)).verify()
    except Exception:  # noqa: BLE001
        raise HTTPException(400, "이미지 파일이 아닙니다.")
    fid = f"up-{hashlib.sha256(data).hexdigest()[:16]}"
    _FACES[fid] = data
    return {"id": fid}


@app.get("/api/swap/face-img/{face_id}")
def face_img(face_id: str):
    data = _FACES.get(face_id)
    if not data:
        raise HTTPException(404, "얼굴 이미지를 찾을 수 없습니다.")
    return Response(_jpeg_thumb(data, 520), media_type="image/jpeg", headers={"Cache-Control": "no-cache"})


# ── 작업 ──────────────────────────────────────────────────────────────────────
def _update(job_id: str, **kw: Any) -> None:
    with _LOCK:
        if job_id in _JOBS:
            _JOBS[job_id].update(kw)


def _gc() -> None:
    """끝난 작업의 '진행 상태' 항목만 정리한다. 결과 파일은 history 가 계속 보관한다."""
    now = time.time()
    with _LOCK:
        for j in [j for j, v in _JOBS.items() if v.get("finished_at") and now - v["finished_at"] > JOB_TTL_SEC]:
            _JOBS.pop(j, None)


def _clean_rel(rel: str, fallback: str) -> str:
    """업로드된 상대경로 → 안전한 'a/b/c.jpg' 형태. 파일명은 그대로 유지."""
    rel = (rel or fallback or "image.png").replace("\\", "/")
    parts = [p for p in rel.split("/") if p not in ("", ".", "..")]
    return "/".join(parts) or "image.png"


_ORPHANS: set = set()      # 결과는 버렸지만 Cloud 에서 아직 돌고 있을 수 있는 '우리' 작업 (실행 중이라 중단하지 못한 것)


def _remote_cancel(prompt_ids: set) -> None:
    """Comfy Cloud 쪽 취소 (네트워크 호출이라 뒤에서 처리).

    interrupt 는 계정의 실행 중 작업 전부를 멈추므로, Cloud 에서 도는 작업이 전부 '우리 것'일 때만 쓴다.
    예전에 취소했지만 중단하지 못해 계속 도는 우리 작업(_ORPHANS)도 우리 것으로 본다.
    """
    own = set(prompt_ids) | set(_ORPHANS)
    for pid in prompt_ids:
        try:
            res = cc.cancel_job(pid, allowed_running=own)
        except Exception as e:  # noqa: BLE001
            print(f"[CANCEL] {pid} 취소 요청 실패: {e}")
            res = "soft"
        if res == "soft":
            _ORPHANS.add(pid)
        else:
            _ORPHANS.discard(pid)
            if res == "interrupted":
                _ORPHANS.clear()           # interrupt 는 실행 중이던 우리 작업을 모두 멈췄다
        with _LOCK:
            for j in _JOBS.values():
                if j.get("promptId") == pid:
                    j["remoteCancel"] = res
        print(f"[CANCEL] prompt {pid} → {res}")


def _input_gone(msg: str) -> bool:
    """Comfy Cloud 에 올려 둔 입력 파일이 없어졌을 때의 오류인지 (실제 Cloud 는 제출은 받아 주고 실행 단계에서 ImageDownloadError 로 실패시킨다)."""
    m = (msg or "").lower()
    return any(k in m for k in ("imagedownloaderror", "doesn't exist", "does not exist", "value_not_in_list", "not in list", "input file"))


def _run_job(job_id: str, image_bytes: Optional[bytes], face_bytes: bytes, settings: Dict[str, Any],
             main_name: Optional[str] = None, source_bytes: Optional[bytes] = None, remember_input: bool = True) -> None:
    """image_bytes: 새로 올린 원본. 재생성(regen)일 때는 None 이고, 대신 이전에 Comfy Cloud 에 올려 둔 입력 파일 이름(main_name)을 쓴다."""
    started = time.time()
    reused = main_name is not None
    with _LOCK:
        ev: threading.Event = _JOBS[job_id]["cancel"]
    cancelled = ev.is_set
    with _SEM:
        try:
            if cancelled():
                raise cc.ComfyCancelled()
            _update(job_id, stage="uploading")
            if main_name is None:
                main_name = cc.ensure_uploaded(image_bytes, "main")
            face_name = cc.ensure_uploaded(face_bytes, "face")
            if cancelled():
                raise cc.ComfyCancelled()

            _update(job_id, stage="submitting")
            wf = cc.build_workflow(main_name, face_name, settings)
            prompt_id = cc.submit(wf)
            _update(job_id, promptId=prompt_id, stage="queued_waiting")
            print(f"[JOB] {job_id} → prompt {prompt_id}")
            if cancelled():                                      # 제출하는 사이에 취소가 들어온 경우
                _remote_cancel({prompt_id})
                raise cc.ComfyCancelled()

            STATUS_TO_STAGE = {"waiting_to_dispatch": "queued_waiting", "pending": "queued_waiting",
                               "in_progress": "executing", "running": "executing"}

            def on_status(st: str) -> None:
                _update(job_id, stage=STATUS_TO_STAGE.get(st) or (st if st in STAGE_LABEL else "executing"))

            result = cc.wait_for_result(prompt_id, on_status=on_status, should_cancel=cancelled)
            _update(job_id, stage="downloading")
            Image.open(io.BytesIO(result)).verify()          # 이미지인지 확인
            elapsed = round(time.time() - started, 1)
            with _LOCK:
                j = dict(_JOBS.get(job_id) or {})
            history.record(job_id, {
                "rel_path": j.get("rel_path", ""), "filename": j.get("filename", ""), "gender": j.get("gender", ""),
                "elapsedSec": elapsed, "comfy_image": main_name if remember_input else None,   # 재생성에 쓰려고 입력 파일 이름을 기억(저화질 미리보기로 만든 건 기억하지 않는다)
                "user_username": (j.get("user") or {}).get("username", ""), "user_name": (j.get("user") or {}).get("name", ""),   # 대시보드용
                "regen": bool(j.get("regen") or j.get("regenOf")),
                "settings": {k: settings.get(k) for k in ("denoise", "prompt", "mask_main", "mask_ref") if settings.get(k) is not None},
            }, result, image_bytes or source_bytes)
            _update(job_id, status="done", stage="done", elapsedSec=elapsed, finished_at=time.time())
            print(f"[JOB] {job_id} done in {elapsed}s")
        except cc.ComfyCancelled:
            if ev.is_set():                      # 사용자가 취소한 경우: Comfy Cloud 쪽 취소 결과가 나온 뒤에 '취소됨'으로 알린다
                for _ in range(60):
                    with _LOCK:
                        j = _JOBS.get(job_id) or {}
                        if j.get("remoteCancel") or not j.get("promptId"):
                            break
                    time.sleep(0.25)
            print(f"[JOB] {job_id} cancelled")
            _update(job_id, status="cancelled", stage="cancelled", finished_at=time.time())
        except cc.ComfyError as e:
            print(f"[JOB] {job_id} error: {e}")
            gone = reused and _input_gone(str(e))
            _update(job_id, status="error", stage="error", finished_at=time.time(), errorCode="input_gone" if gone else None,
                    error="이전에 올려 둔 원본을 Comfy Cloud 에서 찾지 못했어요(보관 기간이 지났을 수 있어요)." if gone else str(e))
        except Exception as e:  # noqa: BLE001
            traceback.print_exc()
            _update(job_id, status="error", stage="error", error=f"예기치 못한 오류: {str(e)[:200]}", finished_at=time.time())


@app.post("/api/swap/jobs")
async def create_job(
    request: Request,
    image: UploadFile = File(...),
    rel_path: str = Form(""),
    gender: str = Form(...),
    face_id: str = Form(""),
    settings: str = Form("{}"),
    preview: str = Form(""),                 # "1" 이면 원본이 아니라 저장된 미리보기로 만든 작업 → 이 입력은 재생성에 쓰지 않는다
    regen: str = Form(""),                   # "1" 이면 마음에 안 들어 다시 만든 작업(통계에 재생성으로 기록)
):
    user = get_current_user(request)
    _require_login(user)
    if gender not in GENDERS:
        raise HTTPException(400, "gender 는 male 또는 female 이어야 합니다.")
    st = cc.status()
    if not st["ok"]:
        raise HTTPException(500, st["error"])
    data = await image.read()
    if not data:
        raise HTTPException(400, "빈 파일입니다.")
    face = _FACES.get(face_id or f"default-{gender}")
    if not face:
        raise HTTPException(400, f"{'남자' if gender == 'male' else '여자'} 얼굴 이미지가 없습니다. 얼굴을 먼저 올려주세요.")
    try:
        opts = json.loads(settings or "{}")
        if not isinstance(opts, dict):
            opts = {}
    except json.JSONDecodeError:
        opts = {}

    _gc()
    rel = _clean_rel(rel_path, image.filename or "")
    job_id = uuid.uuid4().hex[:12]
    with _LOCK:
        _JOBS[job_id] = {
            "status": "running", "stage": "queued_local", "created_at": time.time(),
            "rel_path": rel, "filename": rel.split("/")[-1], "gender": gender,
            "cancel": threading.Event(), "user": user, "regen": bool(regen),
        }
    threading.Thread(target=_run_job, args=(job_id, data, face, opts, None, None, not preview), daemon=True).start()
    return {"jobId": job_id}


@app.post("/api/swap/regen")
def regen_job(request: Request, from_id: str = Form(...), face_id: str = Form(""), settings: str = Form("{}")):
    """지난 결과(from_id)와 같은 원본으로 새 결과를 하나 더 만든다.
    원본을 다시 올리지 않고, 그때 Comfy Cloud 에 올려 둔 입력 파일을 그대로 쓴다 → 원본 화질 유지, 새로고침 후에도 동작."""
    user = get_current_user(request)
    _require_login(user)
    st = cc.status()
    if not st["ok"]:
        raise HTTPException(500, st["error"])
    e = history.entry(from_id)
    if not e:
        raise HTTPException(404, "지난 결과를 찾을 수 없습니다.")
    comfy_image = e.get("comfy_image")
    if not comfy_image:
        raise HTTPException(409, "이 결과는 원본 정보가 저장되지 않았어요(이전 버전으로 만든 결과).")
    gender = e.get("gender")
    if gender not in GENDERS:
        raise HTTPException(400, "성별 정보가 없는 결과예요.")
    face = _FACES.get(face_id or f"default-{gender}")
    if not face:
        raise HTTPException(400, f"{'남자' if gender == 'male' else '여자'} 얼굴 이미지가 없습니다.")
    try:
        opts = json.loads(settings or "{}")
        if not isinstance(opts, dict):
            opts = {}
    except json.JSONDecodeError:
        opts = {}
    src_path = history.path(from_id, "source")                       # 목록 비교용 미리보기만 이어받는다
    src = None
    if src_path:
        with open(src_path, "rb") as f:
            src = f.read()
    _gc()
    job_id = uuid.uuid4().hex[:12]
    rel = e.get("rel_path", "")
    with _LOCK:
        _JOBS[job_id] = {
            "status": "running", "stage": "queued_local", "created_at": time.time(),
            "rel_path": rel, "filename": e.get("filename") or rel.split("/")[-1], "gender": gender,
            "cancel": threading.Event(), "regenOf": from_id, "user": user, "regen": True,
        }
    threading.Thread(target=_run_job, args=(job_id, None, face, opts, comfy_image, src), daemon=True).start()
    return {"jobId": job_id}


@app.post("/api/swap/cancel")
def cancel_jobs(payload: Dict[str, Any] = Body(...)):
    """진행·대기 중인 작업을 취소한다. Comfy Cloud 쪽 취소는 뒤에서 처리한다."""
    ids = [str(i) for i in payload.get("ids", [])]
    prompt_ids = set()
    n = 0
    with _LOCK:
        for i in ids:
            j = _JOBS.get(i)
            if j and j.get("status") == "running":
                j["cancel"].set()
                n += 1
                if j.get("promptId"):
                    prompt_ids.add(j["promptId"])
    if prompt_ids:
        threading.Thread(target=_remote_cancel, args=(prompt_ids,), daemon=True).start()
    return {"cancelled": n}


@app.get("/api/swap/jobs/{job_id}")
def job_info(job_id: str):
    with _LOCK:
        job = _JOBS.get(job_id)
        snap = dict(job) if job else None
    if snap is None:
        e = history.entry(job_id)                  # 재시작 후에도 끝난 작업은 기록에 남아 있다
        if e:
            return {"status": "done", "stage": "done", "stageLabel": STAGE_LABEL["done"], "error": None, "elapsedSec": e.get("elapsedSec", 0)}
        raise HTTPException(404, "작업을 찾을 수 없습니다 (서버가 재시작되었을 수 있습니다).")
    stage = snap.get("stage", "")
    return {
        "status": snap["status"],
        "stage": stage,
        "stageLabel": STAGE_LABEL.get(stage, stage),
        "error": snap.get("error"),
        "errorCode": snap.get("errorCode"),
        "cancelMode": snap.get("remoteCancel"),      # removed | interrupted | soft(Cloud 에서는 이미 실행 중이라 끝까지 진행)
        "elapsedSec": round((snap.get("finished_at") or time.time()) - snap["created_at"], 1),
    }


# ── 결과 파일 (원본 파일명·확장자 유지) ───────────────────────────────────────
def _done_job(job_id: str) -> Dict[str, Any]:
    e = history.entry(job_id)
    path = history.path(job_id, "result") if e else None
    if not e or not path:
        raise HTTPException(404, "저장된 결과가 없습니다.")
    return {**e, "path": path}


def _encode_like_original(job: Dict[str, Any]) -> tuple[bytes, str, str]:
    """결과를 원본과 같은 파일명·확장자로 인코딩. (bytes, filename, media_type)

    jpg/jpeg → JPEG(q95) · png → PNG · webp → WEBP.
    그 외 확장자(bmp 등)는 파일명 줄기를 유지하고 .png 로 저장한다.
    """
    name = job["filename"]
    stem, ext = os.path.splitext(name)
    low = ext.lower()
    img = Image.open(job["path"])
    img.load()
    img = img.convert("RGB")
    out = io.BytesIO()
    if low in (".jpg", ".jpeg"):
        img.save(out, "JPEG", quality=95, subsampling=0, optimize=True)
        return out.getvalue(), name, "image/jpeg"
    if low == ".webp":
        img.save(out, "WEBP", quality=95)
        return out.getvalue(), name, "image/webp"
    img.save(out, "PNG", compress_level=6)
    return out.getvalue(), (name if low == ".png" else f"{stem}.png"), "image/png"


def _content_disposition(kind: str, filename: str) -> str:
    ascii_fallback = re.sub(r"[^A-Za-z0-9._-]", "_", unicodedata.normalize("NFKD", filename).encode("ascii", "ignore").decode()) or "result"
    return f"{kind}; filename=\"{ascii_fallback}\"; filename*=UTF-8''{quote(filename)}"


@app.get("/api/swap/jobs/{job_id}/thumb")
def job_thumb(job_id: str):
    p = history.path(job_id, "thumb")
    if not p:
        raise HTTPException(404, "썸네일이 없습니다.")
    return FileResponse(p, media_type="image/jpeg", headers={"Cache-Control": "max-age=3600"})


@app.get("/api/swap/jobs/{job_id}/source")
def job_source(job_id: str):
    p = history.path(job_id, "source")
    if not p:
        raise HTTPException(404, "원본 미리보기가 없습니다.")
    return FileResponse(p, media_type="image/jpeg", headers={"Cache-Control": "max-age=3600"})


@app.get("/api/swap/jobs/{job_id}/view")
def job_view(job_id: str):
    job = _done_job(job_id)
    data, name, mt = _encode_like_original(job)
    return Response(data, media_type=mt, headers={"Content-Disposition": _content_disposition("inline", name)})


@app.get("/api/swap/jobs/{job_id}/download")
def job_download(job_id: str):
    job = _done_job(job_id)
    data, name, mt = _encode_like_original(job)
    return Response(data, media_type=mt, headers={"Content-Disposition": _content_disposition("attachment", name)})


@app.post("/api/swap/zip")
def zip_download(ids: str = Form(...), name: str = Form("faceswap_results")):
    """선택한 완료 작업을 ZIP 으로. 폴더 구조와 파일명을 업로드한 그대로 유지한다."""
    wanted = [i for i in re.split(r"[,\s]+", ids) if i]
    if not wanted:
        raise HTTPException(400, "다운로드할 작업이 없습니다.")
    tmp = tempfile.TemporaryFile()
    used: set[str] = set()
    count = 0
    with zipfile.ZipFile(tmp, "w", zipfile.ZIP_STORED) as zf:
        for jid in wanted:
            try:
                job = _done_job(jid)
            except HTTPException:
                continue
            data, fname, _ = _encode_like_original(job)
            folder = "/".join(job["rel_path"].split("/")[:-1])
            arc = f"{folder}/{fname}" if folder else fname
            base, ext = os.path.splitext(arc)
            n = 2
            while arc in used:
                arc = f"{base} ({n}){ext}"
                n += 1
            used.add(arc)
            zf.writestr(arc, data)
            count += 1
    if count == 0:
        tmp.close()
        raise HTTPException(404, "완료된 결과가 없습니다.")
    tmp.seek(0)

    def _iter():
        try:
            while chunk := tmp.read(1024 * 1024):
                yield chunk
        finally:
            tmp.close()

    safe = re.sub(r'[\\/:*?"<>|]', "_", name).strip() or "faceswap_results"
    zip_name = f"{safe}.zip"
    return StreamingResponse(_iter(), media_type="application/zip",
                             headers={"Content-Disposition": _content_disposition("attachment", zip_name)})


# ── 지난 결과 ─────────────────────────────────────────────────────────────────
@app.get("/api/history")
def history_list(q: str = "", limit: int = 1000):
    items = history.list_entries(q, min(max(limit, 1), 3000))
    return {"items": [{**{k: e.get(k) for k in ("id", "rel_path", "filename", "folder", "gender", "date", "time", "elapsedSec")}, "canRegen": bool(e.get("comfy_image"))} for e in items],
            "total": len(items)}


@app.post("/api/history/hide")
def history_hide(payload: Dict[str, Any] = Body(...)):
    ids = [str(i) for i in payload.get("ids", [])]
    return {"hidden": history.hide(ids)}


CREDITS_PER_IMAGE = float(os.environ.get("CREDITS_PER_IMAGE", "8.6"))     # 대략치(장당 GPU 약 22초 × 0.39 크레딧/초). 대시보드의 '대략 크레딧'에만 쓴다.


@app.get("/api/dashboard")
def dashboard(request: Request, date_from: str = Query("", alias="from"), date_to: str = Query("", alias="to"), recent: int = 60):
    """생성 기록 대시보드: 지난 결과 기록(완료된 생성만)을 기간별로 집계한다. 목록에서 숨긴 것도 포함해서 센다."""
    _require_login(get_current_user(request))
    df, dt = (date_from or "")[:10], (date_to or "")[:10]
    rows = [e for e in history.all_entries() if (not df or e.get("date", "") >= df) and (not dt or e.get("date", "") <= dt)]
    days: Dict[str, Dict[str, Any]] = {}
    users: Dict[str, Dict[str, Any]] = {}
    folders: Dict[str, Dict[str, Any]] = {}
    files: set = set()
    hours = [0] * 24
    gender = {"male": 0, "female": 0}
    elapsed_all: List[float] = []
    regen_total = 0
    for e in rows:
        day = e.get("date", "")
        is_regen = bool(e.get("regen"))
        rel = unicodedata.normalize("NFC", e.get("rel_path") or e.get("filename") or "")
        stamp = e.get("created_at", "")
        d = days.setdefault(day, {"day": day, "gen": 0, "regen": 0})
        d["gen"] += 1
        d["regen"] += 1 if is_regen else 0
        regen_total += 1 if is_regen else 0
        files.add(rel)
        if e.get("gender") in gender:
            gender[e["gender"]] += 1
        try:
            hours[int((e.get("time") or "00:00")[:2])] += 1
        except ValueError:
            pass
        es = e.get("elapsedSec")
        if isinstance(es, (int, float)):
            elapsed_all.append(float(es))
        uk = e.get("user_username", "")                           # "" = 이 기능 이전 기록(미기록), "anonymous" = 로그인 없이 사용
        u = users.setdefault(uk, {"username": uk, "name": "", "gen": 0, "regen": 0, "files": set(), "days": set(), "elapsed": [], "last": ""})
        u["name"] = e.get("user_name") or u["name"]
        u["gen"] += 1
        u["regen"] += 1 if is_regen else 0
        u["files"].add(rel)
        u["days"].add(day)
        if isinstance(es, (int, float)):
            u["elapsed"].append(float(es))
        u["last"] = max(u["last"], stamp)
        fk = rel.rsplit("/", 1)[0] if "/" in rel else history.NO_FOLDER
        f = folders.setdefault(fk, {"folder": fk, "gen": 0, "files": set(), "last": ""})
        f["gen"] += 1
        f["files"].add(rel)
        f["last"] = max(f["last"], stamp)

    def avg(xs: List[float]) -> Optional[float]:
        return round(sum(xs) / len(xs), 1) if xs else None

    user_list = sorted(({"username": u["username"], "name": u["name"], "gen": u["gen"], "regen": u["regen"], "files": len(u["files"]),
                         "active_days": len(u["days"]), "avg_elapsed": avg(u["elapsed"]), "last": u["last"].replace("T", " ")[:16]} for u in users.values()),
                       key=lambda x: (-x["gen"], x["username"]))
    folder_list = sorted(({"folder": f["folder"], "gen": f["gen"], "files": len(f["files"]), "last": f["last"].replace("T", " ")[:16]} for f in folders.values()),
                         key=lambda x: (-x["gen"], x["folder"]))
    rows.sort(key=lambda e: e.get("created_at", ""), reverse=True)
    recent_list = [{"id": e["id"], "when": f"{e.get('date', '')} {e.get('time', '')}", "rel_path": e.get("rel_path", ""), "filename": e.get("filename", ""),
                    "gender": e.get("gender", ""), "elapsed": e.get("elapsedSec"), "regen": bool(e.get("regen")),
                    "username": e.get("user_username", ""), "name": e.get("user_name", "")} for e in rows[:max(1, min(recent, 200))]]
    st = history.status()
    all_days = sorted(d for d in days if d)
    return {
        "summary": {"gen": len(rows), "files": len(files), "regen": regen_total, "users": len([u for u in users if u and u != "anonymous"]), "days": len(all_days),
                    "avg_elapsed": avg(elapsed_all), "credits_est": round(len(rows) * CREDITS_PER_IMAGE),
                    "first_day": all_days[0] if all_days else "", "last_day": all_days[-1] if all_days else "",
                    "legacy_rows": users.get("", {}).get("gen", 0)},
        "daily": [days[d] for d in sorted(days)], "gender": gender, "hours": hours,
        "users": user_list, "folders": folder_list, "recent": recent_list,
        "filter": {"from": df, "to": dt}, "durable": bool(st.get("durable")), "remote": st.get("remote", ""),
    }


@app.get("/api/history/status")
def history_status():
    return history.status()


@app.on_event("shutdown")
def _on_shutdown() -> None:
    history.flush(20)


# ── 설정 / 상태 ───────────────────────────────────────────────────────────────
@app.get("/api/swap/defaults")
def defaults():
    try:
        return cc.workflow_defaults()
    except cc.ComfyError as e:
        raise HTTPException(500, str(e))


@app.get("/api/health")
def health():
    return {**cc.status(), "maxConcurrent": MAX_CONCURRENT, "oauth": _OAUTH_ATTACHED}


@app.get("/api/whoami")
def whoami(request: Request):
    """화면 위쪽의 로그인 표시용."""
    u = get_current_user(request)
    return {**u, "oauth_enabled": _OAUTH_ATTACHED, "login_required": _login_required(),
            "login_url": "/oauth/huggingface/login" if _OAUTH_ATTACHED else None,
            "logout_url": "/oauth/huggingface/logout" if _OAUTH_ATTACHED else None}


def _find_index() -> Optional[str]:
    """화면 파일 위치: 루트의 index.html 을 우선, 없으면 static/index.html. (둘 중 어디에 올려도 동작)"""
    for cand in (os.path.join(BASE_DIR, "index.html"), os.path.join(BASE_DIR, "static", "index.html")):
        if os.path.exists(cand):
            return cand
    return None


@app.get("/", include_in_schema=False)
@app.get("/index.html", include_in_schema=False)
def index_page(request: Request):
    if _login_required() and not get_current_user(request)["is_logged_in"]:
        return HTMLResponse(_LOGIN_GATE_HTML, status_code=401, headers={"Cache-Control": "no-cache"})
    path = _find_index()
    if not path:
        raise HTTPException(500, "index.html 파일이 없습니다. server.py 와 같은 폴더(또는 static/)에 올려주세요.")
    return FileResponse(path, media_type="text/html", headers={"Cache-Control": "no-cache"})