chore: 공연·전시 API 진단 스크립트 추가

KOPIS(https·조회 기간·건수·시설 좌표·afterdate)와 한눈에보는문화정보(건수·분야·좌표 채움)를 구현 전에 실제로 불러 확인한다. 키는 backend env 에서 읽고 출력하지 않는다. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 22:07:44 +09:00
parent ec4a2a3db8
commit e7c2506ce7
1 changed files with 248 additions and 0 deletions
@@ -0,0 +1,248 @@
+"""공연·전시 API 진단 — KOPIS(공연)·한국문화정보원 한눈에보는문화정보(전시)를 실제로 불러 구현 전에 확인한다.
+
+확인하는 것
+  KOPIS: https 동작 · 조회 기간 제한(90일/31일) · 대구(27)·경북(47) 건수 · 목록·상세·공연시설 응답 칸 ·
+         좌표(la/lo) 채움 · afterdate(최근 수정분) 동작
+  문화정보: 대구·경북 건수 · 분야(realmName) 분포 · 좌표(gpsX=경도, gpsY=위도) 채움 · 응답 칸 이름
+
+키는 backend/.env.development 의 KOPIS_SERVICE_KEY · PUBLIC_DATA_SERVICE_KEY 에서 읽고 출력하지 않는다.
+KOPIS 는 초당 10회에서 차단되므로 0.6초 간격으로 부른다(전체 30회 안팎).
+
+실행(저장소 루트에서): python -I -X utf8 tools/data-check/culture_api_check.py backend/.env.development
+"""
+import collections
+import importlib.util
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+import xml.etree.ElementTree as ET
+from datetime import timedelta
+from pathlib import Path
+
+
+def _load(name):
+    spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(f'{name}.py'))
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+c1 = _load('public_data_check')  # kst_now 만 쓴다
+KOPIS_BASES = ['https://kopis.or.kr/openApi/restful', 'http://www.kopis.or.kr/openApi/restful']
+CULTURE = 'https://apis.data.go.kr/B553457/cultureinfo'
+REGIONS = {'27': '대구', '47': '경북'}
+calls = collections.Counter()
+errors = []
+secrets = []
+
+
+def read_key(env_path, name):
+    for line in Path(env_path).read_text(encoding='utf-8').splitlines():
+        if line.strip().startswith(name + '='):
+            value = line.split('=', 1)[1].strip().strip('"').strip("'")
+            if value and 'change-me' not in value:
+                secrets.extend([value, urllib.parse.quote(value, safe='')])
+                return value
+    return None
+
+
+def mask(text):
+    for s in secrets:
+        if s:
+            text = text.replace(s, '***')
+    return text
+
+
+def get_xml(label, url, params, interval):
+    calls[label] += 1
+    query = urllib.parse.urlencode(params)
+    try:
+        with urllib.request.urlopen(f'{url}?{query}', timeout=25) as r:
+            body = r.read().decode('utf-8', errors='replace')
+    except Exception as e:  # 메시지에 URL(키 포함)이 섞일 수 있어 가린다
+        errors.append(mask(f'{label}: {type(e).__name__} {str(e)[:150]}'))
+        return None, ''
+    finally:
+        time.sleep(interval)
+    try:
+        return ET.fromstring(body), body[:300]
+    except ET.ParseError:
+        errors.append(mask(f'{label}: XML 아님 — {body[:200]}'))
+        return None, body[:300]
+
+
+def leaf(el):
+    """자식 요소 중 글자만 있는 것을 {태그: 값} 으로"""
+    return {ch.tag: (ch.text or '').strip() for ch in el if len(ch) == 0}
+
+
+def short(v, n=60):
+    v = (v or '').replace('\n', ' ').strip()
+    return v if len(v) <= n else v[:n] + '…'
+
+
+# ---------------- KOPIS ----------------
+
+def kopis_list(base, key, label, extra):
+    """공연목록 전체(최대 20쪽). 오류 응답(returncode/errmsg)이면 (None, 메시지)"""
+    items, page = [], 1
+    while page <= 20:
+        root, raw = get_xml(label, f'{base}/pblprfr', {'service': key, 'cpage': page, 'rows': 100, **extra}, 0.6)
+        if root is None:
+            return None, raw
+        dbs = root.findall('db')
+        if dbs and any(ch.tag in ('returncode', 'errmsg') for ch in dbs[0]):
+            return None, short(' / '.join(f'{k}={v}' for k, v in leaf(dbs[0]).items()), 200)
+        items += [leaf(d) for d in dbs]
+        if len(dbs) < 100:
+            break
+        page += 1
+    return items, ''
+
+
+def check_kopis(key, today):
+    print('## KOPIS (공연)')
+    start = today.strftime('%Y%m%d')
+    in90 = (today + timedelta(days=90)).strftime('%Y%m%d')
+    in31 = (today + timedelta(days=31)).strftime('%Y%m%d')
+
+    base = None
+    for b in KOPIS_BASES:
+        items, msg = kopis_list(b, key, 'KOPIS 주소 확인', {'stdate': start, 'eddate': in31, 'signgucode': '27', 'rows': 1, 'cpage': 1})
+        if items is not None:
+            base = b
+            break
+        print(f'- {b} 실패: {msg or "응답 없음"}')
+    if not base:
+        print('- KOPIS 를 부르지 못했다(키·주소 확인 필요)\n')
+        return
+    print(f'- 동작하는 주소: {base}')
+
+    d31, msg31 = kopis_list(base, key, 'KOPIS 대구 31일', {'stdate': start, 'eddate': in31, 'signgucode': '27'})
+    d90, msg90 = kopis_list(base, key, 'KOPIS 대구 90일', {'stdate': start, 'eddate': in90, 'signgucode': '27'})
+    print(f'- 대구 31일: {len(d31) if d31 is not None else "오류 " + msg31}건')
+    print(f'- 대구 90일 한 번에: {len(d90) if d90 is not None else "오류 " + msg90}' + ('건' if d90 is not None else ''))
+    daegu = d90 if d90 is not None else d31 or []
+    gb, msggb = kopis_list(base, key, 'KOPIS 경북', {'stdate': start, 'eddate': in90 if d90 is not None else in31, 'signgucode': '47'})
+    print(f'- 경북({"90" if d90 is not None else "31"}일): {len(gb) if gb is not None else "오류 " + msggb}건')
+
+    if daegu:
+        print(f'- 목록 칸: {sorted(daegu[0].keys())}')
+        for field in ('genrenm', 'prfstate', 'openrun', 'area'):
+            if field in daegu[0]:
+                print(f'  · {field}: {collections.Counter(i.get(field, "") for i in daegu).most_common(8)}')
+        long_run = [i for i in daegu if i.get('prfpdfrom') and i.get('prfpdto')
+                    and (int(i['prfpdto'].replace('.', '')) - int(i['prfpdfrom'].replace('.', ''))) > 200]
+        print(f'  · 기간이 대략 2개월 넘는 공연: {len(long_run)}건 / 포스터 있는 공연: {sum(1 for i in daegu if i.get("poster"))}건')
+        print('  · 예:', ' | '.join(f"{short(i.get('prfnm'), 20)}({i.get('prfpdfrom')}~{i.get('prfpdto')}, {short(i.get('fcltynm'), 14)})" for i in daegu[:5]))
+
+        venues, detail_fields = collections.OrderedDict(), collections.Counter()
+        for it in daegu[:5]:
+            root, _ = get_xml('KOPIS 공연상세', f"{base}/pblprfr/{it.get('mt20id')}", {'service': key}, 0.6)
+            db = root.find('db') if root is not None else None
+            if db is None:
+                continue
+            d = leaf(db)
+            detail_fields.update(k for k, v in d.items() if v)
+            styurls = len(db.findall('.//styurl'))
+            relates = [leaf(r) for r in db.findall('.//relate')]
+            if d.get('mt10id'):
+                venues[d['mt10id']] = d.get('fcltynm')
+            print(f"  · 상세 {short(d.get('prfnm'), 18)}: 가격={short(d.get('pcseguidance'), 30)} · 시간={short(d.get('dtguidance'), 40)} · "
+                  f"시설ID={d.get('mt10id') or '-'} · 수정일={d.get('updatedate') or '-'} · 소개이미지 {styurls}장 · 예매처 {len(relates)}곳")
+        print(f'  · 상세에 값이 있는 칸(5건 중 몇 건): {dict(detail_fields.most_common())}')
+
+        filled = 0
+        for mt10id, name in list(venues.items())[:5]:
+            root, _ = get_xml('KOPIS 공연시설상세', f'{base}/prfplc/{mt10id}', {'service': key}, 0.6)
+            db = root.find('db') if root is not None else None
+            if db is None:
+                continue
+            v = leaf(db)
+            has = bool(v.get('la') and v.get('lo'))
+            filled += has
+            print(f"  · 시설 {short(name, 18)}: 위도={v.get('la') or '-'} 경도={v.get('lo') or '-'} 주소={short(v.get('adres'), 28)} · 칸 {sorted(v.keys())[:12]}…")
+        print(f'  · 시설 좌표 채움: {filled}/{min(len(venues), 5)}')
+
+    yesterday = (today - timedelta(days=1)).strftime('%Y%m%d')
+    aft, msgaft = kopis_list(base, key, 'KOPIS afterdate', {'stdate': start, 'eddate': in31, 'signgucode': '27', 'afterdate': yesterday})
+    print(f'- afterdate={yesterday} (어제 이후 등록·수정, 31일 안): {len(aft) if aft is not None else "오류 " + msgaft}' + ('건' if aft is not None else ''))
+    print()
+
+
+# ---------------- 한눈에보는문화정보 ----------------
+
+def culture_list(key, label, op, extra):
+    items, total, page = [], None, 1
+    while page <= 10:
+        root, raw = get_xml(label, f'{CULTURE}/{op}', {'serviceKey': key, 'PageNo': page, 'numOfrows': 100, **extra}, 0.1)
+        if root is None:
+            return None, None, raw
+        code = root.findtext('.//resultCode')
+        if code and code not in ('00', '0000'):
+            return None, None, f"{code} {root.findtext('.//resultMsg')}"
+        if total is None:
+            t = root.findtext('.//totalCount')
+            total = int(t) if t and t.isdigit() else None
+        batch = [leaf(i) for i in root.iter('item')]
+        items += batch
+        if len(batch) < 100 or (total is not None and len(items) >= total):
+            break
+        page += 1
+    return items, total, ''
+
+
+def check_culture(key, today):
+    print('## 한눈에보는문화정보 (전시·공연)')
+    start = today.strftime('%Y%m%d')
+    end = (today + timedelta(days=90)).strftime('%Y%m%d')
+    for sido in ('대구', '경북'):
+        items, total, msg = culture_list(key, f'문화정보 {sido}', 'area2', {'sido': sido, 'from': start, 'to': end})
+        if items is None:
+            print(f'- {sido}: 오류 {msg}')
+            continue
+        print(f'- {sido} 90일: {len(items)}건 (totalCount {total})')
+        if not items:
+            continue
+        print(f'  · 목록 칸: {sorted(items[0].keys())}')
+        realm = collections.Counter(i.get('realmName', '') for i in items)
+        print(f'  · 분야: {realm.most_common(10)}')
+        gps = sum(1 for i in items if i.get('gpsX') and i.get('gpsY'))
+        thumb = sum(1 for i in items if i.get('thumbnail'))
+        print(f'  · 좌표 채움 {gps}/{len(items)} · 썸네일 {thumb}/{len(items)}')
+        print('  · 예:', ' | '.join(f"{short(i.get('title'), 18)}({i.get('startDate')}~{i.get('endDate')}, {i.get('realmName')}, {short(i.get('place'), 12)})" for i in items[:5]))
+        if sido == '대구':
+            seq = items[0].get('seq')
+            root, _ = get_xml('문화정보 상세', f'{CULTURE}/detail2', {'serviceKey': key, 'seq': seq}, 0.1)
+            item = root.find('.//item') if root is not None else None
+            if item is not None:
+                d = leaf(item)
+                print(f"  · 상세 칸: {sorted(d.keys())}")
+                print(f"  · 상세 예: 가격={short(d.get('price'), 30)} · 주소={short(d.get('placeAddr'), 30)} · 이미지={'있음' if d.get('imgUrl') else '없음'} · 전화={'있음' if d.get('phone') else '없음'}")
+    print()
+
+
+def main(env_path):
+    today = c1.kst_now()
+    kopis_key = read_key(env_path, 'KOPIS_SERVICE_KEY')
+    public_key = read_key(env_path, 'PUBLIC_DATA_SERVICE_KEY')
+    print(f'# 공연·전시 API 진단 — {today.strftime("%Y-%m-%d")} 기준\n')
+    if kopis_key:
+        check_kopis(kopis_key, today)
+    else:
+        print('## KOPIS — KOPIS_SERVICE_KEY 가 backend/.env.development 에 없어 건너뜀\n')
+    if public_key:
+        check_culture(public_key, today)
+    else:
+        print('## 문화정보 — PUBLIC_DATA_SERVICE_KEY 가 없어 건너뜀\n')
+    print(f'호출 수: {dict(calls)}')
+    if errors:
+        print(f'오류 {len(errors)}건(앞 8건):')
+        for e in errors[:8]:
+            print(' -', e)
+
+
+if __name__ == '__main__':
+    main(sys.argv[1] if len(sys.argv) > 1 else 'backend/.env.development')