ttipo/wandergo
1
0
Fork 0
generated from ttipo/project-base
Code Issues Pull Requests 2 Actions 4 Packages Projects Releases Wiki Activity

feat(backend): 장소·축제·공연·전시 데이터 기반(장소 DB·동기화·조회 API) #1

Merged
ttipo merged 30 commits from feat/place-db into main 2026-11-04 13:25:51 +09:00
Conversation 0 Commits 30 Files Changed 208
+1174
6 changed files with 1174 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 82087b7ac7 - Show all commits
.gitignore
+4
View File
@@ -57,3 +57,7 @@ Thumbs.db
.cache/
.eslintcache
.npm/
# Python 캐시 (tools/ 점검 스크립트)
__pycache__/
*.pyc
tools/data-check/curation_candidates.py
+275
View File
@@ -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')
tools/data-check/curation_shortlist.py
+233
View File
@@ -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)
tools/data-check/public_data_check.py
+343
View File
@@ -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')
tools/data-check/public_data_check2.py
+296
View File
@@ -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'<br\s*/?>', ' / ', 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')
tools/data-check/run_region.py
+23
View File
@@ -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)