gahusb
84548a326e
- cover.py: DALL·E 3 → 1792x1024, gpt-image-1 → 1536x1024 (모델별 자동), prompt에 'cinematic landscape composition' 명시. OPENAI_IMAGE_SIZE env로 override 가능. - metadata.py: prompt를 list+join 패턴으로 재구성 (인접 문자열/+ 충돌 해결) + lofi 채널 카피라이터 페르소나 부여. description 5-7섹션 구조 명시: 후크/분위기/사용시나리오/챕터/시청권장/콜투액션/해시태그. mix vs single 분기 + tags 가이드 + 출력 JSON schema 명시. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
185 lines
8.0 KiB
Python
185 lines
8.0 KiB
Python
"""메타데이터 생성 — Claude Haiku + 템플릿 폴백."""
|
||
import os
|
||
import json
|
||
import logging
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger("music-lab.metadata")
|
||
|
||
CLAUDE_HAIKU_MODEL_DEFAULT = "claude-haiku-4-5-20251001"
|
||
TIMEOUT_S = 30
|
||
|
||
|
||
def _get_api_key() -> str:
|
||
return os.getenv("ANTHROPIC_API_KEY", "")
|
||
|
||
|
||
def _get_model() -> str:
|
||
return os.getenv("CLAUDE_HAIKU_MODEL", CLAUDE_HAIKU_MODEL_DEFAULT)
|
||
|
||
|
||
def _format_chapters(tracks: list[dict]) -> str:
|
||
"""YouTube 챕터 자동 인식 형식: '[mm:ss] 제목' 한 줄씩.
|
||
|
||
1시간 이상이면 hh:mm:ss 형식.
|
||
"""
|
||
if not tracks:
|
||
return ""
|
||
lines = []
|
||
for t in tracks:
|
||
offset = int(t.get("start_offset_sec", 0))
|
||
m, s = divmod(offset, 60)
|
||
h, m = divmod(m, 60)
|
||
if h > 0:
|
||
ts = f"{h:02d}:{m:02d}:{s:02d}"
|
||
else:
|
||
ts = f"{m:02d}:{s:02d}"
|
||
lines.append(f"{ts} {t.get('title', '')}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
async def generate(*, track: dict, template: dict, trend_keywords: list[str],
|
||
feedback: str = "", tracks: list[dict] | None = None) -> dict:
|
||
"""메타데이터 생성. 성공 시 LLM, 실패/미설정 시 템플릿 치환 폴백.
|
||
|
||
반환: {"title", "description", "tags", "category_id", "used_fallback", "error"}
|
||
"""
|
||
api_key = _get_api_key()
|
||
if not api_key:
|
||
return {**_fallback_template(track, template, tracks), "used_fallback": True, "error": "no api key"}
|
||
|
||
try:
|
||
result = await _call_claude(track, template, trend_keywords, feedback, tracks,
|
||
api_key=api_key, model=_get_model())
|
||
return {**result, "used_fallback": False, "error": None}
|
||
except (httpx.HTTPError, httpx.TimeoutException, KeyError, ValueError, json.JSONDecodeError) as e:
|
||
logger.warning("메타데이터 LLM 실패 — 폴백: %s", e)
|
||
return {**_fallback_template(track, template, tracks), "used_fallback": True, "error": str(e)}
|
||
|
||
|
||
def _fallback_template(track: dict, template: dict, tracks: list[dict] | None = None) -> dict:
|
||
fmt_vars = {
|
||
"title": track.get("title", ""),
|
||
"genre": track.get("genre", ""),
|
||
"bpm": track.get("bpm", ""),
|
||
"key": track.get("key", ""),
|
||
"scale": track.get("scale", ""),
|
||
}
|
||
title = template.get("title", "{title}").format(**fmt_vars)
|
||
description = template.get("description", "{title}").format(**fmt_vars)
|
||
if tracks and len(tracks) > 1:
|
||
description = description + "\n\n" + _format_chapters(tracks)
|
||
return {
|
||
"title": title[:100],
|
||
"description": description[:5000],
|
||
"tags": (template.get("tags") or [])[:15],
|
||
"category_id": template.get("category_id", 10),
|
||
}
|
||
|
||
|
||
def _build_prompt(track: dict, template: dict, trend_keywords: list[str],
|
||
feedback: str, tracks: list[dict] | None) -> str:
|
||
"""프로페셔널 lofi/ambient 채널 수준의 메타데이터 작성 prompt.
|
||
|
||
list + join 패턴으로 조립 — 인접 문자열 리터럴/+ 충돌 회피, 한글 인코딩 안전.
|
||
"""
|
||
is_mix = bool(tracks and len(tracks) > 1)
|
||
parts: list[str] = []
|
||
|
||
# === 입력 ===
|
||
parts.append("당신은 lo-fi/ambient YouTube 채널의 카피라이터입니다.")
|
||
parts.append("프로페셔널 메타데이터를 작성하세요. JSON으로만 응답.")
|
||
parts.append("")
|
||
parts.append("## 입력")
|
||
parts.append(f"트랙: {json.dumps(track, ensure_ascii=False)}")
|
||
parts.append(f"사용자 템플릿(참고용, 이 정보 포함하되 단순 치환은 X): {json.dumps(template, ensure_ascii=False)}")
|
||
trend_str = ", ".join(trend_keywords) if trend_keywords else "(없음)"
|
||
parts.append(f"트렌드 키워드: {trend_str}")
|
||
|
||
if is_mix:
|
||
chapters = _format_chapters(tracks)
|
||
parts.append("")
|
||
parts.append(f"이 영상은 {len(tracks)}개 트랙의 **mix 컴필레이션**입니다. 챕터 리스트:")
|
||
parts.append(chapters)
|
||
|
||
if feedback:
|
||
parts.append("")
|
||
parts.append(f"사용자 피드백 (이 방향으로 수정): {feedback}")
|
||
|
||
# === title 가이드 ===
|
||
parts.append("")
|
||
parts.append("## title (60자 이내, 클릭률 + SEO)")
|
||
if is_mix:
|
||
parts.append("- mix면 '시간 + 분위기 + 사용 시나리오' 형식. 예: '1 Hour Lo-Fi Chill Mix — Late Night Study & Relaxation'")
|
||
else:
|
||
parts.append("- 단일 트랙이면 '제목 — 분위기' 또는 '[장르] 제목 ({BPM}BPM)' 형식")
|
||
parts.append("- 이모지 1개 정도 OK (🎧 📻 ☕ 🌙 등). 과한 이모지 X")
|
||
parts.append("- 영문 + 한글 혼용 자연스럽게")
|
||
|
||
# === description 가이드 ===
|
||
parts.append("")
|
||
parts.append("## description (1500자 이내, 5–7 섹션, 자연스러운 톤)")
|
||
parts.append("1. **한 문장 후크** — 누구를 위한 무엇인지 (예: '집중과 휴식이 필요한 모든 순간을 위한 차분한 lo-fi 컴필레이션.')")
|
||
parts.append("2. **분위기 묘사** (2–3 문장) — 시각적 imagery 포함 (예: '비 오는 카페 창가의 따뜻한 조명처럼')")
|
||
parts.append("3. **추천 사용 상황** — 공부, 작업, 코딩, 명상, 운전, 카페 BGM 등 구체적 시나리오")
|
||
if is_mix:
|
||
parts.append("4. **트랙 리스트 / 챕터** — 위 챕터 리스트를 그대로 포함 (YouTube 자동 챕터 인식). 각 줄에 `[mm:ss] 제목` 형식 유지")
|
||
else:
|
||
parts.append("4. **음향 정보** — 장르, BPM, Key 등 트랙 정보를 자연스럽게 풀어서 설명")
|
||
parts.append("5. **시청 권장사항** — '🎧 헤드폰으로 들으시면 더 좋은 사운드를 경험하실 수 있습니다' 같은 안내")
|
||
parts.append("6. **콜투액션** — 구독, 좋아요, 댓글로 분위기/요청 공유 유도. 자연스럽고 짧게")
|
||
parts.append("7. **(선택) 해시태그** — 끝에 #lofi #studymusic #공부음악 등 5–10개 (description 본문 안)")
|
||
parts.append("- 톤: 차분하고 진정성 있는 lofi 채널 큐레이터. 광고스러운 과장 X")
|
||
|
||
# === tags 가이드 ===
|
||
parts.append("")
|
||
parts.append("## tags (15개 이내, SEO)")
|
||
parts.append("- 영문 키워드 우선: lofi, lo-fi, chill beats, study music, work music, ambient, instrumental, relaxing, focus, night vibes 등")
|
||
parts.append("- 한글 보조: 공부음악, 작업음악, 카페음악, 집중력, 잔잔한 음악 등")
|
||
parts.append("- 트렌드 키워드(있으면) 반드시 포함")
|
||
if is_mix:
|
||
parts.append("- 'mix', 'compilation', 'long mix' 같은 mix 특화 태그 추가")
|
||
|
||
# === category + 출력 ===
|
||
parts.append("")
|
||
parts.append("## category_id")
|
||
parts.append("10 (Music) 고정")
|
||
parts.append("")
|
||
parts.append("## 출력")
|
||
parts.append("```json")
|
||
parts.append('{"title": "...", "description": "...", "tags": [...], "category_id": 10}')
|
||
parts.append("```")
|
||
parts.append("JSON 외 다른 텍스트는 출력 X.")
|
||
|
||
return "\n".join(parts)
|
||
|
||
|
||
async def _call_claude(track: dict, template: dict, trend_keywords: list[str],
|
||
feedback: str, tracks: list[dict] | None,
|
||
*, api_key: str, model: str) -> dict:
|
||
user_prompt = _build_prompt(track, template, trend_keywords, feedback, tracks)
|
||
|
||
async with httpx.AsyncClient(timeout=TIMEOUT_S) as client:
|
||
resp = await client.post(
|
||
"https://api.anthropic.com/v1/messages",
|
||
headers={
|
||
"x-api-key": api_key,
|
||
"anthropic-version": "2023-06-01",
|
||
"content-type": "application/json",
|
||
},
|
||
json={
|
||
"model": model,
|
||
"max_tokens": 2048, # mix 더 길어서
|
||
"messages": [{"role": "user", "content": user_prompt}],
|
||
},
|
||
)
|
||
resp.raise_for_status()
|
||
text = resp.json()["content"][0]["text"]
|
||
# 가장 첫 JSON 블록 추출
|
||
start = text.find("{")
|
||
end = text.rfind("}") + 1
|
||
if start < 0 or end <= start:
|
||
raise ValueError("Claude 응답에 JSON 블록 없음")
|
||
return json.loads(text[start:end])
|