hehihoho3@gmail.com
005beba4c3
통짜 모드(컷 1개 + 자막 없음)는 시각 슬롯으로 배정, 컷 있는 모드는 Gemini 추천. is_whole() 판별, _whole_picks() 슬롯 채우기, build_highlight_cuts() 조립. import에 match_slots, top_liked 추가. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
204 lines
8.4 KiB
Python
204 lines
8.4 KiB
Python
"""컷별 댓글 추천 — 자동 탭 검토 화면용.
|
|
|
|
왜 별도 모듈인가: 배정 규칙(타임스탬프 우선·중복 제거·모드 분기)은 댓글 수집
|
|
(`comments.py`)과도, HTTP 처리(`server/app.py`)와도 책임이 다르다. 순수 함수로 떼어놔야
|
|
Gemini 없이 단위 검증이 된다.
|
|
|
|
설계 근거는 docs/superpowers/specs/2026-08-04-컷별-댓글-추천-design.md.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Dict, List
|
|
|
|
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, round(컷길이 / CARD_SEC)). 짧은 컷도 최소 1장."""
|
|
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]],
|
|
quotas) -> List[List[dict]]:
|
|
"""컷별 추천 확정 — 타임스탬프 우선, 남는 자리만 AI, 전 컷 통틀어 중복 금지.
|
|
|
|
타임스탬프가 AI보다 먼저인 이유: 그 컷의 **원본 구간**을 콕 집어 언급한 댓글은
|
|
근거가 확실하다. 추측(AI)을 이기게 둘 이유가 없다.
|
|
중복은 앞 컷이 가져간다(뒤 컷은 다음 후보로 밀린다).
|
|
|
|
ai_picks: {컷인덱스: [댓글idx …]} — Gemini 실패 시 {} 를 넘기면 타임스탬프만으로 채운다.
|
|
Returns: 컷별 [{"idx": int, "why": "ts"|"ai"}]
|
|
"""
|
|
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 …]}. 형식이 어긋난 항목은 조용히 버린다."""
|
|
try:
|
|
data = json.loads(raw)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return {}
|
|
if not isinstance(data, list):
|
|
return {}
|
|
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) -> Dict[int, List[int]]:
|
|
"""컷 자막으로 컷별 추천을 받는다. **실패하면 예외 없이 {}** — 호출부가 폴백한다.
|
|
|
|
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 {}
|
|
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 {} # 429 포함 — 폴백은 호출부 몫
|
|
return _parse_ai(raw)
|
|
|
|
|
|
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). 모드는 컷 모양으로 판별한다.
|
|
|
|
Returns: ([{"i","sec","bottom","quota","picks"}], need)
|
|
need = Σ quota — 기존 int(total//3) 을 대체한다(컷 경계에 맞추는 쪽이 맞다).
|
|
Gemini 실패는 여기서 흡수된다(ai_pick_cuts 가 {} 를 준다) — 예외를 올리지 않는다.
|
|
"""
|
|
cuts = (hl.get("paste") or {}).get("cuts") or []
|
|
if not cuts:
|
|
return [], 0
|
|
quotas = quotas_for(cuts)
|
|
if is_whole(cuts):
|
|
picks = [_whole_picks(cuts[0], comments, quotas[0])]
|
|
else:
|
|
ai = ai_pick_cuts(cuts, comments, quotas, key=key)
|
|
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)
|