auto: 일일 백업 2026-07-30 02:00

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-30 02:00:02 +09:00
parent c5dd2f98a8
commit a5f6b0ecd8
181 changed files with 5372 additions and 5251 deletions
@@ -0,0 +1,341 @@
+#!/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()