capcut-agent/docs/superpowers/plans/2026-08-04-1단계-파이프라인-분할.md
hehihoho3@gmail.com 6df1930d06 docs: 1단계 구현 계획 — 파이프라인 분할 + 추천 엔진 준비 + 렌더러 통합
겉보기 동작이 하나도 안 바뀌는 구조 변경이라, 분할 전 이벤트 스트림을 파일로 떠 두고
분할 후와 문자 그대로 비교하는 방식으로 회귀를 검증한다(Task 4·5).

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

32 KiB
Raw Permalink Blame History

1단계: 파이프라인 분할 + 추천 엔진 준비 + 렌더러 통합 — 구현 계획

에이전트 작업자용: 이 계획은 superpowers:subagent-driven-development(권장) 또는 superpowers:executing-plans로 태스크 단위로 실행한다. 단계는 체크박스(- [ ])로 추적한다.

목표: 두 파이프라인을 analyze/draft 두 조각으로 쪼개고, 추천 엔진이 컷별 자막·장수를 밖에서 받을 수 있게 만든다. 사용자에게 보이는 동작은 하나도 바뀌지 않는다.

접근: 각 파이프라인을 "받아쓰기까지"와 "드래프트 생성"으로 나누고, 중간 상태를 {"type":"state"} 이벤트로 넘긴다. 기존 함수는 두 조각을 연달아 부르는 얇은 래퍼로 남겨 SSE 이벤트 스트림이 문자 그대로 같게 유지한다.

기술 스택: Python 3.13 / FastAPI / 바닐라 JS / 표준 라이브러리만

스펙: docs/superpowers/specs/2026-08-04-받아쓰기후-댓글매칭-design.md (§3, §5, §6, §10, 13단계 표의 1단계)

Global Constraints

  • 코드를 고쳤으면 캡컷_에이전트_구간합치기.bat을 재시작한다. uvicorn hot-reload가 없다.
  • 테스트 프레임워크가 없다. pytest를 도입하지 마라. 검증은 인라인 assert 스크립트로 한다. 스크립트는 C:\Users\hehih\AppData\Local\Temp\claude\D-------00----capcut2\d2b52aec-0e0e-4485-b47f-a854f6b9c528\scratchpad 아래에 두고 저장소에 커밋하지 마라(이 프로젝트에는 테스트 디렉터리가 없다).
  • 검증 스크립트 첫 줄에 import sys; sys.stdout.reconfigure(encoding='utf-8') — 콘솔이 cp949다.
  • 네트워크를 쓰지 마라. 회귀 검증은 .downloads/의 기존 mp4를 재사용한다.
  • 이 단계에서 사용자에게 보이는 동작이 바뀌면 결함이다. 새 기능은 2~4단계에서 쓴다.
  • 카드 1장 기준 길이는 3.0초(recommend.CARD_SEC), 컷별 장수는 max(1, floor(길이/3)).
  • 한국어 docstring·주석, 표준 라이브러리만, 기존 capcut_agent/*.py 스타일.
  • 바닐라 JS. 빌드 도구·npm 패키지 도입 금지.
  • git 브랜치 feat/yt-range-comments. 태스크마다 커밋. 메시지는 한국어 한 줄 요약 + 왜 그렇게 했는지 본문, 마지막 줄에 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>.

파일 구조

파일 이 단계에서의 책임
capcut_agent/recommend.py 추천 배정. 컷별 자막·장수를 밖에서 받을 수 있게 한다
capcut_agent/pipeline.py 파이프라인 2개를 analyze/draft로 쪼갠다. 컷별 자막 추출 헬퍼를 더한다
server/static/auto.js 카드 패널 렌더러를 renderCutPanel() 하나로 모은다

server/app.py이 단계에서 안 건드린다 — 래퍼가 기존 시그니처를 유지하므로 호출부가 그대로다. 새 엔드포인트는 2단계부터.


Task 1: is_time_based() — 시각 기반 판별 일반화

지금 is_whole(cuts)컷 1개 + 자막 없음일 때만 참이다. 구간 탭은 자막 없는 구간이 2개 이상이라 안 걸린다. 모든 컷의 자막이 비면 시각 기반으로 일반화한다. 컷 1개 통짜는 그 특수 케이스가 되므로 자동 탭 동작은 그대로다.

Files:

  • Modify: capcut_agent/recommend.py

Interfaces:

  • Consumes: comments.match_slots, comments.top_liked

  • Produces:

    • is_time_based(cuts) -> bool (기존 is_whole 대체)
    • _time_based_picks(cuts, comments, quotas) -> List[List[dict]] — 컷마다 슬롯 배정, 컷 간 중복 금지. 기존 _whole_picks를 여러 컷으로 확장
  • Step 1: 실패하는 검증 스크립트 작성

import sys; sys.stdout.reconfigure(encoding='utf-8')
from capcut_agent.recommend import is_time_based, _time_based_picks

# 자막이 전부 비면 참 — 컷 개수와 무관
assert is_time_based([{"start":0.0,"end":6.0,"bottom":""}]) is True
assert is_time_based([{"start":0.0,"end":6.0,"bottom":""},
                      {"start":100.0,"end":106.0,"bottom":"  "}]) is True
# 하나라도 차 있으면 거짓
assert is_time_based([{"start":0.0,"end":6.0,"bottom":""},
                      {"start":100.0,"end":106.0,"bottom":"가"}]) is False
assert is_time_based([]) is False        # 컷이 없으면 판별 불가 → 거짓

cs = [
    {"idx":0,"likeCount":50,"times":[1.0]},    # 컷0 슬롯0
    {"idx":1,"likeCount":90,"times":[4.0]},    # 컷0 슬롯1
    {"idx":2,"likeCount":70,"times":[101.0]},  # 컷1 슬롯0
    {"idx":3,"likeCount":60,"times":[]},       # 분:초 없음
    {"idx":4,"likeCount":10,"times":[]},
]
cuts = [{"start":0.0,"end":6.0,"bottom":""}, {"start":100.0,"end":106.0,"bottom":""}]
got = _time_based_picks(cuts, cs, [2, 2])
assert got[0] == [{"idx":0,"why":"ts"}, {"idx":1,"why":"ts"}], got[0]
# 컷1: 슬롯0=idx2, 슬롯1 비어 좋아요 상위(분:초 없는 것)로 채움
assert got[1] == [{"idx":2,"why":"ts"}, {"idx":3,"why":"like"}], got[1]

# 컷 간 중복 금지: 같은 댓글이 두 컷 시간대에 걸쳐도 앞 컷이 가져간다
cs2 = [{"idx":0,"likeCount":50,"times":[1.0, 101.0]}]
cuts2 = [{"start":0.0,"end":3.0,"bottom":""}, {"start":100.0,"end":103.0,"bottom":""}]
g2 = _time_based_picks(cuts2, cs2, [1, 1])
assert g2[0] == [{"idx":0,"why":"ts"}] and g2[1] == [], g2

assert _time_based_picks([], cs, []) == []
print("Task1 OK")
  • Step 2: 실패 확인

Run: cd "D:/개인폴더/00.유튭/capcut2" && python <임시경로>/t1.py Expected: FAIL — ImportError: cannot import name 'is_time_based'

  • Step 3: 구현

먼저 옛 이름을 쓰는 곳이 더 없는지 확인한다(있으면 전부 고쳐야 한다):

cd "D:/개인폴더/00.유튭/capcut2" && grep -rn "is_whole\|_whole_picks" --include=*.py --include=*.js .

그다음 capcut_agent/recommend.pyis_whole()_whole_picks()를 아래로 대체한다 (옛 이름은 남기지 마라 — 같은 일을 하는 함수가 둘이면 다음 사람이 틀린 걸 고친다). build_highlight_cuts 안의 호출부도 함께 고친다.

def is_time_based(cuts) -> bool:
    """시각 기반으로 배정할 컷 묶음인가 — **모든 컷의 자막이 비었는가**.

    자막이 없으면 내용 추천의 근거가 없다. 대신 이런 묶음(통짜 모드·유튜브 구간 탭)은
    구간을 그대로 이어붙이므로 타임라인 시각 = 원본 시각이 성립해 시각으로 맞출 수 있다.
    컷 1개짜리 통짜는 이 규칙의 특수 케이스다(스펙 §1).
    """
    if not cuts:
        return False
    return all(not (c.get("bottom") or "").strip() for c in cuts)


def _time_based_picks(cuts, comments, quotas) -> List[List[dict]]:
    """컷마다 3초 슬롯 배정 — 슬롯 시간대 언급 댓글, 빈 슬롯은 좋아요 상위로 채움.

    `used` 를 컷 사이에 공유해 한 댓글이 두 컷에 들어가지 않게 한다(앞 컷 우선).
    """
    used: set = set()
    no_ts = [c for c in comments if not c.get("times")]
    out: List[List[dict]] = []
    for i, cut in enumerate(cuts):
        n = quotas[i] if i < len(quotas) else 0
        slots = match_slots(comments, cut["start"], cut["end"] - cut["start"], n,
                            exclude=used)
        used.update(s for s in slots if s is not None)
        fill = iter(top_liked(no_ts, used, n))
        picks: List[dict] = []
        for s in slots:
            if s is not None:
                picks.append({"idx": s, "why": "ts"})
                continue
            nxt = next(fill, None)
            if nxt is not None:
                picks.append({"idx": nxt, "why": "like"})
                used.add(nxt)
        out.append(picks)
    return out

build_highlight_cuts() 안의 분기를 바꾼다:

    if is_time_based(cuts):
        picks = _time_based_picks(cuts, comments, quotas)
    else:
  • Step 4: 통과 확인

Run: python <임시경로>/t1.pyTask1 OK

  • Step 5: 자동 탭 회귀 확인 (통짜 모드 동작 불변)
import sys; sys.stdout.reconfigure(encoding='utf-8')
from capcut_agent import recommend
cs = [{"idx":0,"likeCount":50,"times":[101.0]}, {"idx":1,"likeCount":90,"times":[104.0]},
      {"idx":2,"likeCount":70,"times":[]}, {"idx":3,"likeCount":60,"times":[]}]
whole = {"paste": {"cuts": [{"start":100.0,"end":106.0,"bottom":"","effect":""}]}}
cuts, need, failed = recommend.build_highlight_cuts(whole, cs)
assert need == 2 and failed is False
assert cuts[0]["picks"] == [{"idx":0,"why":"ts"}, {"idx":1,"why":"ts"}], cuts[0]["picks"]
print("Task1 회귀 OK")
  • Step 6: 구문·임포트 검증 후 커밋
cd "D:/개인폴더/00.유튭/capcut2"
python -c "import ast; ast.parse(open('capcut_agent/recommend.py', encoding='utf-8').read())"
python -c "from server import app; print('import OK')"
git add capcut_agent/recommend.py
git commit -m "..."

Task 2: build_highlight_cutsquotas 주입

무음 제거 후 실제 길이로 계산한 장수를 밖에서 넘길 수 있게 한다. 안 넘기면 기존 동작.

Files:

  • Modify: capcut_agent/recommend.py

Interfaces:

  • Consumes: Task 1의 is_time_based, _time_based_picks

  • Produces: build_highlight_cuts(hl, comments, *, key=None, quotas=None) -> (cuts, need, ai_failed)

  • Step 1: 실패하는 검증 스크립트 작성

import sys; sys.stdout.reconfigure(encoding='utf-8')
from capcut_agent import recommend

hl = {"paste": {"cuts": [
    {"start":0.0,"end":10.0,"bottom":"가","effect":""},
    {"start":100.0,"end":110.0,"bottom":"나","effect":""}]}}
cs = []

# 안 넘기면 기존대로 컷 길이(10초)로 계산 → floor(10/3)=3 씩
c1, n1, _ = recommend.build_highlight_cuts(hl, cs, key="")
assert [c["quota"] for c in c1] == [3, 3], [c["quota"] for c in c1]
assert n1 == 6

# 넘기면 그걸 쓴다 (무음 제거로 짧아진 경우)
c2, n2, _ = recommend.build_highlight_cuts(hl, cs, key="", quotas=[1, 2])
assert [c["quota"] for c in c2] == [1, 2], [c["quota"] for c in c2]
assert n2 == 3
print("Task2 OK")
  • Step 2: 실패 확인

Run: python <임시경로>/t2.py Expected: FAIL — TypeError: build_highlight_cuts() got an unexpected keyword argument 'quotas'

  • Step 3: 구현

build_highlight_cuts의 시그니처와 첫 줄들을 고친다:

def build_highlight_cuts(hl, comments, *, key=None, quotas=None):
    """하이라이트 하나 → (cuts[], need, ai_failed). 모드는 컷 모양으로 판별한다.

    quotas: 컷별 카드 장수를 밖에서 정해 넘길 때 쓴다(무음 제거 후 실제 길이 기준).
      안 넘기면 `quotas_for(cuts)` — 원본 컷 길이 기준.
    Returns: ([{"i","sec","bottom","quota","picks"}], need, ai_failed)
      need = Σ quota. Gemini 실패는 ai_failed=True 로 알린다(예외는 안 올린다).
    """
    cuts = (hl.get("paste") or {}).get("cuts") or []
    if not cuts:
        return [], 0, False
    quotas = list(quotas) if quotas is not None else quotas_for(cuts)

나머지 본문은 그대로 둔다.

  • Step 4: 통과 확인

Run: python <임시경로>/t2.pyTask2 OK

  • Step 5: 구문·임포트 검증 후 커밋 (Task 1 Step 6과 같은 명령)

Task 3: 컷별 자막 추출 헬퍼

받아쓰기 결과에서 "이 컷 구간에 걸친 자막"을 뽑는다. 2단계부터 세 탭이 다 쓴다.

Files:

  • Modify: capcut_agent/pipeline.py

Interfaces:

  • Produces: captions_for_places(captions, places, *, cap=500) -> List[str]places와 같은 길이. 각 원소는 그 구간에 걸친 자막을 공백 하나로 이어붙인 것

  • Step 1: 실패하는 검증 스크립트 작성

import sys; sys.stdout.reconfigure(encoding='utf-8')
from capcut_agent.pipeline import captions_for_places

caps = [(0.0, 2.0, "안녕"), (2.0, 4.0, "반가워"), (5.0, 7.0, "잘가"), (9.0, 11.0, "또봐")]
places = [(0.0, 5.0), (5.0, 10.0)]
got = captions_for_places(caps, places)
assert got == ["안녕 반가워", "잘가 또봐"], got

# 경계에 걸치면 포함(겹치는 부분이 있으면 그 컷의 말이다)
assert captions_for_places([(4.0, 6.0, "걸침")], [(0.0, 5.0), (5.0, 10.0)]) == ["걸침", "걸침"]
# 닿기만 하는 건 제외 (s == p1 또는 e == p0)
assert captions_for_places([(5.0, 6.0, "다음")], [(0.0, 5.0)]) == [""]
# 공백 정규화
assert captions_for_places([(0.0, 1.0, "  여러   칸  ")], [(0.0, 5.0)]) == ["여러 칸"]
# 길이 제한
assert captions_for_places([(0.0, 1.0, "가"*900)], [(0.0, 5.0)], cap=500) == ["가"*500]
assert captions_for_places([], [(0.0, 5.0)]) == [""]
assert captions_for_places(caps, []) == []
print("Task3 OK")
  • Step 2: 실패 확인

Run: python <임시경로>/t3.py Expected: FAIL — ImportError: cannot import name 'captions_for_places'

  • Step 3: 구현

capcut_agent/pipeline.py_remap_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
  • Step 4: 통과 확인

Run: python <임시경로>/t3.pyTask3 OK

  • Step 5: 구문·임포트 검증 후 커밋

Task 4: 회귀 기준선 뜨기 (분할 전 이벤트 스트림 저장)

파이프라인을 쪼개기 전에 실제 이벤트 스트림을 파일로 떠 둔다. 쪼갠 뒤 같은 걸 다시 떠서 비교하면 "겉보기 동작 불변"을 증거로 확인할 수 있다.

Files: 없음 (검증 산출물만, 스크래치패드에 둔다)

  • Step 1: 기준선 스크립트 작성

<스크래치패드>/dump_events.py:

import sys, os, json, asyncio, glob
sys.stdout.reconfigure(encoding='utf-8')
sys.path.insert(0, r"D:\개인폴더\00.유튭\capcut2")
from capcut_agent.pipeline import process_bg_template

OUT = sys.argv[1]
vids = sorted(glob.glob(r"D:\개인폴더\00.유튭\capcut2\.downloads\*.mp4"), key=os.path.getsize)
VIDEO = vids[0]                    # 가장 작은 것 — 받아쓰기가 빨리 끝난다
print("영상:", os.path.basename(VIDEO), round(os.path.getsize(VIDEO)/1e6, 1), "MB")

def scrub(ev):
    """실행마다 달라지는 값(소요시간)을 뺀다 — 비교 대상은 순서와 내용이다."""
    e = dict(ev)
    e.pop("elapsed", None)
    if e.get("type") == "result":
        e["stats"] = {k: v for k, v in e["stats"].items() if k != "elapsed"}
        e.pop("draft_path", None)
    return e

async def main():
    evs = []
    async for ev in process_bg_template(VIDEO, "__회귀테스트",
                                        title_top="윗줄", title_main="아랫줄", channel="@ch"):
        evs.append(scrub(ev))
        print(evs[-1].get("type"), evs[-1].get("id", ""), evs[-1].get("status", ""))
    json.dump(evs, open(OUT, "w", encoding="utf-8"), ensure_ascii=False, indent=1)
    print("저장:", OUT, len(evs), "이벤트")

asyncio.run(main())
  • Step 2: 기준선 뜨기

Run: cd "D:/개인폴더/00.유튭/capcut2" && python <스크래치패드>/dump_events.py <스크래치패드>/before.json

첫 실행은 Whisper 받아쓰기 때문에 몇 분 걸린다(캐시가 없으면). 끝까지 기다려라. 두 번째부터는 .cache/asr_*.json 캐시로 빨라진다.

Expected: 저장: …/before.json N 이벤트

  • Step 3: 만들어진 테스트 드래프트 삭제
cd "D:/개인폴더/00.유튭/capcut2"
python -c "
import sys, shutil, os; sys.stdout.reconfigure(encoding='utf-8')
from capcut_agent.draft import DEFAULT_DRAFT_ROOT as R
for n in os.listdir(R):
    if n.startswith('__회귀테스트'):
        shutil.rmtree(os.path.join(R, n), ignore_errors=True); print('삭제:', n)
"
  • Step 4: 커밋 없음

이 태스크는 산출물이 스크래치패드에만 있다. 커밋할 것이 없다. before.json 경로를 다음 태스크에 넘긴다.


Task 5: process_bg_template 분할

Files:

  • Modify: capcut_agent/pipeline.py:190-330 (process_bg_template)

Interfaces:

  • Produces:

    • bg_analyze(video_path, draft_name, *, title_top="", title_main="", channel="", youtube=None) -> AsyncIterator[dict] — 마지막에 {"type":"state","state":{…}}
    • bg_draft(state, *, video_scale=1.0, flip_horizontal=False, scene_split=False, comments_dir="", cards_fixed=False, bg_white=False, comment_cards=None) -> AsyncIterator[dict]
    • bg_steps(youtube) -> List[dict] — manifest용 스텝 목록
    • process_bg_template(...) — 위 셋을 엮는 래퍼. 시그니처·이벤트 불변
  • state 키: video_path, meta, keep, video_clips, captions, total, draft_name, title_top, title_main, channel, t_all

  • Step 1: 구현

process_bg_template을 아래 넷으로 나눈다. 본문은 옮기기만 하고 로직을 바꾸지 마라.

def bg_steps(youtube) -> List[Dict[str, str]]:
    """배경템플릿 파이프라인의 manifest 스텝. 쪼갠 두 조각이 함께 내는 전체 목록."""
    steps = ([{"id": "download", "label": "유튜브 여러 구간 다운로드·병합"}] if youtube else [])
    return steps + [{"id": "silence", "label": "무음·발화 분석"},
                    {"id": "asr", "label": "받아쓰기 (Gemini/Whisper)"},
                    {"id": "draft", "label": "템플릿 드래프트 생성"}]
  • bg_analyze() = 기존 206~287행([다운로드] → probe → silence → asr) 그대로. yield {"type": "manifest", …} 는 넣지 마라 — 호출부가 낸다. 마지막에 yield {"type": "state", "state": {…}}.
  • bg_draft(state, …) = 기존 289~330행(장면분할 → 카드 → 드래프트 → result) 그대로. comment_cards 인자가 주어지면 _load_comment_cards 대신 그걸 쓴다(2단계 준비).
  • 래퍼:
async def process_bg_template(video_path, draft_name, *, title_top="", title_main="",
                              channel="", video_scale=1.0, flip_horizontal=False,
                              scene_split=False, comments_dir="", cards_fixed=False,
                              bg_white=False, youtube=None) -> AsyncIterator[dict]:
    """배경템플릿 파이프라인 — 📁 파일 탭용 얇은 래퍼.

    analyze/draft 두 조각을 연달아 부른다. `state` 이벤트는 밖으로 안 흘린다
    (기존 UI가 모르는 타입이라 흘리면 로그에 정체불명 이벤트가 찍힌다).
    """
    yield {"type": "manifest", "steps": bg_steps(youtube)}
    state = None
    async for ev in bg_analyze(video_path, draft_name, title_top=title_top,
                               title_main=title_main, channel=channel, youtube=youtube):
        if ev.get("type") == "state":
            state = ev["state"]
            continue
        yield ev
    if state is None:
        return                      # analyze 가 error 로 끝난 경우
    async for ev in bg_draft(state, video_scale=video_scale,
                             flip_horizontal=flip_horizontal, scene_split=scene_split,
                             comments_dir=comments_dir, cards_fixed=cards_fixed,
                             bg_white=bg_white):
        yield ev

t_all(총 소요)은 bg_analyze 시작 시점에 재서 state에 담고, bg_draftresult가 그걸 쓴다. 그래야 래퍼의 result.stats.elapsed가 예전과 같은 의미다.

⚠ 기존 코드의 yield {"type": "error", …}; return 경로(오디오 전부 무음)는 bg_analyze 안에 그대로 둔다. 그때는 state 이벤트가 안 나오고 래퍼가 조용히 끝난다.

  • Step 2: 구문·임포트 검증
cd "D:/개인폴더/00.유튭/capcut2"
python -c "import ast; ast.parse(open('capcut_agent/pipeline.py', encoding='utf-8').read())"
python -c "from server import app; print('import OK')"
  • Step 3: 분할 후 이벤트 스트림 뜨기

Run: python <스크래치패드>/dump_events.py <스크래치패드>/after.json (받아쓰기는 캐시를 타므로 빠르다)

  • Step 4: 기준선과 비교 — 여기가 이 태스크의 핵심 검증
import sys, json; sys.stdout.reconfigure(encoding='utf-8')
a = json.load(open(r"<스크래치패드>\before.json", encoding="utf-8"))
b = json.load(open(r"<스크래치패드>\after.json", encoding="utf-8"))
assert len(a) == len(b), f"이벤트 개수 다름: {len(a)}{len(b)}"
for i, (x, y) in enumerate(zip(a, b)):
    assert x == y, f"{i}번째 이벤트 다름:\n  before={x}\n  after ={y}"
assert not any(e.get("type") == "state" for e in b), "state 이벤트가 밖으로 샜다"
print("Task5 회귀 OK —", len(a), "이벤트 동일")

Expected: Task5 회귀 OK — N 이벤트 동일

다르면 로직을 바꾼 것이다. 옮기기만 해야 한다. 되돌리고 다시 옮겨라.

  • Step 5: 테스트 드래프트 삭제 후 커밋 (Task 4 Step 3의 삭제 명령 재사용)

Task 6: process_paste 분할

Files:

  • Modify: capcut_agent/pipeline.py:374-560 (process_paste)

Interfaces:

  • Produces:

    • paste_analyze(payload, draft_name, *, remove_silence=False, asr_bottom=False, name_suffix="") -> AsyncIterator[dict] — 마지막에 {"type":"state","state":{…}}
    • paste_draft(state, *, video_scale=1.0, flip_horizontal=False, scene_split=False, comments_dir="", cards_fixed=False, card_cuts=None, bg_white=False) -> AsyncIterator[dict]
    • paste_steps(asr_bottom) -> List[dict]
    • process_paste(...) — 래퍼. 시그니처·이벤트 불변
  • state 키: video_path, meta, dur, cuts, placements, card_places, video_clips, timeline_dur, bottom_caps, eff_caps, draft_name, title_top, title_main, channel, t_all

  • Step 1: 구현

def paste_steps(asr_bottom: bool) -> List[Dict[str, str]]:
    """붙여넣기 파이프라인의 manifest 스텝."""
    steps = [{"id": "download", "label": "컷 정밀 다운로드·병합"}]
    if asr_bottom:
        steps.append({"id": "asr", "label": "받아쓰기 (Whisper)"})
    steps.append({"id": "draft", "label": "템플릿 드래프트 생성"})
    return steps
  • paste_analyze() = 기존 395~499행(다운로드·병합 → probe → placements/자막 → [무음 제거] → [asr_bottom]) 그대로. manifest는 넣지 마라. 마지막에 yield {"type": "state", "state": {…}}.
  • paste_draft(state, …) = 기존 501~558행(장면분할 → 카드 → 드래프트 → result) 그대로.
  • 래퍼는 Task 5와 같은 모양으로 쓴다:
async def process_paste(payload, draft_name, *, video_scale=1.0, flip_horizontal=False,
                        scene_split=False, comments_dir="", cards_fixed=False,
                        card_cuts=None, bg_white=False, remove_silence=False,
                        asr_bottom=False, name_suffix="") -> AsyncIterator[dict]:
    """붙여넣기(JSON) 파이프라인 — 기존 호출부용 얇은 래퍼.

    analyze/draft 두 조각을 연달아 부른다. `state` 이벤트는 밖으로 안 흘린다.
    """
    yield {"type": "manifest", "steps": paste_steps(asr_bottom)}
    state = None
    async for ev in paste_analyze(payload, draft_name, remove_silence=remove_silence,
                                  asr_bottom=asr_bottom, name_suffix=name_suffix):
        if ev.get("type") == "state":
            state = ev["state"]
            continue
        yield ev
    if state is None:
        return
    async for ev in paste_draft(state, video_scale=video_scale,
                                flip_horizontal=flip_horizontal, scene_split=scene_split,
                                comments_dir=comments_dir, cards_fixed=cards_fixed,
                                card_cuts=card_cuts, bg_white=bg_white):
        yield ev
  • Step 2: 구문·임포트 검증 (Task 5 Step 2와 같은 명령)

  • Step 3: 래퍼가 내는 manifest가 예전과 같은지 확인

import sys; sys.stdout.reconfigure(encoding='utf-8')
from capcut_agent.pipeline import paste_steps, bg_steps
assert paste_steps(False) == [{"id":"download","label":"컷 정밀 다운로드·병합"},
                              {"id":"draft","label":"템플릿 드래프트 생성"}]
assert paste_steps(True) == [{"id":"download","label":"컷 정밀 다운로드·병합"},
                             {"id":"asr","label":"받아쓰기 (Whisper)"},
                             {"id":"draft","label":"템플릿 드래프트 생성"}]
assert bg_steps(None)[0]["id"] == "silence"
assert bg_steps({"url":"x"})[0]["id"] == "download"
assert [s["id"] for s in bg_steps(None)] == ["silence","asr","draft"]
print("Task6 manifest OK")
  • Step 4: 상태 dict가 draft 단계가 쓰는 키를 다 담는지 확인
import sys, inspect, re; sys.stdout.reconfigure(encoding='utf-8')
from capcut_agent import pipeline
src = inspect.getsource(pipeline.paste_draft)
used = set(re.findall(r'state\["(\w+)"\]', src)) | set(re.findall(r'state\.get\("(\w+)"', src))
asrc = inspect.getsource(pipeline.paste_analyze)
m = re.search(r'"state":\s*\{(.+?)\n\s*\}\s*\}', asrc, re.S)
produced = set(re.findall(r'"(\w+)":', m.group(1)))
missing = used - produced
assert not missing, f"paste_analyze 가 안 만드는 키를 paste_draft 가 씀: {missing}"
print("Task6 상태키 OK —", sorted(produced))
  • Step 5: 커밋

Task 7: renderCutPanel() — 렌더러 통합

카드 패널을 그리는 코드가 자동 탭(onResult)과 구간 탭(ytMatch) 두 갈래다. 2~3단계에서 붙여넣기 탭까지 더하면 세 갈래가 되어 한 곳만 고치는 실수가 난다.

Files:

  • Modify: server/static/auto.js

Interfaces:

  • Produces: renderCutPanel(box, panelId, data, opts)

    • box — 카드 섹션을 붙일 DOM 요소
    • panelIdhl.id / "yt" / "paste"
    • data{cuts, matched, candidates} (cuts가 없으면 폴백 렌더)
    • opts{unit: "컷"|"구간"}
  • Step 1: 함수 추출

지금 onResult 안의 "검색창 → 컷별 섹션 → 후보" 렌더 블록과 ytMatch 안의 "검색창 → " 블록을 하나의 함수로 합친다. 동작 규칙은 지금 자동 탭 것을 따른다:

/* 카드 패널 렌더 — 자동/구간/붙여넣기 세 탭이 공유한다.
   갈라 두면 한 곳만 고치는 실수가 난다(배지·검색·선택 상한이 전부 여기 모여 있다). */
function renderCutPanel(box,panelId,data,opts){
  const unit=(opts&&opts.unit)||"컷";
  const WHY={ts:"⭐",ai:"🤖",like:""};
  box.appendChild(searchBar(panelId));      // cardSection 보다 먼저 — SECS 초기화
  const cand=(data.candidates||[]).filter(i=>byIdx[i]!==undefined);
  if(data.cuts){
    for(const cu of data.cuts){
      const rec=(cu.picks||[]).map(p=>p.idx).filter(i=>byIdx[i]!==undefined);
      const why=(cu.picks||[]).map(p=>WHY[p.why]||"").join("");
      const cs=(data.cutRanges||[])[cu.i]||{start:0,end:-1};
      const rest=(data.matched||[]).filter(i=>
        byIdx[i]!==undefined&&!rec.includes(i)&&
        (byIdx[i].times||[]).some(t=>t>=cs.start&&t<=cs.end));
      box.appendChild(cardSection(
        unit+" "+(cu.i+1)+" · "+cu.sec+"초 · 카드 "+cu.quota+"장"+
        (cu.bottom?"  — "+cu.bottom:"")+(why?"   "+why:""),
        rec.concat(rest),panelId,Math.max(CARD_PAGE,rec.length),cu.i));
      if(cand.length){
        box.appendChild(cardSection(
          "  ↳ 좋아요 상위에서 채우기",cand,panelId,CUT_FILL_PAGE,cu.i));
      }
    }
  }else{
    const m=(data.matched||[]).filter(i=>byIdx[i]!==undefined);
    box.appendChild(cardSection(
      "⭐ "+unit+"을 언급한 댓글 "+m.length+"장 (좋아요순, 자동 선택)",
      m,panelId,Math.max(CARD_PAGE,(sel[panelId]||[]).length)));
    if(cand.length){
      box.appendChild(cardSection(
        " 좋아요 상위 후보 "+cand.length+"장 (부족분 클릭)",cand,panelId,CARD_PAGE));
    }
  }
}

data.cutRanges = hl.paste.cuts(원본 시각 {start,end} 배열). 자동 탭은 hl.paste.cuts를, 구간 탭은 나중에 구간 목록을 넘긴다.

  • Step 2: onResult가 이 함수를 부르게 바꾼다

onResult의 렌더 블록을 아래로 대체한다(선택 상태 초기화·제목 UI는 그대로 둔다):

    renderCutPanel(box,hl.id,
      {cuts:hl.cuts,matched:hl.matched,candidates:hl.candidates,
       cutRanges:hl.paste.cuts},{unit:"컷"});
  • Step 3: ytMatch가 이 함수를 부르게 바꾼다

ytMatch의 렌더 블록(검색창 + + )을 아래로 대체한다:

  renderCutPanel(box,"yt",
    {matched:matched,candidates:cand},{unit:"구간"});

구간 탭은 아직 cuts가 없으므로 폴백 경로로 간다 — 지금과 같은 화면이다.

  • Step 4: 문법 검증

Run: cd "D:/개인폴더/00.유튭/capcut2" && node --check server/static/auto.js Expected: 출력 없음(성공)

  • Step 5: 중복 코드가 없어졌는지 확인
cd "D:/개인폴더/00.유튭/capcut2"
grep -c "좋아요 상위 후보" server/static/auto.js     # 1 이어야 한다 (renderCutPanel 안에만)
grep -c "searchBar(" server/static/auto.js           # 2 (정의 1 + renderCutPanel 안 1)
  • Step 6: 커밋

Task 8: 실사용 회귀 확인 + 문서

Files:

  • Modify: ARCHITECTURE.md

  • Step 1: 서버 재시작

검은 창을 닫고 캡컷_에이전트_구간합치기.bat을 다시 실행한다. 빼먹으면 아래가 헛일이다.

  • Step 2: 📁 파일 탭 확인

.downloads/의 mp4 하나를 파일 탭으로 올려 드래프트를 만든다. 확인:

  1. 진행 표시가 무음·발화 분석 → 받아쓰기 → 템플릿 드래프트 생성 순으로 나오는가
  2. 정체불명 이벤트(state)가 로그에 안 찍히는가
  3. CapCut에서 드래프트가 정상으로 열리는가
  • Step 3: 🤖 자동 탭 확인 (렌더러 통합 회귀)

자동 탭 📋 오팔 JSON 모드로 편집안 하나를 붙여넣어 검토 화면까지 간다. 확인:

  1. 컷별 섹션이 컷 N · X초 · 카드 Q장 — 자막 형태로 뜨는가
  2. 🤖 배지가 붙는가
  3. 검색창이 동작하는가(입력 → 필터, ✕ → 복원)
  4. 컷 장수를 채운 뒤 더 눌러도 선택이 안 되는가
  5. 빌드까지 가서 카드가 정상으로 들어가는가
  • Step 4: ▶ 유튜브 구간 탭 확인

💬 구간 댓글 매칭/ 두 덩어리가 지금과 똑같이 뜨는가. 검색도 되는가.

  • Step 5: 문서 갱신

ARCHITECTURE.md에 추가한다:

  • 모듈 설명에 파이프라인이 analyze/draft 두 조각으로 나뉜 것과 (댓글 매칭을 받아쓰기 뒤로 옮기기 위해)

  • state 이벤트는 내부 전용이며 래퍼가 걸러낸다는 것

  • captions_for_places()가 컷별 추천 근거를 만든다는 것

  • Step 6: 커밋


자체 점검 결과

스펙 커버리지 (1단계 해당분)

스펙 항목 담당 태스크
§3 파이프라인 분할, state 이벤트, 파일 탭 래퍼 Task 5, 6
§5 컷별 자막 추출(500자 컷) Task 3
§6 is_time_based 일반화, _whole_picks 공유 used Task 1
§6 quotas 주입 Task 2
§10 렌더러 통합 Task 7
§14 검증 1·2·3·4 (순수 함수) Task 1, 2, 3
§14 검증 6·7·8 (회귀) Task 4·5(자동 비교), Task 8(실사용)

이 단계에 없는 것 (2~4단계로): build_cut_picks 3순위( 채움) — 자동 탭의 현재 동작을 바꾸므로 실제로 쓰는 2단계에서 넣는다. 새 엔드포인트, 탭별 흐름 변경도 마찬가지. 스펙 §13의 1단계 설명("recommend.py 변경")은 이 범위를 뜻한다.

타입 일관성

  • build_highlight_cuts는 전 구간 (cuts, need, ai_failed) 3-튜플
  • cuts[] 원소 키는 i, sec, bottom, quota, picks; picks 원소는 {"idx","why"}
  • state는 dict 하나이며 {"type":"state","state":{…}}로 감싸 흐른다
  • captions_for_places(captions, places, *, cap) — 둘 다 압축 타임라인 좌표