import os, uuid, time, glob, json, threading import psutil, httpx, aiofiles from pathlib import Path from datetime import datetime from fastapi import FastAPI, UploadFile, File, HTTPException, Depends, Form, Request from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse from typing import List from auth import (authenticate, create_access_token, init_users, require_auth, require_admin, require_stt, require_ocr, require_subtitle, list_users, create_user, update_user, delete_user) from tasks import celery_app, transcribe_task, subtitle_pipeline_task from ocr_tasks import ocr_task app = FastAPI(title="VoiceScript API") UPLOAD_DIR = os.getenv("UPLOAD_DIR", "/data/uploads") OUTPUT_DIR = os.getenv("OUTPUT_DIR", "/data/outputs") OLLAMA_URL = os.getenv("OLLAMA_URL", "http://192.168.0.126:11434") MAX_UPLOAD_BYTES = int(os.getenv("MAX_UPLOAD_MB", "500")) * 1024 * 1024 OUTPUT_KEEP_SECS = int(os.getenv("OUTPUT_KEEP_HOURS", "48")) * 3600 DATA_DIR = Path(UPLOAD_DIR).parent SETTINGS_FILE = DATA_DIR / "settings.json" HISTORY_FILE = DATA_DIR / "history.json" HISTORY_MAX = 500 os.makedirs(UPLOAD_DIR, exist_ok=True) os.makedirs(OUTPUT_DIR, exist_ok=True) AUDIO_EXT = {"mp3","mp4","wav","m4a","ogg","flac","aac","wma","webm", "mkv","avi","mov","ts","mts","m2ts","wmv","flv","h264","h265","hevc","264","265","m4v"} IMAGE_EXT = {"jpg","jpeg","png","bmp","tiff","tif","webp","gif"} _DEFAULT_SETTINGS = { "stt_ollama_model":"","ocr_ollama_model":"granite3.2-vision:latest", "cpu_threads":0,"stt_timeout":0,"ollama_timeout":600,"subtitle_timeout":600, "openrouter_url":"https://openrouter.ai/api/v1", "openrouter_api_key":"","openrouter_stt_model":"","openrouter_ocr_model":"", "groq_api_key":"","openai_api_key":"","default_stt_engine":"local", } _hist_lock = threading.Lock() # ── 설정 I/O ────────────────────────────────────────────────── def _load_settings() -> dict: if not SETTINGS_FILE.exists(): return dict(_DEFAULT_SETTINGS) with open(SETTINGS_FILE,"r",encoding="utf-8") as f: data=json.load(f) for k,v in _DEFAULT_SETTINGS.items(): data.setdefault(k,v) return data def _save_settings(data:dict): SETTINGS_FILE.parent.mkdir(parents=True,exist_ok=True) with open(SETTINGS_FILE,"w",encoding="utf-8") as f: json.dump(data,f,ensure_ascii=False,indent=2) def _mask(key:str)->str: if not key: return "" return key[:6]+"..."+(key[-4:] if len(key)>10 else "") def _keep(new_val:str, field:str, current:dict)->str: return new_val.strip() if new_val.strip() else current.get(field,"") # ── 이력 I/O ────────────────────────────────────────────────── def _load_history()->list: with _hist_lock: if not HISTORY_FILE.exists(): return [] try: with open(HISTORY_FILE,"r",encoding="utf-8") as f: return json.load(f) except: return [] def _write_history(h:list): HISTORY_FILE.parent.mkdir(parents=True,exist_ok=True) with open(HISTORY_FILE,"w",encoding="utf-8") as f: json.dump(h,f,ensure_ascii=False,indent=2) def append_history(record:dict): with _hist_lock: try: history=[] if HISTORY_FILE.exists(): with open(HISTORY_FILE,"r",encoding="utf-8") as f: history=json.load(f) history.insert(0,record); _write_history(history[:HISTORY_MAX]) except: pass def _update_history_by_task(task_id:str, result:dict, success:bool, error_msg:str=""): with _hist_lock: if not HISTORY_FILE.exists(): return try: with open(HISTORY_FILE,"r",encoding="utf-8") as f: history=json.load(f) for h in history: if h.get("task_id")!=task_id: continue if h.get("status") not in ("processing","cancelled"): break h["status"]="failed" if not success else "success" if not success: h["output"]={"error":error_msg[:500]} elif h["type"]=="stt": text=result.get("text","") h["output"]={ "filename":result.get("output_file",""), "language":result.get("language",""), "duration_s":result.get("duration",0), "segments":len(result.get("segments",[])), "text_preview":text[:200]+("…" if len(text)>200 else ""), "ollama_used":result.get("ollama_used",False), "ollama_model":result.get("ollama_model",""), "openrouter_used":result.get("openrouter_used",False), "openrouter_model":result.get("openrouter_model",""), "stt_engine":result.get("stt_engine","local"), } elif h["type"]=="subtitle": h["output"]={ "detected_language":result.get("detected_language",""), "duration_s":result.get("duration",0), "segment_count":result.get("segment_count",0), "stt_engine":result.get("stt_engine","local"), "translated":result.get("translated",False), "translate_to":result.get("translate_to",""), "refine_model":result.get("refine_model",""), "srt_orig":result.get("srt_orig",""), "vtt_orig":result.get("vtt_orig",""), "srt_trans":result.get("srt_trans",""), "vtt_trans":result.get("vtt_trans",""), } else: ft=result.get("full_text","") h["output"]={ "txt_file":result.get("txt_file",""), "xlsx_file":result.get("xlsx_file",""), "line_count":result.get("line_count",0), "table_count":len(result.get("tables",[])), "backend":result.get("backend",""), "ollama_model":result.get("ollama_model",""), "openrouter_model":result.get("openrouter_model",""), "text_preview":ft[:200]+("…" if len(ft)>200 else ""), } break _write_history(history) except: pass def delete_history_item(hid:str)->bool: with _hist_lock: if not HISTORY_FILE.exists(): return False try: with open(HISTORY_FILE,"r",encoding="utf-8") as f: history=json.load(f) new=[h for h in history if h.get("id")!=hid] if len(new)==len(history): return False _write_history(new); return True except: return False def clear_history(): with _hist_lock: if HISTORY_FILE.exists(): HISTORY_FILE.write_text("[]",encoding="utf-8") # ════════════════════════════════════════════════════════════════ # 시작 # ════════════════════════════════════════════════════════════════ @app.on_event("startup") async def on_startup(): init_users(); _cleanup_outputs() # ════════════════════════════════════════════════════════════════ # 인증 # ════════════════════════════════════════════════════════════════ @app.post("/api/login") def login(username:str=Form(...),password:str=Form(...)): user=authenticate(username,password) if not user: raise HTTPException(401,"아이디 또는 비밀번호가 올바르지 않습니다") return {"access_token":create_access_token(username),"token_type":"bearer"} @app.get("/api/me") def me(user:dict=Depends(require_auth)): return {"username":user["username"],"role":user.get("role","user"), "permissions":user.get("permissions",{"stt":False,"ocr":False,"subtitle":False})} # ════════════════════════════════════════════════════════════════ # 시스템 정보 # ════════════════════════════════════════════════════════════════ @app.get("/api/system") def system_info(user:dict=Depends(require_auth)): mem=psutil.virtual_memory();swap=psutil.swap_memory();s=_load_settings() return { "ram_total_gb":round(mem.total/1024**3,1),"ram_used_gb":round(mem.used/1024**3,1), "ram_avail_gb":round(mem.available/1024**3,1),"ram_percent":mem.percent, "swap_total_gb":round(swap.total/1024**3,1),"swap_used_gb":round(swap.used/1024**3,1), "cpu_logical":psutil.cpu_count(logical=True),"cpu_physical":psutil.cpu_count(logical=False), "cpu_percent":psutil.cpu_percent(interval=0.3), "cpu_threads_setting":s.get("cpu_threads",0), "stt_timeout":s.get("stt_timeout",0),"ollama_timeout":s.get("ollama_timeout",600), "subtitle_timeout":s.get("subtitle_timeout",600), } @app.get("/api/stt-engines") def stt_engines(user:dict=Depends(require_auth)): s=_load_settings() return { "local":{"available":True}, "groq":{"available":True,"key_set":bool(s.get("groq_api_key",""))}, "openai":{"available":True,"key_set":bool(s.get("openai_api_key",""))}, "default":s.get("default_stt_engine","local"), } # ════════════════════════════════════════════════════════════════ # 작업 상태 / 취소 # ════════════════════════════════════════════════════════════════ @app.get("/api/status/{task_id}") def get_status(task_id:str, user:dict=Depends(require_auth)): r=celery_app.AsyncResult(task_id) if r.state=="PENDING": return {"state":"pending","progress":0,"message":"대기 중..."} if r.state=="PROGRESS": m=r.info or {} return {"state":"progress","progress":m.get("progress",0), "step":m.get("step",0),"step_msg":m.get("step_msg",""), "message":m.get("message","처리 중...")} if r.state=="SUCCESS": _update_history_by_task(task_id,r.result or {},True) return {"state":"success","progress":100,**(r.result or {})} if r.state=="FAILURE": _update_history_by_task(task_id,{},False,str(r.info)) return {"state":"failure","progress":0,"message":str(r.info)} if r.state=="REVOKED": return {"state":"cancelled","progress":0,"message":"작업이 취소되었습니다"} return {"state":r.state.lower(),"progress":0} @app.post("/api/cancel/{task_id}") def cancel_task(task_id:str, user:dict=Depends(require_auth)): """작업 취소 (Celery revoke)""" try: celery_app.control.revoke(task_id, terminate=True, signal="SIGTERM") # 이력에 취소 표시 with _hist_lock: if HISTORY_FILE.exists(): with open(HISTORY_FILE,"r",encoding="utf-8") as f: history=json.load(f) for h in history: if h.get("task_id")==task_id and h.get("status")=="processing": h["status"]="cancelled" h["output"]={"error":"사용자가 취소했습니다"} break _write_history(history) return {"ok":True,"message":"취소 요청 전송됨"} except Exception as e: return {"ok":False,"message":str(e)} # ════════════════════════════════════════════════════════════════ # STT # ════════════════════════════════════════════════════════════════ async def _dispatch_stt(request,files,use_ollama,ollama_model,use_openrouter,openrouter_model, stt_engine,stt_language,user): s=_load_settings() _uo=use_ollama.lower()=="true"; _uor=use_openrouter.lower()=="true" if _uo and not ollama_model.strip(): ollama_model=s.get("stt_ollama_model","") if _uor and not openrouter_model.strip():openrouter_model=s.get("openrouter_stt_model","") if not stt_engine: stt_engine=s.get("default_stt_engine","local") results=[] for file in files: _check_size(request) ext=_ext(file.filename) if ext not in AUDIO_EXT: results.append({"error":f"{file.filename}: 지원하지 않는 형식","filename":file.filename}); continue file_id=str(uuid.uuid4()) save_path=os.path.join(UPLOAD_DIR,f"{file_id}.{ext}") await _save_upload(file,save_path); file_size=os.path.getsize(save_path) task=transcribe_task.delay( file_id,save_path,_uo,ollama_model,_uor,openrouter_model, s.get("openrouter_url",""),s.get("openrouter_api_key",""), stt_engine,s.get("groq_api_key",""),s.get("openai_api_key",""),stt_language or "", ) append_history({"id":file_id,"task_id":task.id,"type":"stt","status":"processing", "timestamp":datetime.now().strftime("%Y-%m-%d %H:%M:%S"),"username":user["username"], "input":{"filename":file.filename,"size_bytes":file_size,"format":ext.upper()}, "settings":{"model":os.getenv("WHISPER_MODEL","medium"), "language":stt_language or os.getenv("WHISPER_LANGUAGE","ko"), "compute_type":os.getenv("WHISPER_COMPUTE_TYPE","int8"), "cpu_threads":s.get("cpu_threads",0),"stt_engine":stt_engine, "use_ollama":_uo,"ollama_model":ollama_model if _uo else "", "use_openrouter":_uor,"openrouter_model":openrouter_model if _uor else ""}, "output":None}) results.append({"task_id":task.id,"file_id":file_id,"filename":file.filename}) return results @app.post("/api/transcribe") async def transcribe(request:Request,file:UploadFile=File(...), use_ollama:str=Form("false"),ollama_model:str=Form(""), use_openrouter:str=Form("false"),openrouter_model:str=Form(""), stt_engine:str=Form(""),stt_language:str=Form(""), user:dict=Depends(require_stt)): items=await _dispatch_stt(request,[file],use_ollama,ollama_model,use_openrouter,openrouter_model,stt_engine,stt_language,user) return items[0] @app.post("/api/transcribe/batch") async def transcribe_batch(request:Request,files:List[UploadFile]=File(...), use_ollama:str=Form("false"),ollama_model:str=Form(""), use_openrouter:str=Form("false"),openrouter_model:str=Form(""), stt_engine:str=Form(""),stt_language:str=Form(""), user:dict=Depends(require_stt)): if not files: raise HTTPException(400,"파일이 없습니다") if len(files)>20: raise HTTPException(400,"최대 20개까지") items=await _dispatch_stt(request,files,use_ollama,ollama_model,use_openrouter,openrouter_model,stt_engine,stt_language,user) return {"items":items,"total":len(items)} # ════════════════════════════════════════════════════════════════ # 자막 # ════════════════════════════════════════════════════════════════ @app.post("/api/subtitle") async def create_subtitle( request:Request, file:UploadFile=File(...), src_language:str=Form(""),subtitle_fmt:str=Form("srt"), stt_engine:str=Form("local"), refine_model:str=Form(""),refine_via:str=Form("ollama"), translate_to:str=Form(""),trans_model:str=Form(""),trans_via:str=Form("ollama"), user:dict=Depends(require_subtitle), ): _check_size(request) ext=_ext(file.filename) if ext not in AUDIO_EXT: raise HTTPException(400,"지원하지 않는 형식입니다") if subtitle_fmt not in ("srt","vtt","both"): subtitle_fmt="srt" s=_load_settings() if not stt_engine: stt_engine=s.get("default_stt_engine","local") if not refine_model.strip(): refine_model=(s.get("openrouter_stt_model","") if refine_via=="openrouter" else s.get("stt_ollama_model","")) if not trans_model.strip(): trans_model=(s.get("openrouter_stt_model","") if trans_via=="openrouter" else s.get("stt_ollama_model","")) file_id=str(uuid.uuid4()) save_path=os.path.join(UPLOAD_DIR,f"{file_id}.{ext}") await _save_upload(file,save_path) file_size=os.path.getsize(save_path) subtitle_timeout=int(s.get("subtitle_timeout",600)) task=subtitle_pipeline_task.delay( file_id,save_path,src_language,subtitle_fmt, stt_engine,s.get("groq_api_key",""),s.get("openai_api_key",""), refine_model,refine_via,translate_to,trans_model,trans_via, s.get("openrouter_url",""),s.get("openrouter_api_key",""), subtitle_timeout, ) append_history({"id":file_id,"task_id":task.id,"type":"subtitle","status":"processing", "timestamp":datetime.now().strftime("%Y-%m-%d %H:%M:%S"),"username":user["username"], "input":{"filename":file.filename,"size_bytes":file_size,"format":ext.upper()}, "settings":{"src_language":src_language or "auto","subtitle_fmt":subtitle_fmt, "stt_engine":stt_engine,"refine_model":refine_model,"refine_via":refine_via, "translate_to":translate_to,"trans_model":trans_model,"trans_via":trans_via, "subtitle_timeout":subtitle_timeout}, "output":None}) return {"task_id":task.id,"file_id":file_id,"filename":file.filename} # ════════════════════════════════════════════════════════════════ # OCR # ════════════════════════════════════════════════════════════════ async def _dispatch_ocr(request,files,mode,backend,ollama_model,openrouter_model,custom_prompt,user): if mode not in ("text","structure"): mode="text" if backend not in ("paddle","ollama","openrouter"): backend="paddle" s=_load_settings() if backend=="ollama" and not ollama_model.strip(): ollama_model=s.get("ocr_ollama_model","granite3.2-vision:latest") if backend=="openrouter" and not openrouter_model.strip():openrouter_model=s.get("openrouter_ocr_model","") results=[] for file in files: _check_size(request) ext=_ext(file.filename) if ext not in IMAGE_EXT: results.append({"error":f"{file.filename}: 지원하지 않는 형식","filename":file.filename}); continue file_id=str(uuid.uuid4()) save_path=os.path.join(UPLOAD_DIR,f"{file_id}.{ext}") await _save_upload(file,save_path); file_size=os.path.getsize(save_path) task=ocr_task.delay(file_id,save_path,mode,backend,ollama_model,openrouter_model, s.get("openrouter_url",""),s.get("openrouter_api_key",""),custom_prompt) append_history({"id":file_id,"task_id":task.id,"type":"ocr","status":"processing", "timestamp":datetime.now().strftime("%Y-%m-%d %H:%M:%S"),"username":user["username"], "input":{"filename":file.filename,"size_bytes":file_size,"format":ext.upper()}, "settings":{"backend":backend,"mode":mode,"ocr_lang":os.getenv("OCR_LANG","korean"), "ollama_model":ollama_model if backend=="ollama" else "", "openrouter_model":openrouter_model if backend=="openrouter" else "", "custom_prompt":custom_prompt[:200] if custom_prompt else ""}, "output":None}) results.append({"task_id":task.id,"file_id":file_id,"filename":file.filename}) return results @app.post("/api/ocr") async def ocr(request:Request,file:UploadFile=File(...), mode:str=Form("text"),backend:str=Form("paddle"), ollama_model:str=Form(""),openrouter_model:str=Form(""),custom_prompt:str=Form(""), user:dict=Depends(require_ocr)): items=await _dispatch_ocr(request,[file],mode,backend,ollama_model,openrouter_model,custom_prompt,user) return items[0] @app.post("/api/ocr/batch") async def ocr_batch(request:Request,files:List[UploadFile]=File(...), mode:str=Form("text"),backend:str=Form("paddle"), ollama_model:str=Form(""),openrouter_model:str=Form(""),custom_prompt:str=Form(""), user:dict=Depends(require_ocr)): if not files: raise HTTPException(400,"파일이 없습니다") if len(files)>20: raise HTTPException(400,"최대 20개까지") items=await _dispatch_ocr(request,files,mode,backend,ollama_model,openrouter_model,custom_prompt,user) return {"items":items,"total":len(items)} # ════════════════════════════════════════════════════════════════ # 이력 # ════════════════════════════════════════════════════════════════ @app.get("/api/history") def get_history(page:int=1,per_page:int=15,type_:str="",user:dict=Depends(require_auth)): history=_load_history() if user.get("role")!="admin": history=[h for h in history if h.get("username")==user["username"]] if type_ in ("stt","ocr","subtitle"): history=[h for h in history if h.get("type")==type_] total=len(history);start=(page-1)*per_page return {"total":total,"page":page,"per_page":per_page,"items":history[start:start+per_page]} @app.delete("/api/history/{history_id}") def delete_history(history_id:str,user:dict=Depends(require_auth)): if not delete_history_item(history_id): raise HTTPException(404,"이력을 찾을 수 없습니다") return {"ok":True} @app.delete("/api/history") def clear_all_history(user:dict=Depends(require_admin)): clear_history(); return {"ok":True} # ════════════════════════════════════════════════════════════════ # 다운로드 / Ollama / OpenRouter / 설정 / 관리자 # ════════════════════════════════════════════════════════════════ @app.get("/api/download/{filename}") def download(filename:str,user:dict=Depends(require_auth)): if ".." in filename or "/" in filename: raise HTTPException(400,"잘못된 파일명") path=os.path.join(OUTPUT_DIR,filename) if not os.path.exists(path): raise HTTPException(404,"파일을 찾을 수 없습니다") if filename.endswith(".xlsx"): media="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" elif filename.endswith(".vtt"): media="text/vtt" elif filename.endswith(".srt"): media="text/plain; charset=utf-8" else: media="text/plain; charset=utf-8" return FileResponse(path,media_type=media,filename=filename) @app.get("/api/ollama/models") def ollama_models(user:dict=Depends(require_auth)): try: resp=httpx.get(f"{OLLAMA_URL}/api/tags",timeout=8.0); resp.raise_for_status() return {"models":[m["name"] for m in resp.json().get("models",[])], "connected":True} except Exception as e: return {"models":[],"connected":False,"error":str(e)} @app.get("/api/openrouter/models") def openrouter_models(user:dict=Depends(require_auth)): s=_load_settings(); api_key=s.get("openrouter_api_key","") base_url=s.get("openrouter_url","https://openrouter.ai/api/v1").rstrip("/") if not api_key: return {"models":[],"vision_models":[],"text_models":[],"connected":False,"error":"API 키가 설정되지 않았습니다"} try: resp=httpx.get(f"{base_url}/models", headers={"Authorization":f"Bearer {api_key}","HTTP-Referer":"https://voicescript.local"},timeout=12.0) resp.raise_for_status() all_models=resp.json().get("data",[]) vision=[m["id"] for m in all_models if any(k in m["id"].lower() for k in ["vision","claude-3","gemini","gpt-4o","llava","pixtral","qwen-vl","deepseek-vl"])] text=[m["id"] for m in all_models if m["id"] not in vision] return {"models":[m["id"] for m in all_models],"vision_models":vision,"text_models":text, "connected":True,"total":len(all_models)} except httpx.HTTPStatusError as e: return {"models":[],"vision_models":[],"text_models":[],"connected":False,"error":f"HTTP {e.response.status_code}"} except Exception as e: return {"models":[],"vision_models":[],"text_models":[],"connected":False,"error":str(e)} @app.post("/api/openrouter/test") def openrouter_test(api_key:str=Form(...),base_url:str=Form("https://openrouter.ai/api/v1"),user:dict=Depends(require_auth)): try: resp=httpx.get(f"{base_url.rstrip('/')}/models", headers={"Authorization":f"Bearer {api_key}","HTTP-Referer":"https://voicescript.local"},timeout=10.0) resp.raise_for_status(); count=len(resp.json().get("data",[])) return {"ok":True,"message":f"연결 성공 — {count}개 모델 사용 가능"} except httpx.HTTPStatusError as e: return {"ok":False,"message":f"인증 실패 (HTTP {e.response.status_code})"} except Exception as e: return {"ok":False,"message":f"연결 실패: {str(e)}"} @app.get("/api/settings") def get_settings(user:dict=Depends(require_auth)): s=_load_settings(); result=dict(s) for field in ("openrouter_api_key","groq_api_key","openai_api_key"): result[field+"_masked"]=_mask(result.get(field,"")) result[field]="" return result @app.post("/api/settings") def save_settings_endpoint( stt_ollama_model:str=Form(""),ocr_ollama_model:str=Form(""), cpu_threads:str=Form("0"),stt_timeout:str=Form("0"), ollama_timeout:str=Form("600"),subtitle_timeout:str=Form("600"), openrouter_url:str=Form("https://openrouter.ai/api/v1"), openrouter_api_key:str=Form(""),openrouter_stt_model:str=Form(""),openrouter_ocr_model:str=Form(""), groq_api_key:str=Form(""),openai_api_key:str=Form(""), default_stt_engine:str=Form("local"), user:dict=Depends(require_auth), ): def _int(v,d): try: return max(0,int(v)) except: return d current=_load_settings() data={ "stt_ollama_model":stt_ollama_model,"ocr_ollama_model":ocr_ollama_model, "cpu_threads":_int(cpu_threads,0),"stt_timeout":_int(stt_timeout,0), "ollama_timeout":_int(ollama_timeout,600),"subtitle_timeout":_int(subtitle_timeout,600), "openrouter_url":openrouter_url.strip() or "https://openrouter.ai/api/v1", "openrouter_api_key":_keep(openrouter_api_key,"openrouter_api_key",current), "openrouter_stt_model":openrouter_stt_model,"openrouter_ocr_model":openrouter_ocr_model, "groq_api_key":_keep(groq_api_key,"groq_api_key",current), "openai_api_key":_keep(openai_api_key,"openai_api_key",current), "default_stt_engine":default_stt_engine or "local", } _save_settings(data) result={k:v for k,v in data.items() if not k.endswith("_api_key")} for f in ("openrouter_api_key","groq_api_key","openai_api_key"): result[f+"_masked"]=_mask(data.get(f,"")) return {"ok":True,"settings":result} @app.get("/api/admin/users") def admin_list_users(user:dict=Depends(require_admin)): return {"users":list_users()} @app.post("/api/admin/users") def admin_create_user(username:str=Form(...),password:str=Form(...), perm_stt:str=Form("false"),perm_ocr:str=Form("false"),perm_subtitle:str=Form("false"), allowed_stt_models:str=Form(""),allowed_ocr_models:str=Form(""), user:dict=Depends(require_admin)): def _p(s): return [m.strip() for m in s.split(",") if m.strip()] perms={"stt":perm_stt.lower()=="true","ocr":perm_ocr.lower()=="true", "subtitle":perm_subtitle.lower()=="true", "allowed_stt_models":_p(allowed_stt_models),"allowed_ocr_models":_p(allowed_ocr_models)} ok,msg=create_user(username,password,perms) if not ok: raise HTTPException(400,msg) return {"ok":True,"message":msg} @app.put("/api/admin/users/{username}") def admin_update_user(username:str, perm_stt:str=Form("false"),perm_ocr:str=Form("false"),perm_subtitle:str=Form("false"), password:str=Form(""),allowed_stt_models:str=Form(""),allowed_ocr_models:str=Form(""), user:dict=Depends(require_admin)): def _p(s): return [m.strip() for m in s.split(",") if m.strip()] perms={"stt":perm_stt.lower()=="true","ocr":perm_ocr.lower()=="true", "subtitle":perm_subtitle.lower()=="true", "allowed_stt_models":_p(allowed_stt_models),"allowed_ocr_models":_p(allowed_ocr_models)} ok,msg=update_user(username,perms,password or None) if not ok: raise HTTPException(400,msg) return {"ok":True,"message":msg} @app.delete("/api/admin/users/{username}") def admin_delete_user(username:str,user:dict=Depends(require_admin)): ok,msg=delete_user(username) if not ok: raise HTTPException(400,msg) return {"ok":True,"message":msg} @app.post("/api/cleanup") def cleanup(user:dict=Depends(require_auth)): return {"removed":_cleanup_outputs()} @app.get("/") async def index(): import pathlib path=pathlib.Path("static/index.html") resp=FileResponse(path,media_type="text/html") resp.headers["Cache-Control"]="no-cache, no-store, must-revalidate" resp.headers["Pragma"]="no-cache"; resp.headers["Expires"]="0" return resp app.mount("/",StaticFiles(directory="static",html=True),name="static") # ════════════════════════════════════════════════════════════════ # 유틸 # ════════════════════════════════════════════════════════════════ def _check_size(request:Request): cl=request.headers.get("content-length") if cl and int(cl)>MAX_UPLOAD_BYTES*20: raise HTTPException(413,"파일이 너무 큽니다") def _cleanup_outputs(): if OUTPUT_KEEP_SECS==0: return 0 cutoff=time.time()-OUTPUT_KEEP_SECS; removed=0 for f in glob.glob(os.path.join(OUTPUT_DIR,"*")): try: if os.path.getmtime(f)MAX_UPLOAD_BYTES: await f.close(); os.remove(path) raise HTTPException(413,f"파일이 너무 큽니다. 최대 {MAX_UPLOAD_BYTES//1024//1024}MB") await f.write(chunk)