starbucks_crawler.py

#!/usr/bin/env python3
"""Crawl Starbucks Korea stores and print name/address/business hours."""

from __future__ import annotations

import json
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from datetime import datetime


STORE_LIST_URL = "https://www.starbucks.co.kr/store/getStore.do"
STORE_TIME_URL = "https://www.starbucks.co.kr/store/getStoreTime.do"
STORE_SIDO_URL = "https://www.starbucks.co.kr/store/getSidoList.do"

HEADERS = {
    "User-Agent": "Mozilla/5.0",
    "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
    "Referer": "https://www.starbucks.co.kr/store/store_map.do",
}

WEEKDAY_LABELS = {
    "1": "일",
    "2": "월",
    "3": "화",
    "4": "수",
    "5": "목",
    "6": "금",
    "7": "토",
}


@dataclass(frozen=True)
class Store:
    biz_code: str
    name: str
    address: str


def post_json(
    url: str, payload: dict[str, str], retries: int = 3, timeout: int = 20
) -> dict:
    encoded = urllib.parse.urlencode(payload).encode("utf-8")

    for attempt in range(1, retries + 1):
        request = urllib.request.Request(
            url, data=encoded, method="POST", headers=HEADERS
        )
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return json.loads(response.read().decode("utf-8"))
        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError):
            if attempt == retries:
                raise
            time.sleep(0.4 * attempt)

    raise RuntimeError("unreachable")


def format_time_range(raw: str | None, holiday_tag: str | None) -> str:
    if holiday_tag == "4":
        return "휴점"
    if not raw or "-" not in raw:
        return "정보없음"

    start, end = raw.split("-", maxsplit=1)
    start = start.strip()
    end = end.strip()

    if len(start) < 4 or len(end) < 4:
        return "정보없음"

    return f"{start[:2]}:{start[2:4]}-{end[:2]}:{end[2:4]}"


def get_sido_map() -> dict[str, str]:
    response = post_json(STORE_SIDO_URL, {})
    return {
        str(row.get("sido_nm", "")).strip(): str(row.get("sido_cd", "")).strip()
        for row in response.get("list", [])
        if str(row.get("sido_nm", "")).strip() and str(row.get("sido_cd", "")).strip()
    }


def ask_region_code(sido_map: dict[str, str]) -> tuple[str, str]:
    regions = sorted(sido_map.keys())
    print("조회할 지역(시/도)을 입력하세요.")
    print("가능 지역: " + ", ".join(regions))

    normalized_map = {"".join(name.split()): name for name in regions}

    while True:
        user_input = input("지역명: ").strip()
        normalized = "".join(user_input.split())
        region_name = normalized_map.get(normalized)
        if region_name:
            return region_name, sido_map[region_name]
        print("유효하지 않은 지역입니다. 위 목록 중 하나를 다시 입력해 주세요.")


def get_all_stores(sido_code: str) -> list[Store]:
    payload = {
        "ins_lat": "37.56682",
        "ins_lng": "126.97865",
        "search_text": "",
        "p_sido_cd": sido_code,
        "p_gugun_cd": "",
        "in_distance": "0",
        "in_biz_cd": "",
        "iend": "1000",
        "searchType": "A",
        "in_biz_cds": "0",
        "in_scodes": "0",
        "set_date": "",
        "T03": "0",
        "T01": "0",
        "T12": "0",
        "T09": "0",
        "T06": "0",
        "T10": "0",
        "P10": "0",
        "P50": "0",
        "P20": "0",
        "P60": "0",
        "P30": "0",
        "P70": "0",
        "P40": "0",
        "P80": "0",
        "T20": "0",
        "T22": "0",
        "T29": "0",
        "T30": "0",
        "T36": "0",
        "T43": "0",
        "T48": "0",
        "new_bool": "0",
        "T64": "0",
        "T66": "0",
    }
    response = post_json(STORE_LIST_URL, payload)

    stores: list[Store] = []
    seen: set[str] = set()

    for row in response.get("list", []):
        biz_code = str(row.get("s_biz_code", "")).strip()
        if not biz_code or biz_code in seen:
            continue

        seen.add(biz_code)
        doro_address = str(row.get("doro_address", "")).strip()
        address = doro_address or str(row.get("addr", "")).strip()
        stores.append(
            Store(
                biz_code=biz_code,
                name=str(row.get("s_name", "")).strip(),
                address=address,
            )
        )

    return stores


def get_today_week_code() -> str:
    weekday = datetime.now().weekday()
    return str(((weekday + 1) % 7) + 1)


def get_today_store_hours(biz_code: str) -> str:
    payload = {
        "in_biz_cd": biz_code,
        "in_store_type": "C",
    }
    response = post_json(STORE_TIME_URL, payload)
    rows = response.get("list", [])
    if not rows:
        return "정보없음"

    today_code = get_today_week_code()
    today_row = next(
        (
            row
            for row in rows
            if str(row.get("store_time_week", "")).strip() == today_code
        ),
        rows[0],
    )
    return format_time_range(
        str(today_row.get("store_opentime", "")).strip()
        if today_row.get("store_opentime") is not None
        else None,
        str(today_row.get("store_time_hlytag", "")).strip()
        if today_row.get("store_time_hlytag") is not None
        else None,
    )


def main() -> int:
    try:
        sido_map = get_sido_map()
        region_name, region_code = ask_region_code(sido_map)
        stores = get_all_stores(region_code)
    except Exception as error:  # noqa: BLE001
        print(f"매장 목록 조회 실패: {error}", file=sys.stderr)
        return 1

    if not stores:
        print(f"{region_name} 지역 매장을 찾지 못했습니다.", file=sys.stderr)
        return 1

    rows: list[tuple[str, str, str]] = []
    with ThreadPoolExecutor(max_workers=24) as executor:
        future_map = {
            executor.submit(get_today_store_hours, store.biz_code): store
            for store in stores
        }
        for future in as_completed(future_map):
            store = future_map[future]
            try:
                hours = future.result()
            except Exception:
                hours = "정보없음"
            rows.append((store.name, store.address, hours))

    rows.sort(key=lambda row: row[0])

    today_label = WEEKDAY_LABELS.get(get_today_week_code(), "오늘")
    print(f"조회 지역: {region_name}")
    print(f"매장명\t주소\t{today_label} 영업시간")
    for name, address, hours in rows:
        print(f"{name}\t{address}\t{hours}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())