auto: 일일 백업 2026-08-28 02:00

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

agents/stock/workspace/scripts/behive_youtube_digest.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""@비하이브투자자문 신규 '종목분석' 영상 감지 + 자막 수집 + watchlist 저장 + 이메일/텔레그램 발송 + 조회.
+"""@비하이브투자자문 신규 '종목분석'/'관심테마' 영상 감지 + 자막 수집 + watchlist 저장 + 이메일/텔레그램 발송 + 조회.
 
 Subcommands:
   fetch                       — 신규 매칭 영상 메타데이터 JSON 출력(자막 제외) + 자막은 fetch 캐시에만 저장
@@ -42,7 +42,7 @@ CHANNEL_ID = 'UCHTRF5r154igU2gXjudUMzg'
 CHANNEL_NAME = '비하이브 투자자문'
 FEED_URL = f'https://www.youtube.com/feeds/videos.xml?channel_id={CHANNEL_ID}'
 SEARCH_URL = 'https://www.youtube.com/results?search_query={query}&sp=CAISAhAB'
-TITLE_FILTER = '종목분석'
+TITLE_FILTERS = ('종목분석', '관심테마')
 TRANSCRIPT_LANGS = ['ko', 'ko-KR', 'en']
 TRANSCRIPT_CHAR_LIMIT = 8000
 FETCH_LIMIT = 10
@@ -139,8 +139,19 @@ def _is_recent_published_text(text: str) -> bool:
     return False
 
 
+def _matches_title_filter(title: str) -> bool:
+    return any(keyword in title for keyword in TITLE_FILTERS)
+
+
+def _video_kind(title: str) -> str:
+    for keyword in TITLE_FILTERS:
+        if keyword in title:
+            return keyword
+    return '비하이브'
+
+
 def _search_results_fallback() -> list[dict]:
-    query = urllib.parse.quote(f'{CHANNEL_NAME} {TITLE_FILTER}')
+    query = urllib.parse.quote(f'{CHANNEL_NAME} {" OR ".join(TITLE_FILTERS)}')
     url = SEARCH_URL.format(query=query)
     req = urllib.request.Request(url, headers={'User-Agent': FEED_USER_AGENT})
     with urllib.request.urlopen(req, timeout=30) as r:
@@ -178,7 +189,7 @@ def _search_results_fallback() -> list[dict]:
         title = ''.join(run.get('text', '') for run in vr.get('title', {}).get('runs', []))
         vid = vr.get('videoId', '')
         published = vr.get('publishedTimeText', {}).get('simpleText', '')
-        if owner != CHANNEL_NAME or TITLE_FILTER not in title or not vid or vid in seen_ids:
+        if owner != CHANNEL_NAME or not _matches_title_filter(title) or not vid or vid in seen_ids:
             continue
         if not _is_recent_published_text(published):
             continue
@@ -186,6 +197,7 @@ def _search_results_fallback() -> list[dict]:
         entries.append({
             'video_id': vid,
             'title': title.strip(),
+            'kind': _video_kind(title),
             'published': published,
             'url': f'https://www.youtube.com/watch?v={vid}',
         })
@@ -231,6 +243,7 @@ def fetch_feed() -> list[dict]:
             entries.append({
                 'video_id': vid,
                 'title': title.strip(),
+                'kind': _video_kind(title),
                 'published': published,
                 'url': link or f'https://www.youtube.com/watch?v={vid}',
             })
@@ -267,7 +280,7 @@ def cmd_fetch() -> int:
     entries = fetch_feed()
     matched = [
         e for e in entries
-        if TITLE_FILTER in e['title'] and e['video_id'] not in seen
+        if _matches_title_filter(e['title']) and e['video_id'] not in seen
     ]
     matched = list(reversed(matched))[:FETCH_LIMIT]
     for item in matched:
@@ -304,6 +317,7 @@ def load_cached_video(video_id: str) -> dict:
             return {
                 'id': item.get('video_id'),
                 'title': item.get('title'),
+                'kind': item.get('kind') or _video_kind(item.get('title') or ''),
                 'url': item.get('url'),
                 'published': item.get('published'),
             }
@@ -435,7 +449,8 @@ def _buy_primary(entry: dict) -> float | None:
 
 def format_entry_block(entry: dict) -> str:
     stock = entry.get('stock', '')
-    lines = [f'[종목분석] #{stock}', '━━━━━━━━━━']
+    kind = (entry.get('video') or {}).get('kind') or '종목분석'
+    lines = [f'[{kind}] #{stock}', '━━━━━━━━━━']
     target_line = format_price_line('목표가', entry.get('target'))
     upside = entry.get('upside_pct')
     if upside is not None and '언급 없음' not in target_line:
@@ -480,10 +495,11 @@ def _format_current_html(current: str | None) -> str:
 def format_entry_html_block(entry: dict) -> str:
     S = HTML_STYLES
     stock = entry.get('stock', '')
+    kind = (entry.get('video') or {}).get('kind') or '종목분석'
     current = fetch_current_price(entry.get('code') or stock, _buy_primary(entry))
     parts = [f'<div style="{S["card"]}">']
     parts.append(
-        f'<div style="{S["card_title"]}">[종목분석] #{html_escape(stock)} '
+        f'<div style="{S["card_title"]}">[{html_escape(kind)}] #{html_escape(stock)} '
         f'<span style="{S["card_meta"]}">현재가: {_format_current_html(current)}</span></div>'
     )
 
@@ -541,7 +557,7 @@ def build_email_html(entries: list[dict]) -> str:
     S = HTML_STYLES
     parts = [
         f'<div style="{S["wrap"]}">',
-        f'<div style="{S["greet"]}">관리자님, 비하이브투자자문 신규 종목분석 {len(entries)}건을 전달드립니다.</div>',
+        f'<div style="{S["greet"]}">관리자님, 비하이브투자자문 신규 추천 영상 {len(entries)}건을 전달드립니다.</div>',
     ]
     for entry in entries:
         parts.append(format_entry_html_block(entry))
@@ -585,7 +601,7 @@ def cmd_email(video_ids: list[str]) -> int:
         return 1
     date_str = datetime.now(KST).strftime('%Y-%m-%d %H:%M')
     stock_names = ', '.join(e.get('stock', '') for e in entries)
-    subject = f'[비하이브 종목분석] {len(entries)}개 보고서 — {date_str} ({stock_names})'
+    subject = f'[비하이브 추천영상] {len(entries)}개 보고서 — {date_str} ({stock_names})'
     html_body = build_email_html(entries)
     cmd = ['gog', 'gmail', 'send', '--to', EMAIL_RECIPIENT, '--subject', subject, '--body-html', html_body]
     p = subprocess.run(cmd, text=True, capture_output=True)
@@ -652,7 +668,7 @@ def cmd_notify(video_ids: list[str]) -> int:
     if not entries:
         print('no matching entries in watchlist', file=sys.stderr)
         return 1
-    header = f'비하이브 종목분석 {len(entries)}건 제출'
+    header = f'비하이브 추천영상 {len(entries)}건 제출'
     blocks = [format_telegram_block(e) for e in entries]
     text = header + '\n\n' + '\n\n'.join(blocks)
     if send_telegram(text):