auto: 일일 백업 2026-09-30 02:00
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
hyowons
@@ -5,9 +5,10 @@ import argparse
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.parse
|
||||
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
|
||||
@@ -19,9 +20,19 @@ 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'
|
||||
CONFIG_PATH = Path('/Users/snowoyh/.openclaw/openclaw.json')
|
||||
TELEGRAM_ACCOUNT = 'stock'
|
||||
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'
|
||||
LISTING_URL = 'https://www.38.co.kr/html/fund/index.htm?o=nw'
|
||||
NAVER_IPO_API_URL = 'https://stock.naver.com/api/domestic/market/ipo/progress'
|
||||
SOURCE_LABEL = '네이버페이 증권 IPO'
|
||||
LISTING_LOOKBACK_DAYS = 14
|
||||
|
||||
|
||||
class IpoParserError(RuntimeError):
|
||||
def __init__(self, errors: list[str]):
|
||||
self.errors = errors
|
||||
super().__init__('IPO 파서 전부 실패: ' + ' | '.join(errors))
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -32,6 +43,7 @@ class EventSpec:
|
||||
end_date: date # inclusive
|
||||
brokers: str
|
||||
source_url: str
|
||||
source_label: str = SOURCE_LABEL
|
||||
|
||||
@property
|
||||
def summary(self) -> str:
|
||||
@@ -49,7 +61,7 @@ class EventSpec:
|
||||
f'종목명: {self.name}\n'
|
||||
f'증권사: {self.brokers or "미확인"}\n'
|
||||
f'{label}: {date_str}\n'
|
||||
f'기준: {SOURCE_LABEL}\n'
|
||||
f'기준: {self.source_label}\n'
|
||||
f'출처: {self.source_url}'
|
||||
)
|
||||
|
||||
@@ -65,6 +77,38 @@ def run(cmd: list[str]) -> str:
|
||||
return p.stdout
|
||||
|
||||
|
||||
def send_telegram(text: str) -> bool:
|
||||
try:
|
||||
cfg = json.loads(CONFIG_PATH.read_text())
|
||||
acct = cfg['channels']['telegram']['accounts'][TELEGRAM_ACCOUNT]
|
||||
token = acct['botToken']
|
||||
chat_ids = acct.get('allowFrom') or []
|
||||
except Exception as e:
|
||||
print(f'telegram cfg load failed: {e}', file=sys.stderr)
|
||||
return False
|
||||
if not chat_ids:
|
||||
print('no telegram chat_ids', file=sys.stderr)
|
||||
return False
|
||||
|
||||
url = f'https://api.telegram.org/bot{token}/sendMessage'
|
||||
ok = True
|
||||
for chat_id in chat_ids:
|
||||
body = urllib.parse.urlencode({
|
||||
'chat_id': chat_id,
|
||||
'text': text[:4000],
|
||||
'disable_web_page_preview': 'true',
|
||||
}).encode()
|
||||
try:
|
||||
req = urllib.request.Request(url, data=body, method='POST')
|
||||
with urllib.request.urlopen(req, timeout=15) as response:
|
||||
if response.status != 200:
|
||||
ok = False
|
||||
except Exception as e:
|
||||
print(f'telegram send failed: {e}', file=sys.stderr)
|
||||
ok = False
|
||||
return ok
|
||||
|
||||
|
||||
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:
|
||||
@@ -106,108 +150,167 @@ def parse_date_range(text: str) -> tuple[date, date] | None:
|
||||
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 _naver_api_rows(payload: dict) -> list[dict]:
|
||||
"""네이버 IPO progress API의 단계별 배열을 종목 단위로 합친다."""
|
||||
list_keys = (
|
||||
'examinationList', 'demandForecastingList', 'forecastingCompleteList',
|
||||
'subscriptionList', 'subscriptionCompleteList', 'listingList',
|
||||
)
|
||||
if not isinstance(payload, dict) or not any(key in payload for key in list_keys):
|
||||
raise ValueError('네이버 IPO API 스키마 불일치')
|
||||
rows: dict[str, dict] = {}
|
||||
for key in list_keys:
|
||||
items = payload.get(key) or []
|
||||
if not isinstance(items, list):
|
||||
raise ValueError(f'네이버 IPO API {key} 타입 불일치')
|
||||
for item in items:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
code = (item.get('ipoCode') or item.get('itemCode') or '').strip()
|
||||
name = (item.get('compName') or item.get('stockName') or '').strip()
|
||||
if not code or not name:
|
||||
continue
|
||||
rows[code] = {**rows.get(code, {}), **item}
|
||||
return list(rows.values())
|
||||
|
||||
|
||||
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 parse_naver_api_events_from(cutoff: date) -> list[EventSpec]:
|
||||
payload = json.loads(fetch(NAVER_IPO_API_URL, encoding='utf-8'))
|
||||
events: list[EventSpec] = []
|
||||
for info in _naver_api_rows(payload):
|
||||
code = info.get('ipoCode') or info.get('itemCode')
|
||||
name = (info.get('compName') or info.get('stockName') or '').strip()
|
||||
brokers = (info.get('orgNm') or '').strip()
|
||||
source_url = f'https://stock.naver.com/ipo/{code}'
|
||||
|
||||
|
||||
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']))
|
||||
if start_raw and end_raw:
|
||||
try:
|
||||
start_date = date.fromisoformat(start_raw)
|
||||
end_date = date.fromisoformat(end_raw)
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
if end_date >= cutoff:
|
||||
events.append(EventSpec(
|
||||
'subscription', name, start_date, end_date, brokers, source_url
|
||||
))
|
||||
|
||||
listed_raw = (info.get('lcalDate') or info.get('listingDate') or '').strip()
|
||||
if listed_raw:
|
||||
try:
|
||||
listed_date = date.fromisoformat(listed_raw)
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
if listed_date >= cutoff:
|
||||
events.append(EventSpec(
|
||||
'listing', name, listed_date, listed_date, brokers, source_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:
|
||||
def _table_by_summary(html: str, summary: str) -> str:
|
||||
match = re.search(
|
||||
r'<table[^>]*summary=["\']' + re.escape(summary) + r'["\'][^>]*>(.*?)</table>',
|
||||
html,
|
||||
re.S | re.I,
|
||||
)
|
||||
if not match:
|
||||
raise ValueError(f'38커뮤니케이션 테이블 없음: {summary}')
|
||||
return match.group(0)
|
||||
|
||||
|
||||
def parse_38_subscription_html(html: str, cutoff: date) -> list[EventSpec]:
|
||||
events: list[EventSpec] = []
|
||||
for cols in parse_html_rows(_table_by_summary(html, '공모주 청약일정')):
|
||||
if len(cols) < 6 or cols[0] == '종목명':
|
||||
continue
|
||||
parsed = parse_date_range(cols[1])
|
||||
if not parsed:
|
||||
continue
|
||||
start_date, end_date = parsed
|
||||
if end_date < cutoff:
|
||||
continue
|
||||
events.append(EventSpec(
|
||||
'subscription', cols[0], start_date, end_date, cols[5], SUBSCRIPTION_URL,
|
||||
source_label='38커뮤니케이션 IPO',
|
||||
))
|
||||
return events
|
||||
|
||||
|
||||
def parse_38_listing_html(html: str, cutoff: date) -> list[EventSpec]:
|
||||
events: list[EventSpec] = []
|
||||
for cols in parse_html_rows(_table_by_summary(html, '신규상장종목')):
|
||||
if len(cols) < 2 or cols[0] == '기업명':
|
||||
continue
|
||||
try:
|
||||
d = date.fromisoformat(listed_raw)
|
||||
listed_date = date.fromisoformat(cols[1].replace('/', '-'))
|
||||
except ValueError:
|
||||
continue
|
||||
if d <= cutoff:
|
||||
if listed_date < cutoff:
|
||||
continue
|
||||
events.append(EventSpec('listing', info.get('compName') or entry['name'], d, d, brokers, entry['detail_url']))
|
||||
events.append(EventSpec(
|
||||
'listing', cols[0], listed_date, listed_date, '', LISTING_URL,
|
||||
source_label='38커뮤니케이션 IPO',
|
||||
))
|
||||
return events
|
||||
|
||||
|
||||
def parse_38_events_from(cutoff: date) -> list[EventSpec]:
|
||||
subscriptions = parse_38_subscription_html(fetch(SUBSCRIPTION_URL), cutoff)
|
||||
listing_cutoff = cutoff - timedelta(days=LISTING_LOOKBACK_DAYS)
|
||||
listings = parse_38_listing_html(fetch(LISTING_URL), listing_cutoff)
|
||||
# 보조 소스는 스팩 종목명이 주 소스와 크게 달라 과거 건을 중복 생성할 수 있다.
|
||||
listings = [
|
||||
ev for ev in listings
|
||||
if ev.start_date == cutoff or '스팩' not in ev.name
|
||||
]
|
||||
return subscriptions + listings
|
||||
|
||||
|
||||
def collect_ipo_events(cutoff: date) -> tuple[list[EventSpec], str, list[str]]:
|
||||
"""네이버 API를 우선 사용하고 실패·0건이면 38 파서로 자동 전환한다.
|
||||
|
||||
네이버 API는 상장 당일 장 시작 후 종목을 listingList에서 제거하므로, 주 파서가
|
||||
정상이어도 최근 14일 신규상장을 38 목록으로 보강한다.
|
||||
"""
|
||||
errors: list[str] = []
|
||||
try:
|
||||
events = parse_naver_api_events_from(cutoff)
|
||||
except Exception as e:
|
||||
events = []
|
||||
errors.append(f'naver_api: {e}')
|
||||
|
||||
parser_used = 'naver_api'
|
||||
if not events:
|
||||
if not errors:
|
||||
errors.append('naver_api: 수집 결과 0건')
|
||||
try:
|
||||
events = parse_38_events_from(cutoff)
|
||||
parser_used = '38_fallback'
|
||||
except Exception as e:
|
||||
errors.append(f'38_fallback: {e}')
|
||||
raise IpoParserError(errors) from e
|
||||
else:
|
||||
try:
|
||||
listing_cutoff = cutoff - timedelta(days=LISTING_LOOKBACK_DAYS)
|
||||
recent_listings = parse_38_listing_html(fetch(LISTING_URL), listing_cutoff)
|
||||
existing_keys = {ev.state_key for ev in events}
|
||||
events.extend(
|
||||
ev for ev in recent_listings
|
||||
if ev.state_key not in existing_keys
|
||||
and ev.start_date <= cutoff
|
||||
and (ev.start_date == cutoff or '스팩' not in ev.name)
|
||||
)
|
||||
except Exception as e:
|
||||
errors.append(f'38_same_day: {e}')
|
||||
|
||||
deduped = {ev.state_key: ev for ev in events}
|
||||
return list(deduped.values()), parser_used, errors
|
||||
|
||||
|
||||
def load_state() -> dict:
|
||||
if STATE_FILE.exists():
|
||||
try:
|
||||
@@ -221,6 +324,59 @@ def save_state(state: dict):
|
||||
STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2))
|
||||
|
||||
|
||||
def update_parser_health(
|
||||
state: dict,
|
||||
status: str,
|
||||
errors: list[str],
|
||||
total_found: int,
|
||||
dry_run: bool = False,
|
||||
) -> None:
|
||||
"""파서 상태가 바뀔 때만 알리고, 실패한 발송은 다음 실행에서 재시도한다."""
|
||||
now_iso = datetime.now(KST).isoformat()
|
||||
previous = state.get('parser_health_status')
|
||||
state['parser_health_status'] = status
|
||||
state['parser_health_checked_at'] = now_iso
|
||||
state['parser_errors'] = errors
|
||||
|
||||
if status == 'healthy':
|
||||
state['parser_last_success_at'] = now_iso
|
||||
else:
|
||||
state['parser_last_failure_at'] = now_iso
|
||||
|
||||
if dry_run:
|
||||
return
|
||||
|
||||
message = None
|
||||
if status == 'failed' and (
|
||||
previous != status or state.get('parser_alerted_status') != status
|
||||
):
|
||||
message = (
|
||||
'[IPO 일정] 파서 전체 실패\n'
|
||||
+ '\n'.join(errors)
|
||||
+ '\n캘린더 동기화를 중단했고 기존 일정은 유지했습니다.'
|
||||
)
|
||||
elif status == 'degraded' and (
|
||||
previous != status or state.get('parser_alerted_status') != status
|
||||
):
|
||||
source = '38커뮤니케이션으로 자동 전환했습니다.'
|
||||
if not any(error.startswith('naver_api:') for error in errors):
|
||||
source = '네이버 주 파서는 정상이며 당일 상장 보강 파서만 실패했습니다.'
|
||||
message = (
|
||||
'[IPO 일정] 파서 일부 실패\n'
|
||||
+ '\n'.join(errors)
|
||||
+ f'\n{source}\n수집 결과: {total_found}건\n기존 일정 자동 삭제는 중지했습니다.'
|
||||
)
|
||||
elif status == 'healthy' and previous in {'degraded', 'failed'}:
|
||||
message = (
|
||||
'[IPO 일정] 파서 복구\n'
|
||||
f'네이버 IPO API와 당일 상장 보강 파서가 정상입니다.\n수집 결과: {total_found}건'
|
||||
)
|
||||
|
||||
if message and send_telegram(message):
|
||||
state['parser_alerted_status'] = status
|
||||
state['parser_alert_sent_at'] = now_iso
|
||||
|
||||
|
||||
def event_key_from_summary(summary: str) -> tuple[str, str] | None:
|
||||
summary = (summary or '').strip()
|
||||
if summary.startswith('[공모청약] '):
|
||||
@@ -314,13 +470,39 @@ def main():
|
||||
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()
|
||||
previous_parser_used = state.get('parser_used')
|
||||
try:
|
||||
events, parser_used, parser_errors = collect_ipo_events(cutoff)
|
||||
except IpoParserError as e:
|
||||
update_parser_health(state, 'failed', e.errors, 0, dry_run=args.dry_run)
|
||||
state['last_run_at'] = datetime.now(KST).isoformat()
|
||||
state['parser_used'] = 'failed'
|
||||
if not args.dry_run:
|
||||
save_state(state)
|
||||
print(json.dumps({
|
||||
'cutoff_after': cutoff.isoformat(),
|
||||
'parser_used': 'failed',
|
||||
'parser_errors': e.errors,
|
||||
'total_found': 0,
|
||||
'dry_run': args.dry_run,
|
||||
}, ensure_ascii=False), file=sys.stderr)
|
||||
return 1
|
||||
|
||||
events.sort(key=lambda e: (e.start_date, e.kind, e.name))
|
||||
parser_health = 'degraded' if parser_errors else 'healthy'
|
||||
update_parser_health(
|
||||
state, parser_health, parser_errors, len(events), dry_run=args.dry_run
|
||||
)
|
||||
state['parser_used'] = parser_used
|
||||
if not args.dry_run:
|
||||
# 알림 상태를 먼저 저장해 이후 캘린더 작업 실패 시 같은 경보가 반복되지 않게 한다.
|
||||
save_state(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)
|
||||
listing_cutoff = cutoff - timedelta(days=LISTING_LOOKBACK_DAYS)
|
||||
existing = {} if not events else fetch_existing_events(listing_cutoff, end_date)
|
||||
created = []
|
||||
updated = []
|
||||
recreated = []
|
||||
@@ -369,25 +551,32 @@ def main():
|
||||
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})
|
||||
# 보조 소스는 종목명 표기가 다를 수 있어 failover 실행 중에는 stale 삭제를 하지 않는다.
|
||||
# 장기간 장애 후 복구 첫 실행도 오탐 삭제를 막고, 주 파서가 2회 연속 정상일 때만 정리한다.
|
||||
cleanup_allowed = parser_used == 'naver_api' and previous_parser_used == 'naver_api'
|
||||
if cleanup_allowed:
|
||||
for key, existing_list in existing.items():
|
||||
if key in event_keys:
|
||||
continue
|
||||
kind, name = key.split('|', 1)
|
||||
for stale in existing_list:
|
||||
old_start, old_end = event_date_range(stale)
|
||||
# 소급 조회는 누락 상장 보강용이다. 이미 지난 일정은 삭제하지 않는다.
|
||||
if old_end and date.fromisoformat(old_end) < cutoff:
|
||||
continue
|
||||
delete_event(stale['id'], dry_run=args.dry_run)
|
||||
deleted.append(key)
|
||||
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))
|
||||
print(json.dumps({'cutoff_after': cutoff.isoformat(), 'parser_used': parser_used, 'parser_errors': parser_errors, '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))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
raise SystemExit(main())
|
||||
|
||||
@@ -0,0 +1,160 @@
|
||||
import sys
|
||||
import unittest
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
SCRIPTS = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
|
||||
import ipo_calendar_sync as sync
|
||||
|
||||
|
||||
class IpoParserTests(unittest.TestCase):
|
||||
def test_naver_api_parser_includes_cutoff_date(self):
|
||||
payload = {
|
||||
'examinationList': [],
|
||||
'demandForecastingList': [],
|
||||
'forecastingCompleteList': [],
|
||||
'subscriptionList': [{
|
||||
'ipoCode': 'A0035S0',
|
||||
'compName': '빅웨이브로보틱스',
|
||||
'poStartDate': '2026-09-15',
|
||||
'poEndDate': '2026-09-16',
|
||||
'lcalDate': '2026-09-29',
|
||||
'orgNm': '한국투자증권',
|
||||
}],
|
||||
'subscriptionCompleteList': [],
|
||||
'listingList': [],
|
||||
}
|
||||
with patch.object(sync, 'fetch', return_value=sync.json.dumps(payload)):
|
||||
events = sync.parse_naver_api_events_from(date(2026, 9, 29))
|
||||
|
||||
self.assertEqual([(e.kind, e.name) for e in events], [
|
||||
('listing', '빅웨이브로보틱스'),
|
||||
])
|
||||
self.assertEqual(events[0].start_date, date(2026, 9, 29))
|
||||
|
||||
def test_38_parsers_read_subscription_and_listing_tables(self):
|
||||
subscription_html = '''
|
||||
<table summary="공모주 청약일정"><tbody>
|
||||
<tr><th>종목명</th><th>공모주일정</th><th>확정공모가</th><th>희망공모가</th><th>경쟁률</th><th>주간사</th></tr>
|
||||
<tr><td>넥스아이</td><td>2026.11.03~11.04</td><td>-</td><td>-</td><td></td><td>한국투자증권</td></tr>
|
||||
</tbody></table>'''
|
||||
listing_html = '''
|
||||
<table summary="신규상장종목"><tbody>
|
||||
<tr><th>기업명</th><th>신규상장일</th></tr>
|
||||
<tr><td>빅웨이브로보틱스</td><td>2026/09/29</td></tr>
|
||||
</tbody></table>'''
|
||||
|
||||
subscriptions = sync.parse_38_subscription_html(
|
||||
subscription_html, date(2026, 9, 29)
|
||||
)
|
||||
listings = sync.parse_38_listing_html(listing_html, date(2026, 9, 29))
|
||||
|
||||
self.assertEqual(subscriptions[0].start_date, date(2026, 11, 3))
|
||||
self.assertEqual(subscriptions[0].brokers, '한국투자증권')
|
||||
self.assertEqual(listings[0].name, '빅웨이브로보틱스')
|
||||
self.assertEqual(listings[0].start_date, date(2026, 9, 29))
|
||||
|
||||
def test_zero_primary_result_switches_to_38_fallback(self):
|
||||
fallback = sync.EventSpec(
|
||||
'listing', '빅웨이브로보틱스', date(2026, 9, 29),
|
||||
date(2026, 9, 29), '', sync.LISTING_URL,
|
||||
source_label='38커뮤니케이션 IPO',
|
||||
)
|
||||
with patch.object(sync, 'parse_naver_api_events_from', return_value=[]), patch.object(
|
||||
sync, 'parse_38_events_from', return_value=[fallback]
|
||||
):
|
||||
events, parser_used, errors = sync.collect_ipo_events(date(2026, 9, 29))
|
||||
|
||||
self.assertEqual(parser_used, '38_fallback')
|
||||
self.assertEqual(events, [fallback])
|
||||
self.assertEqual(errors, ['naver_api: 수집 결과 0건'])
|
||||
|
||||
def test_all_parser_failures_raise_structured_error(self):
|
||||
with patch.object(
|
||||
sync, 'parse_naver_api_events_from', side_effect=ValueError('schema changed')
|
||||
), patch.object(
|
||||
sync, 'parse_38_events_from', side_effect=ValueError('table changed')
|
||||
):
|
||||
with self.assertRaises(sync.IpoParserError) as caught:
|
||||
sync.collect_ipo_events(date(2026, 9, 29))
|
||||
|
||||
self.assertEqual(caught.exception.errors, [
|
||||
'naver_api: schema changed',
|
||||
'38_fallback: table changed',
|
||||
])
|
||||
|
||||
def test_parser_alert_is_deduplicated_and_recovery_is_reported(self):
|
||||
state = {}
|
||||
errors = ['naver_api: 수집 결과 0건']
|
||||
with patch.object(sync, 'send_telegram', return_value=True) as send:
|
||||
sync.update_parser_health(state, 'degraded', errors, 12)
|
||||
sync.update_parser_health(state, 'degraded', errors, 12)
|
||||
sync.update_parser_health(state, 'healthy', [], 18)
|
||||
|
||||
self.assertEqual(send.call_count, 2)
|
||||
self.assertIn('파서 일부 실패', send.call_args_list[0].args[0])
|
||||
self.assertIn('파서 복구', send.call_args_list[1].args[0])
|
||||
self.assertEqual(state['parser_alerted_status'], 'healthy')
|
||||
|
||||
def test_failed_alert_is_retried_on_next_run(self):
|
||||
state = {}
|
||||
errors = ['naver_api: timeout', '38_fallback: timeout']
|
||||
with patch.object(sync, 'send_telegram', side_effect=[False, True]) as send:
|
||||
sync.update_parser_health(state, 'failed', errors, 0)
|
||||
sync.update_parser_health(state, 'failed', errors, 0)
|
||||
|
||||
self.assertEqual(send.call_count, 2)
|
||||
self.assertEqual(state['parser_alerted_status'], 'failed')
|
||||
|
||||
def test_primary_parser_gets_same_day_listing_augmentation(self):
|
||||
future = sync.EventSpec(
|
||||
'subscription', '넥스아이', date(2026, 11, 3), date(2026, 11, 4),
|
||||
'한국투자증권', 'https://stock.naver.com/ipo/A493410',
|
||||
)
|
||||
listing_html = '''
|
||||
<table summary="신규상장종목"><tbody>
|
||||
<tr><th>기업명</th><th>신규상장일</th></tr>
|
||||
<tr><td>빅웨이브로보틱스</td><td>2026/09/29</td></tr>
|
||||
</tbody></table>'''
|
||||
with patch.object(sync, 'parse_naver_api_events_from', return_value=[future]), patch.object(
|
||||
sync, 'fetch', return_value=listing_html
|
||||
):
|
||||
events, parser_used, errors = sync.collect_ipo_events(date(2026, 9, 29))
|
||||
|
||||
self.assertEqual(parser_used, 'naver_api')
|
||||
self.assertEqual(errors, [])
|
||||
self.assertEqual({e.state_key for e in events}, {
|
||||
'subscription|넥스아이', 'listing|빅웨이브로보틱스',
|
||||
})
|
||||
|
||||
def test_primary_parser_backfills_recent_non_spac_listing(self):
|
||||
future = sync.EventSpec(
|
||||
'subscription', '넥스아이', date(2026, 11, 3), date(2026, 11, 4),
|
||||
'한국투자증권', 'https://stock.naver.com/ipo/A493410',
|
||||
)
|
||||
listing_html = '''
|
||||
<table summary="신규상장종목"><tbody>
|
||||
<tr><th>기업명</th><th>신규상장일</th></tr>
|
||||
<tr><td>와이즈플래닛컴퍼니</td><td>2026/09/23</td></tr>
|
||||
<tr><td>KB스팩34호</td><td>2026/09/22</td></tr>
|
||||
<tr><td>덕산넵코어스(구.넵코어스)</td><td>2026/09/30</td></tr>
|
||||
</tbody></table>'''
|
||||
with patch.object(sync, 'parse_naver_api_events_from', return_value=[future]), patch.object(
|
||||
sync, 'fetch', return_value=listing_html
|
||||
):
|
||||
events, parser_used, errors = sync.collect_ipo_events(date(2026, 9, 29))
|
||||
|
||||
self.assertEqual(parser_used, 'naver_api')
|
||||
self.assertEqual(errors, [])
|
||||
self.assertIn('listing|와이즈플래닛컴퍼니', {e.state_key for e in events})
|
||||
self.assertNotIn('listing|KB스팩34호', {e.state_key for e in events})
|
||||
self.assertNotIn(
|
||||
'listing|덕산넵코어스(구.넵코어스)', {e.state_key for e in events}
|
||||
)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user