hehihoho3@gmail.com
6ab3d998e9
ytRanges()가 구간을 초 숫자로 /yt/analyze에 보내는데 다운로드 경로의 cut_youtube()→valid_time()은 MM:SS/HH:MM:SS만 통과시켜, 모든 분석이 다운로드 단계에서 "시간 형식 오류"로 죽고 있었다. 프런트는 입력칸에 이미 들어있는 fmtTime 포맷 문자열을 그대로 보내도록 되돌리고, 서버 (_parse_ranges)도 parse_time()으로 초 환산 후 valid_time이 통과하는 형식으로 재포맷해 프런트가 무엇을 보내든 다운로드 경로까지 안전하게 닿도록 방어를 추가했다. 스크래치패드 스크립트로 정상 케이스·구 버그 케이스·1시간 이상 구간까지 valid_time 통과를 확인했다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1215 lines
60 KiB
Python
1215 lines
60 KiB
Python
"""캡컷 에이전트 로컬 웹 서버 (FastAPI).
|
||
|
||
실행: python -m uvicorn server.app:app --port 8000
|
||
열기: http://127.0.0.1:8000
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import glob
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
from fastapi import FastAPI, File, Form, UploadFile
|
||
from fastapi.responses import HTMLResponse, JSONResponse, Response, StreamingResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
|
||
from capcut_agent.pipeline import (
|
||
process_bg_template, process_paste, paste_analyze, paste_draft,
|
||
paste_steps as pipeline_paste_steps, _remap_caps,
|
||
bg_analyze, bg_draft, bg_steps, _cards_by_cut, _card_paths,
|
||
)
|
||
from capcut_agent.paste import parse_paste, parse_time
|
||
from capcut_agent.draft import DEFAULT_DRAFT_ROOT, list_drafts, repair_layers
|
||
from capcut_agent import comments as hlab
|
||
from capcut_agent import recommend
|
||
from capcut_agent import plan as autoplan
|
||
from capcut_agent import prompts as prompt_store
|
||
from capcut_agent.correct import GeminiQuotaError, has_gemini_key
|
||
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
STATIC_DIR = os.path.join(BASE_DIR, "static")
|
||
UPLOAD_DIR = os.path.join(os.path.dirname(BASE_DIR), ".uploads")
|
||
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
||
COMMENTS_DIR = os.path.join(os.path.dirname(BASE_DIR), ".comments")
|
||
os.makedirs(COMMENTS_DIR, exist_ok=True)
|
||
|
||
app = FastAPI(title="캡컷 에이전트")
|
||
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
||
|
||
# job_id(content hash) → {path, draft_name, title_top, title_main, channel}
|
||
JOBS: dict[str, dict] = {}
|
||
|
||
# analysis_id → {"url": …} (분석은 SSE 1회성 — 결과는 브라우저가 들고 있음)
|
||
ANALYSES: dict[str, dict] = {}
|
||
|
||
# 📋 붙여넣기 탭(새 흐름) — analysis_id → {"state", "places", "orig", "payload"}
|
||
# /paste/stream 이 채우고 /paste/build 가 꺼내 쓴다(서버 재시작 시 소실 — 재분석 필요).
|
||
PSTATES: dict[str, dict] = {}
|
||
|
||
# ▶ 유튜브 구간 탭(새 흐름) — analysis_id → {"state"}
|
||
# /yt/stream 이 채우고 /yt/build 가 꺼내 쓴다(서버 재시작 시 소실 — 재분석 필요).
|
||
YSTATES: dict[str, dict] = {}
|
||
|
||
# 🤖 자동 탭 2단계(준비) — prepare_id → {"aid", "ids"}
|
||
# /auto/prepare(POST) 가 채우고 /auto/prepare/{pid}(SSE) 가 꺼내 쓴다.
|
||
# 준비된 개별 편집안의 다운로드·받아쓰기 상태는 PSTATES[f"{aid}:{id}"] 에 담긴다
|
||
# (📋 붙여넣기 탭과 같은 저장소를 공유 — /auto/build 가 그 값으로 paste_draft 만 돌린다).
|
||
PREPARES: dict[str, dict] = {}
|
||
|
||
|
||
_DEFAULT_CDIR = os.path.join(os.path.dirname(BASE_DIR), "댓글카드")
|
||
|
||
|
||
@app.get("/")
|
||
async def index() -> HTMLResponse:
|
||
with open(os.path.join(STATIC_DIR, "index.html"), encoding="utf-8") as f:
|
||
html = f.read()
|
||
# 댓글 카드 폴더 기본 경로 주입(그 PC 기준 실제 경로)
|
||
html = html.replace("__CDIR__", _DEFAULT_CDIR.replace("\\", "/"))
|
||
# ?v= 캐시 무력화 — StaticFiles 는 Cache-Control 을 안 붙여서 브라우저가 auto.js 를
|
||
# 옛 버전 그대로 들고 있는 일이 있었다(index.html 만 바뀌어 UI가 반만 갱신됨).
|
||
# 파일 mtime 을 버전으로 박아 파일이 바뀌면 URL 이 달라지게 한다.
|
||
mt = 0.0
|
||
for f in ("auto.js", "modern-screenshot.js"):
|
||
p = os.path.join(STATIC_DIR, f)
|
||
if os.path.isfile(p):
|
||
mt = max(mt, os.path.getmtime(p))
|
||
html = html.replace("__V__", str(int(mt)))
|
||
return HTMLResponse(html, headers={"Cache-Control": "no-store"})
|
||
|
||
|
||
def _scale(pct: str) -> float:
|
||
"""확대 퍼센트 문자열 → scale(0.5~3.0). 잘못되면 1.0(100%)."""
|
||
try:
|
||
return max(0.5, min(3.0, float(pct) / 100))
|
||
except (ValueError, TypeError):
|
||
return 1.0
|
||
|
||
|
||
def _truthy(v: str) -> bool:
|
||
return str(v).strip().lower() in ("1", "true", "on", "yes")
|
||
|
||
|
||
def _fmt_range_time(sec: float) -> str:
|
||
"""초 → `M:SS`(1시간 이상은 `H:MM:SS`). youtube.valid_time()이 강제하는 형식과
|
||
맞춰야 한다 — youtube._fmt_hms()는 밀리초까지 붙어(`H:MM:SS.mmm`) valid_time을
|
||
통과 못 하므로 여기 전용 포맷 함수를 따로 둔다."""
|
||
sec = max(0, round(sec))
|
||
h, rem = divmod(int(sec), 3600)
|
||
m, s = divmod(rem, 60)
|
||
return f"{h}:{m:02d}:{s:02d}" if h else f"{m}:{s:02d}"
|
||
|
||
|
||
def _parse_ranges(raw: str) -> list[tuple[str, str]]:
|
||
"""JSON [[시작,끝],...] → [(M:SS, M:SS)]. 항목은 "16:07" 문자열/967 같은 초 숫자
|
||
모두 허용해 parse_time()으로 초로 바꾼 뒤 valid_time()이 통과하는 M:SS/H:MM:SS로
|
||
재포맷한다.
|
||
|
||
⚠ 함정: 프런트(ytRanges())가 초 숫자를 보내던 시절 이 정규화 없이 그대로
|
||
저장했다가, 다운로드 경로의 cut_youtube()→valid_time()이 "967" 같은 순수
|
||
숫자 문자열을 시간 형식으로 인정하지 않아 매 분석이 다운로드 단계에서
|
||
"시간 형식 오류"로 죽었다(구간 탭 전면 불능). 여기서 형식을 통일해두면
|
||
프런트가 무엇을 보내든(문자열이든 숫자든) 다운로드 경로까지 안전하게 도달한다.
|
||
빈 값/파싱실패 항목은 건너뛴다.
|
||
"""
|
||
raw = (raw or "").strip()
|
||
if not raw:
|
||
return []
|
||
try:
|
||
arr = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return []
|
||
out: list[tuple[str, str]] = []
|
||
for item in arr if isinstance(arr, list) else []:
|
||
if isinstance(item, (list, tuple)) and len(item) == 2:
|
||
try:
|
||
s_sec, e_sec = parse_time(item[0]), parse_time(item[1])
|
||
except (ValueError, TypeError):
|
||
continue
|
||
out.append((_fmt_range_time(s_sec), _fmt_range_time(e_sec)))
|
||
return out
|
||
|
||
|
||
@app.post("/upload")
|
||
async def upload(
|
||
file: UploadFile = File(...),
|
||
title_top: str = Form(""),
|
||
title_main: str = Form(""),
|
||
channel: str = Form(""),
|
||
video_scale: str = Form("100"),
|
||
flip: str = Form(""),
|
||
scene: str = Form(""),
|
||
comments_dir: str = Form(""),
|
||
bg_white: str = Form(""),
|
||
) -> JSONResponse:
|
||
data = await file.read()
|
||
# content hash → 같은 영상 재업로드 시 캐시/멱등 (mtime 아님)
|
||
h = hashlib.sha1(data).hexdigest()[:12]
|
||
ext = os.path.splitext(file.filename or "")[1].lower() or ".mp4"
|
||
path = os.path.join(UPLOAD_DIR, h + ext)
|
||
if not os.path.exists(path):
|
||
with open(path, "wb") as f:
|
||
f.write(data)
|
||
base = os.path.splitext(os.path.basename(file.filename or "video"))[0]
|
||
safe = "".join(c for c in base if c.isalnum() or c in (" ", "_", "-")).strip() or "video"
|
||
JOBS[h] = {
|
||
"path": path, "draft_name": f"{safe}_{h}", "youtube": None,
|
||
"title_top": title_top, "title_main": title_main, "channel": channel,
|
||
"video_scale": _scale(video_scale), "flip": _truthy(flip),
|
||
"scene": _truthy(scene), "comments_dir": comments_dir, "bg_white": _truthy(bg_white),
|
||
}
|
||
return JSONResponse({"job_id": h, "draft_name": JOBS[h]["draft_name"]})
|
||
|
||
|
||
@app.post("/paste")
|
||
async def paste(
|
||
data: str = Form(...),
|
||
video_scale: str = Form("144"),
|
||
flip: str = Form(""),
|
||
scene: str = Form(""),
|
||
comments_dir: str = Form(""),
|
||
bg_white: str = Form(""),
|
||
remove_silence: str = Form(""),
|
||
asr_bottom: str = Form(""),
|
||
) -> JSONResponse:
|
||
"""붙여넣기(JSON) 편집안 → 작업 생성. asr_bottom=1 이면 하단 자막을 Whisper로 자동 생성."""
|
||
try:
|
||
payload = parse_paste(data)
|
||
except ValueError as e:
|
||
return JSONResponse({"error": str(e)}, 400)
|
||
sig = payload["url"] + "|" + "|".join(f"{s:.3f}-{e:.3f}" for s, e, _, _ in payload["cuts"])
|
||
h = hashlib.sha1(sig.encode()).hexdigest()[:12]
|
||
JOBS[h] = {
|
||
"paste": payload, "draft_name": f"paste_{h}",
|
||
"video_scale": _scale(video_scale), "flip": _truthy(flip), "scene": _truthy(scene),
|
||
"comments_dir": comments_dir, "bg_white": _truthy(bg_white),
|
||
"remove_silence": _truthy(remove_silence),
|
||
"asr_bottom": _truthy(asr_bottom),
|
||
}
|
||
return JSONResponse({"job_id": h, "cuts": len(payload["cuts"])})
|
||
|
||
|
||
@app.get("/stream/{job_id}")
|
||
async def stream(job_id: str) -> StreamingResponse:
|
||
job = JOBS.get(job_id)
|
||
|
||
async def gen():
|
||
if not job:
|
||
yield _sse({"type": "error", "message": "알 수 없는 작업입니다."})
|
||
return
|
||
try:
|
||
if job.get("paste_state"): # 📋 붙여넣기 탭(새 흐름) — 이미 분석된 상태로 드래프트만
|
||
# paste_draft 는 manifest 를 내지 않는다(analyze 쪽이 이미 냈던 스텝들과
|
||
# 별개 스트림이라 여기서 새로 내야 함) — 안 내면 step 이벤트만 홀로 와서
|
||
# 화면에 진행 표시가 아예 안 뜬다.
|
||
yield _sse({"type": "manifest",
|
||
"steps": [{"id": "draft", "label": "템플릿 드래프트 생성"}]})
|
||
stream_iter = paste_draft(
|
||
job["paste_state"],
|
||
video_scale=job.get("video_scale", 1.0),
|
||
flip_horizontal=job.get("flip", False),
|
||
scene_split=job.get("scene", False),
|
||
comments_dir=job.get("comments_dir", ""),
|
||
cards_fixed=job.get("cards_fixed", False),
|
||
card_cuts=job.get("card_cuts") or None,
|
||
bg_white=job.get("bg_white", False),
|
||
)
|
||
elif job.get("bg_state"): # ▶ 유튜브 구간 탭(새 흐름) — 이미 분석된 상태로 드래프트만
|
||
# paste_state 와 같은 이유로 draft 단독 manifest 를 여기서 새로 낸다.
|
||
yield _sse({"type": "manifest",
|
||
"steps": [{"id": "draft", "label": "템플릿 드래프트 생성"}]})
|
||
stream_iter = bg_draft(
|
||
job["bg_state"],
|
||
video_scale=job.get("video_scale", 1.0),
|
||
flip_horizontal=job.get("flip", False),
|
||
scene_split=job.get("scene", False),
|
||
comments_dir=job.get("comments_dir", ""),
|
||
cards_fixed=job.get("cards_fixed", False),
|
||
bg_white=job.get("bg_white", False),
|
||
comment_cards=job.get("comment_cards"),
|
||
)
|
||
elif job.get("paste"): # 붙여넣기(JSON) 모드
|
||
stream_iter = process_paste(
|
||
job["paste"], job["draft_name"],
|
||
video_scale=job.get("video_scale", 1.0),
|
||
flip_horizontal=job.get("flip", False),
|
||
scene_split=job.get("scene", False),
|
||
comments_dir=job.get("comments_dir", ""),
|
||
bg_white=job.get("bg_white", False),
|
||
remove_silence=job.get("remove_silence", False),
|
||
asr_bottom=job.get("asr_bottom", False),
|
||
name_suffix=job.get("name_suffix", ""),
|
||
cards_fixed=job.get("cards_fixed", False),
|
||
card_cuts=job.get("card_cuts") or None,
|
||
)
|
||
else:
|
||
stream_iter = process_bg_template(
|
||
job["path"], job["draft_name"],
|
||
title_top=job.get("title_top", ""),
|
||
title_main=job.get("title_main", ""),
|
||
channel=job.get("channel", ""),
|
||
video_scale=job.get("video_scale", 1.0),
|
||
flip_horizontal=job.get("flip", False),
|
||
scene_split=job.get("scene", False),
|
||
comments_dir=job.get("comments_dir", ""),
|
||
bg_white=job.get("bg_white", False),
|
||
youtube=job.get("youtube"),
|
||
cards_fixed=job.get("cards_fixed", False),
|
||
)
|
||
async for ev in stream_iter:
|
||
yield _sse(ev)
|
||
except Exception as exc: # noqa: BLE001 — 사용자에게 그대로 노출
|
||
yield _sse({"type": "error", "message": f"{type(exc).__name__}: {exc}"})
|
||
|
||
return StreamingResponse(gen(), media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no"})
|
||
|
||
|
||
@app.post("/yt/analyze")
|
||
async def yt_analyze(
|
||
url: str = Form(...),
|
||
ranges: str = Form(...),
|
||
title_top: str = Form(""),
|
||
title_main: str = Form(""),
|
||
channel: str = Form(""),
|
||
) -> JSONResponse:
|
||
"""▶ 유튜브 구간 탭(새 흐름) 1단계 — URL·구간 검증 후 분석 예약. 실제 작업은 /yt/stream 에서.
|
||
|
||
title_top/title_main/channel 도 여기서 받는다: bg_analyze 가 title_main 을
|
||
Gemini 자막 교정 힌트로 쓰고, /yt/stream 은 GET(SSE)이라 요청 본문이 없어
|
||
지금 안 받으면 넘길 곳이 없다. video_scale 등 렌더 전용 옵션은 /yt/build 에서.
|
||
"""
|
||
u = url.strip()
|
||
if not (u.startswith("http://") or u.startswith("https://")):
|
||
return JSONResponse({"error": "유튜브 주소를 입력하세요."}, 400)
|
||
rng = _parse_ranges(ranges)
|
||
if not rng:
|
||
return JSONResponse({"error": "구간을 하나 이상 입력하세요."}, 400)
|
||
aid = hashlib.sha1(("yt|" + u + "|" + ranges).encode()).hexdigest()[:12]
|
||
ANALYSES[aid] = {"url": u, "ranges": rng, "title_top": title_top,
|
||
"title_main": title_main, "channel": channel}
|
||
return JSONResponse({"analysis_id": aid})
|
||
|
||
|
||
@app.get("/yt/stream/{aid}")
|
||
async def yt_stream(aid: str) -> StreamingResponse:
|
||
"""▶ 유튜브 구간 탭(새 흐름) — 구간 다운로드·무음·받아쓰기 → h-lab 댓글 매칭 → 검토 화면용 result."""
|
||
a = ANALYSES.get(aid)
|
||
|
||
async def gen():
|
||
if not a:
|
||
yield _sse({"type": "error", "message": "알 수 없는 분석입니다."})
|
||
return
|
||
url = a["url"]
|
||
ranges = a["ranges"]
|
||
youtube = {"url": url, "ranges": ranges}
|
||
warnings: list[str] = []
|
||
# draft 스텝은 이 단계(analyze)에서 안 돈다 — /yt/build 때 별도 스트림으로 실행되므로
|
||
# 여기 manifest 에 넣으면 영원히 start 가 안 와 화면에 대기 상태로 멈춰 보인다.
|
||
yield _sse({"type": "manifest", "steps":
|
||
[s for s in bg_steps(youtube) if s["id"] != "draft"] +
|
||
[{"id": "comments", "label": "댓글 수집 (h-lab)"},
|
||
{"id": "recommend", "label": "컷별 댓글 추천"}]})
|
||
com_task = asyncio.create_task(asyncio.to_thread(hlab.fetch_comments, url))
|
||
yield _sse({"type": "step", "id": "comments", "status": "start"})
|
||
state = None
|
||
try:
|
||
async for ev in bg_analyze(None, "yt_" + aid,
|
||
title_top=a.get("title_top", ""),
|
||
title_main=a.get("title_main", ""),
|
||
channel=a.get("channel", ""),
|
||
youtube=youtube):
|
||
if ev.get("type") == "state": # 내부 전용 — 밖으로 흘리지 않는다
|
||
state = ev["state"]
|
||
continue
|
||
if ev.get("type") == "error":
|
||
com_task.cancel()
|
||
yield _sse(ev)
|
||
return
|
||
yield _sse(ev)
|
||
except Exception as exc: # noqa: BLE001 — 다운로드 등 실패는 드래프트 생성을 막는다
|
||
com_task.cancel()
|
||
yield _sse({"type": "error", "message": f"{type(exc).__name__}: {exc}"})
|
||
return
|
||
if state is None:
|
||
com_task.cancel()
|
||
yield _sse({"type": "error", "message": "분석 상태를 만들지 못했습니다."})
|
||
return
|
||
|
||
comments: list[dict] = []
|
||
try:
|
||
comments = await com_task
|
||
yield _sse({"type": "step", "id": "comments", "status": "done",
|
||
"detail": f"{len(comments)}개"})
|
||
except Exception as exc: # noqa: BLE001 — 댓글 실패는 진행을 막지 않는다
|
||
warnings.append(f"h-lab 연결 실패 — 댓글 없이 진행합니다 ({exc})")
|
||
yield _sse({"type": "step", "id": "comments", "status": "done",
|
||
"detail": "실패(생략)"})
|
||
|
||
yield _sse({"type": "step", "id": "recommend", "status": "start"})
|
||
# ⚠ 좌표계 둘: places/captions = 압축 타임라인(카드·자막 추출용),
|
||
# ranges_sec = 원본 영상 시각(⭐ 분:초 매칭용). 섞으면 카드가 통째로 어긋난다.
|
||
# bg_analyze 의 무음 제거는 항상 켜져 있어 state["places"] 는 이미 압축 좌표다
|
||
# (_remap_placements 를 여기서 또 부르면 두 번 압축된다 — 부르지 않는다).
|
||
places = state["places"]
|
||
orig = state["ranges_sec"]
|
||
try:
|
||
cuts, need, ai_failed = await asyncio.to_thread(
|
||
recommend.cuts_from_state, places, orig, state["captions"], comments)
|
||
if ai_failed:
|
||
warnings.append("AI 추천 실패(Gemini 응답 없음) — 분:초·자막단어·좋아요로 배정했습니다")
|
||
except Exception as exc: # noqa: BLE001
|
||
# cuts=None(빈 리스트 아님)이어야 화면이 "컷 0개"로 비지 않고 ⭐/➕ 폴백으로 넘어간다.
|
||
cuts, need = None, max(1, int(state["total"] // 3))
|
||
warnings.append(f"컷별 추천 실패 — 컷 정보 없이 진행합니다 ({type(exc).__name__}: {exc})")
|
||
yield _sse({"type": "step", "id": "recommend", "status": "done",
|
||
"detail": (f"{len(cuts)}컷 · 카드 {need}장" if cuts is not None
|
||
else f"폴백 · 카드 {need}장")})
|
||
|
||
YSTATES[aid] = {"state": state}
|
||
no_ts = [c for c in comments if not c["times"]]
|
||
matched = hlab.match_ranges(comments, orig) if comments else []
|
||
yield _sse({"type": "result", "cuts": cuts, "need": need,
|
||
"total": round(state["total"], 1),
|
||
"cutRanges": [{"start": s, "end": e} for s, e in orig],
|
||
"matched": matched,
|
||
"candidates": hlab.top_liked(no_ts, set(matched), len(no_ts)),
|
||
"comments": comments, "warnings": warnings,
|
||
# 검토 화면 입력칸을 채우기 위한 확정값 — channel 은 비워 뒀어도
|
||
# bg_analyze 내부에서 유튜브 채널명으로 자동인식됐을 수 있다(state 에 반영됨).
|
||
"title_top": state["title_top"], "title_main": state["title_main"],
|
||
"channel": state["channel"]})
|
||
|
||
return StreamingResponse(gen(), media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no"})
|
||
|
||
|
||
@app.post("/yt/build")
|
||
async def yt_build(
|
||
aid: str = Form(...),
|
||
cards: list[UploadFile] = File(default=[]),
|
||
card_cuts: str = Form(""),
|
||
title_top: str = Form(""),
|
||
title_main: str = Form(""),
|
||
channel: str = Form(""),
|
||
video_scale: str = Form("100"),
|
||
flip: str = Form(""),
|
||
scene: str = Form(""),
|
||
bg_white: str = Form(""),
|
||
cards_fixed: str = Form(""),
|
||
comments_dir: str = Form(""),
|
||
) -> JSONResponse:
|
||
"""▶ 유튜브 구간 탭(새 흐름) 3단계 — 보관한 상태로 드래프트만 만든다(다운로드·받아쓰기 다시 안 함).
|
||
|
||
title_top/title_main/channel 을 여기서도 받아 state 에 덮어쓴다: 검토 화면은 분석
|
||
(/yt/analyze) 시점 값을 그대로 입력칸에 채워 보여주고(자동인식된 channel 포함) 여기서
|
||
고칠 수 있게 하는데, bg_draft 는 title_*/channel 을 오직 state 에서만 읽으므로
|
||
(Task 6) 여기서 덮어쓰지 않으면 검토 화면에서 고친 값이 실제 드래프트에 반영되지
|
||
않는다 — 재분석(다운로드·ASR 재실행) 없이 오타만 고치는 게 이 필드들의 존재 이유다.
|
||
"""
|
||
st = YSTATES.get(aid)
|
||
if not st:
|
||
return JSONResponse({"error": "분석 결과가 만료됐습니다. 다시 분석해 주세요."}, 404)
|
||
state = st["state"]
|
||
state["title_top"] = title_top
|
||
state["title_main"] = title_main
|
||
state["channel"] = channel
|
||
|
||
# 카드별 소속 컷 — 값이 깨져도 빌드를 막지 않는다(없으면 전체 균등 배치로 폴백)
|
||
cut_map: list[int] = []
|
||
try:
|
||
parsed = json.loads(card_cuts) if card_cuts.strip() else []
|
||
if isinstance(parsed, list):
|
||
cut_map = [int(v) for v in parsed if isinstance(v, int) and not isinstance(v, bool)]
|
||
except (json.JSONDecodeError, ValueError, TypeError):
|
||
cut_map = []
|
||
|
||
sig = (aid + "|" + card_cuts + "|" + video_scale + "|" + flip + "|" + scene + "|"
|
||
+ bg_white + "|" + cards_fixed + "|" + str(len(cards)))
|
||
h = hashlib.sha1(sig.encode()).hexdigest()[:12]
|
||
# 카드가 오면 폴더 지정보다 우선(/paste/build 와 같은 규칙). 카드를 하나도 안 골랐으면
|
||
# 화면에서 넘어온 폴더 경로를 그대로 써서 예전(폴더 지정) 동작으로 하위호환한다.
|
||
cdir = comments_dir.strip()
|
||
if cards:
|
||
cdir = os.path.join(COMMENTS_DIR, h)
|
||
if os.path.isdir(cdir): # 재빌드 시 이전 카드 잔재 제거
|
||
shutil.rmtree(cdir, ignore_errors=True)
|
||
os.makedirs(cdir, exist_ok=True)
|
||
for i, f in enumerate(cards, 1):
|
||
body = await f.read()
|
||
with open(os.path.join(cdir, f"{i:03d}.png"), "wb") as out:
|
||
out.write(body)
|
||
|
||
# ⚠ state["places"] 는 bg_analyze 가 이미 무음 제거를 반영해 압축한 좌표다.
|
||
# 여기서 또 재매핑하지 않는다 — 카드 시간은 이 좌표를 그대로 컷 구간으로 쓴다.
|
||
cut_cards = _cards_by_cut(_card_paths(cdir), cut_map, state["places"], state["total"],
|
||
fixed=_truthy(cards_fixed))
|
||
|
||
JOBS[h] = {
|
||
"bg_state": state,
|
||
"video_scale": _scale(video_scale), "flip": _truthy(flip), "scene": _truthy(scene),
|
||
"comments_dir": cdir, "bg_white": _truthy(bg_white), "cards_fixed": _truthy(cards_fixed),
|
||
# cut_cards 가 비면 None → bg_draft 가 comments_dir 기준 전체 균등 배치로 폴백
|
||
# (/paste/build·paste_draft 의 `cards = cut_cards or _load_comment_cards(...)`와 동일 규칙).
|
||
"comment_cards": cut_cards or None,
|
||
}
|
||
return JSONResponse({"job_id": h})
|
||
|
||
|
||
@app.post("/auto/analyze")
|
||
async def auto_analyze(url: str = Form(""), mode: str = Form("full"),
|
||
data: str = Form("")) -> JSONResponse:
|
||
"""자동 탭 1단계 — 분석 예약. 실제 작업은 /auto/stream 에서 SSE 로.
|
||
|
||
mode: full = Step1+Step3 (AI 컷편집, 기본)
|
||
whole = Step1만 — 구간 5개를 통짜로(컷 편집 없음)
|
||
paste = 오팔 JSON 여러 개 붙여넣기 — Gemini 안 씀
|
||
"""
|
||
mode = mode if mode in ("full", "whole", "wpaste", "paste") else "full"
|
||
u = url.strip()
|
||
if mode == "paste":
|
||
if not data.strip():
|
||
return JSONResponse({"error": "오팔 JSON을 붙여넣으세요."}, 400)
|
||
elif mode == "wpaste":
|
||
# 구간 JSON 붙여넣기 — Gemini 안 씀. URL은 JSON에 없으므로 입력칸이 필수.
|
||
if not data.strip():
|
||
return JSONResponse({"error": "구간 JSON을 붙여넣으세요."}, 400)
|
||
if not (u.startswith("http://") or u.startswith("https://")):
|
||
return JSONResponse({"error": "유튜브 주소를 입력하세요. "
|
||
"(구간 JSON에는 URL이 없어 직접 넣어야 합니다)"}, 400)
|
||
else:
|
||
if not has_gemini_key():
|
||
return JSONResponse({"error": "Gemini 키가 없습니다. 프로젝트 루트의 "
|
||
".gemini_key 파일을 확인하세요. (그동안은 오팔 → "
|
||
"📋 오팔 JSON 방식을 쓰면 됩니다)"}, 400)
|
||
if not (u.startswith("http://") or u.startswith("https://")):
|
||
return JSONResponse({"error": "유튜브 주소를 입력하세요."}, 400)
|
||
aid = hashlib.sha1((mode + "|" + u + "|" + data).encode()).hexdigest()[:12]
|
||
ANALYSES[aid] = {"url": u, "mode": mode, "data": data}
|
||
return JSONResponse({"analysis_id": aid})
|
||
|
||
|
||
@app.get("/auto/stream/{aid}")
|
||
async def auto_stream(aid: str) -> StreamingResponse:
|
||
"""자동 탭 1단계 분석 SSE: Step1 → (Step3 ×N) → result(편집안·타이틀 후보만).
|
||
|
||
⚠ 댓글 매칭은 여기서 안 한다(Task 7) — 검토 화면의 ✕ 제외 뒤에 받아쓰기를 돌려야
|
||
낭비가 없으므로, 다운로드·받아쓰기·댓글 매칭은 전부 /auto/prepare 로 미뤘다.
|
||
`highlights` 는 여기서 만든 `paste`(편집안 JSON)·`titles`(제목 후보)까지만 담고,
|
||
/auto/prepare 가 그대로 꺼내 쓸 수 있게 `ANALYSES[aid]` 에도 남겨 둔다.
|
||
"""
|
||
a = ANALYSES.get(aid)
|
||
|
||
async def gen():
|
||
if not a:
|
||
yield _sse({"type": "error", "message": "알 수 없는 분석입니다."})
|
||
return
|
||
url = a["url"]
|
||
mode = a.get("mode", "full")
|
||
highlights: list[dict] = []
|
||
|
||
def _need(total: float) -> int:
|
||
return max(1, int(total // 3))
|
||
|
||
def _cuts_json(cuts) -> list[dict]:
|
||
return [{"start": s, "end": e, "bottom": b, "effect": f}
|
||
for s, e, b, f in cuts]
|
||
|
||
def _whole_hl(c: dict) -> dict:
|
||
"""구간 통짜 하이라이트 — 컷 편집 없이 구간 전체가 컷 1개.
|
||
제목은 비워 두고 `editable_title` 로 UI에서 직접 입력받는다.
|
||
whole(Gemini Step1) 과 wpaste(구간 JSON 붙여넣기) 가 공유."""
|
||
total = c["end"] - c["start"]
|
||
return {
|
||
"id": c["id"], "start": c["start"], "end": c["end"],
|
||
"reason": c["reason"],
|
||
"paste": {"url": url, "title_top": "", "title_main": "",
|
||
"channel": "",
|
||
"cuts": [{"start": c["start"], "end": c["end"],
|
||
"bottom": "", "effect": ""}]},
|
||
"titles": [], "editable_title": True,
|
||
"total": round(total, 1), "need": _need(total),
|
||
}
|
||
|
||
if mode == "paste":
|
||
# ── 오팔 JSON 여러 개 — Gemini 안 씀 ──
|
||
yield _sse({"type": "manifest", "steps": [
|
||
{"id": "parse", "label": "오팔 JSON 파싱"},
|
||
]})
|
||
yield _sse({"type": "step", "id": "parse", "status": "start"})
|
||
import re as _re
|
||
from capcut_agent.paste import split_json_objects
|
||
from capcut_agent.plan import parse_titles
|
||
raw_text = a.get("data", "")
|
||
blocks = split_json_objects(raw_text)
|
||
# 각 JSON 블록의 위치 → 블록 사이 텍스트에서 '타이틀 후보 5선' 추출용
|
||
spans, pos = [], 0
|
||
for b in blocks:
|
||
st = raw_text.find(b, pos)
|
||
spans.append((st, st + len(b)))
|
||
pos = st + len(b)
|
||
# URL 결정: 입력 칸 > 블록들 중 유효한 유튜브 ID(11자)가 있는 첫 URL.
|
||
# LLM이 url 을 ""/플레이스홀더로 주는 일이 흔해서 블록별 url 은 믿지 않는다.
|
||
vid_re = _re.compile(
|
||
r"(?:v=|youtu\.be/|shorts/|embed/)([A-Za-z0-9_-]{11})(?![A-Za-z0-9_-])")
|
||
parsed = []
|
||
best_url = url
|
||
for blk in blocks:
|
||
try:
|
||
d = json.loads(blk, strict=False)
|
||
except json.JSONDecodeError:
|
||
d = None
|
||
parsed.append(d)
|
||
if not best_url and isinstance(d, dict):
|
||
u2 = str(d.get("url") or "")
|
||
if u2.startswith("http") and vid_re.search(u2):
|
||
best_url = u2
|
||
yield _sse({"type": "log", "msg": f"URL 자동 인식: {u2}"})
|
||
if not best_url:
|
||
yield _sse({"type": "error",
|
||
"message": "유튜브 URL을 알 수 없습니다 — JSON들의 url이 비어 있거나 "
|
||
"플레이스홀더입니다. 위 유튜브 URL 칸을 채우고 다시 시도하세요."})
|
||
return
|
||
for i, d in enumerate(parsed, 1):
|
||
if not isinstance(d, dict):
|
||
yield _sse({"type": "log", "msg": f"{i}번 JSON 무시: 형식 오류"})
|
||
continue
|
||
d["url"] = best_url # ""/플레이스홀더 무시하고 검증 전에 강제 주입
|
||
try:
|
||
p = parse_paste(d)
|
||
except ValueError as e:
|
||
yield _sse({"type": "log", "msg": f"{i}번 JSON 무시: {e}"})
|
||
continue
|
||
# 이 JSON 뒤 ~ 다음 JSON 앞 텍스트의 타이틀 후보를 이 편집안에 연결
|
||
tail_end = spans[i][0] if i < len(spans) else len(raw_text)
|
||
titles = parse_titles(raw_text[spans[i - 1][1]:tail_end])
|
||
total = sum(e - s for s, e, _, _ in p["cuts"])
|
||
highlights.append({
|
||
"id": len(highlights) + 1,
|
||
"start": min(s for s, _, _, _ in p["cuts"]),
|
||
"end": max(e for _, e, _, _ in p["cuts"]),
|
||
"reason": (p["title_top"] + " / " + p["title_main"]).strip(" /"),
|
||
"paste": {"url": p["url"], "title_top": p["title_top"],
|
||
"title_main": p["title_main"], "channel": p["channel"],
|
||
"cuts": _cuts_json(p["cuts"])},
|
||
"titles": titles, "total": round(total, 1), "need": _need(total),
|
||
})
|
||
if not highlights:
|
||
yield _sse({"type": "error",
|
||
"message": "붙여넣은 텍스트에서 유효한 편집안 JSON을 찾지 못했습니다."})
|
||
return
|
||
url = best_url
|
||
yield _sse({"type": "step", "id": "parse", "status": "done",
|
||
"detail": f"{len(highlights)}개 편집안"})
|
||
elif mode == "wpaste":
|
||
# ── 구간 JSON 붙여넣기 — Gemini 안 씀. 구간 5개를 그대로 통짜로 ──
|
||
yield _sse({"type": "manifest", "steps": [
|
||
{"id": "parse", "label": "구간 JSON 파싱"},
|
||
]})
|
||
yield _sse({"type": "step", "id": "parse", "status": "start"})
|
||
try:
|
||
cands = autoplan.parse_candidates(a.get("data", ""), src="구간 JSON")
|
||
except Exception as exc: # noqa: BLE001
|
||
yield _sse({"type": "error", "message": str(exc)})
|
||
return
|
||
highlights.extend(_whole_hl(c) for c in cands)
|
||
yield _sse({"type": "step", "id": "parse", "status": "done",
|
||
"detail": f"{len(highlights)}개 구간"})
|
||
else:
|
||
steps = [{"id": "step1", "label": "하이라이트 구간 선정 (Gemini)"}]
|
||
if mode == "full":
|
||
steps.append({"id": "step3", "label": "편집안 생성 (Gemini, 구간별 동시)"})
|
||
yield _sse({"type": "manifest", "steps": steps})
|
||
# ── Step 1 ──
|
||
yield _sse({"type": "step", "id": "step1", "status": "start"})
|
||
try:
|
||
cands = await asyncio.to_thread(autoplan.select_highlights, url)
|
||
except Exception as exc: # noqa: BLE001
|
||
yield _sse({"type": "error",
|
||
"message": f"Step 1 실패 — {type(exc).__name__}: {exc}\n"
|
||
"오팔 → 📋 오팔 JSON 방식으로도 만들 수 있습니다."})
|
||
return
|
||
yield _sse({"type": "step", "id": "step1", "status": "done",
|
||
"detail": f"{len(cands)}개 구간"})
|
||
|
||
if mode == "whole":
|
||
# ── 구간 통짜 — 컷 편집 없이 구간 전체가 컷 1개 ──
|
||
highlights.extend(_whole_hl(c) for c in cands)
|
||
else:
|
||
# ── Step 3 (동시) ──
|
||
yield _sse({"type": "step", "id": "step3", "status": "start"})
|
||
|
||
async def _plan_one(i: int, c: dict):
|
||
# 429는 시차를 두고 재시도(동시 재충돌 방지: 시도×15s + 구간×5s)
|
||
for attempt in range(1, 4):
|
||
try:
|
||
return await asyncio.to_thread(
|
||
autoplan.edit_plan, url, c["start"], c["end"])
|
||
except GeminiQuotaError:
|
||
if attempt == 3:
|
||
raise
|
||
await asyncio.sleep(15 * attempt + i * 5)
|
||
|
||
plan_tasks = [asyncio.create_task(_plan_one(i, c))
|
||
for i, c in enumerate(cands)]
|
||
for c, t in zip(cands, plan_tasks):
|
||
hl = {"id": c["id"], "start": c["start"], "end": c["end"],
|
||
"reason": c["reason"]}
|
||
try:
|
||
r = await t
|
||
p = r["paste"]
|
||
total = sum(e - s for s, e, _, _ in p["cuts"])
|
||
hl.update({
|
||
"paste": {"url": p["url"], "title_top": p["title_top"],
|
||
"title_main": p["title_main"],
|
||
"channel": p["channel"],
|
||
"cuts": _cuts_json(p["cuts"])},
|
||
"titles": r["titles"],
|
||
"total": round(total, 1),
|
||
"need": _need(total),
|
||
})
|
||
if r["time_note"]:
|
||
yield _sse({"type": "log",
|
||
"msg": f"ID {c['id']}: {r['time_note']}"})
|
||
yield _sse({"type": "log",
|
||
"msg": f"ID {c['id']} 편집안 완료 — 컷 "
|
||
f"{len(p['cuts'])}개 · {total:.1f}초"})
|
||
except Exception as exc: # noqa: BLE001
|
||
hl["error"] = f"{type(exc).__name__}: {exc}"
|
||
yield _sse({"type": "log",
|
||
"msg": f"ID {c['id']} 실패: {hl['error']}"})
|
||
highlights.append(hl)
|
||
ok = sum(1 for h in highlights if "paste" in h)
|
||
yield _sse({"type": "step", "id": "step3", "status": "done",
|
||
"detail": f"{ok}/{len(highlights)}개 성공"})
|
||
|
||
# 댓글 매칭·다운로드·받아쓰기는 여기서 안 한다(Task 7) — /auto/prepare 가
|
||
# 검토 화면에서 제외(✕)하지 않은 ID만 이어받아 돌린다. url 은 paste/wpaste 모드에서
|
||
# best_url 로 바뀌었을 수 있어(오팔 JSON 안 url) 여기서 다시 저장해 둔다.
|
||
a["highlights"] = highlights
|
||
a["url"] = url
|
||
yield _sse({"type": "result", "highlights": highlights})
|
||
|
||
return StreamingResponse(gen(), media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no"})
|
||
|
||
|
||
@app.post("/auto/prepare")
|
||
async def auto_prepare(aid: str = Form(...), ids: str = Form(...)) -> JSONResponse:
|
||
"""자동 탭 2단계 — 검토 화면에서 제외하지 않은 ID만 준비 예약. 실제 작업은 /auto/prepare/{pid} 에서.
|
||
|
||
ids: 남길 하이라이트 id의 JSON 배열(예: [1,2,4]) — ✕ 로 제외된 ID는 여기 안 들어온다.
|
||
"""
|
||
a = ANALYSES.get(aid)
|
||
if not a or not a.get("highlights"):
|
||
return JSONResponse({"error": "분석 결과가 만료됐습니다. 다시 분석해 주세요."}, 404)
|
||
try:
|
||
id_list = json.loads(ids)
|
||
if not isinstance(id_list, list) or not id_list:
|
||
raise ValueError
|
||
except (json.JSONDecodeError, ValueError):
|
||
return JSONResponse({"error": "준비할 ID 목록이 올바르지 않습니다."}, 400)
|
||
pid = hashlib.sha1((aid + "|" + ids).encode()).hexdigest()[:12]
|
||
PREPARES[pid] = {"aid": aid, "ids": id_list}
|
||
return JSONResponse({"prepare_id": pid})
|
||
|
||
|
||
@app.get("/auto/prepare/{pid}")
|
||
async def auto_prepare_stream(pid: str) -> StreamingResponse:
|
||
"""자동 탭 2단계 SSE — 남은 ID만 순차로 다운로드·받아쓰기 + 댓글 매칭.
|
||
|
||
⚠ 순차로 돌린다(전역 방침) — yt-dlp·ffmpeg·Whisper 가 CPU 를 다 써서 동시에 여러 개를
|
||
돌리면 서로 느려지기만 한다. ID 하나 실패는 그 ID만 건너뛰고 나머지는 계속 진행한다
|
||
(다운로드 실패 말고는 드래프트 생성을 막지 않는다는 전역 원칙과 같은 이유).
|
||
"""
|
||
p = PREPARES.get(pid)
|
||
|
||
async def gen():
|
||
if not p:
|
||
yield _sse({"type": "error", "message": "알 수 없는 준비 요청입니다."})
|
||
return
|
||
aid = p["aid"]
|
||
a = ANALYSES.get(aid)
|
||
if not a or not a.get("highlights"):
|
||
yield _sse({"type": "error", "message": "분석 결과가 만료됐습니다. 다시 분석해 주세요."})
|
||
return
|
||
by_id = {h["id"]: h for h in a["highlights"] if "paste" in h}
|
||
targets = [by_id[i] for i in p["ids"] if i in by_id]
|
||
if not targets:
|
||
yield _sse({"type": "error", "message": "준비할 편집안이 없습니다."})
|
||
return
|
||
url = a.get("url", "")
|
||
warnings: list[str] = []
|
||
|
||
yield _sse({"type": "manifest", "steps": [
|
||
{"id": "comments", "label": "댓글 수집 (h-lab)"},
|
||
{"id": "prepare", "label": "ID별 순차 준비 (다운로드·받아쓰기)"},
|
||
]})
|
||
|
||
yield _sse({"type": "step", "id": "comments", "status": "start"})
|
||
comments: list[dict] = []
|
||
try:
|
||
comments = await asyncio.to_thread(hlab.fetch_comments, url)
|
||
yield _sse({"type": "step", "id": "comments", "status": "done",
|
||
"detail": f"{len(comments)}개"})
|
||
except Exception as exc: # noqa: BLE001 — 댓글 실패는 진행을 막지 않는다
|
||
warnings.append(f"h-lab 연결 실패 — 댓글 없이 진행합니다 ({exc})")
|
||
yield _sse({"type": "step", "id": "comments", "status": "done",
|
||
"detail": "실패(생략)"})
|
||
|
||
yield _sse({"type": "step", "id": "prepare", "status": "start"})
|
||
highlights_out: list[dict] = []
|
||
all_ranges: list[tuple] = []
|
||
# ⚠ 순차로 돌려라 — yt-dlp·ffmpeg·Whisper 가 CPU 를 다 쓴다(전역 방침).
|
||
for h in targets:
|
||
hid = h["id"]
|
||
state = None
|
||
failed = False
|
||
# ⚠ paste_analyze 내부에서 draft_name = _safe_name(title) or draft_name 로
|
||
# 영상 제목이 항상 draft_name 을 덮어쓴다(_safe_name 은 거의 항상 truthy) —
|
||
# 여기서 넘기는 f"auto_{aid}_{hid}" 는 그래서 절대 안 쓰인다. 같은 URL(한 aid)의
|
||
# 하이라이트 5개는 제목이 똑같으니 name_suffix 없이는 draft_name 이 전부 같아져
|
||
# build_bg_template_draft 의 allow_replace=True 가 뒤엣것으로 앞을 덮어쓴다
|
||
# (옛 /auto/build 의 tag→name_suffix 메커니즘 — 여기서도 반드시 넘겨야 한다).
|
||
safe_tag = "".join(ch for ch in f"하이라이트{hid}" if ch.isalnum() or ch in "-_")[:20]
|
||
try:
|
||
async for ev in paste_analyze(h["paste"], f"auto_{aid}_{hid}",
|
||
remove_silence=True, asr_bottom=True,
|
||
name_suffix=safe_tag):
|
||
if ev.get("type") == "state": # 내부 전용 — 밖으로 흘리지 않는다
|
||
state = ev["state"]
|
||
continue
|
||
if ev.get("type") == "log":
|
||
yield _sse({"type": "log", "msg": f"ID {hid}: {ev.get('msg')}"})
|
||
continue
|
||
yield _sse(ev) # step(download/asr) — manifest 밖 id 는 화면에서 무시됨
|
||
except Exception as exc: # noqa: BLE001 — 이 ID만 건너뛰고 나머지는 계속
|
||
failed = True
|
||
warnings.append(f"ID {hid} 준비 실패 — {type(exc).__name__}: {exc}")
|
||
yield _sse({"type": "log", "msg": f"ID {hid} 실패: {type(exc).__name__}: {exc}"})
|
||
if failed or state is None:
|
||
if not failed:
|
||
warnings.append(f"ID {hid} 준비 실패 — 분석 상태를 만들지 못했습니다")
|
||
continue
|
||
|
||
# ⚠ 좌표계 둘: places = 압축 타임라인(카드·자막 추출용),
|
||
# orig = 원본 영상 시각(⭐ 분:초 매칭용). 섞으면 카드가 통째로 어긋난다.
|
||
places = state["card_places"]
|
||
orig = [(s, e) for s, e, _, _ in state["cuts"]]
|
||
all_ranges.extend(orig)
|
||
try:
|
||
cuts, need, ai_failed = await asyncio.to_thread(
|
||
recommend.cuts_from_state, places, orig, state["bottom_caps"], comments)
|
||
if ai_failed:
|
||
warnings.append(f"ID {hid} AI 추천 실패(Gemini 응답 없음) — "
|
||
"분:초·자막단어·좋아요로 배정했습니다")
|
||
except Exception as exc: # noqa: BLE001
|
||
cuts, need = None, max(1, int(state["timeline_dur"] // 3))
|
||
warnings.append(f"ID {hid} 컷별 추천 실패 — 컷 정보 없이 진행합니다 "
|
||
f"({type(exc).__name__}: {exc})")
|
||
|
||
PSTATES[f"{aid}:{hid}"] = {"state": state, "places": places, "orig": orig,
|
||
"payload": h["paste"]}
|
||
matched = hlab.match_ranges(comments, orig) if comments else []
|
||
highlights_out.append({
|
||
"id": hid, "cuts": cuts, "need": need,
|
||
"cutRanges": [{"start": s, "end": e} for s, e in orig],
|
||
"matched": matched,
|
||
})
|
||
yield _sse({"type": "log", "msg": f"ID {hid} 준비 완료"})
|
||
|
||
yield _sse({"type": "step", "id": "prepare", "status": "done",
|
||
"detail": f"{len(highlights_out)}/{len(targets)}개 완료"})
|
||
|
||
no_ts = [c for c in comments if not c["times"]]
|
||
matched_all = hlab.match_ranges(comments, all_ranges) if all_ranges else []
|
||
yield _sse({"type": "result", "highlights": highlights_out,
|
||
"comments": comments,
|
||
"candidates": hlab.top_liked(no_ts, set(matched_all), len(no_ts)),
|
||
"warnings": warnings})
|
||
|
||
return StreamingResponse(gen(), media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no"})
|
||
|
||
|
||
@app.get("/auto/avatar")
|
||
async def auto_avatar(url: str) -> Response:
|
||
"""프로필 이미지 동일 출처 프록시(canvas taint 회피). 구글 도메인만(SSRF 방지)."""
|
||
try:
|
||
host = (urllib.parse.urlparse(url).hostname or "").lower()
|
||
except ValueError:
|
||
host = ""
|
||
if not (host.endswith("ggpht.com") or host.endswith("googleusercontent.com")):
|
||
return Response(status_code=400)
|
||
|
||
def _get():
|
||
with urllib.request.urlopen(url, timeout=15) as r:
|
||
return r.read(), r.headers.get("Content-Type") or "image/jpeg"
|
||
|
||
try:
|
||
data, ct = await asyncio.to_thread(_get)
|
||
except Exception: # noqa: BLE001
|
||
return Response(status_code=502)
|
||
return Response(content=data, media_type=ct,
|
||
headers={"Cache-Control": "public, max-age=21600"})
|
||
|
||
|
||
@app.get("/prompts")
|
||
async def prompts_get() -> JSONResponse:
|
||
try:
|
||
return JSONResponse({
|
||
"step1": prompt_store.load_step1(),
|
||
"step3": prompt_store.load_step3(),
|
||
"config": prompt_store.load_config(),
|
||
"step3_path": prompt_store.STEP3_PATH,
|
||
})
|
||
except Exception as exc: # noqa: BLE001
|
||
return JSONResponse({"error": str(exc)}, 500)
|
||
|
||
|
||
@app.post("/prompts")
|
||
async def prompts_post(step1: str = Form(None), step3: str = Form(None),
|
||
config: str = Form(None), reset: str = Form("")) -> JSONResponse:
|
||
try:
|
||
if _truthy(reset):
|
||
prompt_store.reset() # Step 3 지침서는 사용자 파일 — 리셋 대상 아님
|
||
else:
|
||
prompt_store.save(step1=step1, step3=step3, config=config)
|
||
except Exception as exc: # noqa: BLE001
|
||
return JSONResponse({"error": f"저장 실패: {exc}"}, 400)
|
||
return JSONResponse({"ok": True})
|
||
|
||
|
||
@app.post("/auto/build")
|
||
async def auto_build(
|
||
aid: str = Form(...),
|
||
hid: str = Form(..., alias="id"),
|
||
title_top: str = Form(""),
|
||
title_main: str = Form(""),
|
||
cards: list[UploadFile] = File(default=[]),
|
||
card_cuts: str = Form(""),
|
||
video_scale: str = Form("144"),
|
||
flip: str = Form(""),
|
||
scene: str = Form(""),
|
||
bg_white: str = Form(""),
|
||
cards_fixed: str = Form(""),
|
||
asr_bottom: str = Form("1"),
|
||
comments_dir: str = Form(""),
|
||
) -> JSONResponse:
|
||
"""자동 탭 3단계 — /auto/prepare 가 PSTATES 에 보관한 상태로 드래프트만 만든다.
|
||
|
||
다운로드·받아쓰기는 다시 하지 않는다(/paste/build 와 같은 방식) — aid+id 로
|
||
PSTATES[f"{aid}:{id}"] 를 찾아 paste_draft 만 돌린다. title_top/title_main 은
|
||
검토 화면(1차)에서 고른 값을 여기서 받아 state 에 덮어쓴다.
|
||
"""
|
||
key = f"{aid}:{hid}"
|
||
st = PSTATES.get(key)
|
||
if not st:
|
||
return JSONResponse({"error": "준비 결과가 만료됐습니다. 다시 준비해 주세요."}, 404)
|
||
state = st["state"]
|
||
state["title_top"] = title_top
|
||
state["title_main"] = title_main
|
||
|
||
# asr_bottom을 끈 경우 — 화면 자막을 받아쓰기 결과가 아니라 원래 JSON bottom으로
|
||
# 되돌린다(/paste/build 와 동일 규칙 — state["bottom_caps"]는 /auto/prepare 가
|
||
# 항상 asr_bottom=True 로 돈 결과라 받아쓰기로 덮여 있다).
|
||
if not _truthy(asr_bottom):
|
||
dur = state["dur"]
|
||
bottom_caps = [(p0, min(p1, dur), b) for (p0, p1), (_, _, b, _)
|
||
in zip(state["placements"], state["cuts"]) if b and p0 < dur]
|
||
bottom_caps = _remap_caps(bottom_caps, state["video_clips"])
|
||
state = {**state, "bottom_caps": bottom_caps}
|
||
|
||
# 카드별 소속 컷 — 값이 깨져도 빌드를 막지 않는다(없으면 전체 균등 배치로 폴백)
|
||
cut_map: list[int] = []
|
||
try:
|
||
parsed = json.loads(card_cuts) if card_cuts.strip() else []
|
||
if isinstance(parsed, list):
|
||
cut_map = [int(v) for v in parsed if isinstance(v, int) and not isinstance(v, bool)]
|
||
except (json.JSONDecodeError, ValueError, TypeError):
|
||
cut_map = []
|
||
|
||
sig = (key + "|" + card_cuts + "|" + video_scale + "|" + flip + "|" + scene + "|"
|
||
+ bg_white + "|" + cards_fixed + "|" + asr_bottom + "|" + str(len(cards)))
|
||
h = hashlib.sha1(sig.encode()).hexdigest()[:12]
|
||
cdir = comments_dir.strip()
|
||
if cards:
|
||
cdir = os.path.join(COMMENTS_DIR, h)
|
||
if os.path.isdir(cdir): # 재빌드 시 이전 카드 잔재 제거
|
||
shutil.rmtree(cdir, ignore_errors=True)
|
||
os.makedirs(cdir, exist_ok=True)
|
||
for i, f in enumerate(cards, 1):
|
||
body = await f.read()
|
||
with open(os.path.join(cdir, f"{i:03d}.png"), "wb") as out:
|
||
out.write(body)
|
||
|
||
JOBS[h] = {
|
||
"paste_state": state, "card_cuts": cut_map,
|
||
"video_scale": _scale(video_scale), "flip": _truthy(flip), "scene": _truthy(scene),
|
||
"comments_dir": cdir, "bg_white": _truthy(bg_white), "cards_fixed": _truthy(cards_fixed),
|
||
}
|
||
return JSONResponse({"job_id": h})
|
||
|
||
|
||
@app.post("/paste/analyze")
|
||
async def paste_analyze_start(data: str = Form(...)) -> JSONResponse:
|
||
"""📋 붙여넣기 탭(새 흐름) 1단계 — 편집안 검증 후 분석 예약. 실제 작업은 /paste/stream 에서."""
|
||
try:
|
||
payload = parse_paste(data)
|
||
except ValueError as e:
|
||
return JSONResponse({"error": str(e)}, 400)
|
||
aid = hashlib.sha1(("paste|" + data).encode()).hexdigest()[:12]
|
||
ANALYSES[aid] = {"payload": payload}
|
||
return JSONResponse({"analysis_id": aid})
|
||
|
||
|
||
@app.get("/paste/stream/{aid}")
|
||
async def paste_stream(aid: str) -> StreamingResponse:
|
||
"""📋 붙여넣기 탭(새 흐름) — 다운로드·무음·받아쓰기 → h-lab 댓글 매칭 → 검토 화면용 result.
|
||
|
||
추천을 하려면 받아쓰기가 항상 필요하므로 remove_silence·asr_bottom 은 여기서
|
||
True 로 고정한다. 사용자가 화면에서 끄는 asr_bottom 옵션은 화면 자막에만
|
||
적용되며 /paste/build 단계에서 처리한다.
|
||
"""
|
||
a = ANALYSES.get(aid)
|
||
|
||
async def gen():
|
||
if not a:
|
||
yield _sse({"type": "error", "message": "알 수 없는 분석입니다."})
|
||
return
|
||
payload = a["payload"]
|
||
warnings: list[str] = []
|
||
# draft 스텝은 이 단계(analyze)에서 안 돈다 — /paste/build 때 별도 스트림으로 실행되므로
|
||
# 여기 manifest 에 넣으면 영원히 start 가 안 와 화면에 대기 상태로 멈춰 보인다.
|
||
yield _sse({"type": "manifest", "steps":
|
||
[s for s in pipeline_paste_steps(True) if s["id"] != "draft"] +
|
||
[{"id": "comments", "label": "댓글 수집 (h-lab)"},
|
||
{"id": "recommend", "label": "컷별 댓글 추천"}]})
|
||
com_task = asyncio.create_task(
|
||
asyncio.to_thread(hlab.fetch_comments, payload["url"]))
|
||
yield _sse({"type": "step", "id": "comments", "status": "start"})
|
||
state = None
|
||
try:
|
||
async for ev in paste_analyze(payload, "paste_" + aid,
|
||
remove_silence=True, asr_bottom=True):
|
||
if ev.get("type") == "state": # 내부 전용 — 밖으로 흘리지 않는다
|
||
state = ev["state"]
|
||
continue
|
||
if ev.get("type") == "error":
|
||
com_task.cancel()
|
||
yield _sse(ev)
|
||
return
|
||
yield _sse(ev)
|
||
except Exception as exc: # noqa: BLE001 — 다운로드 등 실패는 드래프트 생성을 막는다
|
||
com_task.cancel()
|
||
yield _sse({"type": "error", "message": f"{type(exc).__name__}: {exc}"})
|
||
return
|
||
if state is None:
|
||
com_task.cancel()
|
||
yield _sse({"type": "error", "message": "분석 상태를 만들지 못했습니다."})
|
||
return
|
||
|
||
comments: list[dict] = []
|
||
try:
|
||
comments = await com_task
|
||
yield _sse({"type": "step", "id": "comments", "status": "done",
|
||
"detail": f"{len(comments)}개"})
|
||
except Exception as exc: # noqa: BLE001 — 댓글 실패는 진행을 막지 않는다
|
||
warnings.append(f"h-lab 연결 실패 — 댓글 없이 진행합니다 ({exc})")
|
||
yield _sse({"type": "step", "id": "comments", "status": "done",
|
||
"detail": "실패(생략)"})
|
||
|
||
yield _sse({"type": "step", "id": "recommend", "status": "start"})
|
||
# ⚠ 좌표계 둘: places/captions = 압축 타임라인(카드·자막 추출용),
|
||
# orig = 원본 영상 시각(⭐ 분:초 매칭용). 섞으면 카드가 통째로 어긋난다.
|
||
places = state["card_places"]
|
||
orig = [(s, e) for s, e, _, _ in state["cuts"]]
|
||
# 예상 밖 예외가 나면(감싸지 않았을 때) SSE 가 error 이벤트도 없이 끊기고
|
||
# PSTATES 도 안 남아 build 가 404 난다 — 추천 실패가 검토 화면 자체를 막으면 안 된다.
|
||
try:
|
||
cuts, need, ai_failed = await asyncio.to_thread(
|
||
recommend.cuts_from_state, places, orig, state["bottom_caps"], comments)
|
||
if ai_failed:
|
||
warnings.append("AI 추천 실패(Gemini 응답 없음) — 분:초·자막단어·좋아요로 배정했습니다")
|
||
except Exception as exc: # noqa: BLE001
|
||
# cuts=None(빈 리스트 아님)이어야 화면이 "컷 0개"로 비지 않고 ⭐/➕ 폴백
|
||
# 화면(자동/구간 탭과 동일한 hl.cuts 없음 경로)으로 넘어간다.
|
||
cuts, need = None, max(1, int(state["timeline_dur"] // 3))
|
||
warnings.append(f"컷별 추천 실패 — 컷 정보 없이 진행합니다 ({type(exc).__name__}: {exc})")
|
||
yield _sse({"type": "step", "id": "recommend", "status": "done",
|
||
"detail": (f"{len(cuts)}컷 · 카드 {need}장" if cuts is not None
|
||
else f"폴백 · 카드 {need}장")})
|
||
|
||
PSTATES[aid] = {"state": state, "places": places, "orig": orig, "payload": payload}
|
||
no_ts = [c for c in comments if not c["times"]]
|
||
matched = hlab.match_ranges(comments, orig) if comments else []
|
||
yield _sse({"type": "result", "cuts": cuts, "need": need,
|
||
"total": round(state["timeline_dur"], 1),
|
||
"cutRanges": [{"start": s, "end": e} for s, e in orig],
|
||
"matched": matched,
|
||
"candidates": hlab.top_liked(no_ts, set(matched), len(no_ts)),
|
||
"comments": comments, "warnings": warnings})
|
||
|
||
return StreamingResponse(gen(), media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no"})
|
||
|
||
|
||
@app.post("/paste/build")
|
||
async def paste_build(
|
||
aid: str = Form(...),
|
||
cards: list[UploadFile] = File(default=[]),
|
||
card_cuts: str = Form(""),
|
||
video_scale: str = Form("144"),
|
||
flip: str = Form(""),
|
||
scene: str = Form(""),
|
||
bg_white: str = Form(""),
|
||
cards_fixed: str = Form(""),
|
||
asr_bottom: str = Form("1"),
|
||
comments_dir: str = Form(""),
|
||
) -> JSONResponse:
|
||
"""📋 붙여넣기 탭(새 흐름) 3단계 — 보관한 상태로 드래프트만 만든다(다운로드·받아쓰기 다시 안 함)."""
|
||
st = PSTATES.get(aid)
|
||
if not st:
|
||
return JSONResponse({"error": "분석 결과가 만료됐습니다. 다시 분석해 주세요."}, 404)
|
||
state = st["state"]
|
||
|
||
# asr_bottom을 끈 경우 — 화면 자막을 받아쓰기 결과가 아니라 원래 JSON bottom으로
|
||
# 되돌린다. state["bottom_caps"]는 analyze가 항상 asr_bottom=True로 돈 결과라
|
||
# 받아쓰기로 덮여 있다 — paste_analyze가 하던 것과 같은 식으로 원본 JSON 자막을
|
||
# 다시 만들고, 무음 제거로 압축된 타임라인에 맞춰 _remap_caps로 재매핑한다.
|
||
if not _truthy(asr_bottom):
|
||
dur = state["dur"]
|
||
bottom_caps = [(p0, min(p1, dur), b) for (p0, p1), (_, _, b, _)
|
||
in zip(state["placements"], state["cuts"]) if b and p0 < dur]
|
||
bottom_caps = _remap_caps(bottom_caps, state["video_clips"])
|
||
state = {**state, "bottom_caps": bottom_caps}
|
||
|
||
# 카드별 소속 컷 — 값이 깨져도 빌드를 막지 않는다(없으면 전체 균등 배치)
|
||
cut_map: list[int] = []
|
||
try:
|
||
parsed = json.loads(card_cuts) if card_cuts.strip() else []
|
||
if isinstance(parsed, list):
|
||
cut_map = [int(v) for v in parsed if isinstance(v, int) and not isinstance(v, bool)]
|
||
except (json.JSONDecodeError, ValueError, TypeError):
|
||
cut_map = []
|
||
|
||
sig = (aid + "|" + card_cuts + "|" + video_scale + "|" + flip + "|" + scene + "|"
|
||
+ bg_white + "|" + cards_fixed + "|" + asr_bottom + "|" + str(len(cards)))
|
||
h = hashlib.sha1(sig.encode()).hexdigest()[:12]
|
||
# 카드가 오면 폴더 지정보다 우선(/youtube 와 같은 규칙). 카드를 하나도 안 골랐으면
|
||
# 화면에서 넘어온 폴더 경로를 그대로 써서 예전(폴더 지정) 동작으로 하위호환한다.
|
||
cdir = comments_dir.strip()
|
||
if cards:
|
||
cdir = os.path.join(COMMENTS_DIR, h)
|
||
if os.path.isdir(cdir): # 재빌드 시 이전 카드 잔재 제거
|
||
shutil.rmtree(cdir, ignore_errors=True)
|
||
os.makedirs(cdir, exist_ok=True)
|
||
for i, f in enumerate(cards, 1):
|
||
body = await f.read()
|
||
with open(os.path.join(cdir, f"{i:03d}.png"), "wb") as out:
|
||
out.write(body)
|
||
|
||
JOBS[h] = {
|
||
"paste_state": state, "card_cuts": cut_map,
|
||
"video_scale": _scale(video_scale), "flip": _truthy(flip), "scene": _truthy(scene),
|
||
"comments_dir": cdir, "bg_white": _truthy(bg_white), "cards_fixed": _truthy(cards_fixed),
|
||
}
|
||
return JSONResponse({"job_id": h})
|
||
|
||
|
||
@app.post("/open-capcut")
|
||
async def open_capcut() -> JSONResponse:
|
||
"""CapCut 실행 (Start Menu 바로가기 우선, 없으면 최신 버전 exe)."""
|
||
lnk = os.path.join(os.environ.get("APPDATA", ""),
|
||
"Microsoft", "Windows", "Start Menu", "Programs", "CapCut.lnk")
|
||
target = lnk if os.path.isfile(lnk) else None
|
||
if not target:
|
||
exes = sorted(glob.glob(os.path.join(
|
||
os.environ.get("LOCALAPPDATA", ""), "CapCut", "Apps", "*", "CapCut.exe")))
|
||
target = exes[-1] if exes else None
|
||
if not target:
|
||
return JSONResponse({"ok": False, "error": "CapCut 실행 파일을 못 찾았습니다."}, 404)
|
||
try:
|
||
os.startfile(target) # type: ignore[attr-defined] (Windows 전용)
|
||
return JSONResponse({"ok": True})
|
||
except Exception as exc: # noqa: BLE001
|
||
return JSONResponse({"ok": False, "error": str(exc)}, 500)
|
||
|
||
|
||
@app.get("/drafts")
|
||
async def drafts() -> JSONResponse:
|
||
"""드래프트 목록(최근순) + 레이어 꼬임 여부. 수리 UI 용."""
|
||
return JSONResponse({"drafts": list_drafts()[:30]})
|
||
|
||
|
||
def _capcut_running() -> bool:
|
||
"""CapCut.exe 실행 여부. 실행 중이면 수리해도 CapCut 이 덮어써서 헛수고가 된다."""
|
||
try:
|
||
out = subprocess.run(["tasklist", "/FI", "IMAGENAME eq CapCut.exe"],
|
||
capture_output=True, text=True, errors="replace",
|
||
timeout=10).stdout
|
||
return "CapCut.exe" in out
|
||
except Exception: # noqa: BLE001 — 판정 실패 시 막지 않음
|
||
return False
|
||
|
||
|
||
@app.post("/repair")
|
||
async def repair(draft: str = Form(...), force: str = Form("")) -> JSONResponse:
|
||
"""레이어 수리 — 옮긴 영상 클립이 흰 띠·댓글 위로 삐져나온 것을 되돌린다.
|
||
|
||
⚠ CapCut 이 실행 중이면 거부한다. 실측 사례: 11:04:32 수리 → 11:05:39 CapCut 저장으로
|
||
되돌아감. CapCut 은 파일을 다시 읽지 않고 메모리 상태로 덮어쓰기 때문.
|
||
"""
|
||
if _capcut_running() and not _truthy(force):
|
||
return JSONResponse({
|
||
"ok": False,
|
||
"capcut_running": True,
|
||
"error": "CapCut이 실행 중입니다. 수리해도 CapCut이 저장하면서 되돌립니다.\n"
|
||
"CapCut을 완전히 종료한 뒤 다시 눌러주세요.",
|
||
}, 409)
|
||
path = _resolve_draft(draft)
|
||
if not path:
|
||
return JSONResponse({"ok": False, "error": f"드래프트를 못 찾음: {draft}"}, 404)
|
||
try:
|
||
res = repair_layers(path)
|
||
except Exception as exc: # noqa: BLE001
|
||
return JSONResponse({"ok": False, "error": f"{type(exc).__name__}: {exc}"}, 500)
|
||
return JSONResponse({"ok": True, "draft": os.path.basename(path), **res})
|
||
|
||
|
||
def _resolve_draft(key: str) -> str:
|
||
"""폼 값(드래프트 폴더 경로 또는 폴더명)을 실제 폴더로 변환. 못 찾으면 "".
|
||
|
||
드래프트가 기본 경로 밖(CapCut 저장 위치 변경)에도 있어 `basename` 고정은 못 쓴다.
|
||
대신 `list_drafts()` 가 아는 드래프트에만 매칭해 임의 경로 접근을 막는다.
|
||
"""
|
||
key = (key or "").strip()
|
||
if not key:
|
||
return ""
|
||
drafts = list_drafts()
|
||
want = os.path.normcase(os.path.abspath(key))
|
||
for d in drafts: # ① 경로 완전 일치
|
||
if os.path.normcase(os.path.abspath(d["path"])) == want:
|
||
return d["path"]
|
||
base = os.path.basename(key.rstrip("\\/"))
|
||
for d in drafts: # ② 폴더명 (구버전 UI 호환)
|
||
if d["name"] == key or d["name"] == base:
|
||
return d["path"]
|
||
return ""
|
||
|
||
|
||
def _sse(obj: dict) -> str:
|
||
return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n"
|