af0e2fca8a
Some checks failed
news-summary-bot-cicd / build_push_deploy (push) Failing after 3m42s
- 요약 처리 중 에러 발생 시 Discord에 에러 상세 내용 전송 - 에러 타입, 메시지, 영상 정보를 포함한 embed 형식 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
import re
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
|
|
|
|
def _extract_video_id(video_url: str) -> str | None:
|
|
"""URL에서 YouTube 비디오 ID 추출."""
|
|
patterns = [
|
|
r"(?:youtu\.be/)([^?&]+)",
|
|
r"(?:v=)([^?&]+)",
|
|
]
|
|
for p in patterns:
|
|
m = re.search(p, video_url)
|
|
if m:
|
|
return m.group(1)
|
|
return None
|
|
|
|
|
|
def _parse_summary(summary: str) -> dict[str, str]:
|
|
"""요약 텍스트를 섹션별로 파싱."""
|
|
sections: dict[str, str] = {}
|
|
current_key = None
|
|
current_lines: list[str] = []
|
|
|
|
for line in summary.split("\n"):
|
|
# **한줄 요약**: ... 또는 ## 한줄 요약 형태 매칭
|
|
header_match = re.match(
|
|
r"^(?:##\s*|-\s*\*\*|\*\*)(한줄\s*요약|주요\s*내용|결론/?시사점)[:\*\s]*(.*)",
|
|
line,
|
|
)
|
|
if header_match:
|
|
if current_key:
|
|
sections[current_key] = "\n".join(current_lines).strip()
|
|
current_key = header_match.group(1).replace(" ", "")
|
|
rest = re.sub(r"^\*\*:?\s*", "", header_match.group(2)).strip()
|
|
current_lines = [rest] if rest else []
|
|
elif current_key is not None:
|
|
current_lines.append(line)
|
|
|
|
if current_key:
|
|
sections[current_key] = "\n".join(current_lines).strip()
|
|
|
|
return sections
|
|
|
|
|
|
async def send_to_discord(title: str, video_url: str, summary: str) -> None:
|
|
"""Discord 웹훅으로 요약 전송 (임베드 디자인)."""
|
|
video_id = _extract_video_id(video_url)
|
|
thumbnail_url = (
|
|
f"https://img.youtube.com/vi/{video_id}/hqdefault.jpg"
|
|
if video_id
|
|
else None
|
|
)
|
|
|
|
sections = _parse_summary(summary)
|
|
|
|
oneliner = sections.get("한줄요약", "")
|
|
main_points = sections.get("주요내용", "")
|
|
conclusion = sections.get("결론/시사점", sections.get("결론시사점", ""))
|
|
|
|
# 파싱 실패 시 전체 텍스트를 그대로 사용
|
|
if not oneliner and not main_points:
|
|
fields = [{"name": "🔗 원본 영상", "value": video_url, "inline": False}]
|
|
description = summary[:4096]
|
|
else:
|
|
description = f"### 💡 {oneliner}" if oneliner else ""
|
|
fields = []
|
|
if main_points:
|
|
fields.append({
|
|
"name": "📋 주요 내용",
|
|
"value": main_points[:1024],
|
|
"inline": False,
|
|
})
|
|
if conclusion:
|
|
fields.append({
|
|
"name": "🎯 결론 / 시사점",
|
|
"value": conclusion[:1024],
|
|
"inline": False,
|
|
})
|
|
fields.append({
|
|
"name": "🔗 원본 영상",
|
|
"value": video_url,
|
|
"inline": False,
|
|
})
|
|
|
|
embed = {
|
|
"title": f"📰 {title}",
|
|
"url": video_url,
|
|
"description": description,
|
|
"color": 0x2B2D31,
|
|
"fields": fields,
|
|
"footer": {
|
|
"text": "YouTube 뉴스 요약 봇",
|
|
"icon_url": "https://www.youtube.com/s/desktop/f5ced909/img/favicon_144x144.png",
|
|
},
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
if thumbnail_url:
|
|
embed["thumbnail"] = {"url": thumbnail_url}
|
|
|
|
payload = {"embeds": [embed]}
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(settings.discord_webhook_url, json=payload)
|
|
resp.raise_for_status()
|
|
|
|
|
|
async def send_error_to_discord(
|
|
title: str, video_url: str, error: Exception
|
|
) -> None:
|
|
"""에러 발생 시 Discord 웹훅으로 에러 내용 전송."""
|
|
error_type = type(error).__name__
|
|
error_msg = str(error)[:1024]
|
|
|
|
embed = {
|
|
"title": "❌ 뉴스 요약 실패",
|
|
"color": 0xED4245,
|
|
"fields": [
|
|
{"name": "영상 제목", "value": title or "(제목 없음)", "inline": False},
|
|
{"name": "영상 URL", "value": video_url, "inline": False},
|
|
{"name": "에러 타입", "value": f"`{error_type}`", "inline": True},
|
|
{"name": "에러 내용", "value": f"```\n{error_msg}\n```", "inline": False},
|
|
],
|
|
"footer": {"text": "YouTube 뉴스 요약 봇 - 에러 알림"},
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
payload = {"embeds": [embed]}
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
await client.post(settings.discord_webhook_url, json=payload)
|