hehihoho3@gmail.com
aab2a3199b
받아쓰기(ASR) 결과에서 컷 구간별로 자막을 이어붙이는 순수 함수. - 시간축 겹침 판정: 닿기만(경계) 제외, 조금이라도 겹치면 포함 - 공백 정규화: 여러 칸 → 공백 하나 - 토큰 제한: cap 문자에서 자르기 (Gemini 효율) - 테스트 모두 통과 (경계 규칙·공백·길이·엣지 케이스) 다음 단계(2단)에서 세 탭이 모두 호출할 예정. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
638 lines
29 KiB
Python
638 lines
29 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 _card_paths(folder: str):
|
||
"""댓글 카드 이미지 경로 목록. 배치 방식과 무관하게 '순서'만 정한다.
|
||
|
||
정렬 규칙:
|
||
- 모든 파일명이 숫자로 시작하면 → 숫자순(1, 2, 10 …)
|
||
- 아니면 → 저장(생성) 순서 = 다운로드한 순서
|
||
folder 가 비었거나 없으면 [].
|
||
"""
|
||
import re
|
||
folder = (folder or "").strip().strip('"')
|
||
if not folder or not os.path.isdir(folder):
|
||
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)) # 저장(생성) 순서
|
||
return imgs
|
||
|
||
|
||
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 고정 — 모자라면 뒤는 비운다
|
||
(편집하면서 부분삭제를 많이 하는 경우 카드가 늘어나 있으면 타이밍이 꼬여서).
|
||
|
||
⚠ 이 함수는 컷을 모른다. 자동 탭처럼 '어느 컷에 붙일지'가 정해진 경우엔
|
||
`_cards_by_cut()` 을 쓴다. 폴더 지정 경로(파일/유튜브 구간 탭)는 계속 이 함수를 쓴다.
|
||
Returns: [(start, end, path)]
|
||
"""
|
||
imgs = _card_paths(folder)
|
||
if not imgs or dur <= 0:
|
||
return []
|
||
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 _cards_by_cut(paths, card_cuts, placements, dur: float, *,
|
||
min_sec: float = 3.0, fixed: bool = False):
|
||
"""카드를 '소속 컷 구간 안'에 배치 — 자동 탭 전용.
|
||
|
||
card_cuts[i] = i번째 카드가 속한 컷 인덱스. 컷 안에서는 균등 분할
|
||
(fixed=True 면 min_sec 고정, 컷 뒷부분은 비움).
|
||
|
||
⚠ placements/dur 은 **최종(무음 제거 반영) 타임라인** 기준이어야 한다.
|
||
압축 전 시간으로 계산해 놓고 나중에 재매핑하면 fixed=True 의 "정확히 3초" 약속이
|
||
깨지고(3초짜리가 2초로 눌린다) 3초 하한도 사라진다.
|
||
|
||
컷당 장수 상한 = max(1, floor(컷길이/min_sec)) — `_load_comment_cards` 와 같은 규칙.
|
||
초과분은 버린다(카드가 1초씩 번쩍이느니 몇 장 빼는 게 낫다).
|
||
|
||
왜 컷 단위인가: 전체 균등 배치(`_load_comment_cards`)는 컷 경계를 몰라서
|
||
3번 컷 얘기하는 댓글이 7번 컷 위에 뜬다. 컷 안에서 계산하면 한 컷이 덜 차도
|
||
**다음 컷 카드가 앞으로 밀리지 않는다.**
|
||
Returns: [(start, end, path)] 시간순
|
||
"""
|
||
if not paths or not card_cuts or not placements or dur <= 0:
|
||
return []
|
||
groups: dict = {}
|
||
for i, ci in enumerate(card_cuts[:len(paths)]):
|
||
if isinstance(ci, int) and 0 <= ci < len(placements):
|
||
groups.setdefault(ci, []).append(i)
|
||
out = []
|
||
for ci, idxs in groups.items():
|
||
p0, p1 = placements[ci]
|
||
p1 = min(p1, dur)
|
||
if p1 <= p0:
|
||
continue # 영상 실제 길이 밖 컷은 버린다
|
||
n = min(len(idxs), max(1, int((p1 - p0) // min_sec))) # 카드당 min_sec 하한
|
||
idxs = idxs[:n]
|
||
step = min_sec if fixed else (p1 - p0) / n
|
||
for k, i in enumerate(idxs):
|
||
s = p0 + k * step
|
||
if s >= p1:
|
||
break # fixed 로 컷을 넘치면 뒤는 비운다
|
||
out.append((s, min(s + step, p1), paths[i]))
|
||
out.sort(key=lambda t: t[0])
|
||
return out
|
||
|
||
|
||
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 _remap_placements(placements, keep_sorted):
|
||
"""컷 구간 [(p0,p1)] 을 무음 제거 타임라인으로 재매핑.
|
||
|
||
⚠ 자막(`_remap_caps`)과 목적이 다르다. 자막은 '그 말이 나오는 시각'을 따라가면 되지만
|
||
카드는 **컷 구간 자체**를 옮겨야 한다. 카드 시간을 압축 전에 만들어 두고 나중에
|
||
자막처럼 재매핑하면 `cards_fixed`(정확히 3초)가 2초로 눌리고, 3초 하한도 사라진다.
|
||
구간을 먼저 옮기고 그 안에서 나누면 둘 다 지켜진다.
|
||
|
||
통째로 무음이라 사라진 컷은 (x, x) 빈 구간이 되고 `_cards_by_cut` 이 건너뛴다.
|
||
"""
|
||
return [(_elapsed_kept(keep_sorted, p0), _elapsed_kept(keep_sorted, p1))
|
||
for p0, p1 in placements]
|
||
|
||
|
||
def captions_for_places(captions, places, *, cap: int = 500):
|
||
"""컷 구간마다 그 구간에 걸친 자막을 이어붙인다 — 댓글 추천의 근거.
|
||
|
||
`captions`·`places` 둘 다 **같은(압축) 타임라인** 좌표여야 한다. 겹치는 부분이
|
||
조금이라도 있으면 그 컷의 말로 본다(경계에 닿기만 하는 건 제외).
|
||
cap 자에서 자른다 — 그 이상은 Gemini 토큰만 먹고 매칭 정확도가 안 오른다.
|
||
Returns: places 와 같은 길이의 문자열 리스트
|
||
"""
|
||
out = []
|
||
for p0, p1 in places:
|
||
parts = [txt for cs, ce, txt in captions if cs < p1 and ce > p0 and txt]
|
||
out.append(" ".join(" ".join(parts).split())[:cap])
|
||
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,
|
||
card_cuts: Optional[List[int]] = None,
|
||
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]
|
||
|
||
card_places = placements # 카드 배치 기준 구간(무음 제거 시 압축본으로 교체)
|
||
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)
|
||
# 카드는 '시간'이 아니라 '컷 구간'을 옮긴다 — 아래에서 이 구간 안에 나눠 넣는다.
|
||
card_places = _remap_placements(placements, 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)}개"}
|
||
|
||
# 댓글 카드(자동 탭): '몇 번 컷 소속'만 받아 여기서 시간을 만든다.
|
||
# 서버가 시간을 확정하면 무음 제거 때 자막만 당겨지고 카드는 혼자 어긋난다.
|
||
# ⚠ 무음 제거 **뒤**에 계산한다 — 압축 전 시간으로 만들어 재매핑하면
|
||
# cards_fixed(정확히 3초)가 눌리고 카드당 3초 하한이 사라진다.
|
||
cut_cards = _cards_by_cut(_card_paths(comments_dir), card_cuts or [],
|
||
card_places, timeline_dur, fixed=cards_fixed)
|
||
# 컷 소속이 지정됐으면 그 컷 구간 안, 아니면 폴더 전체 균등 배치
|
||
cards = cut_cards or _load_comment_cards(comments_dir, timeline_dur, fixed=cards_fixed)
|
||
if cards:
|
||
msg = f"댓글 카드 {len(cards)}개 하단 삽입"
|
||
if cut_cards:
|
||
# 무음 제거로 컷이 짧아지면 quotas_for(원본 길이)가 고른 장수보다
|
||
# _cards_by_cut(압축 후 길이)의 상한이 작아질 수 있다 — 그 차이를 조용히 삼키지 않는다.
|
||
dropped = len(card_cuts or []) - len(cut_cards)
|
||
msg += "(컷별 배치" + (f", {dropped}장은 컷 길이가 짧아 제외)" if dropped > 0 else ")")
|
||
else:
|
||
msg += "(전체 균등)"
|
||
yield {"type": "log", "msg": msg}
|
||
|
||
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),
|
||
},
|
||
}
|