AI 얼굴 교체
이랜드이노플 조직 멤버만 사용할 수 있습니다.
Hugging Face 계정으로 로그인해 주세요.
🤗 Hugging Face로 로그인
로그인 후 자동으로 돌아옵니다. 조직 멤버가 아니면 접근이 거부됩니다.
"""
_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"})