Initial commit: OpenClaw 워크스페이스 버전관리 시작

설정·스크립트·스킬·문서·큐레이션 메모리 추적. 시크릿(credentials/identity)·런타임 상태(state/logs/sessions/sqlite)· 백업(clobbered/bak)·dream 캐시는 .gitignore로 제외. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-04 15:39:41 +09:00
commit fed3526b20
199 changed files with 49671 additions and 0 deletions
@@ -0,0 +1,393 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import urllib.request
+from dataclasses import dataclass
+from urllib.parse import urljoin
+from datetime import date, datetime, timedelta
+from html import unescape
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+KST = ZoneInfo('Asia/Seoul')
+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 / 'ipo_calendar_sync.json'
+SUBSCRIPTION_URL = 'https://www.38.co.kr/html/fund/index.htm?o=k'
+NAVER_IPO_URL = 'https://finance.naver.com/sise/ipo.naver'
+SOURCE_LABEL = '네이버 금융 IPO'
+
+
+@dataclass
+class EventSpec:
+    kind: str
+    name: str
+    start_date: date
+    end_date: date  # inclusive
+    brokers: str
+    source_url: str
+
+    @property
+    def summary(self) -> str:
+        prefix = '[공모청약]' if self.kind == 'subscription' else '[신규상장]'
+        return f'{prefix} {self.name}'
+
+    @property
+    def description(self) -> str:
+        label = '청약일' if self.kind == 'subscription' else '상장일'
+        if self.start_date == self.end_date:
+            date_str = self.start_date.isoformat()
+        else:
+            date_str = f'{self.start_date.isoformat()} ~ {self.end_date.isoformat()}'
+        return (
+            f'종목명: {self.name}\n'
+            f'증권사: {self.brokers or "미확인"}\n'
+            f'{label}: {date_str}\n'
+            f'기준: {SOURCE_LABEL}\n'
+            f'출처: {self.source_url}'
+        )
+
+    @property
+    def state_key(self) -> str:
+        return f'{self.kind}|{self.name}'
+
+
+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, encoding: str = 'euc-kr') -> str:
+    req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
+    with urllib.request.urlopen(req, timeout=30) as r:
+        raw = r.read()
+    if encoding == 'auto':
+        # 한글 페이지는 utf-8과 cp949 둘 중 하나 — strict 디코드 성공하는 쪽 사용 (네이버가 메타태그로 거짓말하는 경우 대비)
+        for enc in ('utf-8', 'cp949'):
+            try:
+                return raw.decode(enc)
+            except UnicodeDecodeError:
+                continue
+        return raw.decode('utf-8', errors='ignore')
+    return raw.decode(encoding, 'ignore')
+
+
+def clean_text(html_fragment: str) -> str:
+    text = re.sub(r'<br\s*/?>', ' ', html_fragment, flags=re.I)
+    text = re.sub(r'<[^>]+>', ' ', text)
+    text = unescape(text)
+    text = text.replace('\xa0', ' ')
+    return re.sub(r'\s+', ' ', text).strip()
+
+
+def parse_html_rows(table_html: str) -> list[list[str]]:
+    rows = []
+    for row in re.findall(r'<tr[^>]*>(.*?)</tr>', table_html, re.S | re.I):
+        cols = [clean_text(c) for c in re.findall(r'<t[dh][^>]*>(.*?)</t[dh]>', row, re.S | re.I)]
+        if cols:
+            rows.append(cols)
+    return rows
+
+
+def parse_date_range(text: str) -> tuple[date, date] | None:
+    text = text.strip()
+    m = re.match(r'(\d{4})\.(\d{2})\.(\d{2})~(\d{2})\.(\d{2})$', text)
+    if not m:
+        return None
+    y, m1, d1, m2, d2 = map(int, m.groups())
+    return date(y, m1, d1), date(y, m2, d2)
+
+
+def parse_single_date(text: str) -> date | None:
+    m = re.match(r'(\d{4})\.(\d{2})\.(\d{2})$', text.strip())
+    if not m:
+        return None
+    y, mm, dd = map(int, m.groups())
+    return date(y, mm, dd)
+
+
+def extract_next_data_json(html: str) -> dict:
+    m = re.search(r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>', html, re.S)
+    if not m:
+        return {}
+    try:
+        return json.loads(m.group(1))
+    except Exception:
+        return {}
+
+
+def extract_naver_ipo_detail(detail_url: str) -> dict:
+    html = fetch(detail_url, encoding='utf-8')
+    data = extract_next_data_json(html)
+    queries = data.get('props', {}).get('pageProps', {}).get('dehydratedState', {}).get('queries', [])
+    for q in queries:
+        result = q.get('state', {}).get('data', {}).get('result', {})
+        ipo_info = result.get('ipoInfo')
+        if ipo_info:
+            return ipo_info
+    return {}
+
+
+def extract_naver_ipo_entries() -> list[dict]:
+    html = fetch(NAVER_IPO_URL, encoding='auto')
+    entries = []
+    seen = set()
+    for m in re.finditer(r'<div class="item_area" id="([^"]+)">.*?<a href="(https://m\.stock\.naver\.com/ipo/[^"]+)"[^>]*>(.*?)</a>', html, re.S):
+        code = m.group(1).strip()
+        detail_url = m.group(2).strip()
+        name = clean_text(m.group(3))
+        if not code or code in seen or not name:
+            continue
+        seen.add(code)
+        entries.append({'code': code, 'name': name, 'detail_url': detail_url})
+    return entries
+
+
+def extract_brokers_from_naver_detail(detail_url: str) -> tuple[dict, str]:
+    html = fetch(detail_url, encoding='utf-8')
+    data = extract_next_data_json(html)
+    queries = data.get('props', {}).get('pageProps', {}).get('dehydratedState', {}).get('queries', [])
+    ipo_info = {}
+    brokers = []
+    for q in queries:
+        result = q.get('state', {}).get('data', {}).get('result', {})
+        if isinstance(result, dict) and result.get('ipoInfo'):
+            ipo_info = result.get('ipoInfo') or {}
+            join_managers = result.get('joinManagers') or []
+            for item in join_managers:
+                if not isinstance(item, dict):
+                    continue
+                name = (item.get('orgNm') or '').strip()
+                if name and name not in brokers:
+                    brokers.append(name)
+            break
+    return ipo_info, ','.join(brokers)
+
+
+def parse_subscription_events_from(cutoff: date) -> list[EventSpec]:
+    events = []
+    for entry in extract_naver_ipo_entries():
+        info, brokers = extract_brokers_from_naver_detail(entry['detail_url'])
+        start_raw = (info.get('poStartDate') or '').strip()
+        end_raw = (info.get('poEndDate') or '').strip()
+        if not start_raw or not end_raw or '미정' in start_raw or '미정' in end_raw:
+            continue
+        try:
+            start_date = date.fromisoformat(start_raw)
+            end_date = date.fromisoformat(end_raw)
+        except ValueError:
+            continue
+        if end_date <= cutoff:
+            continue
+        events.append(EventSpec('subscription', info.get('compName') or entry['name'], start_date, end_date, brokers, entry['detail_url']))
+    return events
+
+
+def parse_listed_events_from(cutoff: date) -> list[EventSpec]:
+    events = []
+    for entry in extract_naver_ipo_entries():
+        info, brokers = extract_brokers_from_naver_detail(entry['detail_url'])
+        listed_raw = (info.get('lcalDate') or info.get('listingDate') or info.get('listDate') or '').strip()
+        if not listed_raw or '미정' in listed_raw:
+            continue
+        try:
+            d = date.fromisoformat(listed_raw)
+        except ValueError:
+            continue
+        if d <= cutoff:
+            continue
+        events.append(EventSpec('listing', info.get('compName') or entry['name'], d, d, brokers, entry['detail_url']))
+    return events
+
+
+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 event_key_from_summary(summary: str) -> tuple[str, str] | None:
+    summary = (summary or '').strip()
+    if summary.startswith('[공모청약] '):
+        return ('subscription', summary.replace('[공모청약] ', '', 1).strip())
+    if summary.startswith('[신규상장] '):
+        return ('listing', summary.replace('[신규상장] ', '', 1).strip())
+    return None
+
+
+def fetch_existing_events(start_date: date, end_date: date) -> dict[str, 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', []):
+        parsed = event_key_from_summary(ev.get('summary', ''))
+        if not parsed:
+            continue
+        kind, name = parsed
+        key = f'{kind}|{name}'
+        existing.setdefault(key, []).append(ev)
+    return existing
+
+
+def create_event(ev: EventSpec, dry_run: bool = False):
+    start_date = ev.start_date.isoformat()
+    end_date = (ev.end_date + timedelta(days=1)).isoformat()  # Google Calendar 종일 이벤트는 end가 exclusive
+    if dry_run:
+        print(json.dumps({'summary': ev.summary, 'date': start_date, 'description': ev.description, 'all_day': True}, ensure_ascii=False))
+        return
+    run([
+        'gog', 'calendar', 'create', CALENDAR_ID,
+        '--summary', ev.summary,
+        '--description', ev.description,
+        '--from', start_date,
+        '--to', end_date,
+        '--all-day',
+        '--event-color', '5' if ev.kind == 'subscription' else '10'
+    ])
+
+
+def update_event(event_id: str, ev: EventSpec, dry_run: bool = False):
+    start_date = ev.start_date.isoformat()
+    end_date = (ev.end_date + timedelta(days=1)).isoformat()
+    if dry_run:
+        print(json.dumps({'update_event_id': event_id, 'summary': ev.summary, 'date': start_date, 'description': ev.description, 'all_day': True}, ensure_ascii=False))
+        return 'updated'
+    try:
+        run([
+            'gog', 'calendar', 'update', CALENDAR_ID, event_id,
+            '--summary', ev.summary,
+            '--description', ev.description,
+            '--from', start_date,
+            '--to', end_date,
+            '--all-day',
+            '--event-color', '5' if ev.kind == 'subscription' else '10'
+        ])
+        return 'updated'
+    except Exception:
+        run(['gog', 'calendar', 'delete', CALENDAR_ID, event_id, '--force', '--no-input'])
+        create_event(ev, 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')
+    args = parser.parse_args()
+
+    cutoff = datetime.now(KST).date()
+    events = parse_subscription_events_from(cutoff) + parse_listed_events_from(cutoff)
+    events.sort(key=lambda e: (e.start_date, e.kind, e.name))
+
+    state = load_state()
+    end_date = max((ev.end_date for ev in events), default=cutoff) + timedelta(days=1)
+    # 사이트 스크래핑 실패 시 모든 일정이 삭제되는 사고 방지 — events 비면 cleanup 스킵
+    existing = {} if not events else fetch_existing_events(cutoff + timedelta(days=1), end_date)
+    created = []
+    updated = []
+    recreated = []
+    deleted = []
+    duplicates_removed = []
+    unchanged = 0
+    changes = []
+
+    event_keys = {ev.state_key for ev in events}
+
+    for ev in events:
+        existing_list = existing.get(ev.state_key, [])
+        # 같은 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(ev.state_key)
+                old_start, old_end = event_date_range(dup)
+                old_date = old_start if old_start == old_end or not old_end else f'{old_start}~{old_end}'
+                changes.append({'kind': ev.kind, 'name': ev.name, 'action': 'duplicate_removed', 'event_id': dup['id'], 'old_date': old_date})
+
+        existing_ev = existing_list[0] if existing_list else None
+
+        if not existing_ev:
+            create_event(ev, dry_run=args.dry_run)
+            created.append(ev.state_key)
+            changes.append({'kind': ev.kind, 'name': ev.name, 'action': 'created', 'new_date': ev.start_date.isoformat() if ev.start_date == ev.end_date else f'{ev.start_date.isoformat()}~{ev.end_date.isoformat()}'})
+            continue
+
+        existing_start, existing_end = event_date_range(existing_ev)
+        desired_start = ev.start_date.isoformat()
+        desired_end = ev.end_date.isoformat()
+        existing_description = (existing_ev.get('description') or '').strip()
+        desired_description = ev.description.strip()
+
+        if existing_start != desired_start or existing_end != desired_end or existing_description != desired_description:
+            result = update_event(existing_ev['id'], ev, dry_run=args.dry_run)
+            if result == 'recreated':
+                recreated.append(ev.state_key)
+            else:
+                updated.append(ev.state_key)
+            old_date = existing_start if existing_start == existing_end or not existing_end else f'{existing_start}~{existing_end}'
+            new_date = desired_start if desired_start == desired_end else f'{desired_start}~{desired_end}'
+            changes.append({'kind': ev.kind, 'name': ev.name, 'action': result, 'old_date': old_date, 'new_date': new_date})
+        else:
+            unchanged += 1
+
+    # 사이트에서 사라진 일정은 캘린더에서도 삭제 (events 비면 위에서 existing이 빈 dict라 자동으로 스킵됨)
+    for key, existing_list in existing.items():
+        if key in event_keys:
+            continue
+        kind, name = key.split('|', 1)
+        for stale in existing_list:
+            delete_event(stale['id'], dry_run=args.dry_run)
+            deleted.append(key)
+            old_start, old_end = event_date_range(stale)
+            old_date = old_start if old_start == old_end or not old_end else f'{old_start}~{old_end}'
+            changes.append({'kind': kind, 'name': name, 'action': 'deleted', 'event_id': stale['id'], 'old_date': old_date})
+
+    state['last_changes'] = changes
+    state['last_run_at'] = datetime.now(KST).isoformat()
+    if not args.dry_run:
+        save_state(state)
+
+    print(json.dumps({'cutoff_after': cutoff.isoformat(), 'total_found': len(events), '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()