"""컷별 댓글 추천 — 자동 탭 검토 화면용. 왜 별도 모듈인가: 배정 규칙(타임스탬프 우선·중복 제거·모드 분기)은 댓글 수집 (`comments.py`)과도, HTTP 처리(`server/app.py`)와도 책임이 다르다. 순수 함수로 떼어놔야 Gemini 없이 단위 검증이 된다. 설계 근거는 docs/superpowers/specs/2026-08-04-컷별-댓글-추천-design.md. """ from __future__ import annotations import json import urllib.request from typing import Dict, List, Optional from . import prompts from .comments import MAX_TIMES, match_ranges, match_slots, top_liked from .correct import _gemini_key # 카드 1장이 차지하는 기준 시간(초). pipeline._load_comment_cards(min_sec) 과 같은 값. CARD_SEC = 3.0 def quotas_for(cuts) -> List[int]: """컷별 카드 장수 — max(1, floor(컷길이 / CARD_SEC)). 짧은 컷도 최소 1장. round 가 아니라 floor 인 이유: `pipeline._cards_by_cut` 의 컷당 장수 상한(무음 제거 후 길이 기준)·`pipeline._load_comment_cards` 의 규칙과 같은 공식이어야, 여기서 고른 카드가 빌드 단계에서 말없이 잘려나가지 않는다(round 를 쓰면 5초 컷처럼 상한보다 1장 더 골라 조용히 버려지는 경우가 생겼다). """ return [max(1, int((c["end"] - c["start"]) // CARD_SEC)) for c in cuts] def build_cut_picks(cuts, comments, ai_picks: Optional[Dict[int, List[int]]], quotas) -> List[List[dict]]: """컷별 추천 확정 — 타임스탬프 우선, 남는 자리만 AI, 전 컷 통틀어 중복 금지. 타임스탬프가 AI보다 먼저인 이유: 그 컷의 **원본 구간**을 콕 집어 언급한 댓글은 근거가 확실하다. 추측(AI)을 이기게 둘 이유가 없다. 중복은 앞 컷이 가져간다(뒤 컷은 다음 후보로 밀린다). ai_picks: {컷인덱스: [댓글idx …]} — Gemini 실패 시 None/{} 어느 쪽을 넘겨도 타임스탬프만으로 채운다(`ai_pick_cuts` 는 실패를 None 으로 알린다). Returns: 컷별 [{"idx": int, "why": "ts"|"ai"}] """ ai_picks = ai_picks or {} valid = {c["idx"] for c in comments} used: set = set() out: List[List[dict]] = [] for i, cut in enumerate(cuts): quota = quotas[i] if i < len(quotas) else 0 picks: List[dict] = [] for idx in match_ranges(comments, [(cut["start"], cut["end"])]): if len(picks) >= quota: break if idx not in used: picks.append({"idx": idx, "why": "ts"}) used.add(idx) for idx in (ai_picks.get(i) or []): if len(picks) >= quota: break if idx in valid and idx not in used: picks.append({"idx": idx, "why": "ai"}) used.add(idx) out.append(picks) return out # Gemini에 넘길 댓글 후보 수. 더 늘려도 채택률이 안 오르고 토큰만 먹는다. AI_CANDIDATES = 150 TEXT_CAP = 200 # 댓글 본문 절단 길이 AI_MODEL = "gemini-2.5-flash" # correct.py 와 같은 무료 티어 모델 _ENDPOINT = ("https://generativelanguage.googleapis.com/v1beta/models/" "{model}:generateContent?key={key}") _SCHEMA = {"type": "array", "items": {"type": "object", "properties": { "cut": {"type": "integer"}, "picks": {"type": "array", "items": {"type": "integer"}}}, "required": ["cut", "picks"]}} def _candidate_lines(comments, limit: int = AI_CANDIDATES) -> str: """Gemini에 줄 댓글 후보 — 목차 댓글 제외, 좋아요순 상위 limit, 본문 절단.""" usable = [c for c in comments if len(c.get("times") or []) <= MAX_TIMES] usable.sort(key=lambda c: -c.get("likeCount", 0)) out = [] for c in usable[:limit]: text = " ".join(str(c.get("text") or "").split())[:TEXT_CAP] out.append(f"{c['idx']}. 👍{c.get('likeCount', 0)} / {text}") return "\n".join(out) def _cut_lines(cuts) -> str: """컷 목록 — 번호·길이·자막. 자막이 추천의 유일한 근거다.""" out = [] for i, c in enumerate(cuts): bottom = " ".join(str(c.get("bottom") or "").split()) or "(자막 없음)" effect = " ".join(str(c.get("effect") or "").split()) line = f"{i}. ({c['end'] - c['start']:.1f}초) {bottom}" if effect: line += f" [효과자막: {effect}]" out.append(line) return "\n".join(out) def _parse_ai(raw: str) -> Dict[int, List[int]]: """Gemini 응답(문자열) → {컷인덱스: [댓글idx …]}. 형식이 어긋나면 {}. ⚠ 여기서는 '깨진 응답'과 '아무것도 안 고름'이 둘 다 {} 다. 둘을 구분해야 하는 호출부(`ai_pick_cuts`)는 json.loads 를 직접 하고 `_rows_to_picks()` 를 쓴다. """ try: data = json.loads(raw) except (json.JSONDecodeError, TypeError): return {} if not isinstance(data, list): return {} return _rows_to_picks(data) def _rows_to_picks(data: list) -> Dict[int, List[int]]: """파싱된 배열 → {컷인덱스: [댓글idx …]}. 형식이 어긋난 항목은 조용히 버린다.""" out: Dict[int, List[int]] = {} for row in data: if not isinstance(row, dict): continue cut, picks = row.get("cut"), row.get("picks") if not isinstance(cut, int) or isinstance(cut, bool): continue if not isinstance(picks, list): continue idxs = [p for p in picks if isinstance(p, int) and not isinstance(p, bool)] if idxs: out[cut] = idxs return out def ai_pick_cuts(cuts, comments, quotas, *, model: str = AI_MODEL, key=None, timeout: float = 90.0) -> Optional[Dict[int, List[int]]]: """컷 자막으로 컷별 추천을 받는다. **예외를 올리지 않는다** — 호출부가 폴백한다. Returns: - dict — 성공(빈 dict 도 성공: "어울리는 게 없다"는 정상 답이다). 키가 없어도 {} — 키 없이 쓰는 것도 정상 사용이라 실패로 치지 않는다. - None — **실패**. HTTP·타임아웃·응답 파싱·프롬프트 치환 오류. 화면에 "AI 추천 실패" 경고를 띄우려면 이 둘을 구분해야 한다 (실패해도 타임스탬프 배정은 그대로 돌아가서 겉보기엔 멀쩡하다). plan.py._call() 은 parts[0] 에 영상 fileData 가 항상 들어가 재사용할 수 없다. correct.py 의 텍스트 전용 호출 패턴(responseSchema + 실패 시 안전 반환)을 따른다. """ key = key if key is not None else _gemini_key() if not key or not cuts or not comments: return {} # 부를 이유가 없다 — 실패가 아니다 k = max(quotas) + 2 if quotas else 3 # 갈아끼울 여유분 try: prompt = prompts.load_recommend().format( cuts=_cut_lines(cuts), comments=_candidate_lines(comments), k=k) except (KeyError, IndexError, ValueError, OSError): # 사용자가 프롬프트를 깨뜨린 경우 return None body = { "contents": [{"parts": [{"text": prompt}]}], "generationConfig": {"temperature": 0.3, "responseMimeType": "application/json", "responseSchema": _SCHEMA}, } req = urllib.request.Request( _ENDPOINT.format(model=model, key=key), data=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=timeout) as resp: data = json.loads(resp.read().decode("utf-8")) raw = data["candidates"][0]["content"]["parts"][0]["text"] except (KeyError, IndexError, json.JSONDecodeError, ValueError, OSError): # URLError, TimeoutError are OSError subclasses; ValueError covers UnicodeDecodeError return None # 429 포함 — 폴백은 호출부 몫 try: rows = json.loads(raw) except (json.JSONDecodeError, TypeError): return None # 응답이 JSON 이 아니다 = 실패 if not isinstance(rows, list): return None # 스키마와 다른 모양 = 실패 return _rows_to_picks(rows) def is_whole(cuts) -> bool: """통짜 하이라이트인가 — 컷 1개 + 자막 없음(app._whole_hl 이 만드는 모양). 통짜는 자막이 없어 내용 추천의 근거가 없다. 대신 타임라인 시각 = 원본 시각이라 시각으로 정확히 맞출 수 있다(스펙 §4.1). """ return len(cuts) == 1 and not (cuts[0].get("bottom") or "").strip() def _whole_picks(cut, comments, n: int) -> List[dict]: """통짜 — 슬롯마다 그 시간대 언급 댓글, 빈 슬롯은 좋아요 상위로 메움.""" slots = match_slots(comments, cut["start"], cut["end"] - cut["start"], n) used = {i for i in slots if i is not None} no_ts = [c for c in comments if not c.get("times")] fill = iter(top_liked(no_ts, used, n)) out: List[dict] = [] for s in slots: if s is not None: out.append({"idx": s, "why": "ts"}) continue nxt = next(fill, None) if nxt is not None: out.append({"idx": nxt, "why": "like"}) return out def build_highlight_cuts(hl, comments, *, key=None): """하이라이트 하나 → (cuts[], need, ai_failed). 모드는 컷 모양으로 판별한다. Returns: ([{"i","sec","bottom","quota","picks"}], need, ai_failed) need = Σ quota — 기존 int(total//3) 을 대체한다(컷 경계에 맞추는 쪽이 맞다). ai_failed = Gemini 호출이 **실패**했는가(429·타임아웃·파싱). 통짜 모드는 Gemini 를 안 부르므로 항상 False. 호출부는 이 값으로만 경고를 띄운다 — 실패해도 cuts 는 타임스탬프만으로 채워져 비지 않으므로, cuts 가 비었는지로는 실패를 알 수 없다. Gemini 실패는 여기서 흡수된다(ai_pick_cuts 가 None 을 준다) — 예외를 올리지 않는다. """ cuts = (hl.get("paste") or {}).get("cuts") or [] if not cuts: return [], 0, False quotas = quotas_for(cuts) ai_failed = False if is_whole(cuts): picks = [_whole_picks(cuts[0], comments, quotas[0])] else: ai = ai_pick_cuts(cuts, comments, quotas, key=key) ai_failed = ai is None picks = build_cut_picks(cuts, comments, ai, quotas) out = [{"i": i, "sec": round(c["end"] - c["start"], 1), "bottom": c.get("bottom") or "", "quota": quotas[i], "picks": picks[i]} for i, c in enumerate(cuts)] return out, sum(quotas), ai_failed