#!/usr/bin/env python3
import json
import os
import re
import subprocess
import sys
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from email.utils import parsedate_to_datetime
from html import unescape
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional
from zoneinfo import ZoneInfo
KST = ZoneInfo('Asia/Seoul')
WORKSPACE = Path('/Users/snowoyh/.openclaw/workspace')
STATE_DIR = WORKSPACE / 'state'
STATE_DIR.mkdir(parents=True, exist_ok=True)
SEND_STATE = STATE_DIR / 'briefing_send_state.json'
IPO_SYNC_STATE = Path('/Users/snowoyh/.openclaw/agents/stock/workspace/state/ipo_calendar_sync.json')
IPO_SEEN_STATE = STATE_DIR / 'briefing_ipo_seen.json'
NEWS_HISTORY_FILE = STATE_DIR / 'news_sent_history.json'
PENDING_SELECTION_FILE = STATE_DIR / 'briefing_pending_selection.json'
NEWS_HISTORY_LOOKBACK_HOURS = 48
NEWS_HISTORY_RETENTION_DAYS = 7
RECIPIENT = 'mini.snowoyh@gmail.com'
NEWS_FEEDS = [
('국내 경제', 'https://www.yna.co.kr/rss/economy.xml'),
('국제', 'https://www.yna.co.kr/rss/international.xml'),
('국제', 'https://rss.edaily.co.kr/world_news.xml'),
('증시', 'https://rss.edaily.co.kr/stock_news.xml'),
]
MARKET_FEEDS = [
('USD/KRW', 'https://query1.finance.yahoo.com/v8/finance/chart/KRW%3DX?range=2d&interval=1d'),
]
NAVER_INDEX_FEEDS = [
('.INX', 'https://m.stock.naver.com/worldstock/index/.INX/total'),
('.IXIC', 'https://m.stock.naver.com/worldstock/index/.IXIC/total'),
]
NAVER_DOMESTIC_INDEX_FEEDS = [
('KOSPI', 'https://m.stock.naver.com/domestic/index/KOSPI/total'),
('KOSDAQ', 'https://m.stock.naver.com/domestic/index/KOSDAQ/total'),
]
KOSPI_FUTURES_INVESTING_URL = 'https://kr.investing.com/indices/korea-200-futures'
MSCI_KOREA_INVESTING_URL = 'https://kr.investing.com/etfs/ishares-south-korea-index'
def run(cmd: list[str]) -> str:
p = subprocess.run(cmd, capture_output=True, text=True)
if p.returncode != 0:
raise RuntimeError(f"command failed: {' '.join(cmd)}\n{p.stderr.strip()}")
return p.stdout
def load_json(path: Path, default):
if path.exists():
try:
return json.loads(path.read_text())
except Exception as e:
print(f"[briefing_mail:load_json:{path}] {type(e).__name__}: {e}", file=sys.stderr)
return default
return default
def save_json(path: Path, data):
path.write_text(json.dumps(data, ensure_ascii=False, indent=2))
def fetch(url: str) -> str:
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
with urllib.request.urlopen(req, timeout=30) as r:
return r.read().decode('utf-8', 'ignore')
def fetch_bytes(url: str) -> bytes:
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
with urllib.request.urlopen(req, timeout=30) as r:
return r.read()
def get_today_events(target_date: datetime) -> list[str]:
start = target_date.astimezone(KST).replace(hour=0, minute=0, second=0, microsecond=0)
end = start + timedelta(days=1)
out = run([
'gog', 'calendar', 'events', 'mini.snowoyh@gmail.com',
'--from', start.isoformat(), '--to', end.isoformat(), '--json'
])
data = json.loads(out)
items = []
for ev in data.get('events', []):
summary = ev.get('summary', '(제목 없음)')
s = ev.get('start', {})
e = ev.get('end', {})
if 'dateTime' in s:
st = datetime.fromisoformat(s['dateTime']).astimezone(KST)
en = datetime.fromisoformat(e['dateTime']).astimezone(KST) if 'dateTime' in e else None
times = f"{st:%H:%M}"
if en:
times += f"-{en:%H:%M}"
else:
times = '종일'
items.append(f"- {summary} ({times})")
return items
def section(title: str, lines: list[str]) -> list[str]:
if not lines:
return []
return [f'■ {title}'] + lines + ['']
def clean_title(title: str) -> str:
title = re.sub(r'\s+', ' ', title).strip()
title = re.sub(r'\s*-\s*[^-]+$', '', title)
return title
DESCRIPTION_MAX_LEN = 500
def clean_description(desc: str) -> str:
if not desc:
return ''
text = re.sub(r'<[^>]+>', ' ', desc)
text = unescape(text)
text = re.sub(r'\(서울=연합뉴스\)\s*[^=]{0,40}기자\s*=\s*', '', text)
text = re.sub(r'\s+', ' ', text).strip()
if len(text) > DESCRIPTION_MAX_LEN:
text = text[:DESCRIPTION_MAX_LEN].rstrip() + '…'
return text
def source_from_link(link: str) -> str:
m = re.search(r'https?://(?:www\.|rss\.)?([^/]+)', link or '')
if not m:
return ''
host = m.group(1).lower()
mapping = {
'yna.co.kr': '연합뉴스',
'edaily.co.kr': '이데일리',
'mk.co.kr': '매일경제',
'hankyung.com': '한국경제',
}
for key, name in mapping.items():
if host.endswith(key):
return name
return host
def parse_google_news_feed(url: str) -> list[dict]:
xml_text = fetch(url)
try:
root = ET.fromstring(xml_text)
except ET.ParseError:
# Recover from bare-& in attribute values (common RSS breakage, e.g. youtube embed URLs).
fixed = re.sub(r'&(?!#?\w+;)', '&', xml_text)
root = ET.fromstring(fixed)
items = []
for item in root.findall('./channel/item'):
title = clean_title(item.findtext('title', default=''))
link = item.findtext('link', default='').strip()
pub_date = item.findtext('pubDate', default='').strip()
description = clean_description(item.findtext('description', default=''))
source_el = item.find('source')
source = source_el.text.strip() if source_el is not None and source_el.text else ''
if not source:
source = source_from_link(link)
pub_dt = None
if pub_date:
try:
pub_dt = parsedate_to_datetime(pub_date)
if pub_dt.tzinfo is None:
pub_dt = pub_dt.replace(tzinfo=KST)
except (TypeError, ValueError):
pub_dt = None
if title and link:
items.append({
'title': title,
'link': link,
'pubDate': pub_date,
'pub_dt': pub_dt,
'source': source,
'description': description,
})
return items
def news_priority(title: str) -> int:
t = title.lower()
score = 0
keywords = {
'전쟁': 7,
'침공': 6, '공습': 6, '폭격': 6, '포격': 6, '미사일': 6, '휴전': 6, '정전': 6,
'총격': 6, '총기난사': 6, '저격': 6, '암살': 6, '피습': 6, '테러': 6, '핵실험': 6,
'관세': 5, '금리': 5, '환율': 5, '인플레이션': 5, 'cpi': 5, 'fomc': 5,
'지정학': 5, '제재': 5,
'반도체': 4, 'ai': 4, '실적': 4, '수출': 4, '경기': 4, '유가': 4,
'코스피': 3, '코스닥': 3, '나스닥': 3, 's&p': 3, '증시': 3,
'달러': 3, '국채': 3,
'트럼프': 3, '푸틴': 3, '시진핑': 3, '김정은': 3,
}
for k, w in keywords.items():
if k in t:
score += w
return score
def summarize_news_title(title: str) -> tuple[str, str, str, str]:
t = title.strip()
rules = [
(
'관세',
'핵심: 관세가 올라간다는 뜻입니다. 물건을 수출할 때 더 많은 돈을 내야 한다는 얘기입니다.',
'의미: 삼성이나 LG 같은 큰 회사들이 장난감을 팔 때 비용이 더 커집니다. 그럼 이익이 줄어들 수 있어요.',
'체크: 한국 회사들이 미국이나 다른 나라에 물건을 많이 팔기 때문에 관세 소식을 정말 자세히 봐야 합니다.',
'제 의견: 관세는 뉴스가 나온 후 2~3일 더 지켜봐야 돼요. 한 번 뉴스 떴다고 끝이 아니니까요.'
),
(
'금리',
'핵심: 금리가 올라가거나 내려간다는 뜻입니다. 은행에 돈을 맡길 때 받는 이자 같은 거요.',
'의미: 금리가 올라가면 돈을 빌려 쓰는 회사들이 더 많이 내야 해요. 그럼 회사 이익이 줄어들 수 있습니다.',
'체크: 달러 환율과 채권 금리가 함께 움직이는지 봐야 해요. 그럼 시장이 어떻게 될지 더 쉽게 알 수 있어요.',
'제 의견: 금리 뉴스는 한 번으로 끝나지 않아요. 2~3일 동안 계속 주의 깊게 봐야 합니다.'
),
(
'환율',
'핵심: 원화와 달러의 교환 비율이 바뀐다는 뜻입니다. 달러가 올라가면 한국 돈이 약해진다는 거예요.',
'의미: 한국 회사가 미국에 물건을 팔면 원화로 받는 돈이 줄어들어요. 반대로 물건을 싸게 만들 수 있어서 도움도 되고요.',
'체크: 외국 사람들이 한국 주식을 사거나 팔 때 환율이 얼마나 영향을 주는지 봐야 해요.',
'제 의견: 환율은 숫자가 얼마 올랐냐보다 빠르게 변하는 게 더 중요해요. 빠르게 변하면 위험신호입니다.'
),
(
'반도체',
'핵심: 반도체(스마트폰이나 컴퓨터에 들어가는 아주 작은 부품) 소식이 나왔다는 뜻입니다.',
'의미: 반도체 소식은 삼성, SK하이닉스 같은 큰 회사에 직접 영향을 줘요. 그럼 한국 전체 주가에도 미쳐요.',
'체크: 반도체 수요가 정말 늘어나는 건지, 아니면 그냥 하루 뉴스인지 구분해서 봐야 합니다.',
'제 의견: 반도체는 장기로 수요가 늘어날지를 보는 게 가장 중요해요. 한 번 소식보다요.'
),
(
'실적',
'핵심: 회사가 얼마나 돈을 벌었는지 발표한 것입니다. 좋은 결과인지 나쁜 결과인지를 뜻해요.',
'의미: 실적이 좋으면 주가가 올라가고, 나쁘면 내려가요. 아주 중요한 신호입니다.',
'체크: 실적이 시장이 기대했던 것보다 좋았는지 나빴는지를 봐야 해요. 그게 더 중요합니다.',
'제 의견: 실적 뉴스가 나왔을 때는 하루 반응보다 그 다음날 움직임을 봐야 돼요. 정확해집니다.'
),
(
'유가',
'핵심: 석유 값이 올라가거나 내려간다는 뜻입니다. 세계 모든 나라의 물가에 영향을 줘요.',
'의미: 유가가 올라가면 휘발유, 택시, 배송비가 모두 비싸져요. 그럼 회사 비용이 늘어납니다.',
'체크: 환율과 유가가 같이 올라가는지 봐야 해요. 같이 올라가면 한국 회사들 부담이 훨씬 커집니다.',
'제 의견: 유가는 한두 번 변해도 괜찮아요. 추세가 바뀌는 게 더 중요합니다. 계속 올라갈 건지 내려갈 건지요.'
),
(
'나스닥',
'핵심: 미국의 기술주 지수가 올라가거나 내려갔다는 뜻입니다. 애플, 구글, 마이크로소프트 같은 회사들이요.',
'의미: 미국 기술주가 올라가면 한국의 AI, 반도체 회사들도 주가가 올라갈 가능성이 커집니다.',
'체크: 정말 미국 기술주 전체가 올라갔는지, 아니면 몇 개 회사만 올라갔는지 구분해서 봐야 합니다.',
'제 의견: 미국에서 좋은 소식이 나도 한국에선 반대가 될 수 있어요. 항상 신중하게 봐야 합니다.'
),
(
'코스피',
'핵심: 한국 큰 회사들(삼성, SK, 현대 같은)의 주가 지수가 변했다는 뜻입니다.',
'의미: 코스피가 올라가면 한국 경제가 잘 나간다는 신호예요. 외국인도 한국 주식을 사고 싶다는 뜻이기도 해요.',
'체크: 원화 환율과 외국인이 얼마나 사고 파는지를 함께 봐야 해요. 그럼 진짜 강한 건지 약한 건지 알 수 있습니다.',
'제 의견: 코스피 뉴스 하나보다 계속 올라가는 흐름이 중요해요. 한 번 올랐다고 끝이 아니니까요.'
),
(
'코스닥',
'핵심: 한국의 작은 회사, 새로운 회사들(스타트업 같은)의 주가 지수가 변했다는 뜻입니다.',
'의미: 코스닥이 올라가면 젊은 세대, 벤처 회사들이 잘 나간다는 신호예요. 위험도는 높지만요.',
'체크: 코스닥이 올라갈 때 거래량도 함께 늘어나는지 봐야 해요. 거래량이 없으면 진짜가 아니거든요.',
'제 의견: 코스닥은 재미있는 테마가 많지만 아주 빠르게 변해요. 계속 지켜봐야 합니다.'
),
]
for keyword, line1, line2, line3, opinion in rules:
if keyword.lower() in t.lower():
return line1, line2, line3, opinion
return (
'핵심: 시장에 영향을 줄 수 있는 소식이 나왔습니다.',
'의미: 이 소식이 어떤 회사들한테 좋은지 나쁜지 생각해봐야 합니다.',
'체크: 이 뉴스가 그 다음날, 그 다음주에도 계속 영향을 주는지 지켜봐야 합니다.',
'제 의견: 뉴스 제목만 보지 말고 실제로 주가가 어떻게 움직이는지 봐야 가장 정확해요.'
)
NEWS_CUTOFF_SCORE = 5
NEWS_PER_CATEGORY_CAP = 2
NEWS_TOTAL_CAP = 12 # 전체 상한: 카테고리별 cap 통과 후 점수순 상위 N건만
NEWS_DUP_THRESHOLD = 0.40 # min-overlap ratio: overlap / min(|A|,|B|) >= threshold → 같은 사안으로 간주
def _title_tokens(title: str) -> set:
return set(re.findall(r'[가-힣0-9A-Za-z]{2,}', title.lower()))
def _is_similar_to_seen(tokens: set, seen_token_sets: list, threshold: float = NEWS_DUP_THRESHOLD) -> bool:
if not tokens:
return False
for t in seen_token_sets:
if not t:
continue
denom = min(len(tokens), len(t))
if not denom:
continue
if len(tokens & t) / denom >= threshold:
return True
return False
def briefing_cutoff(mode: str, now: datetime) -> datetime:
"""뉴스 발행시각 하한선: 이 시각 이후 발행분만 포함."""
if mode == 'morning':
return (now - timedelta(days=1)).replace(hour=19, minute=0, second=0, microsecond=0)
return now.replace(hour=7, minute=0, second=0, microsecond=0)
def _normalize_news_key(title: str) -> str:
return re.sub(r'[^0-9A-Za-z가-힣]+', '', title).lower()
def _parse_iso(s: Optional[str]) -> datetime:
if not s:
return datetime.min.replace(tzinfo=KST)
try:
dt = datetime.fromisoformat(s)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=KST)
return dt
except Exception:
return datetime.min.replace(tzinfo=KST)
def load_news_history() -> dict:
return load_json(NEWS_HISTORY_FILE, {'items': []})
def gc_news_history(history: dict, now: datetime) -> dict:
cutoff = now - timedelta(days=NEWS_HISTORY_RETENTION_DAYS)
history['items'] = [
it for it in history.get('items', [])
if _parse_iso(it.get('sent_at')) >= cutoff
]
return history
def recent_history_keys(history: dict, now: datetime, lookback_hours: int = NEWS_HISTORY_LOOKBACK_HOURS) -> set:
cutoff = now - timedelta(hours=lookback_hours)
return {
it['key'] for it in history.get('items', [])
if it.get('key') and _parse_iso(it.get('sent_at')) >= cutoff
}
def mark_news_sent_from_pending(mode: str, today_key: str, now: datetime) -> int:
"""발송 성공 후 pending selection을 history로 옮기고 pending에서 제거. 반환: 추가된 항목 수."""
pending = load_json(PENDING_SELECTION_FILE, {})
selection = pending.get(mode) or {}
if selection.get('date') != today_key:
return 0
articles = selection.get('articles', []) or []
history = load_news_history()
history = gc_news_history(history, now)
sent_at = now.isoformat()
added = 0
for a in articles:
key = a.get('key')
if not key:
continue
history['items'].append({
'key': key,
'title': a.get('title'),
'url': a.get('url'),
'mode': mode,
'sent_at': sent_at,
})
added += 1
save_json(NEWS_HISTORY_FILE, history)
pending.pop(mode, None)
save_json(PENDING_SELECTION_FILE, pending)
return added
def collect_news_articles(
cutoff_score: int = NEWS_CUTOFF_SCORE,
per_category_cap: int = NEWS_PER_CATEGORY_CAP,
published_after: Optional[datetime] = None,
exclude_keys: Optional[set] = None,
) -> list[dict]:
"""Return structured news articles passing cutoff + per-category cap + fuzzy title dedup.
published_after: 발행일이 이보다 이전인 항목 제외 (pub_dt 없으면 통과 — history에서 막힘)
exclude_keys: normalized title이 이 set에 있으면 제외 (최근 발송 이력)
"""
exclude_keys = exclude_keys or set()
seen_norm = set()
seen_token_sets = []
result = []
for category, url in NEWS_FEEDS:
try:
items = parse_google_news_feed(url)
except Exception as e:
print(f"[briefing_mail:news_feed:{category}] {type(e).__name__}: {e}", file=sys.stderr)
continue
candidates = []
for item in items:
title = item['title']
normalized = re.sub(r'[^0-9A-Za-z가-힣]+', '', title).lower()
if not normalized:
continue
pub_dt = item.get('pub_dt')
if published_after and pub_dt and pub_dt < published_after:
continue
if normalized in exclude_keys:
continue
score = news_priority(title)
if score < cutoff_score:
continue
candidates.append({
'category': category,
'title': title,
'link': item.get('link', ''),
'source': item.get('source', ''),
'description': item.get('description', ''),
'score': score,
'_norm': normalized,
'_tokens': _title_tokens(title),
})
candidates.sort(key=lambda x: x['score'], reverse=True)
cat_accepted = 0
for c in candidates:
if cat_accepted >= per_category_cap:
break
if c['_norm'] in seen_norm:
continue
if _is_similar_to_seen(c['_tokens'], seen_token_sets):
continue
seen_norm.add(c['_norm'])
seen_token_sets.append(c['_tokens'])
result.append({k: v for k, v in c.items() if not k.startswith('_')})
cat_accepted += 1
if NEWS_TOTAL_CAP and len(result) > NEWS_TOTAL_CAP:
result = sorted(result, key=lambda x: x['score'], reverse=True)[:NEWS_TOTAL_CAP]
return result
def get_simple_news(limit: Optional[int] = None) -> list[str]:
"""Legacy text-line output. Applies new cutoff/cap logic. `limit` caps total items (no forced min)."""
picked = collect_news_articles()
if limit is not None:
picked = sorted(picked, key=lambda x: x['score'], reverse=True)[:limit]
if not picked:
return ['- 지금 기준 통과한 핵심 경제 뉴스가 없습니다.']
lines = []
for item in picked:
lines.append(f"- [{item['category']}] {item['title']}")
if item.get('source'):
lines.append(f" 출처: {item['source']}")
if item.get('link'):
lines.append(f" 원문: {item['link']}")
lines.append('')
if lines and lines[-1] == '':
lines.pop()
return lines
def format_change(symbol: str, prev_close: float, close: float) -> str:
diff = close - prev_close
pct = (diff / prev_close * 100) if prev_close else 0.0
arrow = '▲' if diff > 0 else '▼' if diff < 0 else '-'
return f'- {symbol}: {close:,.2f} ({arrow} {abs(diff):,.2f}, {abs(pct):.2f}%)'
def get_naver_index(code: str, page_url: str, domestic: bool = False):
try:
base = 'https://m.stock.naver.com/api' if domestic else 'https://api.stock.naver.com'
raw = fetch_bytes(f'{base}/index/{code}/basic')
data = json.loads(raw.decode('utf-8', 'ignore'))
name = data.get('indexName') or data.get('stockName') or code
price = data.get('closePrice', '')
change = data.get('compareToPreviousClosePrice', '')
pct = data.get('fluctuationsRatio', '')
if not price:
return None
try:
change_f = float(change.replace(',', ''))
arrow = '▲' if change_f > 0 else '▼' if change_f < 0 else '-'
suffix = f' ({arrow} {abs(change_f):,.2f}, {abs(float(pct)):.2f}%)'
except (ValueError, TypeError):
suffix = ''
return f'- {name}: {price}{suffix}'
except Exception as e:
print(f"[briefing_mail:naver_index:{code}] {type(e).__name__}: {e}", file=sys.stderr)
return None
def get_market_snapshot(include_domestic: bool = False) -> list[str]:
lines = []
fx_lines = []
if include_domestic:
for code, page_url in NAVER_DOMESTIC_INDEX_FEEDS:
line = get_naver_index(code, page_url, domestic=True)
if line:
lines.append(line)
for code, page_url in NAVER_INDEX_FEEDS:
line = get_naver_index(code, page_url)
if line:
lines.append(line)
for symbol, url in MARKET_FEEDS:
try:
raw = fetch_bytes(url)
data = json.loads(raw.decode('utf-8', 'ignore'))
result = data['chart']['result'][0]
closes = [x for x in result['indicators']['quote'][0]['close'] if x is not None]
if len(closes) >= 2:
fx_lines.append(format_change(symbol, closes[-2], closes[-1]))
elif len(closes) == 1:
fx_lines.append(f'- {symbol}: {closes[-1]:,.2f}')
except Exception as e:
print(f"[briefing_mail:market_feed:{symbol}] {type(e).__name__}: {e}", file=sys.stderr)
continue
merged = lines + fx_lines
return merged or ['- 주요 시장 지표를 불러오지 못했습니다.']
def parse_investing_quote(url: str, fallback_title: str) -> Optional[str]:
try:
html = fetch(url)
title_match = re.search(r'\s*([^<]+?)\s*', html, re.IGNORECASE)
title = unescape(title_match.group(1)).strip() if title_match else fallback_title
price_match = re.search(r'data-test="instrument-price-last">\s*([^<]+?)\s*<', html)
change_match = re.search(r'data-test="instrument-price-change">\s*([^<]+?)\s*<', html)
pct_match = re.search(r'data-test="instrument-price-change-percent">\s*([^<]+?)\s*<', html)
price = unescape(price_match.group(1)).strip() if price_match else ''
change = unescape(change_match.group(1)).strip() if change_match else ''
pct = unescape(pct_match.group(1)).strip() if pct_match else ''
if price:
clean_title = re.sub(r'\s*선물 시세\s*$', '', title).strip()
change_clean = change.replace('+', '').strip()
pct_clean = pct.replace('(', '').replace(')', '').strip()
arrow = '-'
if change.startswith('+'):
arrow = '▲'
elif change.startswith('-'):
arrow = '▼'
suffix = ''
if change_clean and pct_clean:
suffix = f' ({arrow} {change_clean}, {pct_clean})'
elif change_clean:
suffix = f' ({arrow} {change_clean})'
elif pct_clean:
suffix = f' ({pct_clean})'
return f'- {clean_title}: {price}{suffix}'
except Exception as e:
print(f"[briefing_mail:investing_quote:{fallback_title}] {type(e).__name__}: {e}", file=sys.stderr)
return None
return None
def get_kospi_night_futures() -> list[str]:
line = parse_investing_quote(KOSPI_FUTURES_INVESTING_URL, '코스피200 선물 (F)')
if line:
line = re.sub(r'^-\s*[^:]+:', '- 코스피 선물:', line)
return [line]
return ['- 코스피 야간선물 데이터를 불러오지 못했습니다.']
def get_msci_korea() -> list[str]:
line = parse_investing_quote(MSCI_KOREA_INVESTING_URL, 'MSCI')
if line:
line = re.sub(r'^-\s*[^:]+:', '- MSCI:', line)
return [line]
return ['- MSCI 데이터를 불러오지 못했습니다.']
def was_already_sent(mode: str, today_key: str) -> bool:
state = load_json(SEND_STATE, {})
return state.get(mode) == today_key
def mark_sent(mode: str, today_key: str):
state = load_json(SEND_STATE, {})
state[mode] = today_key
save_json(SEND_STATE, state)
IPO_KIND_LABEL = {'subscription': '청약', 'listing': '상장'}
IPO_ACTION_LABEL = {'created': '신규', 'updated': '변경', 'recreated': '변경', 'deleted': '취소'}
def _format_ipo_change(item: dict) -> Optional[str]:
kind = (item.get('kind') or '').strip()
action = (item.get('action') or '').strip()
name = (item.get('name') or '').strip()
if not name or action not in IPO_ACTION_LABEL:
return None
kind_label = IPO_KIND_LABEL.get(kind, kind or '?')
new_date = (item.get('new_date') or '').strip()
old_date = (item.get('old_date') or '').strip()
if action in ('updated', 'recreated') and old_date and new_date and old_date != new_date:
return f'- [{kind_label}] {name} (변경, {old_date} → {new_date})'
if action == 'deleted':
return f'- [{kind_label}] {name} (취소)' + (f' — 기존 {old_date}' if old_date else '')
if new_date:
return f'- [{kind_label}] {name} ({IPO_ACTION_LABEL[action]}, {new_date})'
return f'- [{kind_label}] {name} ({IPO_ACTION_LABEL[action]})'
def read_ipo_changes_for_brief() -> list[str]:
"""IPO sync state(`ipo_calendar_sync.json`)에서 아직 브리핑에 노출되지 않은 changes를 줄 목록으로 반환.
sync 자체는 launchd(`ai.openclaw.stock.ipo-calendar-sync`, 매주 금 17:00)가 책임짐. 여기서는 읽기만.
같은 sync 결과(last_run_at 동일)는 한 번만 노출하기 위해 `briefing_ipo_seen.json` dedup 사용.
"""
state = load_json(IPO_SYNC_STATE, {})
last_run = (state.get('last_run_at') or '').strip()
if not last_run:
return []
seen = load_json(IPO_SEEN_STATE, {})
if seen.get('last_seen_run_at') == last_run:
return []
lines = []
for item in state.get('last_changes', []) or []:
line = _format_ipo_change(item)
if line:
lines.append(line)
return lines
def mark_ipo_changes_seen():
state = load_json(IPO_SYNC_STATE, {})
last_run = (state.get('last_run_at') or '').strip()
if not last_run:
return
save_json(IPO_SEEN_STATE, {'last_seen_run_at': last_run})
def get_tomorrow_listings(now: datetime) -> list[str]:
"""내일 [신규상장] 종목을 캘린더에서 추출 (D-1 임박 알림용)."""
start = now.astimezone(KST).replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)
end = start + timedelta(days=1)
try:
out = run([
'gog', 'calendar', 'events', 'mini.snowoyh@gmail.com',
'--from', start.isoformat(), '--to', end.isoformat(), '--json'
])
data = json.loads(out)
except Exception as e:
print(f"[briefing_mail:tomorrow_listings] {type(e).__name__}: {e}", file=sys.stderr)
return []
lines = []
for ev in data.get('events', []):
summary = (ev.get('summary') or '').strip()
if not summary.startswith('[신규상장]'):
continue
name = summary.replace('[신규상장]', '', 1).strip()
date_str = (ev.get('start', {}) or {}).get('date') or ''
lines.append(f'- {name} ({date_str})' if date_str else f'- {name}')
return lines
def build_morning() -> tuple[str, str]:
now = datetime.now(KST)
events = get_today_events(now)
ipo_changes = read_ipo_changes_for_brief()
tomorrow_listings = get_tomorrow_listings(now)
body = [
f'관리자님, 좋은 아침입니다.',
f'{now:%Y-%m-%d} 오전 브리핑만 간단히 정리드리겠습니다.',
''
]
body += section('오늘 일정', events or ['- 오늘 등록된 일정이 없습니다.'])
body += section('내일 상장', tomorrow_listings)
body += section('공모일정 변경', ipo_changes)
body += section('시장 체크', get_kospi_night_futures() + get_msci_korea() + get_market_snapshot())
body += section('중요 뉴스 요약', get_simple_news())
if body and body[-1] == '':
body.pop()
return (f'[오전 브리핑] {now:%Y-%m-%d}', '\n'.join(body))
def build_evening() -> tuple[str, str]:
now = datetime.now(KST)
tomorrow = now + timedelta(days=1)
events = get_today_events(tomorrow)
ipo_changes = read_ipo_changes_for_brief()
tomorrow_listings = get_tomorrow_listings(now)
body = [
f'관리자님, 오늘 마감 브리핑입니다.',
f'{now:%Y-%m-%d} 저녁 기준으로 짧게 정리드리겠습니다.',
''
]
body += section('내일 일정', events or ['- 내일 등록된 일정이 없습니다.'])
body += section('내일 상장', tomorrow_listings)
body += section('공모일정 변경', ipo_changes)
body += section('오늘 증시 간단 요약', get_market_snapshot(include_domestic=True))
body += section('핵심 뉴스', get_simple_news(3))
if body and body[-1] == '':
body.pop()
return (f'[오후 브리핑] {now:%Y-%m-%d}', '\n'.join(body))
def send_mail(subject: str, body: str, html: bool = False) -> str:
base = ['gog', 'gmail', 'send', '--to', RECIPIENT, '--subject', subject]
if html:
cmd = base + ['--body-html', body]
p = subprocess.run(cmd, text=True, capture_output=True)
else:
cmd = base + ['--body-file', '-']
p = subprocess.run(cmd, input=body, text=True, capture_output=True)
if p.returncode != 0:
raise RuntimeError(p.stderr.strip() or p.stdout.strip())
return p.stdout.strip()
WEEKDAY_KR = ['월', '화', '수', '목', '금', '토', '일']
def build_prepare_payload(mode: str) -> dict:
if mode not in {'morning', 'evening'}:
raise SystemExit('mode must be morning or evening')
now = datetime.now(KST)
target = now if mode == 'morning' else (now + timedelta(days=1))
events = get_today_events(target)
ipo_changes = read_ipo_changes_for_brief()
tomorrow_listings = get_tomorrow_listings(now)
cutoff = briefing_cutoff(mode, now)
history = load_news_history()
history = gc_news_history(history, now)
save_json(NEWS_HISTORY_FILE, history)
exclude_keys = recent_history_keys(history, now)
articles = collect_news_articles(published_after=cutoff, exclude_keys=exclude_keys)
today_key = now.strftime('%Y-%m-%d')
pending = load_json(PENDING_SELECTION_FILE, {})
pending[mode] = {
'date': today_key,
'articles': [
{
'key': _normalize_news_key(a['title']),
'title': a['title'],
'url': a.get('link', ''),
}
for a in articles
],
}
save_json(PENDING_SELECTION_FILE, pending)
return {
'mode': mode,
'date': today_key,
'date_target': target.strftime('%Y-%m-%d'),
'weekday_kr': WEEKDAY_KR[now.weekday()],
'events': events,
'ipo_changes': ipo_changes,
'tomorrow_listings': tomorrow_listings,
'articles': articles,
'news_window': {
'published_after': cutoff.isoformat(),
'history_lookback_hours': NEWS_HISTORY_LOOKBACK_HOURS,
'excluded_recent_count': len(exclude_keys),
},
'market': {
'futures': get_kospi_night_futures() if mode == 'morning' else [],
'msci': get_msci_korea() if mode == 'morning' else [],
'indices': get_market_snapshot(include_domestic=(mode == 'evening')),
},
'already_sent': was_already_sent(mode, today_key),
}
def cmd_prepare(mode: str) -> int:
payload = build_prepare_payload(mode)
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 0
def html_escape(s: str) -> str:
return (str(s)
.replace('&', '&')
.replace('<', '<')
.replace('>', '>')
.replace('"', '"')
.replace("'", '''))
HTML_STYLES = {
'wrap': "font-family:-apple-system,BlinkMacSystemFont,'Apple SD Gothic Neo',sans-serif;font-size:14px;color:#222;max-width:680px;line-height:1.55;",
'greet': 'color:#444;margin-bottom:18px;',
'card': 'margin-bottom:18px;border:1px solid #e5e5e5;border-radius:8px;padding:14px 16px;background:#fafafa;',
'card_title': 'font-size:14px;font-weight:700;color:#111;margin-bottom:10px;',
'line': 'color:#333;font-size:13px;margin:3px 0;',
'market_line': 'color:#333;font-size:13px;margin:4px 0;',
'market_label': 'color:#444;',
'market_value': "font-family:ui-monospace,SFMono-Regular,Menlo,monospace;color:#111;",
'pos': 'color:#d24f4f;font-weight:600;',
'neg': 'color:#1565c0;font-weight:600;',
'flat': 'color:#666;',
'news_section': 'font-size:14px;font-weight:700;color:#111;margin:24px 0 10px;border-left:3px solid #333;padding-left:8px;',
'article': 'margin-bottom:12px;border:1px solid #eee;border-radius:6px;padding:12px 14px;background:#fff;',
'category': 'display:inline-block;background:#eef3ff;color:#0066cc;font-size:11px;font-weight:600;padding:2px 8px;border-radius:10px;margin-right:6px;vertical-align:middle;',
'title': 'color:#111;font-weight:600;font-size:14px;',
'source': 'color:#888;font-size:12px;margin:6px 0 8px;',
'link': 'color:#0066cc;text-decoration:none;',
'summary': 'color:#333;font-size:13px;line-height:1.6;',
}
def _strip_bullet(text: str) -> str:
return re.sub(r'^\s*-\s*', '', text).strip()
def _split_market_line(line: str) -> tuple[str, str, str]:
"""`- 코스피: 6,641.02 (▲ 25.99, 0.39%)` → (label, value, change)"""
text = _strip_bullet(line)
label, _, rest = text.partition(':')
rest = rest.strip()
change = ''
value = rest
if '(' in rest and rest.rstrip().endswith(')'):
head, _, tail = rest.rpartition('(')
value = head.strip()
change = tail.rstrip(')').strip()
return label.strip(), value, change
def _market_change_color(change: str) -> str:
if '▲' in change:
return HTML_STYLES['pos']
if '▼' in change:
return HTML_STYLES['neg']
return HTML_STYLES['flat']
def _card(title: str, inner_html: str) -> str:
S = HTML_STYLES
return (
f''
f'
{html_escape(title)}
'
f'{inner_html}'
f'
{html_escape(_strip_bullet(l))}
'
for l in lines
)
return _card(title, body)
def _market_card(title: str, lines: list) -> str:
if not lines:
return ''
S = HTML_STYLES
rows = []
for l in lines:
label, value, change = _split_market_line(l)
if not label and not value:
rows.append(f'{html_escape(_strip_bullet(l))}
')
continue
right = html_escape(value)
if change:
color = _market_change_color(change)
right = (
f'{html_escape(value)} '
f'({html_escape(change)})'
)
rows.append(
f''
f'{html_escape(label)}'
f' '
f'{right}'
f'
'
)
return _card(title, '\n'.join(rows))
def _article_card(a: dict) -> str:
S = HTML_STYLES
title = html_escape(a.get('title', ''))
cat = html_escape(a.get('category', ''))
source = html_escape(a.get('source', ''))
link = a.get('link', '')
summary = a.get('summary') or ''
if isinstance(summary, dict):
summary = ' '.join(
str(v) for v in (summary.get(k) for k in ('core', 'meaning', 'check', 'opinion')) if v
)
link_html = (
f' | 원문 ↗'
if link else ''
)
parts = [f'']
parts.append(
f'
{cat}'
f'{title}
'
)
parts.append(f'
출처: {source}{link_html}
')
if summary:
parts.append(
f'
{html_escape(summary).replace(chr(10), "
")}
'
)
parts.append('
']
out.append(
f'
{html_escape(greet_line1)}
{html_escape(greet_line2)}
'
)
out.append(_list_card(today_title, events, today_empty))
if tomorrow_listings:
out.append(_list_card('내일 상장', tomorrow_listings))
if ipo:
out.append(_list_card('공모일정 변경', ipo))
out.append(_market_card(market_title, market_lines))
if articles:
out.append(f'
주요 뉴스
')
for a in articles:
out.append(_article_card(a))
out.append('