196 lines
7.4 KiB
Python
196 lines
7.4 KiB
Python
"""
|
|
config_loader.py — YAML 설정 파일 로딩 + 캐싱
|
|
|
|
모든 설정 접근의 단일 진입점.
|
|
config/ 폴더의 YAML 파일을 로드하고 lru_cache로 캐싱합니다.
|
|
|
|
YAML 구조: site_label(key) → [alias_1, alias_2, ...] (Closed Set 기반)
|
|
조회 시: alias → site_label (역매핑)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
CONFIG_DIR = Path(__file__).resolve().parent.parent / "config"
|
|
|
|
|
|
@lru_cache(maxsize=None)
|
|
def load_config(name: str) -> dict[str, Any]:
|
|
"""YAML 파일을 로드하여 dict로 반환 (결과 캐싱)"""
|
|
path = CONFIG_DIR / f"{name}.yaml"
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"설정 파일을 찾을 수 없습니다: {path}")
|
|
with open(path, encoding="utf-8") as f:
|
|
data = yaml.safe_load(f) or {}
|
|
return data
|
|
|
|
|
|
def get_mapping(config_name: str, key: str) -> dict[str, Any]:
|
|
"""특정 설정 파일의 특정 섹션을 반환 (원본 구조 그대로)"""
|
|
return load_config(config_name).get(key, {})
|
|
|
|
|
|
def get_list(config_name: str, key: str) -> list:
|
|
"""특정 설정 파일의 특정 리스트 섹션을 반환"""
|
|
return load_config(config_name).get(key, [])
|
|
|
|
|
|
def get_value(config_name: str, key: str, default: Any = None) -> Any:
|
|
"""특정 설정 파일의 단일 값을 반환"""
|
|
return load_config(config_name).get(key, default)
|
|
|
|
|
|
# ──────────────────────────────────────────────
|
|
# Closed Set 역매핑 빌드
|
|
# ──────────────────────────────────────────────
|
|
|
|
@lru_cache(maxsize=None)
|
|
def _build_reverse_map(config_name: str, key: str) -> dict[str, str]:
|
|
"""site_label: [aliases...] 구조를 {alias: site_label} 역매핑으로 변환
|
|
|
|
예: { '패스트볼': ['직구', '패스트볼'] }
|
|
→ { '직구': '패스트볼', '패스트볼': '패스트볼' }
|
|
"""
|
|
raw = get_mapping(config_name, key)
|
|
reverse: dict[str, str] = {}
|
|
for site_label, aliases in raw.items():
|
|
if isinstance(aliases, list):
|
|
for alias in aliases:
|
|
reverse[str(alias)] = str(site_label)
|
|
else:
|
|
# aliases가 리스트가 아닌 경우 (단순 값이면 그대로)
|
|
reverse[str(aliases)] = str(site_label)
|
|
return reverse
|
|
|
|
|
|
def allowed_values(config_name: str, key: str) -> set[str]:
|
|
"""해당 섹션의 관리자 사이트 허용값(Closed Set) 반환"""
|
|
raw = get_mapping(config_name, key)
|
|
return set(raw.keys())
|
|
|
|
|
|
def lookup(config_name: str, key: str, alias: str) -> str | None:
|
|
"""alias → site_label 조회. 없으면 None"""
|
|
return _build_reverse_map(config_name, key).get(alias)
|
|
|
|
|
|
def lookup_or_raise(config_name: str, key: str, alias: str) -> str:
|
|
"""alias → site_label 조회. 없으면 오류"""
|
|
result = lookup(config_name, key, alias)
|
|
if result is None:
|
|
allowed = allowed_values(config_name, key)
|
|
raise ValueError(
|
|
f"매핑 오류: '{alias}'는 {key}의 허용값에 없습니다. "
|
|
f"허용값: {sorted(allowed)}"
|
|
)
|
|
return result
|
|
|
|
|
|
# ──────────────────────────────────────────────
|
|
# 편의 함수: 역매핑 (alias → site_label)
|
|
# ──────────────────────────────────────────────
|
|
|
|
def pitch_type_map() -> dict[str, str]:
|
|
"""네이버 stuff → 사이트 구종 라벨"""
|
|
return _build_reverse_map("pitch_rules", "pitch_type")
|
|
|
|
def pitch_result_map() -> dict[str, str]:
|
|
"""네이버 pitchResultText → 사이트 투구결과 라벨"""
|
|
return _build_reverse_map("pitch_rules", "pitch_result")
|
|
|
|
def batter_result_map() -> dict[str, str]:
|
|
"""result.type → 사이트 타자결과 라벨"""
|
|
return _build_reverse_map("pitch_rules", "batter_result")
|
|
|
|
def runner_event_map() -> dict[str, str]:
|
|
"""runnerEvent.type → 사이트 주루 라벨"""
|
|
return _build_reverse_map("pitch_rules", "runner_event")
|
|
|
|
|
|
def team_name_map() -> dict[str, str]:
|
|
"""네이버 팀명 → 사이트 팀명"""
|
|
return _build_reverse_map("mappings", "team_name")
|
|
|
|
def team_code_map() -> dict[str, str]:
|
|
"""네이버 팀코드 → 한글 팀명"""
|
|
return _build_reverse_map("mappings", "team_code")
|
|
|
|
def stadium_name_map() -> dict[str, str]:
|
|
"""네이버 구장명 → 사이트 구장명"""
|
|
return _build_reverse_map("mappings", "stadium_name")
|
|
|
|
def game_type_map() -> dict[str, str]:
|
|
"""네이버 경기유형 → 사이트 경기유형"""
|
|
return _build_reverse_map("mappings", "game_type")
|
|
|
|
def position_number_map() -> dict[str, str]:
|
|
"""포지션명 → 번호"""
|
|
return _build_reverse_map("mappings", "position_number")
|
|
|
|
def result_labels() -> dict[str, str]:
|
|
"""W/L/H/S → 승리투수/패전투수/홀드/세이브"""
|
|
return _build_reverse_map("mappings", "result_labels")
|
|
|
|
def kbo_sr_id_candidates() -> dict[str, list]:
|
|
"""역매핑 불필요 — 원본 그대로"""
|
|
return get_mapping("mappings", "kbo_sr_id_candidates")
|
|
|
|
|
|
# ──────────────────────────────────────────────
|
|
# 편의 함수: Closed Set 직접 조회
|
|
# ──────────────────────────────────────────────
|
|
|
|
def pitch_type_allowed() -> set[str]:
|
|
return allowed_values("pitch_rules", "pitch_type")
|
|
|
|
def pitch_result_allowed() -> set[str]:
|
|
return allowed_values("pitch_rules", "pitch_result")
|
|
|
|
def batter_result_allowed() -> set[str]:
|
|
return allowed_values("pitch_rules", "batter_result")
|
|
|
|
def runner_event_allowed() -> set[str]:
|
|
return allowed_values("pitch_rules", "runner_event")
|
|
|
|
|
|
# ──────────────────────────────────────────────
|
|
# 편의 함수: 역매핑 불필요한 것들 (원본 구조 그대로)
|
|
# ──────────────────────────────────────────────
|
|
|
|
def field_coordinates() -> dict[str, list]:
|
|
return get_mapping("field_coordinates", "field_coordinates")
|
|
|
|
def hit_ball_type_map() -> dict[str, str]:
|
|
return get_mapping("field_coordinates", "hit_ball_type")
|
|
|
|
def foul_fly_coords() -> dict[str, list]:
|
|
return get_mapping("field_coordinates", "foul_fly")
|
|
|
|
def defense_button_id_map() -> dict[str, str]:
|
|
return get_mapping("site_selectors", "defense_button_id")
|
|
|
|
def position_to_defense_no() -> dict[str, str]:
|
|
return get_mapping("site_selectors", "position_to_defense_no")
|
|
|
|
def review_result_groups() -> dict[str, dict]:
|
|
return get_mapping("review_rules", "review_result_groups")
|
|
|
|
def crawler_headers() -> dict[str, str]:
|
|
return get_mapping("crawler_constants", "headers")
|
|
|
|
def skip_option_types() -> set[int]:
|
|
return set(get_list("crawler_constants", "skip_option_types"))
|
|
|
|
def hidden_event_texts() -> set[str]:
|
|
return set(get_list("crawler_constants", "hidden_event_texts"))
|
|
|
|
def change_keywords() -> tuple[str, ...]:
|
|
return tuple(get_list("crawler_constants", "change_keywords"))
|
|
|
|
def max_inning() -> int:
|
|
return get_value("crawler_constants", "max_inning", 20)
|