import time import requests from bs4 import BeautifulSoup from typing import Optional _cache: dict[str, tuple[Optional[int], float]] = {} # ticker -> (price, timestamp) _CACHE_TTL = 180 # 3분 _HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/90.0.4430.93 Safari/537.36" ) } def _fetch_from_mobile_api(ticker: str) -> Optional[int]: """네이버 모바일 주식 API로 현재가 조회""" url = f"https://m.stock.naver.com/api/stock/{ticker}/basic" try: resp = requests.get(url, headers=_HEADERS, timeout=5) resp.raise_for_status() data = resp.json() price_str = data.get("closePrice") or data.get("stockEndPrice") or "" price_str = str(price_str).replace(",", "").strip() return int(price_str) if price_str.isdigit() else None except Exception: return None def _fetch_from_html_fallback(ticker: str) -> Optional[int]: """네이버 금융 HTML 폴백 (.no_today .blind 파싱)""" url = f"https://finance.naver.com/item/main.naver?code={ticker}" try: resp = requests.get(url, headers=_HEADERS, timeout=5) resp.raise_for_status() soup = BeautifulSoup(resp.content, "html.parser", from_encoding="cp949") tag = soup.select_one(".no_today .blind") if tag: price_str = tag.get_text(strip=True).replace(",", "") return int(price_str) if price_str.isdigit() else None return None except Exception: return None def get_current_price(ticker: str) -> Optional[int]: """단건 현재가 조회 (3분 캐시)""" now = time.time() cached = _cache.get(ticker) if cached and (now - cached[1]) < _CACHE_TTL: return cached[0] price = _fetch_from_mobile_api(ticker) if price is None: price = _fetch_from_html_fallback(ticker) _cache[ticker] = (price, now) return price def get_current_prices(tickers: list[str]) -> dict[str, Optional[int]]: """배치 현재가 조회 (캐시 미스 종목만 실제 호출)""" result: dict[str, Optional[int]] = {} for ticker in tickers: result[ticker] = get_current_price(ticker) return result