hehihoho3@gmail.com
c3556edb9d
- auto_build 시그니처에 card_cuts 폼 필드 추가 - JSON 배열 문자열 파싱 (값 깨져도 빌드 계속) - job dict에 저장 후 process_paste 호출에 전달 - bool 필터링 강화: JSON true/false가 정수로 둔갑하지 않도록 이유: JSON의 true/false가 Python bool로 파싱되는데, bool이 int의 서브클래스라 isinstance(v, int)를 통과해 컷 인덱스 1/0으로 대신 쓰이는 문제 해결. capcut_agent/recommend.py 패턴과 일관성 있음. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
743 lines
34 KiB
Python
743 lines
34 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 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
|
||
from capcut_agent.paste import parse_paste
|
||
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] = {}
|
||
|
||
|
||
_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 _parse_ranges(raw: str) -> list[tuple[str, str]]:
|
||
"""JSON [["mm:ss","mm:ss"],...] → [(start,end)]. 빈 값/파싱실패는 []."""
|
||
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:
|
||
s, e = str(item[0]).strip(), str(item[1]).strip()
|
||
if s and e:
|
||
out.append((s, e))
|
||
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("/youtube")
|
||
async def youtube(
|
||
url: str = Form(...),
|
||
ranges: str = Form(""),
|
||
start: str = Form(""),
|
||
end: str = Form(""),
|
||
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(""),
|
||
cards_fixed: str = Form(""),
|
||
cards: list[UploadFile] = File(default=[]),
|
||
) -> JSONResponse:
|
||
"""유튜브 URL + 여러 구간으로 작업 생성. ranges=JSON [["mm:ss","mm:ss"],...].
|
||
|
||
ranges 없으면 start/end 단일 구간으로 폴백(하위호환).
|
||
cards(댓글 매칭에서 캡처한 PNG들)가 오면 폴더 지정보다 우선한다.
|
||
"""
|
||
rng = _parse_ranges(ranges) or ([(start.strip(), end.strip())] if start and end else [])
|
||
if not rng:
|
||
return JSONResponse({"error": "구간을 하나 이상 입력하세요."}, 400)
|
||
sig = url + "|" + "|".join(f"{s}-{e}" for s, e in rng)
|
||
h = hashlib.sha1(sig.encode()).hexdigest()[:12]
|
||
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)
|
||
comments_dir = cdir
|
||
JOBS[h] = {
|
||
"path": None, "draft_name": f"yt_{h}",
|
||
"youtube": {"url": url.strip(), "ranges": rng},
|
||
"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),
|
||
"cards_fixed": _truthy(cards_fixed),
|
||
}
|
||
return JSONResponse({"job_id": h})
|
||
|
||
|
||
@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"): # 붙여넣기(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/comments")
|
||
async def yt_comments(url: str = Form(...), ranges: str = Form(...)) -> JSONResponse:
|
||
"""유튜브 구간 탭 — 구간 언급 댓글 매칭. 규칙은 자동 탭과 동일:
|
||
⭐ = 어느 구간이든 언급(좋아요순), ➕ = 분:초 없는 댓글(좋아요순), 전체 전송."""
|
||
u = url.strip()
|
||
if not (u.startswith("http://") or u.startswith("https://")):
|
||
return JSONResponse({"error": "유튜브 주소를 입력하세요."}, 400)
|
||
try:
|
||
rng = [(float(s), float(e)) for s, e in json.loads(ranges)]
|
||
rng = [(s, e) for s, e in rng if e > s]
|
||
except (json.JSONDecodeError, TypeError, ValueError):
|
||
rng = []
|
||
if not rng:
|
||
return JSONResponse({"error": "구간을 하나 이상 입력하세요."}, 400)
|
||
try:
|
||
comments = await asyncio.to_thread(hlab.fetch_comments, u)
|
||
except Exception as exc: # noqa: BLE001
|
||
return JSONResponse({"error": f"h-lab 연결 실패: {exc}"}, 502)
|
||
total = sum(e - s for s, e in rng)
|
||
matched = hlab.match_ranges(comments, rng)
|
||
no_ts = [c for c in comments if not c["times"]]
|
||
return JSONResponse({
|
||
"need": max(1, int(total // 3)), "total": round(total, 1),
|
||
"matched": matched,
|
||
"candidates": hlab.top_liked(no_ts, set(matched), len(no_ts)),
|
||
"comments": comments,
|
||
})
|
||
|
||
|
||
@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:
|
||
"""자동 탭 분석 SSE: Step1 → (Step3 ×N ∥ 댓글) → result."""
|
||
a = ANALYSES.get(aid)
|
||
|
||
async def gen():
|
||
if not a:
|
||
yield _sse({"type": "error", "message": "알 수 없는 분석입니다."})
|
||
return
|
||
url = a["url"]
|
||
mode = a.get("mode", "full")
|
||
warnings: list[str] = []
|
||
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 파싱"},
|
||
{"id": "comments", "label": "댓글 수집 (h-lab)"},
|
||
]})
|
||
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)}개 편집안"})
|
||
com_task = asyncio.create_task(asyncio.to_thread(hlab.fetch_comments, url))
|
||
yield _sse({"type": "step", "id": "comments", "status": "start"})
|
||
elif mode == "wpaste":
|
||
# ── 구간 JSON 붙여넣기 — Gemini 안 씀. 구간 5개를 그대로 통짜로 ──
|
||
yield _sse({"type": "manifest", "steps": [
|
||
{"id": "parse", "label": "구간 JSON 파싱"},
|
||
{"id": "comments", "label": "댓글 수집 (h-lab)"},
|
||
]})
|
||
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)}개 구간"})
|
||
com_task = asyncio.create_task(asyncio.to_thread(hlab.fetch_comments, url))
|
||
yield _sse({"type": "step", "id": "comments", "status": "start"})
|
||
else:
|
||
steps = [{"id": "step1", "label": "하이라이트 구간 선정 (Gemini)"}]
|
||
if mode == "full":
|
||
steps.append({"id": "step3", "label": "편집안 생성 (Gemini, 구간별 동시)"})
|
||
steps.append({"id": "comments", "label": "댓글 수집 (h-lab)"})
|
||
yield _sse({"type": "manifest", "steps": steps})
|
||
# 댓글은 URL을 이미 아니까 Step1 과 동시에 수집
|
||
com_task = asyncio.create_task(asyncio.to_thread(hlab.fetch_comments, url))
|
||
yield _sse({"type": "step", "id": "comments", "status": "start"})
|
||
# ── 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
|
||
com_task.cancel()
|
||
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)}개 성공"})
|
||
|
||
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": "실패(생략)"})
|
||
|
||
# 댓글 매칭 — 전체 전송, 브라우저가 '더보기'로 30장씩 나눠 그린다.
|
||
# 후보(candidates)는 분:초 언급이 아예 없는 댓글만 — 타임스탬프 댓글은
|
||
# 자기 구간의 ⭐에서 잡히므로, 다른 구간 얘기하는 댓글이 섞이지 않게.
|
||
no_ts = [c for c in comments if not c["times"]]
|
||
for h in highlights:
|
||
if "paste" not in h:
|
||
continue
|
||
matched = hlab.match_window(comments, h["start"], h["end"])
|
||
h["matched"] = matched
|
||
h["candidates"] = hlab.top_liked(no_ts, set(matched), len(no_ts))
|
||
# 컷별 추천 — 실패해도 위의 matched/candidates 로 화면이 돌아간다.
|
||
# Gemini 호출이 섞여 있어 블로킹이므로 스레드로 뺀다.
|
||
try:
|
||
cuts, need = await asyncio.to_thread(
|
||
recommend.build_highlight_cuts, h, comments)
|
||
if cuts:
|
||
h["cuts"], h["need"] = cuts, need
|
||
except Exception as exc: # noqa: BLE001 — 추천 실패가 생성을 막으면 안 된다
|
||
warnings.append(
|
||
f"ID {h.get('id')} 컷별 추천 실패 — 기존 방식으로 표시 "
|
||
f"({type(exc).__name__}: {exc})")
|
||
yield _sse({"type": "result", "highlights": highlights,
|
||
"comments": comments, "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(
|
||
data: str = Form(...),
|
||
tag: str = Form(""),
|
||
cards: list[UploadFile] = File(default=[]),
|
||
video_scale: str = Form("144"),
|
||
flip: str = Form(""),
|
||
scene: str = Form(""),
|
||
bg_white: str = Form(""),
|
||
remove_silence: str = Form(""),
|
||
asr_bottom: str = Form(""),
|
||
cards_fixed: str = Form(""),
|
||
card_cuts: str = Form(""),
|
||
) -> JSONResponse:
|
||
"""자동 탭 빌드 — 붙여넣기 스키마 JSON + 카드 PNG들 → 기존 paste job.
|
||
|
||
tag(예: '하이라이트1')는 드래프트 이름 꼬리표 — 같은 영상 5개가 서로
|
||
덮어쓰는 것을 막는다. 진행은 기존 /stream/{job_id} 로 본다.
|
||
"""
|
||
try:
|
||
payload = parse_paste(data)
|
||
except ValueError as e:
|
||
return JSONResponse({"error": str(e)}, 400)
|
||
safe_tag = "".join(ch for ch in tag if ch.isalnum() or ch in "-_")[:20]
|
||
sig = (payload["url"] + "|" + safe_tag + "|"
|
||
+ "|".join(f"{s:.3f}-{e:.3f}" for s, e, _, _ in payload["cuts"]))
|
||
h = hashlib.sha1(sig.encode()).hexdigest()[:12]
|
||
cdir = ""
|
||
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)
|
||
# 카드별 소속 컷 — 값이 깨져도 빌드를 막지 않는다(없으면 기존 전체 균등 배치)
|
||
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 = []
|
||
JOBS[h] = {
|
||
"paste": payload, "draft_name": f"auto_{h}",
|
||
"video_scale": _scale(video_scale), "flip": _truthy(flip),
|
||
"scene": _truthy(scene), "comments_dir": cdir,
|
||
"bg_white": _truthy(bg_white), "remove_silence": _truthy(remove_silence),
|
||
"asr_bottom": _truthy(asr_bottom), "name_suffix": safe_tag,
|
||
"cards_fixed": _truthy(cards_fixed),
|
||
"card_cuts": cut_map,
|
||
}
|
||
return JSONResponse({"job_id": h, "cuts": len(payload["cuts"]),
|
||
"cards": len(cards)})
|
||
|
||
|
||
@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"
|