"""모두의 빛길 — 광주 공공데이터 Excel 로더. `data/광주공공데이터_2026-05-04.xlsx` (13시트) → GraphData 변환. 채울 수 있는 노드 ---------------- ✅ VenueNode : 광주 관광명소 487개 (단, 좌표는 별도 지오코딩 필요) ✅ TransitNode : 도시철도 1호선 정거장 (이름은 운행데이터에서, 좌표는 하드코딩) ✅ AmenityNode : 공공화장실 + 무더위쉼터 + CCTV + 가로등 (수만 개) ✅ HazardNode : 점자블록·턱낮춤·조명 미비 횡단보도 채울 수 없는 노드 (별도 수집 필요) -------------------------------- ❌ EventNode : 공연·전시 데이터 — 광주문화재단 API / 문화포털 API ❌ 정확한 HazardNode (사고다발) — 도로교통공단 보행자/노인 사고다발지점 ❌ VenueNode 좌표 — Kakao/Naver 지오코딩 API로 시설명 → 좌표 변환 학습 단계에 추가로 필요 --------------------- - (질문, 정답 시설) 라벨 페어 ~500개 (수동 라벨링) - 시나리오 페이지 130개 발화 BIO 태깅 (수동) """ from __future__ import annotations import os from typing import List, Optional, Tuple, Dict from .data_schema import ( GraphData, VenueNode, EventNode, TransitNode, AmenityNode, HazardNode, ) # ============================================================ # 광주 도시철도 1호선 19개역 좌표 (운영데이터에 좌표가 없어 하드코딩) # 출처: 위키피디아 + 다음맵 검증 # ============================================================ GWANGJU_LINE1_COORDS: Dict[str, Tuple[float, float]] = { "녹동": (35.1182, 126.9479), "소태": (35.1257, 126.9387), "학동": (35.1364, 126.9269), "학동·증심사입구": (35.1364, 126.9269), "증심사입구": (35.1364, 126.9269), "남광주": (35.1417, 126.9209), "문화전당": (35.1473, 126.9181), "금남로4가": (35.1499, 126.9136), "금남로5가": (35.1521, 126.9094), "양동시장": (35.1497, 126.9027), "돌고개": (35.1500, 126.8978), "농성": (35.1511, 126.8866), "화정": (35.1520, 126.8769), "쌍촌": (35.1532, 126.8676), "운천": (35.1535, 126.8554), "상무": (35.1571, 126.8455), "김대중컨벤션센터": (35.1577, 126.8408), "공항": (35.1262, 126.8085), "송정공원": (35.1448, 126.8050), "광주송정역": (35.1399, 126.7916), "도산": (35.1422, 126.8000), "평동": (35.1399, 126.7800), } # 광주 중심 (관광명소 좌표 임시 부여용 — 지오코딩 전 fallback) GWANGJU_CENTER = (35.1595, 126.8526) # ============================================================ # 메인 로더 # ============================================================ def load_gwangju_excel( path: str, streetlight_sample_rate: int = 50, skip_streetlights: bool = False, venue_geocoder: Optional[callable] = None, ) -> GraphData: """엑셀 파일 → GraphData. 파라미터 -------- path : 엑셀 경로 streetlight_sample_rate : 가로등 N개당 1개만 샘플링 (22000개 전부는 너무 많음) skip_streetlights : True면 가로등 제외 (테스트용) venue_geocoder : 시설명 → (lat, lng) 함수. None이면 광주 중심으로 임시 부여. 예: lambda name: kakao_geocode(name + " 광주") """ try: import openpyxl except ImportError as e: raise ImportError( "openpyxl이 필요합니다: pip install openpyxl" ) from e if not os.path.exists(path): raise FileNotFoundError(f"엑셀 파일을 찾을 수 없습니다: {path}") print(f"[load_gwangju_excel] 파일 열기: {path}") wb = openpyxl.load_workbook(path, data_only=True, read_only=True) venues = _load_venues(wb, geocoder=venue_geocoder) transits = _load_transits(wb) amenities = _load_amenities( wb, streetlight_sample_rate=streetlight_sample_rate, skip_streetlights=skip_streetlights, ) hazards = _load_hazards(wb) gdata = GraphData( venues=venues, events=[], # ⚠️ EventNode는 별도 API에서 수집 transits=transits, amenities=amenities, hazards=hazards, ) print(f"[load_gwangju_excel] 완료. 노드 수: {gdata.stats()}") print(f" ⚠️ EventNode 0개 (공연·전시 데이터는 광주문화재단 API에서 별도 수집)") print(f" ⚠️ VenueNode 좌표 — venue_geocoder가 None이면 광주 중심으로 임시 부여됨") return gdata # ============================================================ # 1. VenueNode — 관광명소 # ============================================================ def _load_venues(wb, geocoder: Optional[callable] = None) -> List[VenueNode]: """GJ_Tourism_Attractions 시트 → VenueNode 리스트. 헤더: 연번 | 구분 | 세부구분 | 시설명 | 시설명영문명 | 데이터기준일자 좌표가 없어 (1) geocoder 함수가 있으면 그걸로 변환, (2) 없으면 광주 중심으로 임시 부여. """ if "GJ_Tourism_Attractions" not in wb.sheetnames: return [] ws = wb["GJ_Tourism_Attractions"] venues: List[VenueNode] = [] for i, row in enumerate(ws.iter_rows(min_row=2, values_only=True)): if not row or len(row) < 4: continue cat, sub, name = row[1], row[2], row[3] if not name: continue venue_type = _map_venue_type(str(cat or ""), str(sub or "")) # 좌표 결정 if geocoder is not None: try: lat, lng = geocoder(str(name)) except Exception: lat, lng = GWANGJU_CENTER else: lat, lng = GWANGJU_CENTER # ⚠️ 임시. 실사용 시 지오코딩 필수 # 카테고리에서 indoor/free 추정 indoor = any(kw in str(sub or "") for kw in ["박물관", "미술관", "도서관", "공연장", "실내"]) free = any(kw in str(sub or "") for kw in ["공원", "광장"]) # 무료가 명시된 경우 venues.append(VenueNode( id=i, name=str(name).strip(), lat=lat, lng=lng, venue_type=venue_type, accessibility_score=0.5, # 별도 데이터로 보정 필요 indoor=indoor, free=free, )) return venues def _map_venue_type(cat: str, sub: str) -> str: """관광명소 카테고리 → VenueNode.venue_type.""" if "박물관" in sub or "박물관" in cat: return "박물관" if "미술관" in sub or "갤러리" in sub: return "미술관" if "도서관" in sub: return "도서관" if "공연" in sub or "극장" in sub: return "공연장" if "체험" in sub: return "체험관" if "카페" in sub or "식당" in sub: return "카페" if "문화" in cat or "문화" in sub: return "문화센터" if "공원" in sub or "광장" in sub: return "공원" return "기타" # ============================================================ # 2. TransitNode — 도시철도 1호선 # ============================================================ def _load_transits(wb) -> List[TransitNode]: """N_Urban_Railway_Operation_Stand 시트의 운행구간정거장에서 정거장명 추출. 좌표는 GWANGJU_LINE1_COORDS 딕셔너리에서 룩업. 버스 정류장은 본 엑셀에 없으므로 추후 별도 데이터 추가 필요. """ if "N_Urban_Railway_Operation_Stand" not in wb.sheetnames: return [] ws = wb["N_Urban_Railway_Operation_Stand"] station_set = set() for row in ws.iter_rows(min_row=2, values_only=True): if not row or len(row) < 8: continue stations_field = row[7] # 운행구간정거장 if not stations_field: continue for token in str(stations_field).split(","): token = token.strip() if not token: continue # "녹동-001" → "녹동" name = token.split("-")[0].strip() if name: station_set.add(name) transits: List[TransitNode] = [] for i, name in enumerate(sorted(station_set)): if name not in GWANGJU_LINE1_COORDS: continue # 알 수 없는 역명 (오기 등) 스킵 lat, lng = GWANGJU_LINE1_COORDS[name] transits.append(TransitNode( id=i, name=name, lat=lat, lng=lng, transit_type="지하철", has_elevator=True, # 광주 1호선은 전 역 엘리베이터 보유 has_toilet=True, line="1호선", )) return transits # ============================================================ # 3. AmenityNode — 화장실/쉼터/CCTV/가로등 # ============================================================ def _load_amenities( wb, streetlight_sample_rate: int = 50, skip_streetlights: bool = False, ) -> List[AmenityNode]: amenities: List[AmenityNode] = [] next_id = [0] # nonlocal 대신 list로 카운터 def _add(name: str, lat: float, lng: float, atype: str, accessible: bool = False, open_24h: bool = False): amenities.append(AmenityNode( id=next_id[0], name=name, lat=lat, lng=lng, amenity_type=atype, accessible=accessible, open_24h=open_24h, )) next_id[0] += 1 # ---- 1) 공공화장실 ---- if "N_Public_Toilet_Standard_GJMetr" in wb.sheetnames: ws = wb["N_Public_Toilet_Standard_GJMetr"] # 헤더: 구분, 화장실명, ..., 위도(18), 경도(19), ..., 개방시간(16), 남성장애인대변기수(7), 여성장애인대변기수(12) for row in ws.iter_rows(min_row=2, values_only=True): if not row or len(row) < 20: continue try: name = str(row[1] or "공공화장실").strip() lat = _safe_float(row[18]) lng = _safe_float(row[19]) male_disabled = _safe_int(row[7]) fem_disabled = _safe_int(row[12]) hours = str(row[16] or "") except (TypeError, ValueError, IndexError): continue if lat is None or lng is None: continue disabled = (male_disabled > 0) or (fem_disabled > 0) open_24 = "24" in hours or "00:00~24:00" in hours _add(name, lat, lng, atype="장애인화장실" if disabled else "공공화장실", accessible=disabled, open_24h=open_24) # ---- 2) 무더위쉼터 ---- if "N_Heat_Shelter_Standard_GJMetro" in wb.sheetnames: ws = wb["N_Heat_Shelter_Standard_GJMetro"] # 헤더: 쉼터명(0), ..., 위도(17), 경도(18) for row in ws.iter_rows(min_row=2, values_only=True): if not row or len(row) < 19: continue try: name = str(row[0] or "쉼터").strip() lat = _safe_float(row[17]) lng = _safe_float(row[18]) except (TypeError, ValueError, IndexError): continue if lat is None or lng is None: continue _add(name, lat, lng, atype="무더위쉼터", accessible=True) # ---- 3) CCTV (표준 1137개) ---- if "N_CCTV_Standard_GJMetro" in wb.sheetnames: ws = wb["N_CCTV_Standard_GJMetro"] # 헤더: 관리기관명(0), ..., 위도(10), 경도(11) for row in ws.iter_rows(min_row=2, values_only=True): if not row or len(row) < 12: continue try: lat = _safe_float(row[10]) lng = _safe_float(row[11]) except (TypeError, ValueError, IndexError): continue if lat is None or lng is None: continue _add(f"CCTV-{next_id[0]}", lat, lng, atype="CCTV") # ---- 4) CCTV (구별 6749개 — 너무 많아 샘플링) ---- if "GJ_CCTV" in wb.sheetnames: ws = wb["GJ_CCTV"] # 헤더: ..., 위도(8), 경도(9) for i, row in enumerate(ws.iter_rows(min_row=2, values_only=True)): if i % 5 != 0: # 5개당 1개 continue if not row or len(row) < 10: continue try: lat = _safe_float(row[8]) lng = _safe_float(row[9]) except (TypeError, ValueError, IndexError): continue if lat is None or lng is None: continue _add(f"CCTV-{next_id[0]}", lat, lng, atype="CCTV") # ---- 5) 가로등 (북구·서구 합 22000개 — 큰 폭으로 샘플링) ---- if not skip_streetlights: if "GJ_Bukgu_Streetlight_Locations" in wb.sheetnames: ws = wb["GJ_Bukgu_Streetlight_Locations"] # 헤더: 등록번호, 표찰번호, 지번주소, 경도(3), 위도(4), 데이터제공일자 for i, row in enumerate(ws.iter_rows(min_row=2, values_only=True)): if i % streetlight_sample_rate != 0: continue if not row or len(row) < 5: continue try: lng = _safe_float(row[3]) lat = _safe_float(row[4]) except (TypeError, ValueError, IndexError): continue if lat is None or lng is None: continue _add(f"가로등-{next_id[0]}", lat, lng, atype="가로등") if "GJ_Seogu_Streetlight_Status" in wb.sheetnames: ws = wb["GJ_Seogu_Streetlight_Status"] # 헤더: 표찰번호, 지번주소, 위도(2), 경도(3) for i, row in enumerate(ws.iter_rows(min_row=2, values_only=True)): if i % streetlight_sample_rate != 0: continue if not row or len(row) < 4: continue try: lat = _safe_float(row[2]) lng = _safe_float(row[3]) except (TypeError, ValueError, IndexError): continue if lat is None or lng is None: continue _add(f"가로등-{next_id[0]}", lat, lng, atype="가로등") if "GJ_Namgu_Security_Lights" in wb.sheetnames: ws = wb["GJ_Namgu_Security_Lights"] # 헤더: 보안등위치명(0), 설치개수, 주소, 위도(3), 경도(4) for row in ws.iter_rows(min_row=2, values_only=True): if not row or len(row) < 5: continue try: name = str(row[0] or "보안등").strip() lat = _safe_float(row[3]) lng = _safe_float(row[4]) except (TypeError, ValueError, IndexError): continue if lat is None or lng is None: continue _add(f"보안등-{name}", lat, lng, atype="가로등") return amenities # ============================================================ # 4. HazardNode — 점자/턱/조명 미비 횡단보도 # ============================================================ def _load_hazards(wb) -> List[HazardNode]: """N_Crosswalk_Standard_GJMetro에서 위험 횡단보도 추출. 헤더: ..., 위도(9), 경도(10), ..., 보도턱낮춤여부(20), 점자블록유무(21), 집중조명시설유무(22) Y/N 값. 셋 다 Y면 안전 → 제외. 하나라도 N이면 위험구간으로 등록. """ if "N_Crosswalk_Standard_GJMetro" not in wb.sheetnames: return [] ws = wb["N_Crosswalk_Standard_GJMetro"] hazards: List[HazardNode] = [] next_id = 0 for row in ws.iter_rows(min_row=2, values_only=True): if not row or len(row) < 23: continue try: lat = _safe_float(row[9]) lng = _safe_float(row[10]) except (TypeError, ValueError, IndexError): continue if lat is None or lng is None: continue curb_low = _yn(row[20]) # 보도턱낮춤 tactile = _yn(row[21]) # 점자블록 bright = _yn(row[22]) # 집중조명 if curb_low and tactile and bright: continue # 안전한 곳 severity = 0.0 types = [] if not curb_low: severity += 0.3 types.append("턱") if not tactile: severity += 0.3 types.append("점자블록미비") if not bright: severity += 0.2 types.append("저조도") hazards.append(HazardNode( id=next_id, lat=lat, lng=lng, hazard_type="턱" if "턱" in types else ("저조도" if "저조도" in types else "공사"), severity=min(severity, 1.0), night_only=(not bright) and curb_low and tactile, )) next_id += 1 return hazards # ============================================================ # 헬퍼 # ============================================================ def _safe_float(v) -> Optional[float]: if v is None: return None try: f = float(str(v).strip()) if f == 0.0: return None return f except (TypeError, ValueError): return None def _safe_int(v) -> int: if v is None: return 0 try: return int(float(str(v).strip())) except (TypeError, ValueError): return 0 def _yn(v) -> bool: if v is None: return False s = str(v).strip().upper() return s == "Y" # ============================================================ # 지오코더 어댑터 (선택) — 실사용 시 주석 해제하고 API 키 입력 # ============================================================ def make_kakao_geocoder(rest_api_key: str) -> callable: """카카오 로컬 API 지오코더 생성. 사용법: from src.data_loader import load_gwangju_excel, make_kakao_geocoder geocoder = make_kakao_geocoder("YOUR_REST_API_KEY") gdata = load_gwangju_excel("data/광주공공데이터.xlsx", venue_geocoder=geocoder) 카카오 개발자센터에서 REST API 키 발급 필요 (무료, 일 300,000회). """ import urllib.parse import urllib.request import json import time cache: Dict[str, Tuple[float, float]] = {} def geocode(query: str) -> Tuple[float, float]: if query in cache: return cache[query] url = "https://dapi.kakao.com/v2/local/search/keyword.json?query=" + \ urllib.parse.quote(query) req = urllib.request.Request(url, headers={ "Authorization": f"KakaoAK {rest_api_key}", }) try: with urllib.request.urlopen(req, timeout=5) as r: data = json.loads(r.read().decode("utf-8")) docs = data.get("documents", []) if not docs: raise ValueError(f"No result for: {query}") lat = float(docs[0]["y"]) lng = float(docs[0]["x"]) cache[query] = (lat, lng) time.sleep(0.05) # rate limit 보호 return lat, lng except Exception: return GWANGJU_CENTER return geocode __all__ = [ "load_gwangju_excel", "make_kakao_geocoder", "GWANGJU_LINE1_COORDS", "GWANGJU_CENTER", ]