hyowons
210 lines
7.3 KiB
Python
210 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""오선의 미국 증시 라이브(@futuresnow) '오늘의 요약' 최신 영상 감지 + 자막 수집.
|
|
|
|
오전 브리핑(briefing_mail.py)이 import 해서 메일에 '미국증시 요약' 카드를 넣는다.
|
|
미국 정규장 마감(약 05:00 KST) 후 매일 아침 07:20~08:30 KST 사이에 한국어 요약 영상이
|
|
올라온다. 영상 제목 날짜는 '미국장 세션일'이라 헷갈리므로, '오늘 올라온 영상인지'는
|
|
watch 페이지의 publishDate(절대시각)를 KST로 변환해 판정한다.
|
|
|
|
CLI:
|
|
python3 us_market_digest.py latest — 최신 요약 영상 메타 + 자막 + 오늘게시여부 JSON
|
|
python3 us_market_digest.py transcript — 최신 영상 자막만 출력
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
KST = timezone(timedelta(hours=9))
|
|
|
|
CHANNEL_HANDLE = 'futuresnow'
|
|
CHANNEL_NAME = '오선의 미국 증시 라이브'
|
|
CHANNEL_VIDEOS_URL = f'https://www.youtube.com/@{CHANNEL_HANDLE}/videos'
|
|
WATCH_URL = 'https://www.youtube.com/watch?v={vid}'
|
|
TITLE_FILTER = '오늘의 요약' # 공지/광고 영상 제외
|
|
TRANSCRIPT_LANGS = ['ko', 'ko-KR', 'en']
|
|
TRANSCRIPT_CHAR_LIMIT = 8000
|
|
|
|
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'
|
|
)
|
|
|
|
|
|
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:
|
|
"""lockupMetadataViewModel title 등에서 평문 텍스트 추출."""
|
|
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() -> list[dict]:
|
|
"""채널 영상 페이지를 긁어 최신 영상 목록을 [{video_id, title}]로 반환.
|
|
|
|
2026-06 YouTube가 videoRenderer → lockupViewModel 구조로 바뀌어 그에 맞춰 파싱한다.
|
|
RSS 피드(feeds/videos.xml)는 현재 IP 차원에서 404가 잦아 사용하지 않는다.
|
|
"""
|
|
html = _get(CHANNEL_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:
|
|
dt = datetime.fromisoformat(m.group(1))
|
|
return dt.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_us_summary() -> dict:
|
|
"""오전 브리핑용 미국증시 요약 데이터.
|
|
|
|
반환 키:
|
|
channel — 채널명
|
|
recap_expected_today — 오늘(KST) 미국장 요약이 올라올 예정인 날인지 (화~토)
|
|
available_today — 오늘(KST) 게시된 최신 요약 영상이 실제로 있는지
|
|
video — {id, title, url, published_kst} (없으면 None)
|
|
transcript — available_today일 때만 자막 본문
|
|
transcript_status — ok / unavailable / error...
|
|
error — 수집 실패 시 메시지 (그 외 키 없음)
|
|
"""
|
|
now = datetime.now(KST)
|
|
# 미국장(월~금 ET) 요약은 다음날 KST 아침(화~토)에 올라온다.
|
|
recap_expected_today = now.weekday() in {1, 2, 3, 4, 5}
|
|
result = {
|
|
'channel': CHANNEL_NAME,
|
|
'recap_expected_today': recap_expected_today,
|
|
'available_today': False,
|
|
'video': None,
|
|
'transcript': '',
|
|
'transcript_status': 'not_fetched',
|
|
}
|
|
try:
|
|
videos = fetch_channel_videos()
|
|
except Exception as e:
|
|
result['error'] = f'channel fetch 실패: {type(e).__name__}: {e}'
|
|
return result
|
|
|
|
latest = next((v for v in videos if TITLE_FILTER in v.get('title', '')), None)
|
|
if not latest:
|
|
result['error'] = f"'{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 main() -> int:
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else 'latest'
|
|
if cmd == 'transcript':
|
|
data = get_us_summary()
|
|
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_us_summary(), ensure_ascii=False, indent=2))
|
|
return 0
|
|
print(f'unknown command: {cmd}', file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|