hehihoho3@gmail.com
bf1b387d6d
컷별 댓글 추천 작업을 태스크 단위로 되돌릴 수 있게 버전관리를 시작한다. .gitignore 로 영상·캐시(.downloads 2.7G, .comments 72M, .media 28M)와 비밀키(.gemini_key)를 제외했다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
536 lines
24 KiB
Python
536 lines
24 KiB
Python
"""런타임 파이프라인 — SSE 이벤트를 내보내는 async 제너레이터.
|
||
|
||
현재 단계(1단 기준): silence → draft.
|
||
3·4단 추가 시 asr / filler 스텝을 STEPS 와 본문에 삽입.
|
||
|
||
함정 메모:
|
||
- ASR(추후)은 numba 비안전 → asyncio.Lock 으로 직렬화(동시 호출 segfault 방지). 자리만 마련.
|
||
- 단계당 최소 MIN_STEP 지연 → 캐시 hit 시에도 애니메이션 가시화.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import os
|
||
import time
|
||
from typing import AsyncIterator, Dict, List
|
||
|
||
from typing import Optional
|
||
|
||
from .probe import probe
|
||
from .silence import detect_speech_segments
|
||
from .transcribe import transcribe
|
||
from .highlight import cut_plan
|
||
from .media import make_frame, make_solid
|
||
from .scene import detect_scene_changes, split_clips_at_scenes
|
||
from .youtube import cut_youtube, cut_youtube_multi, download_paste_cuts, REPAIR_LOG
|
||
from .correct import has_gemini_key, correct_captions
|
||
from .draft import build_jumpcut_draft, build_bg_template_draft, _ty
|
||
|
||
MIN_STEP = 0.5 # 초
|
||
|
||
# ASR(numba) 직렬화 — 동시 호출 시 segfault 방지
|
||
ASR_LOCK = asyncio.Lock()
|
||
|
||
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
_DOWNLOADS = os.path.join(_ROOT, ".downloads")
|
||
|
||
|
||
def _load_comment_cards(folder: str, dur: float, min_sec: float = 3.0,
|
||
fixed: bool = False):
|
||
"""지정 폴더의 이미지를 영상 길이에 맞춰 하단에 균등 배치.
|
||
|
||
장수: n = min(카드 수, max(1, floor(dur/min_sec))) — 카드 하나가 min_sec 밑으로
|
||
내려가지 않는 상한. 배치 간격은 항상 dur/n → 카드가 모자라도 끝까지 빈 곳 없이
|
||
채워지고(간격이 3초 이상으로 늘어남), 넘치면 초과분은 버린다.
|
||
fixed=True 면 늘리지 않고 min_sec 고정 — 모자라면 뒤는 비운다
|
||
(편집하면서 부분삭제를 많이 하는 경우 카드가 늘어나 있으면 타이밍이 꼬여서).
|
||
정렬 규칙:
|
||
- 모든 파일명이 숫자로 시작하면 → 숫자순(1, 2, 10 …)
|
||
- 아니면 → 저장(생성) 순서 = 다운로드한 순서
|
||
folder 가 비었거나 없으면 [] (댓글 카드 없음).
|
||
Returns: [(start, end, path)]
|
||
"""
|
||
import re
|
||
folder = (folder or "").strip().strip('"')
|
||
if not folder or not os.path.isdir(folder) or dur <= 0:
|
||
return []
|
||
imgs = [os.path.join(folder, f) for f in os.listdir(folder)
|
||
if os.path.splitext(f)[1].lower() in (".png", ".jpg", ".jpeg", ".webp")]
|
||
if not imgs:
|
||
return []
|
||
|
||
def _leadnum(path):
|
||
m = re.match(r"\s*0*(\d+)", os.path.splitext(os.path.basename(path))[0])
|
||
return int(m.group(1)) if m else None
|
||
|
||
if all(_leadnum(p) is not None for p in imgs):
|
||
imgs.sort(key=_leadnum) # 파일명 숫자순
|
||
else:
|
||
imgs.sort(key=lambda p: os.path.getctime(p)) # 저장(생성) 순서
|
||
|
||
n = min(len(imgs), max(1, int(dur // min_sec)))
|
||
if fixed: # 3초 고정 — 뒤가 비어도 늘리지 않음
|
||
return [(i * min_sec, min((i + 1) * min_sec, dur), path)
|
||
for i, path in enumerate(imgs[:n])]
|
||
interval = dur / n
|
||
return [(i * interval, (i + 1) * interval, path)
|
||
for i, path in enumerate(imgs[:n])]
|
||
|
||
|
||
def _elapsed_kept(keep_sorted, t: float) -> float:
|
||
"""원본 시간 t 가 무음 제거 후(압축) 타임라인에서 놓이는 위치 = t 이전의 보존 길이 합."""
|
||
tot = 0.0
|
||
for s, e in keep_sorted:
|
||
if s >= t:
|
||
break
|
||
tot += min(t, e) - s
|
||
return tot
|
||
|
||
|
||
def _remap_caps(caps, keep_sorted):
|
||
"""자막 [(s,e,txt)] 를 무음 제거 타임라인으로 재매핑. 전부 무음이면 버림."""
|
||
out = []
|
||
for cs, ce, txt in caps:
|
||
ns = _elapsed_kept(keep_sorted, cs)
|
||
ne = _elapsed_kept(keep_sorted, ce)
|
||
if ne - ns > 0.05:
|
||
out.append((ns, ne, txt))
|
||
return out
|
||
|
||
|
||
def _safe_name(name: str) -> str:
|
||
s = "".join(c for c in name if c.isalnum() or c in (" ", "_", "-", ".")).strip()
|
||
return s[:60] or "video"
|
||
|
||
|
||
def _template_path() -> str:
|
||
for n in ("배경.png", "배경템플릿.png", os.path.join("assets", "bg_template.png")):
|
||
p = os.path.join(_ROOT, n)
|
||
if os.path.isfile(p):
|
||
return p
|
||
return os.path.join(_ROOT, "배경.png")
|
||
|
||
|
||
BG_STEPS: List[Dict[str, str]] = [
|
||
{"id": "silence", "label": "무음·발화 분석"},
|
||
{"id": "asr", "label": "음성 인식(자막)"},
|
||
{"id": "draft", "label": "템플릿 드래프트 생성"},
|
||
]
|
||
|
||
|
||
async def process_bg_template(
|
||
video_path: Optional[str],
|
||
draft_name: str,
|
||
*,
|
||
title_top: str = "",
|
||
title_main: str = "",
|
||
channel: str = "",
|
||
video_scale: float = 1.0,
|
||
flip_horizontal: bool = False,
|
||
scene_split: bool = False,
|
||
comments_dir: str = "",
|
||
cards_fixed: bool = False,
|
||
bg_white: bool = False,
|
||
youtube: Optional[dict] = None,
|
||
) -> AsyncIterator[dict]:
|
||
"""배경템플릿 파이프라인: [유튜브 구간 다운로드] → 무음컷 → 자막 → 드래프트. SSE 이벤트."""
|
||
t_all = time.perf_counter()
|
||
use_gemini = has_gemini_key()
|
||
steps = ([{"id": "download", "label": "유튜브 여러 구간 다운로드·병합"}] if youtube else [])
|
||
steps += [{"id": "silence", "label": "무음·발화 분석"},
|
||
{"id": "asr", "label": "받아쓰기 (Gemini/Whisper)"},
|
||
{"id": "draft", "label": "템플릿 드래프트 생성"}]
|
||
yield {"type": "manifest", "steps": steps}
|
||
|
||
# ── 유튜브 여러 구간 다운로드 + 병합 (URL 입력 시) ──
|
||
if youtube:
|
||
ranges = youtube.get("ranges") or [(youtube.get("start", ""), youtube.get("end", ""))]
|
||
yield {"type": "step", "id": "download", "status": "start"}
|
||
rng_txt = ", ".join(f"{s}~{e}" for s, e in ranges)
|
||
yield {"type": "log", "msg": f"유튜브 {len(ranges)}개 구간 다운로드·병합 중… [{rng_txt}]"}
|
||
t = time.perf_counter()
|
||
REPAIR_LOG.clear()
|
||
video_path, title, yt_channel = await asyncio.to_thread(
|
||
cut_youtube_multi, youtube["url"], ranges, _DOWNLOADS,
|
||
)
|
||
for msg in REPAIR_LOG: # 초록 깨짐 자동 수리 내역
|
||
yield {"type": "log", "msg": f"🩹 {msg}"}
|
||
draft_name = _safe_name(title) or draft_name
|
||
# 출처를 안 적었으면 유튜브 채널명으로 자동 채움
|
||
if not channel and yt_channel:
|
||
channel = f"@{yt_channel}"
|
||
yield {"type": "log", "msg": f"출처 자동: {channel}"}
|
||
await _floor(t)
|
||
yield {"type": "step", "id": "download", "status": "done",
|
||
"elapsed": round(time.perf_counter() - t, 1), "detail": title}
|
||
|
||
yield {"type": "log", "msg": f"입력: {os.path.basename(video_path)}"}
|
||
meta = await asyncio.to_thread(probe, video_path)
|
||
yield {"type": "log", "msg": f"{meta.width}×{meta.height} · {meta.fps}fps · {meta.duration:.1f}s"}
|
||
|
||
# ── silence (오디오 무음 컷) ──
|
||
yield {"type": "step", "id": "silence", "status": "start"}
|
||
t = time.perf_counter()
|
||
keep = await asyncio.to_thread(
|
||
detect_speech_segments, video_path, meta.duration,
|
||
noise_db=-28.0, min_silence=0.3, pad=0.04,
|
||
)
|
||
if not keep:
|
||
yield {"type": "error", "message": "오디오가 없거나 전부 무음입니다."}
|
||
return
|
||
kept = sum(e - s for s, e in keep)
|
||
await _floor(t)
|
||
yield {"type": "step", "id": "silence", "status": "done",
|
||
"elapsed": round(time.perf_counter() - t, 1),
|
||
"detail": f"보존 {len(keep)}구간 · {meta.duration - kept:.1f}s 무음 제거"}
|
||
|
||
# ── asr (받아쓰기): 타이밍=Whisper 단어 타임스탬프(정확, 드리프트 없음).
|
||
# 글자 품질만 Gemini로 제자리 교정(1:1, 시간은 절대 안 건드림) → 싱크 유지 ──
|
||
yield {"type": "step", "id": "asr", "status": "start"}
|
||
t = time.perf_counter()
|
||
|
||
yield {"type": "log", "msg": "Whisper로 받아쓰기·타이밍 분석 중… (1분당 ≈30초)"}
|
||
async with ASR_LOCK:
|
||
tr = await asyncio.to_thread(
|
||
transcribe, video_path,
|
||
model_size="medium", language="ko", use_cache=True, vad_filter=False,
|
||
)
|
||
video_clips, captions, total = cut_plan(keep, tr) # Whisper 타이밍 자막
|
||
method = "Whisper 추출"
|
||
|
||
# Gemini 교정: 자막 글자만 다듬고 개수·순서·시간 그대로 유지(싱크 불변).
|
||
if use_gemini and captions:
|
||
yield {"type": "log", "msg": "Gemini로 자막 글자 교정 중… (시간은 그대로)"}
|
||
try:
|
||
texts = [c[2] for c in captions]
|
||
fixed = await asyncio.to_thread(
|
||
correct_captions, texts, title=(title_main or draft_name))
|
||
if isinstance(fixed, list) and len(fixed) == len(captions):
|
||
captions = [(s, e, (ft.strip() or captions[i][2]))
|
||
for i, ((s, e, _), ft) in enumerate(zip(captions, fixed))]
|
||
method = "Whisper 타이밍 + Gemini 교정"
|
||
except Exception as exc: # noqa: BLE001 — 교정 실패해도 Whisper 원문 유지
|
||
yield {"type": "log", "msg": f"⚠ Gemini 교정 실패({type(exc).__name__}) → Whisper 원문"}
|
||
|
||
await _floor(t)
|
||
yield {"type": "step", "id": "asr", "status": "done",
|
||
"elapsed": round(time.perf_counter() - t, 1),
|
||
"detail": f"{method} · 자막 {len(captions)}개"}
|
||
|
||
# ── 장면전환 분할 (선택): 컷이 바뀌는 지점에서 세그먼트 추가 분할 ──
|
||
if scene_split:
|
||
yield {"type": "log", "msg": "장면전환 감지 중… (화면 바뀌는 컷 찾기)"}
|
||
before = len(video_clips)
|
||
scenes = await asyncio.to_thread(detect_scene_changes, video_path)
|
||
video_clips = split_clips_at_scenes(video_clips, scenes)
|
||
yield {"type": "log",
|
||
"msg": f"장면전환 {len(scenes)}곳 → 세그먼트 {before} → {len(video_clips)}개"}
|
||
|
||
# ── draft (배경템플릿) ──
|
||
yield {"type": "step", "id": "draft", "status": "start"}
|
||
t = time.perf_counter()
|
||
frame, bg, pos = await asyncio.to_thread(_template_pos, bg_white)
|
||
cards = _load_comment_cards(comments_dir, total, fixed=cards_fixed)
|
||
if cards:
|
||
yield {"type": "log", "msg": f"댓글 카드 {len(cards)}개 하단 삽입(3초 간격)"}
|
||
path = await asyncio.to_thread(
|
||
lambda: build_bg_template_draft(
|
||
video_path, bg, frame, video_clips, captions, meta, draft_name,
|
||
title_top=title_top or None, title_main=title_main or None,
|
||
channel=channel or None, video_scale=video_scale,
|
||
flip_horizontal=flip_horizontal, comment_cards=cards, bg_white=bg_white, **pos,
|
||
) # video_clips 는 장면분할 반영된 최신 리스트 사용
|
||
)
|
||
await _floor(t)
|
||
yield {"type": "step", "id": "draft", "status": "done",
|
||
"elapsed": round(time.perf_counter() - t, 1), "detail": draft_name}
|
||
|
||
yield {
|
||
"type": "result",
|
||
"draft_name": draft_name,
|
||
"draft_path": path,
|
||
"stats": {
|
||
"duration": round(meta.duration, 1),
|
||
"kept": round(total, 1),
|
||
"cut": round(meta.duration - total, 1),
|
||
"segments": len(video_clips),
|
||
"captions": len(captions),
|
||
"elapsed": round(time.perf_counter() - t_all, 1),
|
||
},
|
||
}
|
||
|
||
|
||
# ── 템플릿 레이아웃 (캔버스 1080×1920, 단위 = 픽셀, 위가 0) ───────────────────
|
||
# 레퍼런스 템플릿을 실측해 잡은 값. **여기만 고치면 전체 배치가 같이 움직인다.**
|
||
# (예전엔 배경.png 흰밴드 자동감지였는데, 좌표를 정확히 통제하려고 상수로 바꿨다)
|
||
CANVAS_H = 1920
|
||
VIDEO_TOP = 323 # 영상 창 시작 = 위 흰 띠가 끝나는 지점
|
||
VIDEO_BOTTOM = 1122 # 영상 창 끝 = 아래 흰 띠가 시작하는 지점
|
||
TITLE_TOP_Y = 109 # 서브제목(주황) 중앙
|
||
TITLE_MAIN_Y = 252 # 메인제목(흰색) 중앙
|
||
CAPTION_GAP = 72 # 하단 자막 중앙 = VIDEO_BOTTOM − 이 값 (영상 창 안쪽 아래)
|
||
EFFECT_GAP = 25 # 효과자막 중앙 = VIDEO_TOP + 이 값 (영상 창 안쪽 위)
|
||
COMMENT_TOP = VIDEO_BOTTOM # 댓글 카드 윗변 = 영상 바로 아래(딱 붙음)
|
||
CHANNEL_RATIO = 0.85 # 출처: 아래 띠에서 85% 내려간 지점
|
||
|
||
|
||
def _template_pos(white: bool = False):
|
||
"""배경템플릿 프레임/배경 생성 + 위치 dict 계산. (frame_path, bg_path, pos) 반환.
|
||
|
||
white=True 면 상하 띠·빈 곳을 흰색으로(흰 띠 프레임 + 흰 배경 레이어),
|
||
아니면 검은색(기본, 배경 레이어 없음 → 빈 곳 검정).
|
||
좌표는 전부 위 레이아웃 상수에서 파생 — 한 곳만 고치면 된다.
|
||
"""
|
||
band = (255, 255, 255) if white else (0, 0, 0)
|
||
fname = "frame_template_white.png" if white else "frame_template.png"
|
||
frame = os.path.join(_ROOT, "assets", fname)
|
||
os.makedirs(os.path.dirname(frame), exist_ok=True)
|
||
make_frame(frame, top_bar=VIDEO_TOP, bottom_top=VIDEO_BOTTOM, band_color=band)
|
||
bg = None
|
||
if white:
|
||
bg = os.path.join(_ROOT, "assets", "bg_white.png")
|
||
make_solid(bg, (255, 255, 255))
|
||
pos = dict(
|
||
video_y=_ty((VIDEO_TOP + VIDEO_BOTTOM) / 2),
|
||
title_top_y=_ty(TITLE_TOP_Y),
|
||
title_main_y=_ty(TITLE_MAIN_Y),
|
||
caption_y=_ty(VIDEO_BOTTOM - CAPTION_GAP),
|
||
effect_y=_ty(VIDEO_TOP + EFFECT_GAP),
|
||
channel_y=_ty(VIDEO_BOTTOM + (CANVAS_H - VIDEO_BOTTOM) * CHANNEL_RATIO),
|
||
comment_top=COMMENT_TOP,
|
||
)
|
||
return frame, bg, pos
|
||
|
||
|
||
async def process_paste(
|
||
payload: dict,
|
||
draft_name: str,
|
||
*,
|
||
video_scale: float = 1.0,
|
||
flip_horizontal: bool = False,
|
||
scene_split: bool = False,
|
||
comments_dir: str = "",
|
||
cards_fixed: bool = False,
|
||
bg_white: bool = False,
|
||
remove_silence: bool = False,
|
||
asr_bottom: bool = False,
|
||
name_suffix: str = "",
|
||
) -> AsyncIterator[dict]:
|
||
"""붙여넣기(JSON) 파이프라인: 컷 정밀 다운로드·병합 → 공급된 자막 2트랙으로 드래프트.
|
||
|
||
자막 배치 시간 = 컷 순서 누적. asr_bottom=True 면 JSON bottom 대신 병합본을
|
||
Whisper로 받아써 실제 발화 타이밍에 맞춘 하단 자막을 생성(effect/제목은 JSON 유지).
|
||
payload: paste.parse_paste 결과 dict.
|
||
"""
|
||
t_all = time.perf_counter()
|
||
cuts = payload["cuts"] # [(s, e, bottom, effect)]
|
||
url = payload["url"]
|
||
title_top = payload.get("title_top", "")
|
||
title_main = payload.get("title_main", "")
|
||
channel = payload.get("channel", "")
|
||
|
||
steps = [{"id": "download", "label": "컷 정밀 다운로드·병합"}]
|
||
if asr_bottom:
|
||
steps.append({"id": "asr", "label": "받아쓰기 (Whisper)"})
|
||
steps.append({"id": "draft", "label": "템플릿 드래프트 생성"})
|
||
yield {"type": "manifest", "steps": steps}
|
||
|
||
# ── 컷 정밀 다운로드 + 병합 ──
|
||
yield {"type": "step", "id": "download", "status": "start"}
|
||
yield {"type": "log", "msg": f"{len(cuts)}개 컷 정밀 다운로드·병합 중… (프레임 정확 컷)"}
|
||
t = time.perf_counter()
|
||
ranges = [(s, e) for s, e, _, _ in cuts]
|
||
REPAIR_LOG.clear()
|
||
video_path, title, yt_channel = await asyncio.to_thread(
|
||
download_paste_cuts, url, ranges, _DOWNLOADS,
|
||
)
|
||
for msg in REPAIR_LOG: # 초록 깨짐 자동 수리 내역
|
||
yield {"type": "log", "msg": f"🩹 {msg}"}
|
||
draft_name = _safe_name(title) or draft_name
|
||
if name_suffix: # 같은 영상에서 여럿 만들 때 이름 충돌(=드래프트 교체) 방지
|
||
draft_name = f"{draft_name}_{name_suffix}"
|
||
if not channel and yt_channel:
|
||
channel = f"@{yt_channel}"
|
||
yield {"type": "log", "msg": f"출처 자동: {channel}"}
|
||
await _floor(t)
|
||
yield {"type": "step", "id": "download", "status": "done",
|
||
"elapsed": round(time.perf_counter() - t, 1), "detail": title}
|
||
|
||
meta = await asyncio.to_thread(probe, video_path)
|
||
dur = meta.duration
|
||
yield {"type": "log", "msg": f"병합 결과 {meta.width}×{meta.height} · {dur:.1f}s"}
|
||
|
||
# 자막 배치 = 컷 순서 누적(공급된 컷 길이 기준). 영상 실제 길이로 클램프.
|
||
placements, c = [], 0.0
|
||
for s, e, _, _ in cuts:
|
||
placements.append((c, c + (e - s)))
|
||
c += (e - s)
|
||
bottom_caps = [(p0, min(p1, dur), b) for (p0, p1), (_, _, b, _) in zip(placements, cuts)
|
||
if b and p0 < dur]
|
||
eff_caps = [(p0, min(p1, dur), ef) for (p0, p1), (_, _, _, ef) in zip(placements, cuts)
|
||
if ef and p0 < dur]
|
||
|
||
video_clips = [(0.0, dur)] # 병합본 = 한 덩어리(재컷 없음)
|
||
timeline_dur = dur
|
||
|
||
# 무음 제거(선택): 컷 안의 무음까지 잘라내고 자막 시간을 압축 타임라인으로 재매핑
|
||
if remove_silence:
|
||
yield {"type": "log", "msg": "무음 분석 중… (컷 안의 무음 제거)"}
|
||
keep = await asyncio.to_thread(
|
||
detect_speech_segments, video_path, dur,
|
||
noise_db=-28.0, min_silence=0.3, pad=0.04,
|
||
)
|
||
if keep:
|
||
keep = sorted(keep)
|
||
bottom_caps = _remap_caps(bottom_caps, keep)
|
||
eff_caps = _remap_caps(eff_caps, keep)
|
||
video_clips = keep
|
||
timeline_dur = sum(e - s for s, e in keep)
|
||
yield {"type": "log",
|
||
"msg": f"무음 {dur - timeline_dur:.1f}s 제거 → {timeline_dur:.1f}s"}
|
||
|
||
# ── asr_bottom(선택): 병합본을 Whisper로 받아써 하단 자막 자동 생성 ──
|
||
# Whisper 타임스탬프는 '병합 파일' 기준. cut_plan(video_clips, tr)이 파일 시간을
|
||
# 최종(무음제거 반영) 타임라인으로 매핑 + 짧은 자막 청킹까지 처리한다.
|
||
# (무음제거 안 켰으면 video_clips=[(0,dur)] → 항등 매핑. 파일/유튜브 탭과 동일 패턴)
|
||
if asr_bottom:
|
||
yield {"type": "step", "id": "asr", "status": "start"}
|
||
t = time.perf_counter()
|
||
yield {"type": "log", "msg": "Whisper로 받아쓰기 중… (1분당 ≈30초)"}
|
||
async with ASR_LOCK:
|
||
tr = await asyncio.to_thread(
|
||
transcribe, video_path,
|
||
model_size="medium", language="ko", use_cache=True, vad_filter=False,
|
||
)
|
||
_, asr_caps, _ = cut_plan(video_clips, tr)
|
||
method = "Whisper 추출"
|
||
if asr_caps and has_gemini_key():
|
||
yield {"type": "log", "msg": "Gemini로 자막 글자 교정 중… (시간은 그대로)"}
|
||
try:
|
||
texts = [ct for _, _, ct in asr_caps]
|
||
fixed = await asyncio.to_thread(
|
||
correct_captions, texts, title=(title_main or draft_name))
|
||
if isinstance(fixed, list) and len(fixed) == len(asr_caps):
|
||
asr_caps = [(s, e, (ft.strip() or asr_caps[i][2]))
|
||
for i, ((s, e, _), ft) in enumerate(zip(asr_caps, fixed))]
|
||
method = "Whisper 타이밍 + Gemini 교정"
|
||
except Exception as exc: # noqa: BLE001 — 교정 실패해도 원문 유지
|
||
yield {"type": "log", "msg": f"⚠ Gemini 교정 실패({type(exc).__name__}) → Whisper 원문"}
|
||
if asr_caps:
|
||
bottom_caps = asr_caps # JSON bottom 완전 대체
|
||
else:
|
||
yield {"type": "log", "msg": "⚠ 받아쓰기 결과 없음 → JSON bottom 자막 사용"}
|
||
await _floor(t)
|
||
yield {"type": "step", "id": "asr", "status": "done",
|
||
"elapsed": round(time.perf_counter() - t, 1),
|
||
"detail": f"{method} · 자막 {len(bottom_caps)}개"}
|
||
|
||
# ── draft ──
|
||
yield {"type": "step", "id": "draft", "status": "start"}
|
||
t = time.perf_counter()
|
||
frame, bg, pos = await asyncio.to_thread(_template_pos, bg_white)
|
||
|
||
# 장면분할(선택): 화면 바뀌는 지점마다 세그먼트 추가 분할(자막 시간 불변)
|
||
if scene_split:
|
||
yield {"type": "log", "msg": "장면전환 감지 중… (화면 바뀌는 컷 찾기)"}
|
||
scenes = await asyncio.to_thread(detect_scene_changes, video_path)
|
||
video_clips = split_clips_at_scenes(video_clips, scenes)
|
||
yield {"type": "log", "msg": f"장면전환 {len(scenes)}곳 → 세그먼트 {len(video_clips)}개"}
|
||
|
||
# 댓글 카드(선택): 지정 폴더의 1,2,3… 을 3초씩 하단에 순서대로
|
||
cards = _load_comment_cards(comments_dir, timeline_dur, fixed=cards_fixed)
|
||
if cards:
|
||
yield {"type": "log", "msg": f"댓글 카드 {len(cards)}개 하단 삽입(3초 간격)"}
|
||
|
||
path = await asyncio.to_thread(
|
||
lambda: build_bg_template_draft(
|
||
video_path, bg, frame, video_clips, bottom_caps, meta, draft_name,
|
||
title_top=title_top or None, title_main=title_main or None,
|
||
channel=channel or None, video_scale=video_scale,
|
||
flip_horizontal=flip_horizontal, effect_captions=eff_caps,
|
||
comment_cards=cards, bg_white=bg_white, **pos,
|
||
)
|
||
)
|
||
await _floor(t)
|
||
yield {"type": "step", "id": "draft", "status": "done",
|
||
"elapsed": round(time.perf_counter() - t, 1), "detail": draft_name}
|
||
|
||
yield {
|
||
"type": "result",
|
||
"draft_name": draft_name,
|
||
"draft_path": path,
|
||
"stats": {
|
||
"duration": round(dur, 1),
|
||
"kept": round(dur, 1),
|
||
"cut": 0.0,
|
||
"segments": len(cuts),
|
||
"captions": len(bottom_caps),
|
||
"elapsed": round(time.perf_counter() - t_all, 1),
|
||
},
|
||
}
|
||
|
||
|
||
STEPS: List[Dict[str, str]] = [
|
||
{"id": "silence", "label": "무음·발화 분석"},
|
||
{"id": "draft", "label": "점프컷 드래프트 생성"},
|
||
]
|
||
|
||
|
||
async def _floor(t0: float) -> None:
|
||
"""최소 단계 시간 보장."""
|
||
dt = time.perf_counter() - t0
|
||
if dt < MIN_STEP:
|
||
await asyncio.sleep(MIN_STEP - dt)
|
||
|
||
|
||
async def process_video(video_path: str, draft_name: str) -> AsyncIterator[dict]:
|
||
"""영상 1개 처리. SSE 로 흘려보낼 dict 이벤트를 yield."""
|
||
t_all = time.perf_counter()
|
||
yield {"type": "manifest", "steps": STEPS}
|
||
|
||
meta = await asyncio.to_thread(probe, video_path)
|
||
yield {"type": "meta", "resolution": f"{meta.width}×{meta.height}",
|
||
"fps": meta.fps, "duration": round(meta.duration, 1)}
|
||
|
||
# ── silence ──────────────────────────────────────────
|
||
yield {"type": "step", "id": "silence", "status": "start"}
|
||
t = time.perf_counter()
|
||
segments = await asyncio.to_thread(detect_speech_segments, video_path, meta.duration)
|
||
if not segments:
|
||
await _floor(t)
|
||
yield {"type": "error", "message": "발화 구간이 감지되지 않았습니다 (무음 임계값 확인)."}
|
||
return
|
||
speech_total = sum(e - s for s, e in segments)
|
||
cut = meta.duration - speech_total
|
||
await _floor(t)
|
||
yield {"type": "step", "id": "silence", "status": "done",
|
||
"elapsed": round(time.perf_counter() - t, 1),
|
||
"detail": f"발화 {len(segments)}구간 · {cut:.1f}s 컷"}
|
||
|
||
# ── draft ────────────────────────────────────────────
|
||
yield {"type": "step", "id": "draft", "status": "start"}
|
||
t = time.perf_counter()
|
||
path = await asyncio.to_thread(
|
||
build_jumpcut_draft, video_path, segments, meta, draft_name
|
||
)
|
||
await _floor(t)
|
||
yield {"type": "step", "id": "draft", "status": "done",
|
||
"elapsed": round(time.perf_counter() - t, 1),
|
||
"detail": draft_name}
|
||
|
||
# ── result ───────────────────────────────────────────
|
||
yield {
|
||
"type": "result",
|
||
"draft_name": draft_name,
|
||
"draft_path": path,
|
||
"stats": {
|
||
"duration": round(meta.duration, 1),
|
||
"kept": round(speech_total, 1),
|
||
"cut": round(cut, 1),
|
||
"cut_pct": round(cut / meta.duration * 100) if meta.duration else 0,
|
||
"segments": len(segments),
|
||
"resolution": f"{meta.width}×{meta.height}",
|
||
"elapsed": round(time.perf_counter() - t_all, 1),
|
||
},
|
||
}
|