hyowons
342 lines
13 KiB
Python
342 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""미 연준 FOMC 회의 일정 → Google 캘린더 동기화 (조회+등록 전용).
|
|
|
|
출처: federalreserve.gov 공식 FOMC 일정표. 회의 날짜는 미 동부 기준 그대로 등록한다
|
|
(기존 수동 등록분·시황 자막 관례와 동일).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime, timedelta
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo
|
|
|
|
KST = ZoneInfo('Asia/Seoul')
|
|
ET = ZoneInfo('America/New_York')
|
|
CALENDAR_ID = 'mini.snowoyh@gmail.com'
|
|
WORKSPACE = Path('/Users/snowoyh/.openclaw/agents/stock/workspace')
|
|
STATE_DIR = WORKSPACE / 'state'
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
STATE_FILE = STATE_DIR / 'fomc_calendar_sync.json'
|
|
SOURCE_URL = 'https://www.federalreserve.gov/monetarypolicy/fomccalendars.htm'
|
|
|
|
MONTHS = {
|
|
'january': 1, 'february': 2, 'march': 3, 'april': 4, 'may': 5, 'june': 6,
|
|
'july': 7, 'august': 8, 'september': 9, 'october': 10, 'november': 11, 'december': 12,
|
|
}
|
|
|
|
YEAR_PANEL_RE = re.compile(r'(\d{4})\s+FOMC\s+Meetings')
|
|
MEETING_ROW_RE = re.compile(
|
|
r'fomc-meeting__month[^>]*>\s*(?:<strong>)?(.*?)(?:</strong>)?\s*</div>'
|
|
r'.*?fomc-meeting__date[^>]*>\s*(.*?)\s*</div>',
|
|
re.S,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Meeting:
|
|
start_date: date # 회의 시작 (미 동부)
|
|
end_date: date # 회의 종료 (미 동부, inclusive)
|
|
is_sep: bool
|
|
|
|
@property
|
|
def state_key(self) -> str:
|
|
# FOMC는 한 달에 두 번 열리지 않으므로 시작월이 고유 키
|
|
return f'{self.start_date:%Y-%m}'
|
|
|
|
@property
|
|
def summary(self) -> str:
|
|
base = f'[FOMC] {self.state_key} 회의'
|
|
return f'{base} (SEP)' if self.is_sep else base
|
|
|
|
@property
|
|
def announce_kst(self) -> datetime:
|
|
"""성명 발표 = 회의 마지막날 미 동부 14:00. 서머타임은 zoneinfo가 처리."""
|
|
et = datetime(self.end_date.year, self.end_date.month, self.end_date.day, 14, 0, tzinfo=ET)
|
|
return et.astimezone(KST)
|
|
|
|
@property
|
|
def cal_start(self) -> date:
|
|
return self.start_date
|
|
|
|
@property
|
|
def cal_end(self) -> date:
|
|
"""캘린더 이벤트 마지막날 = 발표일(KST). 미 동부 회의 종료 다음날이라
|
|
한국 장이 결과를 반영하는 날까지 일정이 덮인다."""
|
|
return self.announce_kst.date()
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
if self.start_date == self.end_date:
|
|
days = self.start_date.isoformat()
|
|
else:
|
|
days = f'{self.start_date.isoformat()} ~ {self.end_date.isoformat()}'
|
|
lines = [
|
|
f'회의일(미 동부): {days}',
|
|
f'성명 발표: {self.announce_kst:%Y-%m-%d %H:%M} KST (한국 장 반영일)',
|
|
]
|
|
if self.is_sep:
|
|
lines.append('경제전망(SEP)·점도표 공개 회의')
|
|
lines.append(f'출처: {SOURCE_URL}')
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def run(cmd: list[str]) -> str:
|
|
p = subprocess.run(cmd, capture_output=True, text=True)
|
|
if p.returncode != 0:
|
|
raise RuntimeError(p.stderr.strip() or p.stdout.strip() or 'command failed')
|
|
return p.stdout
|
|
|
|
|
|
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 parse_meetings(html: str, after: date) -> list[Meeting]:
|
|
"""연도 패널별로 회의 행을 훑어 캘린더 기간이 아직 안 끝난 회의만 반환.
|
|
|
|
기준을 회의 시작일이 아니라 `cal_end`(발표일 KST)로 두는 이유: 미 동부 회의가
|
|
끝났어도 한국 장 반영일이 남아 있으면 아직 관리 대상이다.
|
|
"""
|
|
panels = [(m.group(1), m.start()) for m in YEAR_PANEL_RE.finditer(html)]
|
|
if not panels:
|
|
return []
|
|
bounds = panels + [('', len(html))]
|
|
|
|
meetings: list[Meeting] = []
|
|
for (year_raw, start), (_, end) in zip(bounds[:-1], bounds[1:]):
|
|
year = int(year_raw)
|
|
if year < after.year:
|
|
continue
|
|
for row in MEETING_ROW_RE.finditer(html[start:end]):
|
|
month_txt = re.sub(r'<[^>]+>', '', row.group(1)).strip()
|
|
day_txt = re.sub(r'<[^>]+>', '', row.group(2)).strip()
|
|
m = parse_row(year, month_txt, day_txt)
|
|
if m and m.cal_end > after:
|
|
meetings.append(m)
|
|
meetings.sort(key=lambda m: m.start_date)
|
|
return meetings
|
|
|
|
|
|
def parse_row(year: int, month_txt: str, day_txt: str) -> Meeting | None:
|
|
"""'April/May' + '28-1*' 같은 월 경계 표기까지 처리."""
|
|
months = [MONTHS[p.strip().lower()] for p in month_txt.split('/') if p.strip().lower() in MONTHS]
|
|
if not months:
|
|
return None
|
|
|
|
is_sep = '*' in day_txt
|
|
days = [int(d) for d in re.findall(r'\d+', day_txt)]
|
|
if not days:
|
|
return None
|
|
|
|
start_month = months[0]
|
|
end_month = months[-1]
|
|
start_year = year
|
|
end_year = year if end_month >= start_month else year + 1 # 12월/1월 회의 대비
|
|
|
|
try:
|
|
start = date(start_year, start_month, days[0])
|
|
end = date(end_year, end_month, days[-1])
|
|
except ValueError:
|
|
return None
|
|
if end < start:
|
|
return None
|
|
return Meeting(start, end, is_sep)
|
|
|
|
|
|
def load_state() -> dict:
|
|
if STATE_FILE.exists():
|
|
try:
|
|
return json.loads(STATE_FILE.read_text())
|
|
except Exception:
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def save_state(state: dict):
|
|
STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2))
|
|
|
|
|
|
def state_key_from_summary(summary: str) -> str | None:
|
|
m = re.match(r'\[FOMC\]\s*(\d{4}-\d{2})\b', (summary or '').strip())
|
|
return m.group(1) if m else None
|
|
|
|
|
|
def fetch_existing_events(start_date: date, end_date: date) -> dict[str, list[dict]]:
|
|
start_dt = datetime(start_date.year, start_date.month, start_date.day, 0, 0, tzinfo=KST)
|
|
end_dt = datetime(end_date.year, end_date.month, end_date.day, 0, 0, tzinfo=KST)
|
|
out = run([
|
|
'gog', 'calendar', 'events', CALENDAR_ID,
|
|
'--from', start_dt.isoformat(), '--to', end_dt.isoformat(),
|
|
'--all-pages', '--max', '250', '--json'
|
|
])
|
|
data = json.loads(out)
|
|
existing: dict[str, list[dict]] = {}
|
|
for ev in data.get('events', []):
|
|
key = state_key_from_summary(ev.get('summary', ''))
|
|
if key:
|
|
existing.setdefault(key, []).append(ev)
|
|
return existing
|
|
|
|
|
|
def create_event(m: Meeting, dry_run: bool = False):
|
|
start = m.cal_start.isoformat()
|
|
end = (m.cal_end + timedelta(days=1)).isoformat() # 종일 이벤트의 end는 exclusive
|
|
if dry_run:
|
|
print(json.dumps({'create': m.summary, 'from': start, 'to': m.cal_end.isoformat()}, ensure_ascii=False))
|
|
return
|
|
# --event-color 미지정: 기존 이벤트에 지정된 색을 덮어쓰지 않는다
|
|
run([
|
|
'gog', 'calendar', 'create', CALENDAR_ID,
|
|
'--summary', m.summary,
|
|
'--description', m.description,
|
|
'--from', start,
|
|
'--to', end,
|
|
'--all-day',
|
|
])
|
|
|
|
|
|
def update_event(event_id: str, m: Meeting, dry_run: bool = False) -> str:
|
|
start = m.cal_start.isoformat()
|
|
end = (m.cal_end + timedelta(days=1)).isoformat()
|
|
if dry_run:
|
|
print(json.dumps({'update': m.summary, 'event_id': event_id, 'from': start, 'to': m.cal_end.isoformat()}, ensure_ascii=False))
|
|
return 'updated'
|
|
try:
|
|
run([
|
|
'gog', 'calendar', 'update', CALENDAR_ID, event_id,
|
|
'--summary', m.summary,
|
|
'--description', m.description,
|
|
'--from', start,
|
|
'--to', end,
|
|
'--all-day',
|
|
])
|
|
return 'updated'
|
|
except Exception:
|
|
run(['gog', 'calendar', 'delete', CALENDAR_ID, event_id, '--force', '--no-input'])
|
|
create_event(m, dry_run=False)
|
|
return 'recreated'
|
|
|
|
|
|
def delete_event(event_id: str, dry_run: bool = False):
|
|
if dry_run:
|
|
print(json.dumps({'delete_event_id': event_id}, ensure_ascii=False))
|
|
return
|
|
run(['gog', 'calendar', 'delete', CALENDAR_ID, event_id, '--force', '--no-input'])
|
|
|
|
|
|
def event_date_range(ev: dict) -> tuple[str, str]:
|
|
start_info = ev.get('start', {})
|
|
end_info = ev.get('end', {})
|
|
start = start_info.get('date') or start_info.get('dateTime', '')[:10]
|
|
end_excl = end_info.get('date') or end_info.get('dateTime', '')[:10]
|
|
end = ''
|
|
if end_excl:
|
|
end = (date.fromisoformat(end_excl) - timedelta(days=1)).isoformat()
|
|
return start, end
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--dry-run', action='store_true')
|
|
parser.add_argument('--list', action='store_true', help='파싱 결과만 출력 (캘린더 미접근)')
|
|
args = parser.parse_args()
|
|
|
|
today = datetime.now(KST).date()
|
|
meetings = parse_meetings(fetch(SOURCE_URL), today)
|
|
|
|
if args.list:
|
|
print(json.dumps({
|
|
'today': today.isoformat(),
|
|
'total': len(meetings),
|
|
'meetings': [
|
|
{'key': m.state_key, 'summary': m.summary,
|
|
'meeting_et': f'{m.start_date.isoformat()}~{m.end_date.isoformat()}',
|
|
'announce_kst': m.announce_kst.strftime('%Y-%m-%d %H:%M'),
|
|
'cal_start': m.cal_start.isoformat(), 'cal_end': m.cal_end.isoformat()}
|
|
for m in meetings
|
|
],
|
|
}, ensure_ascii=False, indent=2))
|
|
return
|
|
|
|
state = load_state()
|
|
end_date = max((m.cal_end for m in meetings), default=today) + timedelta(days=1)
|
|
# 스크래핑 실패 시 기존 일정이 몰살되는 사고 방지 — meetings 비면 cleanup 스킵
|
|
# 조회 시작이 today인 건 오늘 걸쳐 있는 진행 중 회의도 갱신 대상이기 때문
|
|
existing = {} if not meetings else fetch_existing_events(today, end_date)
|
|
|
|
created, updated, recreated, deleted, duplicates_removed = [], [], [], [], []
|
|
unchanged = 0
|
|
changes = []
|
|
keys = {m.state_key for m in meetings}
|
|
|
|
for m in meetings:
|
|
existing_list = existing.get(m.state_key, [])
|
|
# 같은 키로 여러 건이면 가장 최근 created를 남기고 나머지 삭제
|
|
if len(existing_list) > 1:
|
|
existing_list.sort(key=lambda e: e.get('created', ''), reverse=True)
|
|
for dup in existing_list[1:]:
|
|
delete_event(dup['id'], dry_run=args.dry_run)
|
|
duplicates_removed.append(m.state_key)
|
|
changes.append({'key': m.state_key, 'action': 'duplicate_removed', 'event_id': dup['id']})
|
|
|
|
existing_ev = existing_list[0] if existing_list else None
|
|
|
|
if not existing_ev:
|
|
create_event(m, dry_run=args.dry_run)
|
|
created.append(m.state_key)
|
|
changes.append({'key': m.state_key, 'action': 'created',
|
|
'new_date': f'{m.cal_start.isoformat()}~{m.cal_end.isoformat()}'})
|
|
continue
|
|
|
|
existing_start, existing_end = event_date_range(existing_ev)
|
|
desired_start, desired_end = m.cal_start.isoformat(), m.cal_end.isoformat()
|
|
same_summary = (existing_ev.get('summary') or '').strip() == m.summary
|
|
same_desc = (existing_ev.get('description') or '').strip() == m.description.strip()
|
|
|
|
if existing_start != desired_start or existing_end != desired_end or not same_summary or not same_desc:
|
|
result = update_event(existing_ev['id'], m, dry_run=args.dry_run)
|
|
(recreated if result == 'recreated' else updated).append(m.state_key)
|
|
changes.append({'key': m.state_key, 'action': result,
|
|
'old_date': f'{existing_start}~{existing_end}',
|
|
'new_date': f'{desired_start}~{desired_end}'})
|
|
else:
|
|
unchanged += 1
|
|
|
|
# 연준 일정표에서 사라진 미래 회의는 캘린더에서도 삭제 (meetings 비면 existing이 빈 dict라 자동 스킵)
|
|
for key, existing_list in existing.items():
|
|
if key in keys:
|
|
continue
|
|
for stale in existing_list:
|
|
old_start, old_end = event_date_range(stale)
|
|
# 오늘 진행 중이거나 지난 회의는 parse 대상(오늘 이후)에서 빠져 있을 뿐이므로 삭제 금지
|
|
if old_start and old_start <= today.isoformat():
|
|
continue
|
|
delete_event(stale['id'], dry_run=args.dry_run)
|
|
deleted.append(key)
|
|
changes.append({'key': key, 'action': 'deleted', 'event_id': stale['id'],
|
|
'old_date': f'{old_start}~{old_end}'})
|
|
|
|
state['last_changes'] = changes
|
|
state['last_run_at'] = datetime.now(KST).isoformat()
|
|
if not args.dry_run:
|
|
save_state(state)
|
|
|
|
print(json.dumps({
|
|
'today': today.isoformat(), 'total_found': len(meetings),
|
|
'newly_created': len(created), 'updated': len(updated), 'recreated': len(recreated),
|
|
'deleted': len(deleted), 'duplicates_removed': len(duplicates_removed),
|
|
'unchanged': unchanged, 'dry_run': args.dry_run, 'changes': changes,
|
|
}, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|