From 82087b7ac7998348dd4f7e0af4895dac6b191ef4 Mon Sep 17 00:00:00 2001 From: ttipo Date: Mon, 2 Nov 2026 14:53:09 +0900 Subject: [PATCH] =?UTF-8?q?chore:=20=EA=B3=B5=EA=B3=B5=EB=8D=B0=EC=9D=B4?= =?UTF-8?q?=ED=84=B0=20=EC=A0=90=EA=B2=80=C2=B7=ED=81=90=EB=A0=88=EC=9D=B4?= =?UTF-8?q?=EC=85=98=20=ED=9B=84=EB=B3=B4=20=EB=8F=84=EA=B5=AC=20=EC=B6=94?= =?UTF-8?q?=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 대구 1차 출시 지역의 공공데이터를 점검하고 큐레이션 작업표를 만드는 스크립트다. 키는 backend/.env.development 에서 읽고 출력·오류 문구에서 가린다. - public_data_check(2).py · run_region.py: 관광공사·반려동물·축제·상가정보· 단기예보 점검 (결과는 docs/research/data-check/) - curation_candidates.py: 동네별 큐레이션 후보표(CSV). 관광공사는 위치 검색이 반경 안 장소의 15–38%만 돌려줘서 시·도 지역 목록을 받아 반경을 직접 거른다 - curation_shortlist.py: 후보표에 우선순위(1~4·제외)를 매기고 시범표를 만든다 (키 불필요) 결과 CSV 는 data/ 아래에 생기며 .gitignore 대상이다. 파이썬 캐시도 무시한다. Co-Authored-By: Claude Opus 5.5 --- .gitignore | 4 + tools/data-check/curation_candidates.py | 275 +++++++++++++++++++ tools/data-check/curation_shortlist.py | 233 ++++++++++++++++ tools/data-check/public_data_check.py | 343 ++++++++++++++++++++++++ tools/data-check/public_data_check2.py | 296 ++++++++++++++++++++ tools/data-check/run_region.py | 23 ++ 6 files changed, 1174 insertions(+) create mode 100644 tools/data-check/curation_candidates.py create mode 100644 tools/data-check/curation_shortlist.py create mode 100644 tools/data-check/public_data_check.py create mode 100644 tools/data-check/public_data_check2.py create mode 100644 tools/data-check/run_region.py diff --git a/.gitignore b/.gitignore index df744a5..4c630d2 100644 --- a/.gitignore +++ b/.gitignore @@ -57,3 +57,7 @@ Thumbs.db .cache/ .eslintcache .npm/ + +# Python 캐시 (tools/ 점검 스크립트) +__pycache__/ +*.pyc diff --git a/tools/data-check/curation_candidates.py b/tools/data-check/curation_candidates.py new file mode 100644 index 0000000..60e67b8 --- /dev/null +++ b/tools/data-check/curation_candidates.py @@ -0,0 +1,275 @@ +"""큐레이션 후보표(CSV) 만들기 — 동네별 데이트 후보를 한 표로 뽑는다. + +- 상가정보: 데이트 업종(카페·빵/도넛·아이스크림/빙수·샌드위치/샐러드·요리 주점·생맥주) 업소 +- 관광공사: 동네 반경 안 장소 + 유형별 영업시간·쉬는날 원문 + 반려동물 동반 정보 + 시·도 단위 지역 목록(areaBasedList2)을 전부 받아 반경을 직접 거른다. 위치 검색(locationBasedList2)은 + 반경 안의 장소를 빠뜨린다(2026-11-02 대구: 김광석다시그리기길·수성못 유원지 등 7곳). 위치 검색은 비교용으로만 부른다. +- 대표 명소는 키워드 검색으로 한 번 더 확인한다(빠진 것이 있으면 더한다) +- 큐레이터가 채울 칸(데이트 추천·확인한 영업시간·반려동물·공식 채널·확인 방법·확인일 등)은 비워 둔다 +- 같은 이름의 파일이 이미 있으면 덮어쓰지 않고 -2, -3 을 붙여 새로 만든다(큐레이션을 적던 표를 지키기 위해) + +공공데이터만 담는다. 카카오·네이버 등 다른 서비스의 정보는 이 표에 넣지 않는다(큐레이터가 직접 확인한 내용만 적는다). +엑셀에서 한글이 깨지지 않도록 UTF-8(BOM)으로 저장한다. 키는 출력하지 않는다. + +실행(저장소 루트에서): python -I -X utf8 tools/data-check/curation_candidates.py daegu backend/.env.development +결과: data/curation/<지역>/candidates-<날짜>.csv +""" +import csv +import importlib.util +import sys +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') +c2 = _load('public_data_check2') +STORE_RADIUS_URL = f'{c2.STORE_BASE}/storeListInRadius' + +# 동네 중심을 관광공사 장소 좌표로 맞출 때 쓰는 검색어(점검에서 앞산 중심이 카페거리를 벗어났다) +ANCHORS = {'daegu': {'앞산': '앞산카페거리'}} + +# 관광공사 지역 목록에 쓸 법정동 시·도 코드(lDongRegnCd). 없는 지역은 위치 검색으로 대신한다 +TOUR_REGION_CODES = {'daegu': '27'} +INTRO_BUDGET = 600 # 한 번 실행에 쓸 소개정보 호출 상한(개발 계정 오퍼레이션별 하루 1,000건) + +# 대표 명소 — 키워드로 찾아 동네 안인데 후보에 없으면 더한다 +LANDMARKS = {'daegu': { + '대구 도심': ['김광석', '청라언덕', '계산성당', '방천시장'], + '수성못': ['수성못', '수성유원지'], + '앞산': ['앞산공원', '앞산전망대', '앞산케이블카'], +}} +LANDMARK_TYPES = {'12', '14', '28', '38', '39'} # 장소 유형만(축제·숙박·여행코스 제외) +LANDMARK_RADIUS_FACTOR = 1.5 # 동네 반경의 몇 배까지 동네 안으로 볼지 + +# 관광공사 유형별 영업시간·쉬는날 필드(매뉴얼 v4.4). 15 축제는 공연시간. 32 숙박은 코스 장소가 아니라 부르지 않는다 +INTRO_FIELDS = { + '12': ('usetime', 'restdate'), '14': ('usetimeculture', 'restdateculture'), '15': ('playtime', None), + '28': ('usetimeleports', 'restdateleports'), '38': ('opentime', 'restdateshopping'), + '39': ('opentimefood', 'restdatefood'), +} + +PUBLIC_COLS = ['동네', '출처', '출처ID', '이름', '지점명', '분류', '도로명주소', '지번주소', '위도', '경도', '행정동', '전화', + '공공_영업시간(원문)', '공공_쉬는날(원문)', '공공_반려동물', '사진저작권'] +CURATOR_COLS = ['데이트추천(O/X)', '추천이유', '영업시간(확인)', '휴무(확인)', '반려동물(가능/일부/불가/미확인)', '반려동물 조건', + '실내/실외', '공식채널URL', '확인방법(전화/공식채널/방문)', '확인일', '확인자', '메모'] + + +def anchor_center(key, keyword, fallback): + """관광공사 키워드 검색으로 동네 중심 좌표를 찾는다. 못 찾으면 기본 좌표.""" + d = c1.call(key, '국문 키워드', f'{c1.TOUR}/searchKeyword2', {**c1.COMMON, 'keyword': keyword, 'numOfRows': 10, 'pageNo': 1}) + for it in c1.items_of(d): + try: + if keyword.replace(' ', '') in str(it.get('title')).replace(' ', ''): + return float(it['mapx']), float(it['mapy']) + except (KeyError, TypeError, ValueError): + continue + return fallback + + +def store_rows(key, area, lon, lat, radius): + rows = [] + for label, code in c2.DATE_CODES.items(): + got, page = [], 1 + while page <= 10: + d = c1.call(key, '상가 반경', STORE_RADIUS_URL, + {'radius': radius, 'cx': lon, 'cy': lat, 'indsSclsCd': code, 'numOfRows': 1000, 'pageNo': page, 'type': 'json'}) + its = c1.items_of(d) + got += its + if not its or len(got) >= (c1.total_of(d) or 0): + break + page += 1 + for s in got: + rows.append({'동네': area, '출처': '상가정보', '출처ID': s.get('bizesId'), '이름': s.get('bizesNm'), '지점명': s.get('brchNm') or '', + '분류': f"{label} ({s.get('indsSclsNm')})", '도로명주소': s.get('rdnmAdr'), '지번주소': s.get('lnoAdr'), + '위도': s.get('lat'), '경도': s.get('lon'), '행정동': s.get('adongNm'), '전화': '', + '공공_영업시간(원문)': '', '공공_쉬는날(원문)': '', '공공_반려동물': '', '사진저작권': ''}) + return rows + + +def tour_row(key, area, it, pets): + """관광공사 목록 항목 하나를 후보표 행으로 만든다(유형별 영업시간·쉬는날 원문과 반려동물 정보를 붙인다).""" + t = str(it.get('contenttypeid')) + hours = rest = '' + if t in INTRO_FIELDS and c1.calls.get('국문 소개정보', 0) < INTRO_BUDGET: + rec = c1.items_of(c1.call(key, '국문 소개정보', f'{c1.TOUR}/detailIntro2', {**c1.COMMON, 'contentId': it['contentid'], 'contentTypeId': t})) + if rec: + hf, rf = INTRO_FIELDS[t] + hours = c2.short(rec[0].get(hf, ''), 300) + rest = c2.short(rec[0].get(rf, ''), 200) if rf else '' + pet = pets.get(str(it.get('contentid'))) + pet_text = ' / '.join(x for x in (c2.short(pet.get('acmpyTypeCd'), 40), c2.short(pet.get('acmpyPsblCpam'), 80), + c2.short(pet.get('acmpyNeedMtr'), 80)) if x) if pet else '' + return {'동네': area, '출처': '관광공사', '출처ID': it.get('contentid'), '이름': it.get('title'), '지점명': '', + '분류': f"{c1.TYPE_NAMES.get(t, t)} ({it.get('lclsSystm3') or it.get('lclsSystm2') or ''})", + '도로명주소': it.get('addr1'), '지번주소': it.get('addr2') or '', '위도': it.get('mapy'), '경도': it.get('mapx'), + '행정동': '', '전화': it.get('tel') or '', '공공_영업시간(원문)': hours, '공공_쉬는날(원문)': rest, + '공공_반려동물': pet_text, '사진저작권': it.get('cpyrhtDivCd') or ''} + + +def region_tour_items(key, region): + """시·도 단위 관광공사 지역 목록 전체와 totalCount 를 돌려준다. 코드가 없는 지역이면 (None, 0).""" + code = TOUR_REGION_CODES.get(region) + if not code: + return None, 0 + out, page, total = [], 1, 0 + while page <= 30: + d = c1.call(key, '국문 지역 목록', f'{c1.TOUR}/areaBasedList2', + {**c1.COMMON, 'lDongRegnCd': code, 'numOfRows': 1000, 'pageNo': page, 'arrange': 'A'}) + its = c1.items_of(d) + out += its + total = c1.total_of(d) or total + if not its or len(out) >= total: + break + page += 1 + return out, total + + +def within(items, lon, lat, radius): + """좌표가 반경 안인 항목만 남긴다.""" + out = [] + for it in items: + try: + if c1.haversine(lon, lat, float(it['mapx']), float(it['mapy'])) <= radius: + out.append(it) + except (KeyError, TypeError, ValueError): + continue + return out + + +def landmark_items(key, region, area, lon, lat, radius, have_ids): + """대표 명소를 키워드로 찾아, 동네 안에 있는데 아직 후보에 없는 항목을 돌려준다. 찾은 과정도 함께 돌려준다.""" + found, report = [], [] + for kw in LANDMARKS.get(region, {}).get(area, []): + d = c1.call(key, '국문 키워드', f'{c1.TOUR}/searchKeyword2', {**c1.COMMON, 'keyword': kw, 'numOfRows': 30, 'pageNo': 1}) + hits = [] + for it in c1.items_of(d): + if kw.replace(' ', '') not in str(it.get('title', '')).replace(' ', ''): + continue + if str(it.get('contenttypeid')) not in LANDMARK_TYPES: + continue + try: + dist = c1.haversine(lon, lat, float(it['mapx']), float(it['mapy'])) + except (KeyError, TypeError, ValueError): + hits.append(f"{it.get('title')}(좌표 없음)") + continue + cid = str(it.get('contentid')) + if dist > radius * LANDMARK_RADIUS_FACTOR: + state = '동네 밖' + elif cid in have_ids: + state = '이미 있음' + else: + # 반경 안인데 위치 검색 결과에 없었다면 위치 검색 누락이다 + state = '추가 · 위치 검색 누락' if dist <= radius else '추가' + have_ids.add(cid) + found.append(it) + hits.append(f"{it.get('title')}({round(dist)}m, {state})") + report.append(f'{kw}: ' + (', '.join(hits[:6]) if hits else '검색 결과 없음')) + return found, report + + +def next_free(path): + """같은 이름의 파일이 있으면 -2, -3 … 을 붙인 새 경로를 돌려준다(덮어쓰지 않는다).""" + if not path.exists(): + return path + n = 2 + while (cand := path.with_name(f'{path.stem}-{n}{path.suffix}')).exists(): + n += 1 + return cand + + +def pet_index(key): + """반려동물 동반 정보 전체를 contentid 로 묶는다(contentId 없이 부르면 전부 온다).""" + out, page = [], 1 + while page <= 12: + d = c1.call(key, '국문 반려동물 일괄', f'{c1.TOUR}/detailPetTour2', {**c1.COMMON, 'numOfRows': 1000, 'pageNo': page}) + its = c1.items_of(d) + out += its + if not its or len(out) >= (c1.total_of(d) or 0): + break + page += 1 + return {str(x.get('contentid')): x for x in out} + + +def main(region, env_path): + c1.use_region(region) + key = c1.load_key(env_path) + print('… 반려동물 동반 정보 받는 중', flush=True) + pets = pet_index(key) + print('… 관광공사 지역 목록 받는 중', flush=True) + region_items, region_total = region_tour_items(key, region) + rows, seen = [], set() + summary, landmark_report, compare = [], [], [] + for name, lon, lat, radius in c1.AREAS: + if name in ANCHORS.get(region, {}): + lon, lat = anchor_center(key, ANCHORS[region][name], (lon, lat)) + print(f'… {name} 후보 모으는 중 (중심 {lon:.4f}, {lat:.4f})', flush=True) + loc_items = c1.list_all(key, '국문 위치 목록(비교용)', c1.TOUR, lon, lat, radius) + if region_items is not None: + items = within(region_items, lon, lat, radius) + ids, loc_ids = {str(i.get('contentid')) for i in items}, {str(i.get('contentid')) for i in loc_items} + compare.append((name, len(loc_ids), len(ids), len(ids - loc_ids), len(loc_ids - ids))) + # 지역 목록에 없고 위치 검색에만 있는 항목도 버리지 않는다 + items += [i for i in loc_items if str(i.get('contentid')) not in ids] + else: + items = loc_items + tours = [tour_row(key, name, it, pets) for it in items] + have ={cid for src, cid in seen if src == '관광공사'} | {str(r['출처ID']) for r in tours} + extra, report = landmark_items(key, region, name, lon, lat, radius, have) + tours += [tour_row(key, name, it, pets) for it in extra] + landmark_report.append((name, report)) + area_rows = tours + store_rows(key, name, lon, lat, radius) + added = 0 + for r in area_rows: + uid = (r['출처'], str(r['출처ID'])) + if uid in seen: # 동네 반경이 겹치면 먼저 나온 동네에만 둔다 + continue + seen.add(uid) + rows.append(r) + added += 1 + kinds = {} + for r in area_rows: + k = r['분류'].split(' (')[0] + kinds[k] = kinds.get(k, 0) + 1 + summary.append((name, added, kinds)) + + now = c1.kst_now() + out_dir = Path('data/curation') / region + out_dir.mkdir(parents=True, exist_ok=True) + out_file = next_free(out_dir / f'candidates-{now.strftime("%Y%m%d")}.csv') + with out_file.open('w', encoding='utf-8-sig', newline='') as f: + w = csv.DictWriter(f, fieldnames=PUBLIC_COLS + CURATOR_COLS) + w.writeheader() + for r in rows: + w.writerow({**{c: '' for c in CURATOR_COLS}, **r}) + + print(f'\n# 큐레이션 후보표 ({region}) — {len(rows)}곳') + for name, added, kinds in summary: + print(f'- {name}: {added}곳 · ' + ', '.join(f'{k} {v}' for k, v in kinds.items())) + if region_items is not None: + print(f'\n# 관광공사 위치 검색 vs 지역 목록 (지역 목록 {len(region_items)}/{region_total}건 받음)') + for name, n_loc, n_area, only_area, only_loc in compare: + print(f'- {name}: 위치 검색 {n_loc} · 지역 목록 반경 안 {n_area} ' + f'(위치 검색이 빠뜨린 곳 {only_area}, 지역 목록에 없던 곳 {only_loc})') + if c1.calls.get('국문 소개정보', 0) >= INTRO_BUDGET: + print(f'\n⚠ 소개정보 호출이 상한({INTRO_BUDGET})에 닿아 일부 장소의 영업시간 원문이 비어 있다') + if any(report for _, report in landmark_report): + print('\n# 명소 키워드 보강') + for name, report in landmark_report: + for line in report: + print(f'- [{name}] {line}') + print(f'\n호출 수: {c1.calls}') + if c1.errors: + print(f'오류 {len(c1.errors)}건 (앞 5건):') + for e in c1.errors[:5]: + print(' -', e) + print(f'\n결과: {out_file} (엑셀로 열 수 있음)') + + +if __name__ == '__main__': + main(sys.argv[1] if len(sys.argv) > 1 else 'daegu', sys.argv[2] if len(sys.argv) > 2 else 'backend/.env.development') diff --git a/tools/data-check/curation_shortlist.py b/tools/data-check/curation_shortlist.py new file mode 100644 index 0000000..58199fc --- /dev/null +++ b/tools/data-check/curation_shortlist.py @@ -0,0 +1,233 @@ +"""큐레이션 1차 자동 정리 — 후보표(CSV)에 우선순위를 매겨 사람이 볼 순서대로 정렬한다. + +지우지 않고 표시만 한다. 각 행 앞에 우선순위·정리 사유·가장 가까운 기준점·기준점 거리를 붙인다. +- 제외: 숙박, 축제·행사(축제·행사 영역에서 다룸), 코스형 항목(좌표가 등록 기관 주소인 경우가 있다), + 관광공사 항목과 같은 곳으로 보이는 상가정보 +- 1순위: 관광공사 장소(영업시간 원문이 있어 확인이 쉽다), 거리·골목 단위 장소(코스 경유지 후보) +- 2순위: 기준점 가까이 있는 개인 매장(카페·디저트·요리 주점) +- 3순위: 나머지 개인 매장, 같은 상호가 여러 곳인 매장(지역 체인일 수 있어 사람이 판단) +- 4순위: 전국 프랜차이즈, 생맥주(호프) + +기준점은 후보표 안의 관광공사 명소 좌표에서 찾는다(좌표를 코드에 적지 않는다). 못 찾으면 동네 중심만 쓴다. +공공데이터 파일만 읽으므로 키가 필요 없다. + +실행(저장소 루트에서): python -I -X utf8 tools/data-check/curation_shortlist.py daegu [후보표.csv] +결과: data/curation/<지역>/shortlist-<날짜>.csv — 전체 정렬표 + data/curation/<지역>/pilot-<동네>-<날짜>.csv — 시범 확인용(한 곳 확인에 걸리는 시간을 재는 용도) +""" +import csv +import importlib.util +import re +import sys +from collections import Counter +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 + + +cc = _load('curation_candidates') # 동네 중심 좌표·거리 계산·파일 이름 규칙만 쓴다(키를 읽지 않는다) +c1 = cc.c1 + +TIERS = ['1', '2', '3', '4', '제외'] +# 기준점에서 이 거리 안이면 2순위. 동네마다 2순위가 100~150곳 안팎이 되도록 맞춘 값(2026-11-02 대구 후보표 기준) +# - 도심: 매장이 촘촘하고 기준점(동성로·근대골목·김광석길·청라언덕 등)이 많아 150m 로 충분하다 +# - 수성못: 동네 중심이 호수 한가운데라 둘레 매장까지 닿으려면 600m 가 필요하다 +# - 앞산: 카페거리가 길게 이어져 600m +NEAR_M = {'daegu': {'대구 도심': 150, '수성못': 600, '앞산': 600}} +NEAR_M_DEFAULT = 400 +CAFE_KINDS = {'카페', '빵/도넛', '아이스크림/빙수', '샌드위치/샐러드', '요리 주점'} +REPEAT_MIN = 3 # 같은 상호가 이만큼 있으면 지역 체인일 수 있다 + +# 기준점: 데이트 동선의 중심이 되는 명소. 후보표의 관광공사 장소 이름에서 찾는다 +HOTSPOTS = {'daegu': { + '대구 도심': ['동성로거리', '근대골목', '김광석', '봉산문화거리', '청라언덕'], + '수성못': ['수성못', '들안길'], + '앞산': ['앞산카페거리'], +}} +HOTSPOT_KINDS = {'관광지', '문화시설', '레포츠', '쇼핑'} + +# 전국 프랜차이즈(이름에 들어 있으면). 지역 체인은 넣지 않는다 — 같은 상호 여러 곳은 3순위로 사람이 본다 +CHAINS = ('스타벅스', '투썸', '이디야', '메가커피', '메가엠지씨', '컴포즈', '빽다방', '파리바게', '뚜레쥬르', '배스킨', + '설빙', '할리스', '폴바셋', '커피빈', '엔제리너스', '탐앤탐스', '파스쿠찌', '더벤티', '매머드', '공차', '던킨', + '크리스피크림', '카페봄봄', '요거프레소', '쥬씨', '감성커피', '텐퍼센트', '블루샥', '하삼동', '달콤커피', + '바나프레소', '서브웨이', '써브웨이', '이삭토스트', '커피에반하다', '디저트39', '와플대학', '투다리', '역전할머니맥주', + '봉구비어', '생활맥주', '한신포차', '청춘포차', '빌리엔젤') + +STREET_RE = re.compile(r'(거리|골목|길|타운|시장)\s*(\(.*\))?$') +COURSE_RE = re.compile(r'코스|올레') +# 관광공사 쇼핑 SH040300 은 대부분 백화점 입점·브랜드 매장이다(2026-11-02 대구 도심 135곳). 백화점·쇼핑몰 자체만 1순위로 둔다 +SHOP_STORE_CODE = 'SH040300' +MALL_RE = re.compile(r'^(현대백화점|더현대|롯데백화점|대구백화점|신세계|대구신세계|동아백화점)') +CONVENIENCE_RE = re.compile(r'이마트24|GS25|세븐일레븐|미니스톱|씨유|CU\b') # 명소 키워드 검색에 섞여 들어온다(예: 이마트24 청라언덕점) + +PILOT = {'daegu': ('앞산', 20)} +LEAD_COLS = ['우선순위', '정리사유', '가까운 기준점', '기준점 거리(m)'] + + +def norm(name): + """상호 비교용 — 공백을 없애고 끝의 '대구'를 뗀다(예: 파스쿠찌대구 → 파스쿠찌).""" + n = re.sub(r'\s+', '', name or '') + return n[:-2] if n.endswith('대구') and len(n) > 4 else n + + +def kind_of(row): + return row['분류'].split(' (')[0] + + +def coords(row): + try: + return float(row['경도']), float(row['위도']) + except (TypeError, ValueError): + return None + + +def find_hotspots(rows, region): + """동네별 기준점 목록(이름, 경도, 위도)과 못 찾은 이름을 돌려준다. 동네 중심은 항상 포함한다.""" + spots, missing = {}, [] + for name, lon, lat, _ in c1.REGIONS[region]: + spots[name] = [('동네 중심', lon, lat)] + for kw in HOTSPOTS.get(region, {}).get(name, []): + # 같은 키워드가 여러 곳에 걸리면 관광지를 먼저 고른다(예: '동성로'가 쇼핑 매장 이름에도 들어 있다) + hits = [r for r in rows if r['출처'] == '관광공사' and kind_of(r) in HOTSPOT_KINDS + and kw in r['이름'].replace(' ', '') and coords(r)] + hit = min(hits, key=lambda r: kind_of(r) != '관광지', default=None) + if hit: + spots[name].append((hit['이름'], *coords(hit))) + else: + missing.append(f'{name}/{kw}') + return spots, missing + + +def nearest(spots, xy): + if not xy: + return '', None + best = min(spots, key=lambda s: c1.haversine(s[1], s[2], *xy)) + return best[0], round(c1.haversine(best[1], best[2], *xy)) + + +def classify(row, ctx): + """(우선순위, 정리 사유) 를 정한다.""" + kind, name = kind_of(row), row['이름'] + dist = row['기준점 거리(m)'] + if row['출처'] == '관광공사': + if kind == '숙박': + return '제외', '숙박 — 코스 장소가 아님' + if kind == '축제·행사': + return '제외', '행사 — 축제·행사 영역에서 다룸' + if kind == '여행코스' or COURSE_RE.search(name): + return '제외', '코스형 항목 — 좌표가 실제 위치가 아닐 수 있음' + if CONVENIENCE_RE.search(name): + return '제외', '편의점' + pet = ' · 반려동물 동반 정보 있음' if row['공공_반려동물'] else '' + if SHOP_STORE_CODE in row['분류'] and not MALL_RE.search(name): + return '4', '쇼핑 매장(입점·브랜드 매장) — 코스 장소로 약함' + pet + if STREET_RE.search(name) or '상이' in row['공공_영업시간(원문)']: + return '1', '거리·골목 단위 — 영업시간 확인 불필요, 경유지 후보' + pet + has_hours = '영업시간 원문 있음' if row['공공_영업시간(원문)'] else '영업시간 원문 없음' + return '1', f'관광공사 {kind} — {has_hours}' + pet + + n = norm(name) + if ctx['dup'](row): + return '제외', '관광공사 항목과 같은 곳으로 보임' + if any(c in n for c in CHAINS): + return '4', '전국 프랜차이즈' + if kind == '생맥주': + return '4', '생맥주(호프) — 데이트 적합도 낮음' + if ctx['repeat'][n] >= REPEAT_MIN: + return '3', f'같은 상호 {ctx["repeat"][n]}곳 — 지역 체인일 수 있음' + near = ctx['near'].get(row['동네'], NEAR_M_DEFAULT) + if kind in CAFE_KINDS and dist is not None and dist <= near: + return '2', f'기준점 {near}m 안 개인 매장' + return '3', '개인 매장' + + +def make_dup_checker(rows): + """상가정보 행이 같은 동네 관광공사 항목과 같은 곳인지(이름이 서로 포함되고 80m 안) 판단하는 함수를 만든다.""" + tours = {} + for r in rows: + if r['출처'] == '관광공사' and coords(r): + tours.setdefault(r['동네'], []).append((norm(r['이름']), coords(r))) + + def dup(row): + xy, n = coords(row), norm(row['이름']) + if not xy or len(n) < 2: + return False + return any((n in tn or tn in n) and c1.haversine(*txy, *xy) <= 80 for tn, txy in tours.get(row['동네'], [])) + return dup + + +def latest_candidates(region): + files = sorted(Path('data/curation', region).glob('candidates-*.csv'), key=lambda p: p.stat().st_mtime) + if not files: + sys.exit(f'후보표가 없습니다: data/curation/{region}/candidates-*.csv — curation_candidates.py 를 먼저 실행하세요') + return files[-1] + + +def write(path, cols, rows): + with path.open('w', encoding='utf-8-sig', newline='') as f: + w = csv.DictWriter(f, fieldnames=cols) + w.writeheader() + w.writerows(rows) + + +def main(region, src=None): + src = Path(src) if src else latest_candidates(region) + with src.open(encoding='utf-8-sig', newline='') as f: + reader = csv.DictReader(f) + cols, rows = reader.fieldnames, list(reader) + + spots, missing = find_hotspots(rows, region) + for r in rows: + r['가까운 기준점'], r['기준점 거리(m)'] = nearest(spots.get(r['동네'], []), coords(r)) + ctx = {'dup': make_dup_checker(rows), + 'repeat': Counter(norm(r['이름']) for r in rows if r['출처'] == '상가정보'), + 'near': NEAR_M.get(region, {})} + for r in rows: + r['우선순위'], r['정리사유'] = classify(r, ctx) + + area_order = {name: i for i, (name, *_) in enumerate(c1.REGIONS[region])} + rows.sort(key=lambda r: (area_order.get(r['동네'], 99), TIERS.index(r['우선순위']), + r['기준점 거리(m)'] if r['기준점 거리(m)'] is not None else 10 ** 9)) + + out_dir = src.parent + stamp = c1.kst_now().strftime('%Y%m%d') + out = cc.next_free(out_dir / f'shortlist-{stamp}.csv') + write(out, LEAD_COLS + cols, rows) + + pilot_out = None + if region in PILOT: + area, n = PILOT[region] + # 1순위(관광공사 장소)와 2순위(개인 매장)를 반씩 — 확인 방법이 달라 둘 다 시간을 재 본다 + pool = [r for r in rows if r['동네'] == area and not r['정리사유'].startswith('거리·골목')] + firsts = [r for r in pool if r['우선순위'] == '1'] + seconds = [r for r in pool if r['우선순위'] == '2'] + pilot = firsts[:n // 2] + seconds[:n - min(len(firsts), n // 2)] + pilot_out = cc.next_free(out_dir / f'pilot-{area}-{stamp}.csv') + write(pilot_out, LEAD_COLS + cols, pilot) + + print(f'# 1차 자동 정리 ({region}) — 원본 {src.name}, {len(rows)}곳') + for area in area_order: + a = [r for r in rows if r['동네'] == area] + cnt = Counter(r['우선순위'] for r in a) + print(f'- {area}: ' + ' · '.join(f'{t}{"순위" if t != "제외" else ""} {cnt.get(t, 0)}' for t in TIERS) + + f' (기준점: {", ".join(s[0] for s in spots[area])})') + if missing: + print(f'\n기준점 못 찾음: {", ".join(missing)} — 후보표에 해당 관광공사 장소가 없다(동네 중심으로 대신함)') + reasons = Counter(r['정리사유'] for r in rows if r['우선순위'] == '제외') + print('\n제외 사유: ' + ', '.join(f'{k} {v}' for k, v in reasons.most_common())) + repeats = sorted({(norm(r['이름']), ctx['repeat'][norm(r['이름'])]) for r in rows + if r['정리사유'].startswith('같은 상호')}, key=lambda x: -x[1]) + if repeats: + print('같은 상호 여러 곳(3순위, 체인 여부 사람이 판단): ' + ', '.join(f'{n} {c}' for n, c in repeats[:15])) + print(f'\n결과: {out}') + if pilot_out: + print(f'시범: {pilot_out}') + + +if __name__ == '__main__': + main(sys.argv[1] if len(sys.argv) > 1 else 'daegu', sys.argv[2] if len(sys.argv) > 2 else None) diff --git a/tools/data-check/public_data_check.py b/tools/data-check/public_data_check.py new file mode 100644 index 0000000..c8a76e9 --- /dev/null +++ b/tools/data-check/public_data_check.py @@ -0,0 +1,343 @@ +"""공공데이터 커버리지 점검 — 지역 묶음(REGIONS)별로 잰다. + +AI 코스 품질은 우리가 저장할 수 있는 공공데이터가 얼마나 채워져 있는지에 달려 있다. +지역마다 장소 수, 영업시간·반려동물·영어 정보가 있는 비율, 이번 주말 축제, 날씨 예보 범위를 잰다. + +- 키는 env 파일에서 읽기만 하고 출력하지 않는다. 오류 메시지에서도 가린다. +- 관광공사 개발 계정(하루 1,000건) 안에서 끝나도록 상세 조회는 표본만 부른다(약 400건). +- 응답 필드 이름은 짐작하지 않고 실제 응답에서 찾는다. + +실행(저장소 루트에서): python -I -X utf8 tools/data-check/public_data_check.py backend/.env.development [daegu|seoul-busan] +결과: 화면에 요약, docs/research/data-check/ 에 상세 JSON +""" +import datetime as dt +import json +import math +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path + +KEY_NAME = 'PUBLIC_DATA_SERVICE_KEY' +TOUR = 'https://apis.data.go.kr/B551011/KorService2' +ENG = 'https://apis.data.go.kr/B551011/EngService2' +KMA = 'https://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getVilageFcst' +STORE = 'https://apis.data.go.kr/B553077/api/open/sdsc2/storeListInRadius' +COMMON = {'MobileOS': 'ETC', 'MobileApp': 'WanderGo', '_type': 'json'} + +# 지역 묶음 — (이름, 경도, 위도, 반경 m). 좌표는 동네 중심의 대략값이다. +REGIONS = { + 'seoul-busan': [('성수', 127.0557, 37.5446, 2000), ('연남', 126.9235, 37.5627, 1500), ('해운대', 129.1604, 35.1587, 2000)], + 'daegu': [('대구 도심', 128.5966, 35.8650, 1500), # 동성로·김광석길·근대골목 + ('수성못', 128.6176, 35.8282, 1500), # 수성못·들안길 + ('앞산', 128.5845, 35.8325, 1500)], # 앞산 카페거리(대명동) +} +AREAS = REGIONS['seoul-busan'] + + +def use_region(name): + """점검할 지역 묶음을 바꾼다. 다른 함수는 실행 시점의 AREAS 를 읽는다.""" + global AREAS + if name not in REGIONS: + sys.exit(f'지역 묶음은 {", ".join(REGIONS)} 중 하나여야 합니다') + AREAS = REGIONS[name] +TYPE_NAMES = {'12': '관광지', '14': '문화시설', '15': '축제·행사', '25': '여행코스', '28': '레포츠', + '32': '숙박', '38': '쇼핑', '39': '음식점'} +HOURS_TYPES = ['12', '14', '28', '38', '39'] # 영업시간을 볼 유형 +HOURS_SAMPLE = 20 # 유형별 표본 수 +PET_SAMPLE = 30 # 지역별 반려동물 상세 표본 수 +STORE_RADIUS = 500 # 상가정보 비교 반경(m) + +calls = {} +errors = [] + + +def kst_now(): + return dt.datetime.now(dt.timezone.utc) + dt.timedelta(hours=9) + + +def load_key(env_path): + for line in Path(env_path).read_text(encoding='utf-8').splitlines(): + if line.strip().startswith(KEY_NAME + '='): + return line.split('=', 1)[1].strip().strip('"').strip("'") + sys.exit(f'{KEY_NAME} 가 {env_path} 에 없습니다') + + +def call(key, label, url, params): + """API 를 부르고 (상태, JSON 또는 None) 를 돌려준다. 키는 어떤 경우에도 밖으로 나가지 않는다.""" + calls[label] = calls.get(label, 0) + 1 + q = urllib.parse.urlencode({'serviceKey': key, **params}) + try: + with urllib.request.urlopen(f'{url}?{q}', timeout=25) as r: + body = r.read().decode('utf-8', errors='replace') + except urllib.error.HTTPError as e: + errors.append(f'{label}: HTTP {e.code} {e.read().decode("utf-8", "replace")[:120]}'.replace(key, '***')) + return None + except Exception as e: # 네트워크 오류 메시지에 URL(키 포함)이 섞일 수 있다 + errors.append(f'{label}: {type(e).__name__} {str(e)[:120]}'.replace(key, '***')) + return None + finally: + time.sleep(0.08) + try: + data = json.loads(body) + except json.JSONDecodeError: + errors.append(f'{label}: JSON 아님 {body[:160]}'.replace(key, '***')) + return None + head = (data.get('response') or {}).get('header') or data.get('header') or {} + code = str(head.get('resultCode', '')) + if code and code not in ('0000', '00'): + errors.append(f'{label}: {code} {head.get("resultMsg")}') + return data + + +def body_of(data): + if not data: + return {} + return (data.get('response') or {}).get('body') or data.get('body') or {} + + +def items_of(data): + b = body_of(data) + it = b.get('items') + if not it: + return [] + if isinstance(it, dict): + it = it.get('item', []) + return it if isinstance(it, list) else [it] + + +def total_of(data): + try: + return int(body_of(data).get('totalCount')) + except (TypeError, ValueError): + return None + + +def haversine(lon1, lat1, lon2, lat2): + r = 6371000 + p1, p2 = math.radians(lat1), math.radians(lat2) + dp, dl = p2 - p1, math.radians(lon2 - lon1) + a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2 + return 2 * r * math.asin(math.sqrt(a)) + + +def kma_grid(lon, lat): + """기상청 단기예보 격자(nx, ny) 변환 — 기상청 공개 LCC 공식.""" + re_, grid, slat1, slat2, olon, olat, xo, yo = 6371.00877, 5.0, 30.0, 60.0, 126.0, 38.0, 43, 136 + d = math.pi / 180 + re_ /= grid + s1, s2, ol, oa = slat1 * d, slat2 * d, olon * d, olat * d + sn = math.log(math.cos(s1) / math.cos(s2)) / math.log(math.tan(math.pi / 4 + s2 / 2) / math.tan(math.pi / 4 + s1 / 2)) + sf = math.tan(math.pi / 4 + s1 / 2) ** sn * math.cos(s1) / sn + ro = re_ * sf / math.tan(math.pi / 4 + oa / 2) ** sn + ra = re_ * sf / math.tan(math.pi / 4 + lat * d / 2) ** sn + th = lon * d - ol + th = (th + math.pi) % (2 * math.pi) - math.pi + th *= sn + return int(math.floor(ra * math.sin(th) + xo + 0.5)), int(math.floor(ro - ra * math.cos(th) + yo + 0.5)) + + +def kma_base(now): + """발표 후 10분이 지난 가장 최근 단기예보 발표 시각.""" + for h in (23, 20, 17, 14, 11, 8, 5, 2): + if now.hour > h or (now.hour == h and now.minute >= 10): + return now.strftime('%Y%m%d'), f'{h:02d}00' + y = now - dt.timedelta(days=1) + return y.strftime('%Y%m%d'), '2300' + + +def filled(v): + return v not in (None, '', '0') and str(v).strip() not in ('', '-', '없음') + + +def pct(a, b): + return '-' if not b else f'{round(100 * a / b)}%' + + +def list_all(key, label, url, lon, lat, radius): + out, page = [], 1 + while page <= 4: + d = call(key, label, f'{url}/locationBasedList2', + {**COMMON, 'mapX': lon, 'mapY': lat, 'radius': radius, 'numOfRows': 500, 'pageNo': page, 'arrange': 'E'}) + its = items_of(d) + out += its + total = total_of(d) or 0 + if not its or len(out) >= total: + break + page += 1 + return out + + +def check_area(key, name, lon, lat, radius, weekend): + r = {'area': name, 'center': [lon, lat], 'radius_m': radius} + kor = list_all(key, '국문 목록', TOUR, lon, lat, radius) + by_type = {} + for it in kor: + by_type.setdefault(str(it.get('contenttypeid')), []).append(it) + r['kor_total'] = len(kor) + r['kor_by_type'] = {TYPE_NAMES.get(t, t): len(v) for t, v in sorted(by_type.items())} + r['kor_list_fields'] = sorted(kor[0].keys()) if kor else [] + + # 영업시간: 유형별 표본의 소개정보에서 시간 관련 필드가 채워져 있는지 + hours = {} + intro_fields = set() + examples = [] + for t in HOURS_TYPES: + sample = by_type.get(t, [])[:HOURS_SAMPLE] + have = rest = 0 + for it in sample: + d = call(key, '국문 소개정보', f'{TOUR}/detailIntro2', {**COMMON, 'contentId': it['contentid'], 'contentTypeId': t}) + its = items_of(d) + if not its: + continue + rec = its[0] + intro_fields |= set(rec.keys()) + hk = [k for k in rec if any(s in k.lower() for s in ('usetime', 'opentime', 'playtime'))] + rk = [k for k in rec if 'restdate' in k.lower()] + if any(filled(rec[k]) for k in hk): + have += 1 + if len(examples) < 4: + examples.append({'title': it.get('title'), 'type': TYPE_NAMES.get(t), 'hours': ' / '.join(str(rec[k])[:60] for k in hk if filled(rec[k]))}) + if any(filled(rec[k]) for k in rk): + rest += 1 + hours[TYPE_NAMES[t]] = {'sampled': len(sample), 'with_hours': have, 'with_restdate': rest} + r['hours'] = hours + r['hours_examples'] = examples + r['intro_fields'] = sorted(intro_fields) + + # 반려동물: 가까운 장소 표본에 반려동물 상세가 있는지, 크기 정보가 있는지 + pet_sample = [it for it in kor if str(it.get('contenttypeid')) in ('12', '14', '28', '32', '38', '39')][:PET_SAMPLE] + pet_have = pet_size = 0 + pet_fields, pet_examples = set(), [] + for it in pet_sample: + d = call(key, '국문 반려동물', f'{TOUR}/detailPetTour2', {**COMMON, 'contentId': it['contentid']}) + its = items_of(d) + if not its: + continue + rec = its[0] + vals = {k: v for k, v in rec.items() if k != 'contentid' and filled(v)} + if not vals: + continue + pet_have += 1 + pet_fields |= set(rec.keys()) + text = ' '.join(str(v) for v in vals.values()) + if any(s in text for s in ('소형', '중형', '대형', 'kg', 'KG')): + pet_size += 1 + if len(pet_examples) < 3: + pet_examples.append({'title': it.get('title'), **{k: str(v)[:80] for k, v in vals.items()}}) + r['pet'] = {'sampled': len(pet_sample), 'with_pet_info': pet_have, 'with_size_info': pet_size} + r['pet_fields'] = sorted(pet_fields) + r['pet_examples'] = pet_examples + + # 영어: 같은 반경의 영문 목록 수와, 국문과 같은 contentid 가 있는지 + eng = list_all(key, '영문 목록', ENG, lon, lat, radius) + kor_ids = {str(it.get('contentid')) for it in kor} + r['eng_total'] = len(eng) + r['eng_by_type'] = {} + for it in eng: + t = TYPE_NAMES.get(str(it.get('contenttypeid')), str(it.get('contenttypeid'))) + r['eng_by_type'][t] = r['eng_by_type'].get(t, 0) + 1 + r['eng_same_contentid'] = sum(1 for it in eng if str(it.get('contentid')) in kor_ids) + r['eng_examples'] = [{'title': it.get('title'), 'addr1': it.get('addr1')} for it in eng[:3]] + + # 날씨: 단기예보가 이번 주말 오후를 덮는지 + nx, ny = kma_grid(lon, lat) + bd, bt = kma_base(kst_now()) + d = call(key, '기상청 단기예보', KMA, {'pageNo': 1, 'numOfRows': 1500, 'dataType': 'JSON', 'base_date': bd, 'base_time': bt, 'nx': nx, 'ny': ny}) + fc = items_of(d) + dates = sorted({x.get('fcstDate') for x in fc if x.get('fcstDate')}) + pop = [x.get('fcstValue') for x in fc if x.get('category') == 'POP' and x.get('fcstDate') == weekend and x.get('fcstTime') == '1500'] + r['weather'] = {'grid': [nx, ny], 'base': f'{bd} {bt}', 'forecast_dates': dates, + 'covers_weekend': weekend in dates, 'weekend_1500_pop': pop[0] if pop else None} + + # 상가정보: 같은 반경에서 관광공사 음식점과 상가정보 음식 업소 수 비교 + stores, page = [], 1 + while page <= 5: + d = call(key, '상가정보', STORE, {'radius': STORE_RADIUS, 'cx': lon, 'cy': lat, 'type': 'json', 'numOfRows': 1000, 'pageNo': page}) + its = items_of(d) + stores += its + if not its or len(stores) >= (total_of(d) or 0): + break + page += 1 + food = [s for s in stores if '음식' in str(s.get('indsLclsNm', ''))] + cafe = [s for s in food if any(w in str(s.get('indsSclsNm', '')) + str(s.get('indsMclsNm', '')) for w in ('카페', '커피', '비알코올'))] + tour_food_near = sum(1 for it in by_type.get('39', []) if float(it.get('dist') or 1e9) <= STORE_RADIUS) + r['store'] = {'radius_m': STORE_RADIUS, 'total': len(stores), 'food': len(food), 'cafe': len(cafe), + 'tour_food_same_radius': tour_food_near, 'fields': sorted(stores[0].keys()) if stores else []} + return r + + +def festivals(key, weekend): + out, page = [], 1 + start = (dt.date.fromisoformat(f'{weekend[:4]}-{weekend[4:6]}-{weekend[6:]}') - dt.timedelta(days=60)).strftime('%Y%m%d') + while page <= 4: + d = call(key, '국문 축제', f'{TOUR}/searchFestival2', {**COMMON, 'eventStartDate': start, 'numOfRows': 500, 'pageNo': page, 'arrange': 'A'}) + its = items_of(d) + out += its + if not its or len(out) >= (total_of(d) or 0): + break + page += 1 + active = [f for f in out if str(f.get('eventstartdate', '9')) <= weekend <= str(f.get('eventenddate', '0'))] + near = {} + for name, lon, lat, _ in AREAS: + lst = [] + for f in active: + try: + dist = haversine(lon, lat, float(f['mapx']), float(f['mapy'])) + except (KeyError, TypeError, ValueError): + continue + if dist <= 5000: + lst.append({'title': f.get('title'), 'period': f"{f.get('eventstartdate')}–{f.get('eventenddate')}", 'km': round(dist / 1000, 1)}) + near[name] = sorted(lst, key=lambda x: x['km']) + return {'fetched': len(out), 'active_on_weekend_nationwide': len(active), 'near_5km': near, + 'list_fields': sorted(out[0].keys()) if out else []} + + +def main(env_path, region='seoul-busan'): + use_region(region) + key = load_key(env_path) + now = kst_now() + weekend = (now.date() + dt.timedelta(days=(5 - now.weekday()) % 7)).strftime('%Y%m%d') # 다가오는 토요일 + report = {'checked_at_kst': now.strftime('%Y-%m-%d %H:%M'), 'region': region, 'weekend_saturday': weekend, 'areas': []} + for name, lon, lat, radius in AREAS: + print(f'… {name} 점검 중', flush=True) + report['areas'].append(check_area(key, name, lon, lat, radius, weekend)) + print('… 축제 점검 중', flush=True) + report['festivals'] = festivals(key, weekend) + report['calls'] = calls + report['errors'] = errors + + out_dir = Path('docs/research/data-check') + out_dir.mkdir(parents=True, exist_ok=True) + out_file = out_dir / f'public-data-check-{region}-{now.strftime("%Y%m%d-%H%M")}.json' + out_file.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8') + + # ── 화면 요약 ── + print(f'\n# 공공데이터 점검 {report["checked_at_kst"]} KST · 주말 기준 {weekend}') + for a in report['areas']: + print(f'\n## {a["area"]} (반경 {a["radius_m"]}m)') + print(f'- 국문 장소 {a["kor_total"]}곳: ' + ', '.join(f'{k} {v}' for k, v in a['kor_by_type'].items())) + print('- 영업시간(표본): ' + ', '.join(f'{k} {pct(v["with_hours"], v["sampled"])}({v["with_hours"]}/{v["sampled"]})' for k, v in a['hours'].items())) + p = a['pet'] + print(f'- 반려동물 상세(가까운 {p["sampled"]}곳): 있음 {p["with_pet_info"]}곳, 크기 정보 {p["with_size_info"]}곳') + print(f'- 영문 장소 {a["eng_total"]}곳 (국문 대비 {pct(a["eng_total"], a["kor_total"])}), 국문과 같은 ID {a["eng_same_contentid"]}곳') + w = a['weather'] + print(f'- 단기예보 격자 {w["grid"]}, 예보 날짜 {w["forecast_dates"][:1]}~{w["forecast_dates"][-1:]}, 주말 포함 {w["covers_weekend"]}, 주말 15시 강수확률 {w["weekend_1500_pop"]}') + s = a['store'] + print(f'- 반경 {s["radius_m"]}m 상가정보 음식 {s["food"]}곳(카페 {s["cafe"]}) vs 관광공사 음식점 {s["tour_food_same_radius"]}곳') + f = report['festivals'] + print(f'\n## 축제: 받아온 {f["fetched"]}건, 주말에 열리는 전국 {f["active_on_weekend_nationwide"]}건') + for k, v in f['near_5km'].items(): + print(f'- {k} 5km 안: ' + (', '.join(x['title'] for x in v[:4]) or '없음')) + print(f'\n호출 수: {calls}') + if errors: + print(f'오류 {len(errors)}건 (앞 5건):') + for e in errors[:5]: + print(' -', e) + print(f'\n상세 결과: {out_file}') + + +if __name__ == '__main__': + main(sys.argv[1] if len(sys.argv) > 1 else 'backend/.env.development', sys.argv[2] if len(sys.argv) > 2 else 'seoul-busan') diff --git a/tools/data-check/public_data_check2.py b/tools/data-check/public_data_check2.py new file mode 100644 index 0000000..3a672f1 --- /dev/null +++ b/tools/data-check/public_data_check2.py @@ -0,0 +1,296 @@ +"""공공데이터 2차 점검 — 매뉴얼 검토(docs/research/data-openapi-review/)로 바로잡은 방식으로 다시 잰다. + +1. 반려동물: 동반 가능 장소 목록(KorPetTourService2)과 동반 정보 일괄 조회(KorService2 detailPetTour2, contentId 생략) +2. 축제: eventStartDate=eventEndDate=토요일 동작 확인, 축제(EV01)만 거르기 +3. 영업시간: 유형별 정확한 필드로 음식점 영업시간 형식, 관광지·문화시설 동반 여부(chkpet*) 값 +4. 상가정보: 지역 묶음의 행정동 코드(baroApi)와 동별 업소 수, 동네 반경 안 데이트 업종 수 +5. 날씨: 예보 날짜별 시각 수(마지막 날이 하루 전체인지) + +키는 출력하지 않는다. 관광공사 개발 계정 한도는 오퍼레이션별 하루 1,000건이며 이 점검은 오퍼레이션마다 수십 건 안쪽이다. +실행(저장소 루트에서): python -I -X utf8 tools/data-check/public_data_check2.py backend/.env.development [daegu|seoul-busan] +""" +import datetime as dt +import importlib.util +import json +import re +import sys +from collections import Counter +from pathlib import Path + +# 1차 점검 스크립트의 공통 함수(키 읽기·호출·파싱)를 그대로 쓴다 +_spec = importlib.util.spec_from_file_location('check1', Path(__file__).with_name('public_data_check.py')) +c1 = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(c1) + +TOUR, COMMON = c1.TOUR, c1.COMMON # 지역 묶음은 실행 시점에 c1.AREAS 로 읽는다 +PET = 'https://apis.data.go.kr/B551011/KorPetTourService2' +STORE_BASE = 'https://apis.data.go.kr/B553077/api/open/sdsc2' +SIZE_RE = re.compile(r'소형|중형|대형|\d+\s*(kg|KG|㎏|키로)') +# 지역 묶음별 상가정보 행정구역 — (시군구 코드, 행정동 이름에 들어가는 말) +STORE_DONGS = { + 'seoul-busan': [('11200', ['성수'])], + 'daegu': [('27110', ['성내', '삼덕', '대봉', '남산', '동인']), # 중구 도심 + ('27200', ['대명']), # 남구 앞산 + ('27260', ['두산', '상동', '수성', '지산'])], # 수성구 수성못 +} +# 데이트 후보 업종(상가정보 음식 소분류, 2차 점검에서 확인). 유흥 주점은 넣지 않는다 +DATE_CODES = {'카페': 'I21201', '빵/도넛': 'I21001', '아이스크림/빙수': 'I21008', '샌드위치/샐러드': 'I21005', + '요리 주점': 'I21104', '생맥주': 'I21103'} +# 유형별 영업시간·동반 여부 필드(매뉴얼 v4.4 기준) +HOURS_FIELD = {'12': 'usetime', '14': 'usetimeculture', '28': 'usetimeleports', '38': 'opentime', '39': 'opentimefood'} +PET_FIELD = {'12': 'chkpet', '14': 'chkpetculture', '28': 'chkpetleports', '38': 'chkpetshopping'} + + +def find_list(data, must_key=None): + """중첩 JSON 에서 dict 목록을 찾는다(응답 구조가 API 마다 달라서).""" + if isinstance(data, list) and data and isinstance(data[0], dict): + if must_key is None or must_key in data[0]: + return data + if isinstance(data, dict): + for v in data.values(): + found = find_list(v, must_key) + if found: + return found + return [] + + +def short(v, n=70): + return re.sub(r'\s+', ' ', re.sub(r'', ' / ', str(v))).strip()[:n] + + +def pet_check(key): + r = {} + # 1) 동반 정보 일괄 — contentId 를 비우면 전부 나온다(매뉴얼) + bulk, page = [], 1 + while page <= 12: + d = c1.call(key, '국문 반려동물 일괄', f'{TOUR}/detailPetTour2', {**COMMON, 'numOfRows': 1000, 'pageNo': page}) + its = c1.items_of(d) + bulk += its + if not its or len(bulk) >= (c1.total_of(d) or 0): + break + page += 1 + by_id = {str(x.get('contentid')): x for x in bulk} + type_counter = Counter(short(x.get('acmpyTypeCd'), 20) for x in bulk) + size_n = sum(1 for x in bulk if SIZE_RE.search(' '.join(str(v) for v in x.values()))) + r['bulk_total'] = len(bulk) + r['bulk_acmpyTypeCd'] = type_counter.most_common(8) + r['bulk_with_size_text'] = size_n + r['bulk_psblcpam_examples'] = [short(x.get('acmpyPsblCpam')) for x in bulk if c1.filled(x.get('acmpyPsblCpam'))][:6] + + # 2) 지역별 동반 가능 장소 목록 + areas = [] + for name, lon, lat, radius in c1.AREAS: + lst = c1.list_all(key, '반려 위치목록', PET, lon, lat, radius) + types = Counter(c1.TYPE_NAMES.get(str(x.get('contenttypeid')), str(x.get('contenttypeid'))) for x in lst) + joined = [by_id[str(x.get('contentid'))] for x in lst if str(x.get('contentid')) in by_id] + areas.append({ + 'area': name, 'pet_places': len(lst), 'by_type': dict(types), + 'with_conditions_in_bulk': len(joined), + 'with_size_text': sum(1 for x in joined if SIZE_RE.search(' '.join(str(v) for v in x.values()))), + 'examples': [{'title': x.get('title'), 'type': c1.TYPE_NAMES.get(str(x.get('contenttypeid'))), + 'cond': short(by_id.get(str(x.get('contentid')), {}).get('acmpyTypeCd'), 20), + 'who': short(by_id.get(str(x.get('contentid')), {}).get('acmpyPsblCpam'), 50)} for x in lst[:5]], + }) + r['areas'] = areas + + # 3) 두 서비스의 detailPetTour2 가 같은 데이터인지 한 곳으로 비교 + sample = next((a for a in areas if a['pet_places']), None) + if sample: + lon, lat, radius = next((x[1], x[2], x[3]) for x in c1.AREAS if x[0] == sample['area']) + first = c1.list_all(key, '반려 위치목록', PET, lon, lat, radius)[0] + cid = first.get('contentid') + a = c1.items_of(c1.call(key, '국문 반려동물 단건', f'{TOUR}/detailPetTour2', {**COMMON, 'contentId': cid})) + b = c1.items_of(c1.call(key, '반려 상세', f'{PET}/detailPetTour2', {**COMMON, 'contentId': cid})) + r['same_record_check'] = {'contentid': cid, 'title': first.get('title'), + 'kor_service': {k: short(v, 40) for k, v in (a[0] if a else {}).items()}, + 'pet_service': {k: short(v, 40) for k, v in (b[0] if b else {}).items()}, + 'identical': bool(a and b and a[0] == b[0])} + return r + + +def festival_check(key, weekend): + r = {} + # 1) 그날 열리는 행사만 받기 + same_day, page = [], 1 + while page <= 3: + d = c1.call(key, '국문 축제', f'{TOUR}/searchFestival2', + {**COMMON, 'eventStartDate': weekend, 'eventEndDate': weekend, 'numOfRows': 500, 'pageNo': page, 'arrange': 'A'}) + its = c1.items_of(d) + same_day += its + if not its or len(same_day) >= (c1.total_of(d) or 0): + break + page += 1 + def span_days(x): + try: + return (dt.datetime.strptime(str(x.get('eventenddate')), '%Y%m%d') - + dt.datetime.strptime(str(x.get('eventstartdate')), '%Y%m%d')).days + except ValueError: + return None + + outside = [x for x in same_day if not (str(x.get('eventstartdate')) <= weekend <= str(x.get('eventenddate')))] + long_runs = [x for x in same_day if (span_days(x) or 0) > 60] + r['same_day_total'] = len(same_day) + r['same_day_not_overlapping'] = len(outside) + r['same_day_over_60_days'] = len(long_runs) + r['lclsSystm2_counts'] = Counter(str(x.get('lclsSystm2')) for x in same_day).most_common() + r['progresstype'] = Counter(str(x.get('progresstype')) for x in same_day).most_common(6) + r['festivaltype'] = Counter(str(x.get('festivaltype')) for x in same_day).most_common(6) + # 2) 축제(EV01)만 + d = c1.call(key, '국문 축제', f'{TOUR}/searchFestival2', + {**COMMON, 'eventStartDate': weekend, 'eventEndDate': weekend, 'lclsSystm1': 'EV', 'lclsSystm2': 'EV01', + 'numOfRows': 500, 'pageNo': 1, 'arrange': 'A'}) + ev01 = c1.items_of(d) + r['ev01_total'] = c1.total_of(d) + near = {} + for name, lon, lat, _ in c1.AREAS: + lst = [] + for f in ev01: + try: + km = c1.haversine(lon, lat, float(f['mapx']), float(f['mapy'])) / 1000 + except (KeyError, TypeError, ValueError): + continue + if km <= 15: + lst.append(f"{f.get('title')} ({f.get('eventstartdate')}–{f.get('eventenddate')}, {km:.1f}km)") + near[name] = sorted(lst, key=lambda s: float(s.rsplit(', ', 1)[1][:-3]))[:6] + r['ev01_within_15km'] = near + r['ev01_examples'] = [f"{x.get('title')} ({x.get('eventstartdate')}–{x.get('eventenddate')})" for x in ev01[:8]] + return r + + +def hours_check(key): + r = {'food_hours_examples': [], 'chkpet_values': Counter(), 'chkpet_sampled': 0} + for name, lon, lat, radius in c1.AREAS: + kor = c1.list_all(key, '국문 목록', TOUR, lon, lat, radius) + food = [x for x in kor if str(x.get('contenttypeid')) == '39'][:4] + for it in food: + rec = c1.items_of(c1.call(key, '국문 소개정보', f'{TOUR}/detailIntro2', {**COMMON, 'contentId': it['contentid'], 'contentTypeId': '39'})) + if rec: + r['food_hours_examples'].append({'area': name, 'title': it.get('title'), + 'opentimefood': short(rec[0].get('opentimefood'), 90), + 'restdatefood': short(rec[0].get('restdatefood'), 40)}) + for it in [x for x in kor if str(x.get('contenttypeid')) in PET_FIELD][:10]: + t = str(it.get('contenttypeid')) + rec = c1.items_of(c1.call(key, '국문 소개정보', f'{TOUR}/detailIntro2', {**COMMON, 'contentId': it['contentid'], 'contentTypeId': t})) + if rec: + r['chkpet_sampled'] += 1 + r['chkpet_values'][short(rec[0].get(PET_FIELD[t]), 30) or '(빈 값)'] += 1 + r['chkpet_values'] = r['chkpet_values'].most_common(10) + return r + + +def store_check(key, region): + r = {'dongs': [], 'radius_counts': []} + # 1) 행정동 코드와 동별 업소 수(전체·음식·카페) — 총건수만 보므로 numOfRows=1 + for signgu, words in STORE_DONGS[region]: + dongs = find_list(c1.call(key, '상가 코드', f'{STORE_BASE}/baroApi', + {'resId': 'dong', 'catId': 'admi', 'signguCd': signgu, 'type': 'json'}), 'adongCd') + for dg in [d for d in dongs if any(w in str(d.get('adongNm')) for w in words)]: + row = {'signguCd': signgu, 'adongCd': dg.get('adongCd'), 'adongNm': dg.get('adongNm')} + for label, extra in [('all', {}), ('food', {'indsLclsCd': 'I2'}), ('cafe', {'indsSclsCd': DATE_CODES['카페']})]: + d = c1.call(key, '상가 동별', f'{STORE_BASE}/storeListInDong', + {'divId': 'adongCd', 'key': dg.get('adongCd'), 'numOfRows': 1, 'pageNo': 1, 'type': 'json', **extra}) + row[label] = c1.total_of(d) or 0 + r['dongs'].append(row) + # 2) 동네 반경 안 데이트 업종 수 + for name, lon, lat, radius in c1.AREAS: + row = {'area': name, 'radius_m': radius} + for label, code in DATE_CODES.items(): + d = c1.call(key, '상가 반경', f'{STORE_BASE}/storeListInRadius', + {'radius': radius, 'cx': lon, 'cy': lat, 'indsSclsCd': code, 'numOfRows': 1, 'pageNo': 1, 'type': 'json'}) + row[label] = c1.total_of(d) or 0 + r['radius_counts'].append(row) + return r + + +def weather_check(key): + lon, lat = c1.AREAS[0][1], c1.AREAS[0][2] + nx, ny = c1.kma_grid(lon, lat) + bd, bt = c1.kma_base(c1.kst_now()) + fc = c1.items_of(c1.call(key, '기상청 단기예보', c1.KMA, {'pageNo': 1, 'numOfRows': 1500, 'dataType': 'JSON', + 'base_date': bd, 'base_time': bt, 'nx': nx, 'ny': ny})) + per_date = {} + for x in fc: + per_date.setdefault(x.get('fcstDate'), set()).add(x.get('fcstTime')) + return {'base': f'{bd} {bt}', 'times_per_date': {k: len(v) for k, v in sorted(per_date.items())}, + 'last_date_times': sorted(per_date.get(max(per_date), [])) if per_date else [], + 'pcp_examples': sorted({str(x.get('fcstValue')) for x in fc if x.get('category') == 'PCP'})[:8]} + + +def main(env_path, region='seoul-busan'): + c1.use_region(region) + key = c1.load_key(env_path) + now = c1.kst_now() + weekend = (now.date() + dt.timedelta(days=(5 - now.weekday()) % 7)).strftime('%Y%m%d') + report = {'checked_at_kst': now.strftime('%Y-%m-%d %H:%M'), 'region': region, 'weekend_saturday': weekend} + for label, fn in [('반려동물', lambda: pet_check(key)), ('축제', lambda: festival_check(key, weekend)), + ('영업시간·동반여부', lambda: hours_check(key)), ('상가정보', lambda: store_check(key, region)), + ('날씨', lambda: weather_check(key))]: + print(f'… {label} 점검 중', flush=True) + try: + report[label] = fn() + except Exception as e: # 한 항목이 실패해도 나머지는 계속한다 + report[label] = {'error': f'{type(e).__name__}: {str(e)[:200]}'.replace(key, '***')} + report['calls'] = c1.calls + report['errors'] = c1.errors + + out_dir = Path('docs/research/data-check') + out_dir.mkdir(parents=True, exist_ok=True) + out_file = out_dir / f'public-data-check2-{region}-{now.strftime("%Y%m%d-%H%M")}.json' + out_file.write_text(json.dumps(report, ensure_ascii=False, indent=2, default=list), encoding='utf-8') + + # ── 화면 요약 ── + p = report['반려동물'] + print(f'\n# 2차 점검 {report["checked_at_kst"]} KST · 주말 {weekend}') + if 'error' in p: + print('\n## 반려동물: 실패', p['error']) + else: + print(f'\n## 반려동물 — 전국 동반 정보 {p["bulk_total"]}건, 크기·무게 문구 {p["bulk_with_size_text"]}건') + print('- 동반 구분:', p['bulk_acmpyTypeCd']) + print('- 동반 가능 동물 예:', p['bulk_psblcpam_examples'][:4]) + for a in p['areas']: + print(f'- {a["area"]}: 동반 가능 장소 {a["pet_places"]}곳 {a["by_type"]}, 조건 있음 {a["with_conditions_in_bulk"]}, 크기 문구 {a["with_size_text"]}') + for e in a['examples'][:3]: + print(f' · {e["title"]} [{e["type"]}] {e["cond"]} / {e["who"]}') + s = p.get('same_record_check') + if s: + print(f'- 두 서비스 같은 데이터? {s["identical"]} ({s["title"]})') + f = report['축제'] + if 'error' not in f: + print(f'\n## 축제 — 시작=종료=토요일 조회 {f["same_day_total"]}건, 그날과 안 겹치는 것 {f["same_day_not_overlapping"]}건, 60일 넘는 장기 {f["same_day_over_60_days"]}건') + print('- 분류(lclsSystm2):', f['lclsSystm2_counts']) + print(f'- 축제(EV01)만: {f["ev01_total"]}건. 예:', f['ev01_examples'][:4]) + for k, v in f['ev01_within_15km'].items(): + print(f'- {k} 15km 안 축제:', v[:3] or '없음') + else: + print('\n## 축제: 실패', f['error']) + h = report['영업시간·동반여부'] + if 'error' not in h: + print('\n## 음식점 영업시간 예') + for e in h['food_hours_examples'][:6]: + print(f'- [{e["area"]}] {e["title"]}: {e["opentimefood"]} | 쉬는날 {e["restdatefood"]}') + print(f'- 관광지·문화시설 동반 여부(chkpet*) 값 ({h["chkpet_sampled"]}곳):', h['chkpet_values']) + else: + print('\n## 영업시간: 실패', h['error']) + st = report['상가정보'] + if 'error' not in st: + print('\n## 상가정보 — 행정동별 업소 수') + for row in st['dongs']: + print(f"- {row['adongNm']} ({row['adongCd']}): 전체 {row['all']}, 음식 {row['food']}, 카페 {row['cafe']}") + print('- 동네 반경 안 데이트 업종:') + for row in st['radius_counts']: + print(' ', row) + else: + print('\n## 상가정보: 실패', st['error']) + w = report['날씨'] + if 'error' not in w: + print(f'\n## 날씨 — 발표 {w["base"]}, 날짜별 예보 시각 수 {w["times_per_date"]}, 마지막 날 시각 {w["last_date_times"]}, 강수량 값 예 {w["pcp_examples"]}') + print(f'\n호출 수: {report["calls"]}') + if report['errors']: + print(f'오류 {len(report["errors"])}건 (앞 5건):') + for e in report['errors'][:5]: + print(' -', e) + print(f'\n상세 결과: {out_file}') + + +if __name__ == '__main__': + main(sys.argv[1] if len(sys.argv) > 1 else 'backend/.env.development', sys.argv[2] if len(sys.argv) > 2 else 'seoul-busan') diff --git a/tools/data-check/run_region.py b/tools/data-check/run_region.py new file mode 100644 index 0000000..77d2594 --- /dev/null +++ b/tools/data-check/run_region.py @@ -0,0 +1,23 @@ +"""한 지역 묶음에 대해 1차·2차 공공데이터 점검을 차례로 돌린다. + +실행(저장소 루트에서): python -I -X utf8 tools/data-check/run_region.py daegu backend/.env.development +""" +import importlib.util +import sys +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 + + +if __name__ == '__main__': + region = sys.argv[1] if len(sys.argv) > 1 else 'daegu' + env_path = sys.argv[2] if len(sys.argv) > 2 else 'backend/.env.development' + print(f'===== 1차 점검 ({region}) =====', flush=True) + load('public_data_check').main(env_path, region) + print(f'\n===== 2차 점검 ({region}) =====', flush=True) + load('public_data_check2').main(env_path, region)