diff --git a/capcut_agent/recommend.py b/capcut_agent/recommend.py index 0d3865a..f7e4b60 100644 --- a/capcut_agent/recommend.py +++ b/capcut_agent/recommend.py @@ -9,9 +9,8 @@ Gemini 없이 단위 검증이 된다. from __future__ import annotations import json -import urllib.error import urllib.request -from typing import Dict, List +from typing import Dict, List, Optional from . import prompts from .comments import MAX_TIMES, match_ranges, match_slots, top_liked @@ -26,7 +25,7 @@ def quotas_for(cuts) -> List[int]: return [max(1, round((c["end"] - c["start"]) / CARD_SEC)) for c in cuts] -def build_cut_picks(cuts, comments, ai_picks: Dict[int, List[int]], +def build_cut_picks(cuts, comments, ai_picks: Optional[Dict[int, List[int]]], quotas) -> List[List[dict]]: """컷별 추천 확정 — 타임스탬프 우선, 남는 자리만 AI, 전 컷 통틀어 중복 금지. @@ -34,9 +33,11 @@ def build_cut_picks(cuts, comments, ai_picks: Dict[int, List[int]], 근거가 확실하다. 추측(AI)을 이기게 둘 이유가 없다. 중복은 앞 컷이 가져간다(뒤 컷은 다음 후보로 밀린다). - ai_picks: {컷인덱스: [댓글idx …]} — Gemini 실패 시 {} 를 넘기면 타임스탬프만으로 채운다. + 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]] = [] @@ -97,13 +98,22 @@ def _cut_lines(cuts) -> str: def _parse_ai(raw: str) -> Dict[int, List[int]]: - """Gemini 응답 → {컷인덱스: [댓글idx …]}. 형식이 어긋난 항목은 조용히 버린다.""" + """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): @@ -120,21 +130,28 @@ def _parse_ai(raw: str) -> Dict[int, List[int]]: def ai_pick_cuts(cuts, comments, quotas, *, model: str = AI_MODEL, - key=None, timeout: float = 90.0) -> Dict[int, List[int]]: - """컷 자막으로 컷별 추천을 받는다. **실패하면 예외 없이 {}** — 호출부가 폴백한다. + 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 {} + 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 {} + return None body = { "contents": [{"parts": [{"text": prompt}]}], "generationConfig": {"temperature": 0.3, @@ -151,8 +168,14 @@ def ai_pick_cuts(cuts, comments, quotas, *, model: str = AI_MODEL, raw = data["candidates"][0]["content"]["parts"][0]["text"] except (KeyError, IndexError, json.JSONDecodeError, ValueError, OSError): # URLError, TimeoutError are OSError subclasses; ValueError covers UnicodeDecodeError - return {} # 429 포함 — 폴백은 호출부 몫 - return _parse_ai(raw) + 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: @@ -182,22 +205,27 @@ def _whole_picks(cut, comments, n: int) -> List[dict]: def build_highlight_cuts(hl, comments, *, key=None): - """하이라이트 하나 → (cuts[], need). 모드는 컷 모양으로 판별한다. + """하이라이트 하나 → (cuts[], need, ai_failed). 모드는 컷 모양으로 판별한다. - Returns: ([{"i","sec","bottom","quota","picks"}], need) + Returns: ([{"i","sec","bottom","quota","picks"}], need, ai_failed) need = Σ quota — 기존 int(total//3) 을 대체한다(컷 경계에 맞추는 쪽이 맞다). - Gemini 실패는 여기서 흡수된다(ai_pick_cuts 가 {} 를 준다) — 예외를 올리지 않는다. + 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 + 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) + return out, sum(quotas), ai_failed diff --git a/server/app.py b/server/app.py index 3e85c73..4d48125 100644 --- a/server/app.py +++ b/server/app.py @@ -525,6 +525,7 @@ async def auto_stream(aid: str) -> StreamingResponse: # 댓글 매칭 — 전체 전송, 브라우저가 '더보기'로 30장씩 나눠 그린다. # 후보(candidates)는 분:초 언급이 아예 없는 댓글만 — 타임스탬프 댓글은 # 자기 구간의 ⭐에서 잡히므로, 다른 구간 얘기하는 댓글이 섞이지 않게. + # no_ts = [c for c in comments if not c["times"]] for h in highlights: if "paste" not in h: @@ -535,10 +536,15 @@ async def auto_stream(aid: str) -> StreamingResponse: # 컷별 추천 — 실패해도 위의 matched/candidates 로 화면이 돌아간다. # Gemini 호출이 섞여 있어 블로킹이므로 스레드로 뺀다. try: - cuts, need = await asyncio.to_thread( + cuts, need, ai_failed = await asyncio.to_thread( recommend.build_highlight_cuts, h, comments) if cuts: h["cuts"], h["need"] = cuts, need + if ai_failed: + # cuts 는 타임스탬프만으로 채워져 비지 않는다 — 이 신호가 없으면 + # 429·타임아웃이 조용히 삼켜져 아무도 모른다. + warnings.append( + f"ID {h.get('id')} AI 추천 실패 — 타임스탬프만으로 배정했습니다") except Exception as exc: # noqa: BLE001 — 추천 실패가 생성을 막으면 안 된다 warnings.append( f"ID {h.get('id')} 컷별 추천 실패 — 기존 방식으로 표시 "