# 2~4단계: 세 탭 모두 받아쓰기 후 댓글 매칭 — 구현 계획 > **에이전트 작업자용:** 이 계획은 `superpowers:subagent-driven-development`(권장) 또는 > `superpowers:executing-plans`로 태스크 단위로 실행한다. 단계는 체크박스(`- [ ]`)로 추적한다. **목표:** ▶ 유튜브 구간 · 📋 붙여넣기 · 🤖 자동 세 탭이 모두 `다운로드 → 무음 → 받아쓰기 → 댓글 매칭 → 검토 → 드래프트` 순으로 돌게 한다. **접근:** 1단계에서 쪼갠 `*_analyze`/`*_draft` 위에 얹는다. 서버 쪽 조립 로직을 `recommend.cuts_from_state()` 하나로 모아 세 탭이 같은 코드를 쓰고, 화면은 이미 통합된 `renderCutPanel()`을 재사용한다. 탭마다 다른 것은 **편집안을 어디서 얻느냐**뿐이다. **기술 스택:** Python 3.13 / FastAPI(SSE) / 바닐라 JS / 표준 라이브러리만 **스펙:** `docs/superpowers/specs/2026-08-04-받아쓰기후-댓글매칭-design.md` **선행:** `docs/superpowers/plans/2026-08-04-1단계-파이프라인-분할.md` (완료) ## Global Constraints - **코드를 고쳤으면 `캡컷_에이전트_구간합치기.bat`을 재시작한다.** 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). - **서버 실행·브라우저 열기 금지.** 실사용 확인은 사용자가 한다. - **네트워크는 꼭 필요할 때만.** 순수 함수 검증에는 쓰지 마라. - **좌표계를 섞지 마라 (이 계획에서 가장 위험):** - **매칭(⭐)** = 원본 영상 시각 — 댓글의 `9:05`와 맞춰야 한다 - **자막 추출·배치·장수** = 압축 타임라인(무음 제거 후) — 실제 자리다 - 배정 순위는 `⭐ts → 🤖ai → 🔤word → ➕like` (라운드 방식, 이미 구현됨). - 카드 1장 기준 3.0초(`recommend.CARD_SEC`), 컷별 장수 `max(1, floor(길이/3))`. - **한국어 docstring·주석**, 표준 라이브러리만, 기존 스타일. 바닐라 JS(빌드 도구 금지). - **어떤 경우에도 드래프트 생성을 막지 마라**(다운로드 실패 제외). - git 브랜치 `feat/yt-range-comments`. 태스크마다 커밋. 메시지는 한국어 한 줄 요약 + 왜, 마지막 줄에 `Co-Authored-By: Claude Opus 5 (1M context) `. - `assets/frame_template.png`는 파이프라인이 매 실행 다시 굽는 산출물 — 더러워지면 `git checkout --`로 되돌려라. --- ## 이미 있는 부품 (전부 검증 완료, 그대로 쓴다) | 부품 | 위치 | 하는 일 | |---|---|---| | `bg_analyze` / `bg_draft` / `bg_steps` | `pipeline.py` | 파일·구간 탭 파이프라인 두 조각 | | `paste_analyze` / `paste_draft` / `paste_steps` | `pipeline.py` | 자동(컷 모드)·붙여넣기 탭 두 조각 | | `captions_for_places(captions, places, *, cap=500)` | `pipeline.py` | 압축 좌표 구간별 자막 이어붙이기 | | `_remap_placements(placements, keep)` | `pipeline.py` | 원본 누적 구간 → 압축 타임라인 | | `_cards_by_cut(paths, card_cuts, places, dur, *, fixed)` | `pipeline.py` | 카드를 컷 구간 안에 배치 | | `build_highlight_cuts(hl, comments, *, key, quotas)` | `recommend.py` | `(cuts, need, ai_failed)` | | `is_time_based(cuts)` | `recommend.py` | 모든 컷 자막이 비면 시각 기반 | | `renderCutPanel(box, panelId, data, opts)` | `auto.js` | 컷별 섹션·배지·검색·상한 | | `/auto/analyze` → `/auto/stream/{aid}` → `/auto/build` | `app.py` | **세 탭이 따라야 할 본보기** | | `ANALYSES` / `JOBS` 메모리 dict | `app.py` | 분석·작업 보관 | --- ## 파일 구조 | 파일 | 이 계획에서의 책임 | |---|---| | `capcut_agent/pipeline.py` | `bg_analyze`가 구간 위치를 state에 담게 한다 (Task 1) | | `capcut_agent/recommend.py` | **`cuts_from_state()`** — 상태+댓글 → `cuts[]`. 세 탭 공통 (Task 2) | | `server/app.py` | 탭별 analyze/stream/build 엔드포인트 (Task 3·5·7) | | `server/static/index.html` | 탭별 화면 골격 (Task 4·6·7) | | `server/static/auto.js` | 탭별 흐름 제어. 렌더는 `renderCutPanel` 재사용 (Task 4·6·7) | --- ## Task 1: `bg_analyze`가 구간 위치를 state에 담는다 ▶ 구간 탭이 "구간 N이 압축 타임라인의 어디인지"를 알아야 자막 추출·배치·장수가 된다. 지금 `bg_analyze`는 유튜브 구간을 다운로드만 하고 그 경계를 안 남긴다. **Files:** - Modify: `capcut_agent/pipeline.py` (`bg_analyze`) **Interfaces:** - Produces: `bg_analyze` state에 세 키 추가 - `ranges_sec: List[Tuple[float,float]]` — 구간의 **원본 영상 시각**(초). 유튜브가 아니면 `[]` - `raw_places: List[Tuple[float,float]]` — 병합본(무음 제거 전) 누적 구간 - `places: List[Tuple[float,float]]` — **압축 타임라인** 구간 (`_remap_placements(raw_places, keep)`) - 파일 탭(유튜브 아님)은 세 키 모두 `[]` — 래퍼가 안 쓰므로 동작 불변 - [ ] **Step 1: 검증 스크립트 작성** ```python import sys; sys.stdout.reconfigure(encoding='utf-8') from capcut_agent.pipeline import _remap_placements # 구간 3개(원본 16:15~16:35, 15:30~16:15, 23:20~23:52) → 병합본 누적 rs=[(975.0,995.0),(930.0,975.0),(1400.0,1432.0)] raw,c=[],0.0 for s,e in rs: raw.append((c,c+(e-s))); c+=e-s assert raw==[(0.0,20.0),(20.0,65.0),(65.0,97.0)], raw # 무음 제거로 앞 10초가 통째로 잘린 경우 keep=[(10.0,97.0)] assert _remap_placements(raw,keep)==[(0.0,10.0),(10.0,55.0),(55.0,87.0)], _remap_placements(raw,keep) print("Task1 계산식 OK") ``` - [ ] **Step 2: 실패 확인** — 위 스크립트는 기존 함수만 쓰므로 **통과한다.** 이 태스크의 실패 확인은 state 키다: ```python import sys, inspect, re; sys.stdout.reconfigure(encoding='utf-8') from capcut_agent import pipeline src=inspect.getsource(pipeline.bg_analyze) m=re.search(r'"state":\s*\{(.+?)\n\s*\}\s*\}', src, re.S) keys=set(re.findall(r'"(\w+)":', m.group(1))) missing={"ranges_sec","raw_places","places"}-keys assert not missing, f"state 에 없음: {missing}" print("Task1 상태키 OK") ``` Run 후 Expected: FAIL — `AssertionError: state 에 없음: {...}` - [ ] **Step 3: 구현** `bg_analyze`의 유튜브 다운로드 분기에서 구간을 **초**로 파싱해 보관하고, 무음 분석 뒤 압축 좌표로 옮긴다. 시간 문자열 → 초 변환은 `capcut_agent/paste.py`의 `parse_time`을 쓴다 (이미 `mm:ss` / `hh:mm:ss`를 처리한다). ```python # 구간 탭이 "구간 N이 타임라인의 어디인지" 알아야 자막·배치·장수를 컷 단위로 낼 수 있다. # 파일 탭(유튜브 아님)은 빈 리스트 → 아무 데서도 안 쓰인다. ranges_sec: List[Tuple[float, float]] = [] raw_places: List[Tuple[float, float]] = [] if youtube: from .paste import parse_time c = 0.0 for s, e in (youtube.get("ranges") or []): ss, ee = parse_time(str(s)), parse_time(str(e)) ranges_sec.append((ss, ee)) raw_places.append((c, c + (ee - ss))) c += ee - ss ``` 무음 분석(`keep`)이 끝난 뒤: ```python places = _remap_placements(raw_places, keep) if raw_places else [] ``` state에 `ranges_sec`, `raw_places`, `places` 세 키를 더한다. - [ ] **Step 4: 통과 확인** — 두 스크립트 모두 통과 - [ ] **Step 5: 회귀 확인 — 파일 탭 이벤트 스트림 불변** 1단계에서 쓴 덤프 스크립트를 그대로 재사용한다: ```bash cd "D:/개인폴더/00.유튭/capcut2" python <스크래치패드>/dump_events.py <스크래치패드>/after-t1.json python -c " import json,sys; sys.stdout.reconfigure(encoding='utf-8') a=json.load(open(r'<스크래치패드>\before.json',encoding='utf-8')) b=json.load(open(r'<스크래치패드>\after-t1.json',encoding='utf-8')) assert a==b, '이벤트가 달라졌다' print('파일 탭 회귀 OK —',len(a),'이벤트 동일') " ``` 만든 테스트 드래프트(`__회귀테스트`)를 지운다. - [ ] **Step 6: 구문·임포트 검증 후 커밋** --- ## Task 2: `cuts_from_state()` — 세 탭 공통 조립 상태(받아쓰기 결과 + 구간 위치)와 댓글을 받아 검토 화면이 쓸 `cuts[]`를 만든다. **세 탭이 이 함수 하나만 부른다.** **Files:** - Modify: `capcut_agent/recommend.py` **Interfaces:** - Consumes: `pipeline.captions_for_places`, `build_highlight_cuts` - Produces: `cuts_from_state(places, orig_ranges, captions, comments, *, key=None) -> (cuts, need, ai_failed)` - `places` — 압축 타임라인 구간 `[(p0,p1)]` (자막·장수·배치 기준) - `orig_ranges` — 같은 길이의 **원본 영상 시각** `[(s,e)]` (⭐ 매칭 기준) - `captions` — 압축 타임라인 기준 `[(s,e,text)]` - 반환은 `build_highlight_cuts`와 같은 3-튜플 - [ ] **Step 1: 실패하는 검증 스크립트 작성** ```python import sys; sys.stdout.reconfigure(encoding='utf-8') from capcut_agent.recommend import cuts_from_state places=[(0.0,10.0),(10.0,22.0)] # 압축 타임라인 orig =[(975.0,995.0),(930.0,975.0)] # 원본 시각 (구간 순서대로) caps =[(1.0,3.0,"이거 100만원짜리 와인이야"), (11.0,13.0,"넉살이 랩을 진짜 잘하더라고")] cs=[{"idx":0,"likeCount":145,"times":[],"text":"요즘 넉살님 좋아요 랩도 잘하신다"}, {"idx":1,"likeCount":95,"times":[],"text":"100만원짜리 와인 맛이 다른가요"}, {"idx":2,"likeCount":80,"times":[980.0],"text":"여기 진짜 웃김"}, {"idx":3,"likeCount":50,"times":[],"text":"봄 대단하넼"}] cuts,need,failed=cuts_from_state(places,orig,caps,cs,key="") # key="" → Gemini 안 씀 assert [c["i"] for c in cuts]==[0,1] assert [c["quota"] for c in cuts]==[3,4], [c["quota"] for c in cuts] # floor(10/3), floor(12/3) assert need==7 and failed is False assert cuts[0]["bottom"]=="이거 100만원짜리 와인이야", cuts[0]["bottom"] assert cuts[1]["bottom"]=="넉살이 랩을 진짜 잘하더라고" # ⭐ 는 원본 시각 기준 — 980초는 구간0(975~995) 안이다 w0=[p for p in cuts[0]["picks"] if p["idx"]==2] assert w0 and w0[0]["why"]=="ts", cuts[0]["picks"] # 🔤 는 자막 단어 기준 — '넉살' 이 구간1 에 붙어야 한다 assert any(p["idx"]==0 and p["why"]=="word" for p in cuts[1]["picks"]), cuts[1]["picks"] # 자막이 하나도 없으면 시각 기반 경로로 떨어진다(예외 없이) c2,n2,f2=cuts_from_state(places,orig,[],cs,key="") assert n2==7 and f2 is False and all(c["bottom"]=="" for c in c2) assert cuts_from_state([],[],[],cs,key="")==([],0,False) print("Task2 OK") ``` - [ ] **Step 2: 실패 확인** Expected: FAIL — `ImportError: cannot import name 'cuts_from_state'` - [ ] **Step 3: 구현** `capcut_agent/recommend.py` 끝에 추가한다. ```python def cuts_from_state(places, orig_ranges, captions, comments, *, key=None): """받아쓰기 상태 + 댓글 → 검토 화면용 (cuts[], need, ai_failed). 세 탭 공통. ⚠ 좌표계가 둘이다. 섞으면 카드가 통째로 어긋난다(스펙 §4): - `places`·`captions` = 압축 타임라인 → 자막 추출·장수·배치 - `orig_ranges` = 원본 영상 시각 → ⭐ 분:초 매칭 둘은 같은 길이여야 하고 인덱스로만 짝지어 다닌다. """ from .pipeline import captions_for_places # 순환 임포트 회피 — 호출 시점에 가져온다 if not places or len(places) != len(orig_ranges): return [], 0, False bottoms = captions_for_places(captions, places) quotas = [max(1, int((p1 - p0) // CARD_SEC)) for p0, p1 in places] cuts = [{"start": s, "end": e, "bottom": b, "effect": ""} for (s, e), b in zip(orig_ranges, bottoms)] return build_highlight_cuts({"paste": {"cuts": cuts}}, comments, key=key, quotas=quotas) ``` ⚠ `pipeline`을 파일 상단에서 임포트하면 순환이 된다(`pipeline` → `draft`·`comments`, 그리고 `app.py`가 둘 다 임포트). **함수 안에서 임포트해라.** 그 이유를 주석에 남겨라. - [ ] **Step 4: 통과 확인** → `Task2 OK` - [ ] **Step 5: 구문·임포트 검증 후 커밋** --- ## Task 3: 📋 붙여넣기 탭 — 서버 **Files:** - Modify: `server/app.py` **Interfaces:** - Produces: - `POST /paste/analyze` (form `data`) → `{"analysis_id": aid}` — `parse_paste`로 검증만 - `GET /paste/stream/{aid}` (SSE) — `paste_analyze` 실행 + h-lab 댓글 수집 → `result` 이벤트 `{cuts, need, total, comments, candidates, warnings}` - `POST /paste/build` (form `aid`, `cards`, `card_cuts`, 옵션들) → `{"job_id": h}` - `PSTATES: dict[str, dict]` — `aid` → `{"state": …, "places": …, "orig": …, "payload": …}` - [ ] **Step 1: `/paste/analyze` 추가** `/auto/analyze` 바로 아래에 둔다. `parse_paste(data)`가 던지면 400. ```python @app.post("/paste/analyze") async def paste_analyze_start(data: str = Form(...)) -> JSONResponse: """📋 붙여넣기 탭 1단계 — 편집안 검증 후 분석 예약. 실제 작업은 /paste/stream 에서.""" try: payload = parse_paste(data) except ValueError as e: return JSONResponse({"error": str(e)}, 400) aid = hashlib.sha1(("paste|" + data).encode()).hexdigest()[:12] ANALYSES[aid] = {"payload": payload} return JSONResponse({"analysis_id": aid}) ``` - [ ] **Step 2: `/paste/stream/{aid}` 추가** `paste_analyze`를 돌리면서 이벤트를 흘리고, `state` 이벤트를 잡아 보관한 뒤 h-lab 댓글과 합쳐 `cuts[]`를 만든다. manifest에는 `recommend` 스텝을 넣는다. 핵심 골격(기존 `/auto/stream`의 패턴을 따른다): ```python @app.get("/paste/stream/{aid}") async def paste_stream(aid: str) -> StreamingResponse: """📋 붙여넣기 탭 — 다운로드·무음·받아쓰기 → 댓글 매칭 → 검토 화면용 result.""" a = ANALYSES.get(aid) async def gen(): if not a: yield _sse({"type": "error", "message": "알 수 없는 분석입니다."}) return payload = a["payload"] warnings: list[str] = [] yield _sse({"type": "manifest", "steps": pipeline_paste_steps(True) + [{"id": "comments", "label": "댓글 수집 (h-lab)"}, {"id": "recommend", "label": "컷별 댓글 추천"}]}) com_task = asyncio.create_task( asyncio.to_thread(hlab.fetch_comments, payload["url"])) state = None async for ev in paste_analyze(payload, "paste_" + aid, remove_silence=True, asr_bottom=True): if ev.get("type") == "state": state = ev["state"] continue if ev.get("type") == "error": com_task.cancel() yield _sse(ev) return yield _sse(ev) … ``` ⚠ **`remove_silence=True, asr_bottom=True` 로 고정한다.** 추천을 위해 받아쓰기가 항상 필요하기 때문이다(스펙 §5). 사용자가 화면에서 끄는 옵션은 **화면 자막에만** 적용되며 그건 build 단계에서 처리한다 — 이 태스크에서는 analyze를 항상 그렇게 돌린다는 것만 지킨다. 댓글 수집 후: ```python comments = [] 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": "실패(생략)"}) yield _sse({"type": "step", "id": "recommend", "status": "start"}) places = state["card_places"] orig = [(s, e) for s, e, _, _ in state["cuts"]] cuts, need, ai_failed = await asyncio.to_thread( recommend.cuts_from_state, places, orig, state["bottom_caps"], comments) if ai_failed: warnings.append("AI 추천 실패(Gemini 응답 없음) — 분:초·자막단어·좋아요로 배정했습니다") yield _sse({"type": "step", "id": "recommend", "status": "done", "detail": f"{len(cuts)}컷 · 카드 {need}장"}) PSTATES[aid] = {"state": state, "places": places} no_ts = [c for c in comments if not c["times"]] matched = hlab.match_ranges(comments, orig) if comments else [] yield _sse({"type": "result", "cuts": cuts, "need": need, "total": round(state["timeline_dur"], 1), "cutRanges": [{"start": s, "end": e} for s, e in orig], "matched": matched, "candidates": hlab.top_liked(no_ts, set(matched), len(no_ts)), "comments": comments, "warnings": warnings}) ``` `PSTATES: dict[str, dict] = {}` 를 `ANALYSES` 옆에 선언한다. `pipeline_paste_steps`는 `from capcut_agent.pipeline import paste_steps as pipeline_paste_steps` 같은 식으로 가져온다(기존 import 줄에 더해라). - [ ] **Step 3: `/paste/build` 추가** `aid`로 보관한 상태를 꺼내 `paste_draft`만 돌리는 job을 만든다. 카드 저장은 `/auto/build`의 코드를 그대로 따른다(`COMMENTS_DIR//001.png…`). ```python @app.post("/paste/build") async def paste_build(aid: str = Form(...), cards: list[UploadFile] = File(default=[]), card_cuts: str = Form(""), video_scale: str = Form("144"), flip: str = Form(""), scene: str = Form(""), bg_white: str = Form(""), cards_fixed: str = Form(""), asr_bottom: str = Form("1")) -> JSONResponse: """📋 붙여넣기 탭 3단계 — 보관한 상태로 드래프트만 만든다(다운로드·받아쓰기 안 함).""" st = PSTATES.get(aid) if not st: return JSONResponse({"error": "분석 결과가 만료됐습니다. 다시 분석해 주세요."}, 404) … JOBS[h] = {"paste_state": st["state"], "card_cuts": cut_map, …} return JSONResponse({"job_id": h}) ``` `/stream/{job_id}`가 `paste_state`가 있으면 `paste_draft`를 부르도록 분기를 더한다 (`job.get("paste")` 분기 옆). ⚠ **`asr_bottom`이 꺼져 있으면** 화면 자막을 JSON `bottom`으로 되돌려야 한다. `state["bottom_caps"]`는 받아쓰기 결과로 덮여 있으므로, 끈 경우 원래 JSON 자막으로 다시 만들어 넣어라(`state["cuts"]`와 `state["placements"]`로 계산 — `paste_analyze`가 하던 것과 같은 식). 이 처리를 빼면 옵션이 무시된다. - [ ] **Step 4: 구문·임포트 검증** ```bash python -c "import ast; ast.parse(open('server/app.py', encoding='utf-8').read())" python -c "from server import app; print('import OK')" ``` - [ ] **Step 5: 라우트 등록 확인** ```python import sys; sys.stdout.reconfigure(encoding='utf-8') from server.app import app paths={r.path for r in app.routes} for p in ("/paste/analyze","/paste/stream/{aid}","/paste/build"): assert p in paths, f"{p} 없음: {sorted(paths)}" print("Task3 라우트 OK") ``` - [ ] **Step 6: 커밋** --- ## Task 4: 📋 붙여넣기 탭 — 화면 **Files:** - Modify: `server/static/index.html`, `server/static/auto.js` **Interfaces:** - Consumes: Task 3의 세 엔드포인트, 기존 `renderCutPanel(box, panelId, data, opts)` - Produces: `panelId = "paste"` 패널. 기존 붙여넣기 탭 실행 버튼이 분석을 돌린다 - [ ] **Step 1: 화면 골격** 붙여넣기 탭의 실행 버튼을 **`댓글 매칭 시작`**으로 바꾸고, 그 아래에 결과 영역 `
`와 생성 버튼 `