diff --git a/tools/data-check/culture_api_check.py b/tools/data-check/culture_api_check.py new file mode 100644 index 0000000..caaeb22 --- /dev/null +++ b/tools/data-check/culture_api_check.py @@ -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')