Gemini 추천 실패를 화면에 알린다 (조용히 삼키지 않음)

ai_pick_cuts 가 429·타임아웃·JSON 파싱 실패를 전부 {} 로 삼켜서,
build_highlight_cuts 는 여전히 타임스탬프만으로 채운 cuts 를 돌려주고
app.py 의 `if cuts:` 가 참이 되어 except 가 절대 타지 않았다.
스펙 §7이 요구한 "⚠ 추천 실패 → 기존 방식" 경고가 화면에 안 떴다.

- ai_pick_cuts: 실패는 None, 성공은 dict(빈 dict 포함).
  키가 없는 경우는 실패가 아니다 — 키 없이 쓰는 것도 정상 사용이라 {} 를 준다.
  실패로 치는 건 HTTP·타임아웃·응답 파싱·프롬프트 치환 오류뿐.
- 응답 파싱을 _rows_to_picks() 로 떼어내 '깨진 응답'(None)과
  '어울리는 게 없어 빈 배열'({})을 구분한다. _parse_ai() 계약은 그대로 둔다.
- build_highlight_cuts → (cuts, need, ai_failed) 3-튜플.
  통짜 모드는 Gemini 를 안 부르므로 ai_failed=False.
- build_cut_picks 는 ai_picks=None 이어도 안전(`ai_picks or {}`).
- app.py: ai_failed 면 warnings 에 "AI 추천 실패 — 타임스탬프만으로 배정" 한 줄.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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 None                                # 429 포함 — 폴백은 호출부 몫
-    return _parse_ai(raw)
+    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

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')} 컷별 추천 실패 — 기존 방식으로 표시 "