capcut-agent/capcut_agent/youtube.py
hehihoho3@gmail.com bf1b387d6d chore: git 저장소 초기화 (기존 코드 스냅샷)
컷별 댓글 추천 작업을 태스크 단위로 되돌릴 수 있게 버전관리를 시작한다.
.gitignore 로 영상·캐시(.downloads 2.7G, .comments 72M, .media 28M)와
비밀키(.gemini_key)를 제외했다.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-04 11:36:05 +09:00

443 lines
19 KiB
Python

"""유튜브 구간 잘라받기 (ClipCut 로직 흡수). yt-dlp --download-sections.
캡컷 에이전트 앞단: URL + 시작/끝 → 그 구간만 h264 mp4 로 받아 파이프라인에 투입.
"""
from __future__ import annotations
import glob
import hashlib
import json
import os
import re
import shutil
import subprocess
from typing import List, Tuple
def _js_runtime_args() -> List[str]:
"""yt-dlp 유튜브 추출용 JS 런타임 지정. 설치된 것(deno/node/bun) 자동 선택.
최신 유튜브는 JS 챌린지 때문에 런타임이 없으면 포맷 누락→ffmpeg 크래시가 난다.
없으면 빈 리스트(사용자가 Node.js 등 설치 필요).
"""
for rt in ("deno", "node", "bun"):
if shutil.which(rt):
return ["--js-runtimes", rt]
return []
# MM:SS 또는 HH:MM:SS
TIME_RE = re.compile(r"^(?:\d{1,2}:)?\d{1,2}:[0-5]\d$")
TIMEOUT_SEC = 1800
def valid_time(t: str) -> bool:
return bool(TIME_RE.match(t.strip()))
def _video_codec(path: str) -> str:
try:
out = subprocess.run(
["ffprobe", "-v", "error", "-select_streams", "v:0",
"-show_entries", "stream=codec_name", "-of", "json", path],
capture_output=True, text=True, encoding="utf-8", errors="replace", check=True,
).stdout
return json.loads(out)["streams"][0]["codec_name"]
except Exception:
return ""
# ── 초록 프레임(GOP 중간 컷) 검증·복구 ──────────────────────────────────
# yt-dlp --download-sections 가 가끔 키프레임이 아닌 위치에서 스트림 복사로 잘라
# 파일을 만든다. 그러면 첫 키프레임 전까지는 참조 프레임이 없어 디코딩이 안 되고,
# 그 파일을 concat 재인코딩할 때 그 구간이 통째로 '초록 화면'으로 구워진다.
# ⚠ 함정: 파트를 단독 재생하면 ffmpeg 가 깨진 앞부분을 건너뛰어서 멀쩡해 보이고,
# ffprobe -read_intervals 도 키프레임으로 시크해버려 못 잡는다.
# → 시크 없이 프레임을 훑어 '첫 키프레임 시각'을 봐야 한다(0 이 아니면 깨진 것).
KEYFRAME_TOL = 0.05 # 첫 키프레임이 이보다 늦으면 앞부분 깨짐으로 판정(초)
DL_ATTEMPTS = 2 # 깨진 결과 재다운로드 횟수(간헐적 실패용). 그 뒤엔 재컷으로 확정 해결
RECUT_LEAD = 6.0 # 최후 수단: 앞에 이만큼 여유를 받아 로컬에서 다시 자름(초)
# 이번 요청에서 초록 깨짐을 고친 내역(파이프라인이 읽어 SSE 로그로 흘림).
# 로컬 단일 사용자 앱이라 모듈 전역으로 둔다 — 동시 작업 시 로그가 섞일 수 있으나 무해.
REPAIR_LOG: List[str] = []
def _first_keyframe_sec(path: str, max_frames: int = 3000) -> float:
"""첫 비디오 키프레임의 시각(초). 0 이면 맨 앞이 키프레임 = 정상.
시크를 쓰지 않고(=`-read_intervals` 금지) 앞에서부터 훑는다. 첫 키프레임을 만나면
즉시 종료하므로 정상 파일에서는 한 줄만 읽고 끝난다.
판정 불가(ffprobe 실패 등)면 0.0 → 통과시킨다(다운로드를 살리는 쪽으로).
"""
try:
proc = subprocess.Popen(
["ffprobe", "-v", "error", "-select_streams", "v:0",
"-show_entries", "frame=key_frame,pts_time", "-of", "csv=p=0", path],
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
text=True, encoding="utf-8", errors="replace",
)
except OSError:
return 0.0
try:
for i, line in enumerate(proc.stdout or []):
if i >= max_frames:
break
f = line.strip().split(",")
if len(f) >= 2 and f[0] == "1":
try:
return float(f[1])
except ValueError:
return 0.0
return 0.0
finally:
# ⚠ Windows: 파이프를 안 닫으면 ffprobe 가 파일을 계속 잡고 있어
# 바로 뒤따르는 삭제/덮어쓰기가 PermissionError 로 실패한다.
try:
if proc.stdout:
proc.stdout.close()
except OSError:
pass
proc.kill()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
pass
def _duration(path: str) -> float:
try:
out = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "json", path],
capture_output=True, text=True, encoding="utf-8", errors="replace", check=True,
).stdout
return float(json.loads(out)["format"]["duration"])
except Exception:
return 0.0
def _encode_h264(src: str, dst: str, *, ss: float = 0.0, t: float = 0.0) -> str:
"""h264/aac 로 재인코딩. ss/t 는 '-i 뒤'에 둬서 프레임 정확(출력 시크)."""
cmd = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", src]
if ss > 0:
cmd += ["-ss", f"{ss:.3f}"]
if t > 0:
cmd += ["-t", f"{t:.3f}"]
cmd += ["-c:v", "libx264", "-preset", "veryfast", "-crf", "20",
"-pix_fmt", "yuv420p", "-c:a", "aac", "-ar", "44100", dst]
subprocess.run(cmd, check=True, encoding="utf-8", errors="replace")
return dst
def _unlink(path: str) -> None:
try:
os.remove(path)
except OSError:
pass
def cut_youtube(url: str, start: str, end: str, out_dir: str) -> Tuple[str, str, str]:
"""유튜브 [start,end] 구간을 h264 mp4 로 받아 (경로, 제목, 채널명) 반환.
- h264(avc1)+aac 우선 → pycapcut/CapCut 호환(AV1/VP9 함정 회피). 안 되면 best.
- 받은 게 h264 아니면 ffmpeg 로 h264 재인코딩.
- 채널명(uploader)도 함께 받아 출처 자동 입력에 사용.
"""
url, start, end = url.strip(), start.strip(), end.strip()
if not (url.startswith("http://") or url.startswith("https://")):
raise ValueError("올바른 URL이 아닙니다.")
if not valid_time(start) or not valid_time(end):
raise ValueError("시간 형식 오류. 예) 03:30 또는 01:03:30")
os.makedirs(out_dir, exist_ok=True)
base_suffix = f"{start}-{end}".replace(":", "")
def _dl(section: str, tag: str = "") -> Tuple[str, str, str]:
"""tag 는 재시도용 파일명 구분자. 제목에선 떼어내 드래프트 이름을 깨끗이 유지."""
suffix = base_suffix + tag
outtmpl = os.path.join(out_dir, f"%(title).80s_{suffix}.%(ext)s")
cmd = [
"yt-dlp", *_js_runtime_args(),
"--download-sections", section,
# h264(avc1) + aac 우선, 안되면 best
"-f", "bv*[vcodec^=avc1]+ba[acodec^=mp4a]/bv*[vcodec^=avc1]+ba/b[ext=mp4]/b",
"--merge-output-format", "mp4", "--no-playlist",
# 채널명(uploader) + 최종 경로 출력 (탭 구분, after_move 시점)
"--no-simulate", "--print", "after_move:%(uploader)s\t%(filepath)s",
"-o", outtmpl, url,
]
# yt-dlp 가 한글 경로를 UTF-8 로 출력하도록 강제(Windows cp949 디코드 깨짐 방지)
env = {**os.environ, "PYTHONIOENCODING": "utf-8", "PYTHONUTF8": "1"}
proc = subprocess.run(cmd, capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=TIMEOUT_SEC, env=env)
if proc.returncode != 0:
tail = (proc.stderr or proc.stdout or "").strip()[-800:]
raise RuntimeError(f"yt-dlp 실패:\n{tail}")
# stdout 에서 채널명 파싱 (경로는 폴더 스캔으로 확정 — 인코딩 의존 제거)
ch = ""
for line in (proc.stdout or "").splitlines():
if "\t" in line:
ch = line.split("\t", 1)[0].strip()
# 출력 폴더에서 suffix 매칭 최신 파일 = 결과물 (stdout 경로 인코딩에 의존 안 함)
cands = [c for c in glob.glob(os.path.join(out_dir, f"*_{suffix}.*"))
if c.lower().endswith((".mp4", ".mkv", ".webm"))]
if not cands:
raise RuntimeError("다운로드 결과 파일을 찾지 못했습니다.")
p = max(cands, key=os.path.getmtime)
t = os.path.splitext(os.path.basename(p))[0]
if tag and t.endswith(tag):
t = t[: -len(tag)]
if _video_codec(p) != "h264": # CapCut 호환(AV1/VP9 회피)
h264 = os.path.splitext(p)[0] + "_h264.mp4"
_encode_h264(p, h264)
_unlink(p)
p = h264
return p, t, ch
# GOP 중간 컷(앞부분 초록) 검증 → 재시도 → 최후엔 앞 여유 붙여 로컬 재컷.
# 상세는 cut_youtube_precise 주석 참고(같은 문제·같은 전략).
last_lead = 0.0
for attempt in range(1, DL_ATTEMPTS + 1):
path, title, channel = _dl(f"*{start}-{end}", "" if attempt == 1 else f"r{attempt}")
last_lead = _first_keyframe_sec(path)
if last_lead <= KEYFRAME_TOL:
if attempt > 1:
REPAIR_LOG.append(f"{start}~{end}: 앞부분 초록 → 재다운로드로 해결")
return path, title, channel
_unlink(path)
REPAIR_LOG.append(f"{start}~{end}: 앞부분 초록 {last_lead:.1f}s → 여유분 재다운로드 후 정밀 재컷")
start_sec, end_sec = _hms_to_sec(start), _hms_to_sec(end)
want = end_sec - start_sec
lead = min(max(RECUT_LEAD, last_lead * 2 + 2.0), start_sec)
if lead <= 0:
raise RuntimeError(
f"구간 앞부분이 계속 깨집니다(첫 키프레임 {last_lead:.2f}s). "
"시작 시각을 조금 뒤로 옮겨 다시 시도하세요.")
path, title, channel = _dl(f"*{_fmt_hms(start_sec - lead)}-{_fmt_hms(end_sec)}", "lead")
recut = os.path.splitext(path)[0] + "_recut.mp4"
_encode_h264(path, recut, ss=max(0.0, _duration(path) - want), t=want)
_unlink(path)
return recut, title, channel
def _concat_parts(parts: List[str], out_dir: str, key: str) -> str:
"""h264 mp4 조각들을 concat demuxer 로 재인코딩 병합 → 합친 파일 경로.
⚠ Windows ffmpeg 의 concat demuxer 는 목록 파일 '내부'의 non-ASCII(한글) 경로를
열지 못한다("Impossible to open ... Invalid argument"). argv 로 직접 넘기는 경로는
되지만 목록 안 경로는 유니코드 변환이 안 되기 때문. → 조각들을 ASCII 임시 이름
(하드링크, 같은 볼륨이라 데이터 복사 없음)으로 가리켜 목록에 넣고, 병합 후 임시
링크·목록만 지운다. 목록의 상대 basename 은 목록 파일이 있는 폴더 기준으로 해석된다.
"""
list_txt = os.path.join(out_dir, f"_concat_{key}.txt")
merged = os.path.join(out_dir, f"merged_{key}.mp4")
tmp_links: List[str] = []
try:
with open(list_txt, "w", encoding="utf-8") as f:
for i, p in enumerate(parts):
ext = os.path.splitext(p)[1] or ".mp4"
link = os.path.join(out_dir, f"_cpart_{key}_{i}{ext}")
if os.path.abspath(link) == os.path.abspath(p):
name = os.path.basename(p) # 이미 ASCII 임시명 (이론상 없음)
else:
try:
if os.path.exists(link):
os.remove(link)
os.link(p, link) # 하드링크 (데이터 복사 없음)
except OSError:
shutil.copy2(p, link) # 폴백: 복사 (다른 볼륨 등)
tmp_links.append(link)
name = os.path.basename(link)
f.write(f"file '{name}'\n")
subprocess.run(
["ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
"-f", "concat", "-safe", "0", "-i", list_txt,
"-c:v", "libx264", "-preset", "veryfast", "-crf", "20",
"-pix_fmt", "yuv420p", "-c:a", "aac", "-ar", "44100", merged],
check=True, encoding="utf-8", errors="replace",
)
return merged
finally:
for link in tmp_links:
try:
os.remove(link)
except OSError:
pass
try:
os.remove(list_txt)
except OSError:
pass
def cut_youtube_multi(
url: str,
ranges: List[Tuple[str, str]],
out_dir: str,
) -> Tuple[str, str, str]:
"""한 URL의 여러 [start,end] 구간을 각각 받아 순서대로 이어붙인 h264 mp4 반환.
Returns: (합친영상경로, 제목, 채널명). 구간이 1개면 그대로 반환(병합 생략).
각 구간은 cut_youtube 로 h264/aac 통일 → concat demuxer 재인코딩으로 안전 병합.
"""
if not ranges:
raise ValueError("구간이 하나도 없습니다.")
os.makedirs(out_dir, exist_ok=True)
parts: List[str] = []
title = channel = ""
for i, (s, e) in enumerate(ranges):
p, t, ch = cut_youtube(url, s, e, out_dir)
parts.append(p)
if i == 0:
title, channel = t, ch
if len(parts) == 1:
return parts[0], title, channel
key = hashlib.sha1("|".join(f"{s}-{e}" for s, e in ranges).encode()).hexdigest()[:8]
merged = _concat_parts(parts, out_dir, key)
merged_title = f"{title}_{len(ranges)}구간" if title else f"merged_{key}"
return merged, merged_title, channel
def _hms_to_sec(t: str) -> float:
"""MM:SS 또는 HH:MM:SS → 초. (구간 탭 입력 → 로컬 재컷 계산용)"""
parts = [float(x) for x in t.strip().split(":")]
sec = 0.0
for p in parts:
sec = sec * 60 + p
return sec
def _fmt_hms(sec: float) -> str:
"""초(float) → HH:MM:SS.mmm (yt-dlp download-sections·정밀 컷용)."""
sec = max(0.0, float(sec))
h = int(sec // 3600)
m = int((sec % 3600) // 60)
s = sec % 60
return f"{h:02d}:{m:02d}:{s:06.3f}"
def _dl_section(url: str, start_sec: float, end_sec: float, out_dir: str,
tag: str = "") -> Tuple[str, str, str]:
"""[start_sec, end_sec] 구간 1회 다운로드 → (경로, 제목, 채널). h264 로 정규화.
1) --force-keyframes-at-cuts (프레임 정확) → 2) 실패 시 키프레임 컷 폴백.
tag 는 파일명 suffix 에 섞어 재시도 결과가 이전 파일과 안 섞이게 한다.
"""
section = f"*{_fmt_hms(start_sec)}-{_fmt_hms(end_sec)}"
suffix = hashlib.sha1((section + tag).encode()).hexdigest()[:10]
outtmpl = os.path.join(out_dir, f"%(title).60s_{suffix}.%(ext)s")
env = {**os.environ, "PYTHONIOENCODING": "utf-8", "PYTHONUTF8": "1"}
def _run(extra):
cmd = [
"yt-dlp", *_js_runtime_args(),
"--download-sections", section, *extra,
"-f", "bv*[vcodec^=avc1]+ba[acodec^=mp4a]/bv*[vcodec^=avc1]+ba/b[ext=mp4]/b",
"--merge-output-format", "mp4", "--no-playlist",
"--no-simulate", "--print", "after_move:%(uploader)s\t%(filepath)s",
"-o", outtmpl, url,
]
return subprocess.run(cmd, capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=TIMEOUT_SEC, env=env)
proc = _run(["--force-keyframes-at-cuts"])
if proc.returncode != 0:
proc = _run([])
if proc.returncode != 0:
tail = (proc.stderr or proc.stdout or "").strip()[-800:]
raise RuntimeError(f"yt-dlp 실패:\n{tail}")
channel = ""
for line in (proc.stdout or "").splitlines():
if "\t" in line:
channel = line.split("\t", 1)[0].strip()
cands = [c for c in glob.glob(os.path.join(out_dir, f"*_{suffix}.*"))
if c.lower().endswith((".mp4", ".mkv", ".webm"))]
if not cands:
raise RuntimeError("다운로드 결과 파일을 찾지 못했습니다.")
path = max(cands, key=os.path.getmtime)
title = os.path.splitext(os.path.basename(path))[0]
if _video_codec(path) != "h264": # CapCut 호환(AV1/VP9 회피)
h264 = os.path.splitext(path)[0] + "_h264.mp4"
_encode_h264(path, h264)
_unlink(path)
path = h264
return path, title, channel
def cut_youtube_precise(url: str, start_sec: float, end_sec: float,
out_dir: str) -> Tuple[str, str, str]:
"""[start_sec, end_sec] (밀리초 정밀) 구간을 프레임 정확히·초록 없이 잘라 h264 mp4 로.
yt-dlp 가 가끔 GOP 중간에서 잘라 앞부분이 깨진(초록) 파일을 준다. 그대로 두면
병합 재인코딩 때 초록 화면이 구워지므로 매번 검증한다:
1) 받은 파일의 첫 키프레임이 맨 앞이면 정상 → 그대로 사용
2) 아니면 버리고 재다운로드 (DL_ATTEMPTS 회) — 간헐적 실패라 대개 여기서 해결
3) 그래도 깨지면 앞에 RECUT_LEAD 초 여유를 붙여 받아 로컬에서 뒤쪽 want 초만
재인코딩으로 잘라낸다. 깨진 앞부분은 버리는 여유 구간에 들어가므로 항상 깨끗하다.
"""
url = url.strip()
if not (url.startswith("http://") or url.startswith("https://")):
raise ValueError("올바른 URL이 아닙니다.")
if end_sec <= start_sec:
raise ValueError("end 는 start 보다 커야 합니다.")
os.makedirs(out_dir, exist_ok=True)
want = end_sec - start_sec
last_lead = 0.0
for attempt in range(1, DL_ATTEMPTS + 1):
path, title, channel = _dl_section(
url, start_sec, end_sec, out_dir, tag="" if attempt == 1 else f"r{attempt}")
last_lead = _first_keyframe_sec(path)
if last_lead <= KEYFRAME_TOL:
if attempt > 1:
REPAIR_LOG.append(f"{start_sec:.1f}~{end_sec:.1f}s: 앞부분 초록 → 재다운로드로 해결")
return path, title, channel
_unlink(path) # 앞 last_lead 초가 초록 → 버리고 다시
# 최후 수단: 앞 여유 + 로컬 정밀 재컷 (여유가 깨져도 어차피 버리는 부분)
REPAIR_LOG.append(
f"{start_sec:.1f}~{end_sec:.1f}s: 앞부분 초록 {last_lead:.1f}s → 여유분 재다운로드 후 정밀 재컷")
lead = min(max(RECUT_LEAD, last_lead * 2 + 2.0), start_sec)
if lead <= 0:
raise RuntimeError(
f"구간 앞부분이 계속 깨집니다(첫 키프레임 {last_lead:.2f}s). "
"시작 시각을 조금 뒤로 옮겨 다시 시도하세요.")
path, title, channel = _dl_section(url, start_sec - lead, end_sec, out_dir, tag="lead")
dur = _duration(path)
recut = os.path.splitext(path)[0] + "_recut.mp4"
_encode_h264(path, recut, ss=max(0.0, dur - want), t=want) # 뒤에서 want 초만
_unlink(path)
return recut, title, channel
def download_paste_cuts(url: str, cuts: List[Tuple[float, float]],
out_dir: str) -> Tuple[str, str, str]:
"""붙여넣기 컷들(초 단위, 정밀)을 각각 정확히 잘라 순서대로 병합.
Returns: (합친영상경로, 제목, 채널명). 1개면 병합 생략.
"""
if not cuts:
raise ValueError("컷이 하나도 없습니다.")
os.makedirs(out_dir, exist_ok=True)
parts: List[str] = []
title = channel = ""
for i, (s, e) in enumerate(cuts):
p, t, ch = cut_youtube_precise(url, s, e, out_dir)
parts.append(p)
if i == 0:
title, channel = t, ch
if len(parts) == 1:
return parts[0], title, channel
key = hashlib.sha1("|".join(f"{s:.3f}-{e:.3f}" for s, e in cuts).encode()).hexdigest()[:8]
merged = _concat_parts(parts, out_dir, key)
return merged, (f"{title}_{len(cuts)}" if title else f"paste_{key}"), channel