hehihoho3@gmail.com
7c74fec079
컷 하나를 4순위까지 다 채운 뒤 다음 컷으로 넘어가면, 앞 컷의 약한 근거(4순위 좋아요)가 뒤 컷의 강한 근거(3순위 단어 겹침)보다 먼저 댓글을 가로챈다. 라운드를 바깥 루프로 두어(전 컷의 ts → 전 컷의 ai → 전 컷의 word → 전 컷의 like) 순위가 컷 순서를 이기게 했다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
352 lines
16 KiB
Python
352 lines
16 KiB
Python
"""컷별 댓글 추천 — 자동 탭 검토 화면용.
|
|
|
|
왜 별도 모듈인가: 배정 규칙(타임스탬프 우선·중복 제거·모드 분기)은 댓글 수집
|
|
(`comments.py`)과도, HTTP 처리(`server/app.py`)와도 책임이 다르다. 순수 함수로 떼어놔야
|
|
Gemini 없이 단위 검증이 된다.
|
|
|
|
설계 근거는 docs/superpowers/specs/2026-08-04-컷별-댓글-추천-design.md.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import time
|
|
import urllib.error
|
|
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
|
|
|
|
# ── 단어 겹침 매칭(3순위) — Gemini 가 실패해도(503 등) 네트워크 없이 항상 채워지는 폴백 ──
|
|
# 흔해서 아무 컷에나 걸리는 일반 단어. 걸리면 오탐이 늘어날 뿐이라 미리 뺀다.
|
|
STOP = {"진짜", "너무", "정말", "그냥", "이거", "저거", "우리", "사람", "이번", "그거", "완전", "진심",
|
|
"이렇게", "그렇게", "하는", "했다", "있는", "없는", "보고", "보는", "같아", "같은", "이건", "저건",
|
|
"근데", "그리고", "하지만", "합니다", "입니다"}
|
|
# 한글 2글자 이상 / 알파벳 3글자 이상 / 숫자 2글자 이상만 키워드 후보로 본다(1글자는 아무 데나 걸린다).
|
|
TOK = re.compile(r"[가-힣]{2,}|[A-Za-z]{3,}|\d{2,}")
|
|
# 긴 조사부터 검사해야 짧은 조사가 먼저 걸려 어간이 덜 잘리는 일이 없다.
|
|
JOSA = ("이야", "에서", "으로", "까지", "부터", "라고", "이고", "은", "는", "이", "가", "을", "를",
|
|
"의", "에", "도", "만", "와", "과", "랑")
|
|
|
|
|
|
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 _cut_keywords(text: str) -> List[str]:
|
|
"""자막 한 줄 → 댓글 매칭용 키워드(원본 토큰 + 조사 뗀 어간 후보).
|
|
|
|
"넉살이"(자막) 를 "넉살님"(댓글) 에 걸리게 하려면 조사를 떼야 한다. 뗀 결과가
|
|
2글자 미만이면 버린다(1글자 어간은 아무 댓글에나 걸려 오탐만 늘린다).
|
|
"""
|
|
out: List[str] = []
|
|
seen = set()
|
|
for tok in TOK.findall(text or ""):
|
|
if tok not in STOP and tok not in seen:
|
|
seen.add(tok)
|
|
out.append(tok)
|
|
for josa in JOSA:
|
|
if tok.endswith(josa) and len(tok) - len(josa) >= 2:
|
|
stem = tok[:-len(josa)]
|
|
if stem not in STOP and stem not in seen:
|
|
seen.add(stem)
|
|
out.append(stem)
|
|
break
|
|
return out
|
|
|
|
|
|
def _word_match_ranked(cut, comments, used: set) -> List[int]:
|
|
"""단어 겹침 순위(3순위) — 자막 키워드가 댓글 본문에 부분 문자열로 들어 있는 개수.
|
|
|
|
부분 문자열로 보는 이유: 조사·어미가 댓글 쪽에 붙어 있어도(예: "넉살" ⊂ "넉살님")
|
|
흡수하려는 것. 점수 0은 제외, 점수 내림차순 → 좋아요 내림차순으로 정렬한다.
|
|
"""
|
|
keywords = _cut_keywords(cut.get("bottom") or "")
|
|
if not keywords:
|
|
return []
|
|
scored = []
|
|
for c in comments:
|
|
if c["idx"] in used:
|
|
continue
|
|
text = str(c.get("text") or "")
|
|
score = sum(1 for kw in keywords if kw in text)
|
|
if score > 0:
|
|
scored.append((score, c.get("likeCount", 0), c["idx"]))
|
|
scored.sort(key=lambda t: (-t[0], -t[1]))
|
|
return [idx for _, _, idx in scored]
|
|
|
|
|
|
def build_cut_picks(cuts, comments, ai_picks: Optional[Dict[int, List[int]]],
|
|
quotas) -> List[List[dict]]:
|
|
"""컷별 추천 확정 — ⭐시각 » 🤖AI » 🔤단어겹침 » ➕좋아요 순, 전 컷 통틀어 중복 금지.
|
|
|
|
타임스탬프가 AI보다 먼저인 이유: 그 컷의 **원본 구간**을 콕 집어 언급한 댓글은
|
|
근거가 확실하다. 추측(AI·단어)을 이기게 둘 이유가 없다.
|
|
단어 겹침이 AI 다음인 이유: Gemini 는 문맥까지 보고 고르니 더 정확하지만, 503 등으로
|
|
실패하거나(`ai_picks=None`) 문맥상 연결을 놓칠 때(예: "와인 뱉는 장면" ↔ "싱크대로 달려간
|
|
이유"는 겹치는 단어가 없다) 그물을 하나 더 치는 것 — 네트워크 없이도 항상 동작한다.
|
|
|
|
⚠ 컷별로 4단계를 다 채우고 다음 컷으로 넘어가면 안 된다 — 앞 컷의 약한 근거(4순위
|
|
좋아요)가 뒤 컷의 강한 근거(3순위 단어 겹침)보다 먼저 댓글을 가져가 버린다. 그래서
|
|
**단계(라운드)를 바깥 루프, 컷을 안쪽 루프**로 둔다 — 전 컷의 ts를 다 채운 뒤에야
|
|
전 컷의 ai로, 그다음에야 word로 넘어간다. 순위가 컷 순서보다 우선한다.
|
|
중복은 앞 컷이 가져간다(뒤 컷은 다음 후보로 밀린다) — `used` 는 라운드·컷을 통틀어 공유.
|
|
|
|
ai_picks: {컷인덱스: [댓글idx …]} — Gemini 실패 시 None/{} 어느 쪽을 넘겨도
|
|
타임스탬프만으로 채운다(`ai_pick_cuts` 는 실패를 None 으로 알린다).
|
|
Returns: 컷별 [{"idx": int, "why": "ts"|"ai"|"word"|"like"}]
|
|
"""
|
|
ai_picks = ai_picks or {}
|
|
valid = {c["idx"] for c in comments}
|
|
no_ts = [c for c in comments if not c.get("times")]
|
|
used: set = set()
|
|
quota = [quotas[i] if i < len(quotas) else 0 for i in range(len(cuts))]
|
|
out: List[List[dict]] = [[] for _ in cuts]
|
|
|
|
# 1라운드: ⭐시각
|
|
for i, cut in enumerate(cuts):
|
|
for idx in match_ranges(comments, [(cut["start"], cut["end"])]):
|
|
if len(out[i]) >= quota[i]:
|
|
break
|
|
if idx not in used:
|
|
out[i].append({"idx": idx, "why": "ts"})
|
|
used.add(idx)
|
|
# 2라운드: 🤖AI
|
|
for i in range(len(cuts)):
|
|
if len(out[i]) >= quota[i]:
|
|
continue
|
|
for idx in (ai_picks.get(i) or []):
|
|
if len(out[i]) >= quota[i]:
|
|
break
|
|
if idx in valid and idx not in used:
|
|
out[i].append({"idx": idx, "why": "ai"})
|
|
used.add(idx)
|
|
# 3라운드: 🔤단어겹침
|
|
for i, cut in enumerate(cuts):
|
|
if len(out[i]) >= quota[i]:
|
|
continue
|
|
for idx in _word_match_ranked(cut, comments, used):
|
|
if len(out[i]) >= quota[i]:
|
|
break
|
|
out[i].append({"idx": idx, "why": "word"})
|
|
used.add(idx)
|
|
# 4라운드: ➕좋아요
|
|
for i in range(len(cuts)):
|
|
if len(out[i]) >= quota[i]:
|
|
continue
|
|
for idx in top_liked(no_ts, used, quota[i] - len(out[i])):
|
|
out[i].append({"idx": idx, "why": "like"})
|
|
used.add(idx)
|
|
return out
|
|
|
|
|
|
# Gemini에 넘길 댓글 후보 수. 더 늘려도 채택률이 안 오르고 토큰만 먹는다.
|
|
AI_CANDIDATES = 150
|
|
TEXT_CAP = 200 # 댓글 본문 절단 길이
|
|
# gemini-2.5-flash 는 실측상 HTTP 503("high demand")이 잦아(3연속 503도 관측됨) 컷별
|
|
# 추천이 통째로 비는 원인이었다. plan.py 가 이미 쓰는 gemini-3.5-flash 로 통일 —
|
|
# 같은 프롬프트·같은 컷으로 실측해도 결과가 동등해 품질 손해는 없다.
|
|
AI_MODEL = "gemini-3.5-flash"
|
|
_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"})
|
|
# 429("quota")·503("high demand")만 짧게 쉬고 재시도한다(최대 2회 더 = 총 3회 시도).
|
|
# 둘 다 "지금 당장은 안 됨"이지 "영영 안 됨"이 아니라서다. 그 외 오류는 재시도해 봐야
|
|
# 같은 결과라 즉시 포기한다.
|
|
RETRY_WAITS = (3.0, 8.0)
|
|
raw = None
|
|
for attempt in range(1 + len(RETRY_WAITS)):
|
|
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"]
|
|
break
|
|
except urllib.error.HTTPError as e:
|
|
if e.code in (429, 503) and attempt < len(RETRY_WAITS):
|
|
time.sleep(RETRY_WAITS[attempt])
|
|
continue
|
|
return None # 재시도 소진, 또는 재시도 대상이 아닌 HTTP 오류
|
|
except (KeyError, IndexError, json.JSONDecodeError, ValueError,
|
|
OSError): # URLError, TimeoutError are OSError subclasses; ValueError covers UnicodeDecodeError
|
|
return None # 폴백은 호출부 몫
|
|
if raw is None: # 방어적 — 위 루프는 항상 break/return 으로 빠진다
|
|
return None
|
|
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_time_based(cuts) -> bool:
|
|
"""시각 기반으로 배정할 컷 묶음인가 — **모든 컷의 자막이 비었는가**.
|
|
|
|
자막이 없으면 내용 추천의 근거가 없다. 대신 이런 묶음(통짜 모드·유튜브 구간 탭)은
|
|
구간을 그대로 이어붙이므로 타임라인 시각 = 원본 시각이 성립해 시각으로 맞출 수 있다.
|
|
컷 1개짜리 통짜는 이 규칙의 특수 케이스다(스펙 §1).
|
|
"""
|
|
if not cuts:
|
|
return False
|
|
return all(not (c.get("bottom") or "").strip() for c in cuts)
|
|
|
|
|
|
def _time_based_picks(cuts, comments, quotas) -> List[List[dict]]:
|
|
"""컷마다 3초 슬롯 배정 — 슬롯 시간대 언급 댓글, 빈 슬롯은 좋아요 상위로 채움.
|
|
|
|
`used` 를 컷 사이에 공유해 한 댓글이 두 컷에 들어가지 않게 한다(앞 컷 우선).
|
|
"""
|
|
used: set = set()
|
|
no_ts = [c for c in comments if not c.get("times")]
|
|
out: List[List[dict]] = []
|
|
for i, cut in enumerate(cuts):
|
|
n = quotas[i] if i < len(quotas) else 0
|
|
slots = match_slots(comments, cut["start"], cut["end"] - cut["start"], n,
|
|
exclude=used)
|
|
used.update(s for s in slots if s is not None)
|
|
fill = iter(top_liked(no_ts, used, n))
|
|
picks: List[dict] = []
|
|
for s in slots:
|
|
if s is not None:
|
|
picks.append({"idx": s, "why": "ts"})
|
|
continue
|
|
nxt = next(fill, None)
|
|
if nxt is not None:
|
|
picks.append({"idx": nxt, "why": "like"})
|
|
used.add(nxt)
|
|
out.append(picks)
|
|
return out
|
|
|
|
|
|
def build_highlight_cuts(hl, comments, *, key=None, quotas=None):
|
|
"""하이라이트 하나 → (cuts[], need, ai_failed). 모드는 컷 모양으로 판별한다.
|
|
|
|
quotas: 컷별 카드 장수를 밖에서 정해 넘길 때 쓴다(무음 제거 후 실제 길이 기준).
|
|
안 넘기면 `quotas_for(cuts)` — 원본 컷 길이 기준.
|
|
Returns: ([{"i","sec","bottom","quota","picks"}], need, ai_failed)
|
|
need = Σ quota. Gemini 실패는 ai_failed=True 로 알린다(예외는 안 올린다).
|
|
"""
|
|
cuts = (hl.get("paste") or {}).get("cuts") or []
|
|
if not cuts:
|
|
return [], 0, False
|
|
quotas = list(quotas) if quotas is not None else quotas_for(cuts)
|
|
ai_failed = False
|
|
if is_time_based(cuts):
|
|
picks = _time_based_picks(cuts, comments, quotas)
|
|
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
|