auto: 일일 백업 2026-06-18 02:00
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
hyowons
@@ -0,0 +1,238 @@
|
||||
#!/usr/bin/env python3
|
||||
"""오전 브리핑용 유튜브 시황 영상 감지 + 자막 수집.
|
||||
|
||||
두 채널을 같은 로직으로 다룬다 (briefing_mail.py가 import):
|
||||
- 오선의 미국 증시 라이브(@futuresnow) '오늘의 요약' → 🇺🇸 미국증시 요약 카드
|
||||
- 비하이브 투자자문 '주식시황'(장전 시황) → 📈 국내 증시 시황 카드
|
||||
|
||||
공통 흐름: 채널 영상 페이지 `lockupViewModel` 스크랩(RSS 피드는 IP 404 잦아 미사용)으로
|
||||
제목필터에 맞는 최신 영상을 찾고, watch 페이지 `publishDate`(절대시각 PT→KST 변환)로
|
||||
'오늘 게시여부'를 판정한다(제목 날짜는 신뢰 불가). 이어 자막을 수집한다.
|
||||
|
||||
CLI:
|
||||
python3 youtube_briefing_digest.py latest [us|behive] — 최신 영상 메타+자막+오늘게시여부 JSON
|
||||
python3 youtube_briefing_digest.py transcript [us|behive] — 최신 영상 자막만 출력
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone, timedelta
|
||||
|
||||
KST = timezone(timedelta(hours=9))
|
||||
|
||||
USER_AGENT = (
|
||||
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
|
||||
'AppleWebKit/537.36 (KHTML, like Gecko) '
|
||||
'Chrome/122.0.0.0 Safari/537.36'
|
||||
)
|
||||
TRANSCRIPT_LANGS = ['ko', 'ko-KR', 'en']
|
||||
TRANSCRIPT_CHAR_LIMIT = 8000
|
||||
WATCH_URL = 'https://www.youtube.com/watch?v={vid}'
|
||||
|
||||
# 채널 설정. expected_weekdays = 시황 영상이 올라올 예정인 KST 요일 (월=0 … 일=6)
|
||||
CHANNELS = {
|
||||
'us': {
|
||||
'name': '오선의 미국 증시 라이브',
|
||||
'videos_url': 'https://www.youtube.com/@futuresnow/videos',
|
||||
'title_filter': '오늘의 요약', # 공지/광고 제외
|
||||
# 미국장(월~금 ET) 요약은 다음날 KST 아침(화~토)에 게시
|
||||
'expected_weekdays': {1, 2, 3, 4, 5},
|
||||
},
|
||||
'behive': {
|
||||
'name': '비하이브 투자자문',
|
||||
'videos_url': 'https://www.youtube.com/channel/UCHTRF5r154igU2gXjudUMzg/videos',
|
||||
'title_filter': '주식시황', # '마감시황'·'주간시황'·'종목분석'·'공모청약' 제외
|
||||
# 국내 장전 시황은 평일(월~금) 새벽 KST에 게시
|
||||
'expected_weekdays': {0, 1, 2, 3, 4},
|
||||
},
|
||||
'behive_close': {
|
||||
'name': '비하이브 투자자문',
|
||||
'videos_url': 'https://www.youtube.com/channel/UCHTRF5r154igU2gXjudUMzg/videos',
|
||||
'title_filter': '마감시황', # 국내 장 마감 해설 (오후 브리핑용)
|
||||
# 마감 시황은 평일(월~금) 16:45~18:10 KST 게시
|
||||
'expected_weekdays': {0, 1, 2, 3, 4},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _get(url: str) -> str:
|
||||
req = urllib.request.Request(url, headers={'User-Agent': USER_AGENT})
|
||||
with urllib.request.urlopen(req, timeout=30) as r:
|
||||
return r.read().decode('utf-8', 'ignore')
|
||||
|
||||
|
||||
def _lockup_text(o) -> str:
|
||||
if isinstance(o, dict):
|
||||
if isinstance(o.get('content'), str):
|
||||
return o['content']
|
||||
for k in ('text', 'simpleText'):
|
||||
if isinstance(o.get(k), str):
|
||||
return o[k]
|
||||
return ''
|
||||
|
||||
|
||||
def fetch_channel_videos(videos_url: str) -> list[dict]:
|
||||
"""채널 영상 페이지를 긁어 최신 영상 목록을 [{video_id, title}]로 반환.
|
||||
|
||||
2026-06 YouTube가 videoRenderer → lockupViewModel 구조로 바뀌어 그에 맞춰 파싱한다.
|
||||
"""
|
||||
html = _get(videos_url)
|
||||
marker = 'var ytInitialData = '
|
||||
start = html.find(marker)
|
||||
if start < 0:
|
||||
raise RuntimeError('ytInitialData not found in channel page')
|
||||
start += len(marker)
|
||||
end = html.find(';</script>', start)
|
||||
if end < 0:
|
||||
raise RuntimeError('ytInitialData terminator not found in channel page')
|
||||
data = json.loads(html[start:end])
|
||||
|
||||
rows: list[dict] = []
|
||||
seen: set[str] = set()
|
||||
|
||||
def walk(node):
|
||||
if isinstance(node, dict):
|
||||
lvm = node.get('lockupViewModel')
|
||||
if lvm and lvm.get('contentId'):
|
||||
vid = lvm['contentId']
|
||||
if vid not in seen:
|
||||
meta = lvm.get('metadata', {}).get('lockupMetadataViewModel', {})
|
||||
title = _lockup_text(meta.get('title', {})).strip()
|
||||
seen.add(vid)
|
||||
rows.append({'video_id': vid, 'title': title})
|
||||
for v in node.values():
|
||||
walk(v)
|
||||
elif isinstance(node, list):
|
||||
for v in node:
|
||||
walk(v)
|
||||
|
||||
walk(data)
|
||||
return rows
|
||||
|
||||
|
||||
def get_publish_kst(video_id: str) -> datetime | None:
|
||||
"""watch 페이지 publishDate(절대시각, PT)를 KST datetime으로 변환. 실패 시 None."""
|
||||
try:
|
||||
html = _get(WATCH_URL.format(vid=video_id))
|
||||
except Exception:
|
||||
return None
|
||||
m = re.search(r'"publishDate":"([^"]+)"', html)
|
||||
if not m:
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(m.group(1)).astimezone(KST)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def fetch_transcript(video_id: str) -> tuple[str, str]:
|
||||
try:
|
||||
from youtube_transcript_api import YouTubeTranscriptApi
|
||||
except Exception as e:
|
||||
return '', f'error: youtube_transcript_api import 실패 ({e})'
|
||||
try:
|
||||
api = YouTubeTranscriptApi()
|
||||
fetched = api.fetch(video_id, languages=TRANSCRIPT_LANGS)
|
||||
parts = []
|
||||
for snippet in fetched:
|
||||
text = getattr(snippet, 'text', None)
|
||||
if text is None and isinstance(snippet, dict):
|
||||
text = snippet.get('text', '')
|
||||
if text:
|
||||
parts.append(text.strip())
|
||||
full = ' '.join(parts).strip()
|
||||
if not full:
|
||||
return '', 'unavailable'
|
||||
if len(full) > TRANSCRIPT_CHAR_LIMIT:
|
||||
full = full[:TRANSCRIPT_CHAR_LIMIT] + '... [자막 일부 생략]'
|
||||
return full, 'ok'
|
||||
except Exception as e:
|
||||
return '', f'unavailable: {type(e).__name__}'
|
||||
|
||||
|
||||
def get_channel_summary(key: str) -> dict:
|
||||
"""브리핑용 시황 요약 데이터. key는 CHANNELS의 'us' / 'behive'.
|
||||
|
||||
반환 키:
|
||||
channel, recap_expected_today, available_today,
|
||||
video {id,title,url,published_kst} | None,
|
||||
transcript (available_today일 때만), transcript_status, [error]
|
||||
"""
|
||||
cfg = CHANNELS[key]
|
||||
now = datetime.now(KST)
|
||||
result = {
|
||||
'channel': cfg['name'],
|
||||
'recap_expected_today': now.weekday() in cfg['expected_weekdays'],
|
||||
'available_today': False,
|
||||
'video': None,
|
||||
'transcript': '',
|
||||
'transcript_status': 'not_fetched',
|
||||
}
|
||||
try:
|
||||
videos = fetch_channel_videos(cfg['videos_url'])
|
||||
except Exception as e:
|
||||
result['error'] = f'channel fetch 실패: {type(e).__name__}: {e}'
|
||||
return result
|
||||
|
||||
latest = next((v for v in videos if cfg['title_filter'] in v.get('title', '')), None)
|
||||
if not latest:
|
||||
result['error'] = f"'{cfg['title_filter']}' 영상을 찾지 못함"
|
||||
return result
|
||||
|
||||
vid = latest['video_id']
|
||||
published = get_publish_kst(vid)
|
||||
result['video'] = {
|
||||
'id': vid,
|
||||
'title': latest['title'],
|
||||
'url': f'https://www.youtube.com/watch?v={vid}',
|
||||
'published_kst': published.isoformat() if published else None,
|
||||
}
|
||||
result['available_today'] = bool(published and published.date() == now.date())
|
||||
if result['available_today']:
|
||||
text, status = fetch_transcript(vid)
|
||||
result['transcript'] = text
|
||||
result['transcript_status'] = status
|
||||
return result
|
||||
|
||||
|
||||
def get_us_summary() -> dict:
|
||||
return get_channel_summary('us')
|
||||
|
||||
|
||||
def get_behive_market_summary() -> dict:
|
||||
return get_channel_summary('behive')
|
||||
|
||||
|
||||
def get_behive_close_summary() -> dict:
|
||||
return get_channel_summary('behive_close')
|
||||
|
||||
|
||||
def main() -> int:
|
||||
cmd = sys.argv[1] if len(sys.argv) > 1 else 'latest'
|
||||
key = sys.argv[2] if len(sys.argv) > 2 else 'us'
|
||||
if key not in CHANNELS:
|
||||
print(f'unknown channel: {key} (us|behive)', file=sys.stderr)
|
||||
return 2
|
||||
if cmd == 'transcript':
|
||||
data = get_channel_summary(key)
|
||||
vid = (data.get('video') or {}).get('id')
|
||||
if not vid:
|
||||
print(data.get('error', 'no video'), file=sys.stderr)
|
||||
return 1
|
||||
text, status = fetch_transcript(vid)
|
||||
print(f'transcript_status: {status}')
|
||||
print('---')
|
||||
print(text)
|
||||
return 0
|
||||
if cmd == 'latest':
|
||||
print(json.dumps(get_channel_summary(key), ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
print(f'unknown command: {cmd}', file=sys.stderr)
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user