Foresttrip Vacancy

#!/usr/bin/env python3
"""Read-only foresttrip.go.kr vacancy lookup helper.

The script logs in with Playwright to obtain a CSRF token and session cookies,
extracts forest IDs from the official monthly reservation status page, then
queries the read-only monthly availability JSON endpoint.

It intentionally does not click booking buttons, submit reservation forms,
handle payment, solve captcha, or bypass queues.
"""
from __future__ import annotations

import argparse
import concurrent.futures
import json
import os
import sys
import time
import urllib.error
import urllib.request
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

LOGIN_URL = "https://www.foresttrip.go.kr/com/login.do"
RSRVT_PAGE = "https://www.foresttrip.go.kr/rep/or/sssn/monthRsrvtSmplStatus.do"
POST_URL = "https://www.foresttrip.go.kr/rep/or/selectRsrvtAvailInfoListForMonthRsrvtSmpl.do"
DEFAULT_CONCURRENCY = 4
MAX_CONCURRENCY = 5
DEFAULT_WEEK_RANGE = 1
CATEGORY_CODES = {"01", "02"}
RESERVE_ROOM_MARKER = "예비"


@dataclass
class Session:
    cookies: dict[str, str]
    csrf: str
    user_agent: str
    forests: dict[str, str]
    expires_at: float


def parse_csv(value: str) -> list[str]:
    return [part.strip() for part in value.split(",") if part.strip()]


def parse_categories(value: str) -> tuple[str, ...]:
    categories = parse_csv(value)
    if not categories:
        raise argparse.ArgumentTypeError("must include at least one category code")
    invalid = [category for category in categories if category not in CATEGORY_CODES]
    if invalid:
        raise argparse.ArgumentTypeError(
            "unknown category code(s): "
            + ", ".join(invalid)
            + " (allowed: 01=lodging, 02=camping)"
        )
    return tuple(dict.fromkeys(categories))


def parse_dates(value: str) -> tuple[str, ...]:
    dates = parse_csv(value)
    if not dates:
        raise argparse.ArgumentTypeError("must include at least one YYYYMMDD date")

    today = datetime.now().date()
    normalized: list[str] = []
    for raw_date in dates:
        try:
            parsed = datetime.strptime(raw_date, "%Y%m%d").date()
        except ValueError as exc:
            raise argparse.ArgumentTypeError(f"invalid YYYYMMDD date: {raw_date}") from exc
        if parsed.strftime("%Y%m%d") != raw_date:
            raise argparse.ArgumentTypeError(f"invalid YYYYMMDD date: {raw_date}")
        if parsed < today:
            raise argparse.ArgumentTypeError(f"date is in the past: {raw_date}")
        normalized.append(raw_date)
    return tuple(sorted(dict.fromkeys(normalized)))


def parse_concurrency(value: str) -> int:
    try:
        concurrency = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("must be an integer") from exc
    if not 1 <= concurrency <= MAX_CONCURRENCY:
        raise argparse.ArgumentTypeError(f"must be between 1 and {MAX_CONCURRENCY}")
    return concurrency


def parse_week_range(value: str) -> int:
    try:
        week_range = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("must be an integer") from exc
    if week_range < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return week_range


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Read-only foresttrip.go.kr vacancy lookup.",
    )
    target = parser.add_argument_group("target selection")
    target.add_argument("--all", action="store_true", help="Scan all extracted forest IDs.")
    target.add_argument(
        "--forest-id",
        action="append",
        help="ForestTrip insttId. Can be passed multiple times or comma-separated.",
    )
    target.add_argument(
        "--forest-name",
        action="append",
        help="Substring to match against official forest names.",
    )

    output = parser.add_mutually_exclusive_group()
    output.add_argument("--json", action="store_true", help="Print JSON output.")
    output.add_argument("--text", action="store_true", help="Print human-readable output.")
    parser.add_argument("--dates", type=parse_dates, help="Comma-separated YYYYMMDD dates.")
    parser.add_argument(
        "--categories",
        type=parse_categories,
        default=("01", "02"),
        help="Comma-separated category codes: 01=lodging, 02=camping.",
    )
    parser.add_argument(
        "--concurrency",
        type=parse_concurrency,
        default=DEFAULT_CONCURRENCY,
        help=f"Parallel POST workers, 1-{MAX_CONCURRENCY}.",
    )
    parser.add_argument("--week-range", type=parse_week_range, help="Weeks ahead to scan when --dates is omitted.")
    parser.add_argument("--refresh-session", action="store_true", help="Ignore session cache.")
    parser.add_argument("--check-deps", action="store_true", help="Check Python and Playwright runtime dependencies.")
    parser.add_argument(
        "--session-cache",
        default="~/.cache/k-skill/foresttrip-vacancy/session.json",
        help="Session cache path.",
    )
    args = parser.parse_args()
    if args.all and (args.forest_id or args.forest_name):
        parser.error("--all cannot be combined with --forest-id or --forest-name")
    if args.dates and args.week_range is not None:
        parser.error("--week-range cannot be combined with --dates; the lookup range is derived from --dates")
    return args


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise SystemExit(f"missing required environment variable: {name}")
    return value


def check_dependencies(*, launch_browser: bool = True) -> None:
    if sys.version_info < (3, 9):
        raise SystemExit("python 3.9+ is required")
    try:
        from playwright.sync_api import Error as PlaywrightError  # type: ignore[reportMissingImports]
        from playwright.sync_api import sync_playwright  # type: ignore[reportMissingImports]
    except ImportError as exc:
        raise SystemExit(
            "playwright is required. Install with: python3 -m pip install playwright"
        ) from exc

    if not launch_browser:
        return

    try:
        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True)
            browser.close()
    except PlaywrightError as exc:
        raise SystemExit(
            "playwright chromium browser is required. Install with: "
            "python3 -m playwright install chromium"
        ) from exc


def load_session_cache(path: Path) -> Session | None:
    if not path.exists():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        if time.time() > float(data.get("expires_at", 0)):
            return None
        return Session(
            cookies=dict(data["cookies"]),
            csrf=str(data["csrf"]),
            user_agent=str(data["user_agent"]),
            forests=dict(data["forests"]),
            expires_at=float(data["expires_at"]),
        )
    except Exception:
        return None


def save_session_cache(path: Path, session: Session) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(asdict(session), ensure_ascii=False), encoding="utf-8")
    try:
        path.chmod(0o600)
    except OSError:
        pass


def bootstrap_session(*, forest_id: str, forest_pw: str, ttl_sec: int = 600) -> Session:
    try:
        from playwright.sync_api import Error as PlaywrightError  # type: ignore[reportMissingImports]
        from playwright.sync_api import sync_playwright  # type: ignore[reportMissingImports]
    except ImportError as exc:
        raise SystemExit(
            "playwright is required. Install with: python3 -m pip install playwright "
            "&& python3 -m playwright install chromium"
        ) from exc

    with sync_playwright() as p:
        try:
            browser = p.chromium.launch(headless=True)
        except PlaywrightError as exc:
            raise SystemExit(
                "playwright chromium browser is required. Install with: "
                "python3 -m playwright install chromium"
            ) from exc
        page = browser.new_page()
        page.goto(LOGIN_URL)
        page.fill("#mmberId", forest_id)
        page.fill("#gnrlMmberPssrd", forest_pw)
        page.click("input.loginBtn")
        page.wait_for_load_state("networkidle")
        page.goto(RSRVT_PAGE)
        page.wait_for_load_state("networkidle")

        csrf_locator = page.locator('input[name="_csrf"]')
        if csrf_locator.count() == 0:
            browser.close()
            raise SystemExit("login succeeded page did not expose a CSRF token")
        csrf = csrf_locator.first.get_attribute("value") or ""

        forests: dict[str, str] = {}
        regions = page.evaluate(
            """
            () => Array.from(document.querySelector('#srchSido').options)
              .slice(1)
              .map(o => ({ value: o.value, text: o.textContent.trim() }))
            """
        )
        for region in regions:
            value = region.get("value")
            if not value:
                continue
            page.select_option("#srchSido", value=value)
            page.wait_for_timeout(500)
            options = page.evaluate(
                """
                () => Array.from(document.querySelector('#srchInstt').options)
                  .slice(1)
                  .map(o => ({ value: o.value, text: o.textContent.trim() }))
                """
            )
            for opt in options:
                fid = str(opt.get("value") or "").strip()
                name = str(opt.get("text") or "").strip()
                if fid and name:
                    forests[fid] = name

        cookies = {cookie["name"]: cookie["value"] for cookie in page.context.cookies()}
        user_agent = page.evaluate("() => navigator.userAgent")
        browser.close()

    if not csrf or not cookies:
        raise SystemExit("failed to bootstrap foresttrip session")
    if not forests:
        raise SystemExit("failed to extract forest list from reservation page")
    return Session(
        cookies=cookies,
        csrf=csrf,
        user_agent=user_agent,
        forests=forests,
        expires_at=time.time() + ttl_sec,
    )


def get_session(args: argparse.Namespace) -> Session:
    cache_path = Path(args.session_cache).expanduser()
    if not args.refresh_session:
        cached = load_session_cache(cache_path)
        if cached is not None:
            return cached
    session = bootstrap_session(
        forest_id=require_env("KSKILL_FORESTTRIP_ID"),
        forest_pw=require_env("KSKILL_FORESTTRIP_PASSWORD"),
    )
    save_session_cache(cache_path, session)
    return session


def split_csv(values: list[str] | None) -> list[str]:
    out: list[str] = []
    for value in values or []:
        out.extend(part.strip() for part in value.split(",") if part.strip())
    return out


def resolve_targets(args: argparse.Namespace, forests: dict[str, str]) -> dict[str, str]:
    if args.all:
        return dict(sorted(forests.items(), key=lambda item: item[1]))

    requested_ids = split_csv(args.forest_id)
    requested_names = split_csv(args.forest_name)
    targets: dict[str, str] = {}
    for fid in requested_ids:
        targets[fid] = forests.get(fid, fid)
    for needle in requested_names:
        matches = {
            fid: name
            for fid, name in forests.items()
            if needle.replace(" ", "") in name.replace(" ", "")
        }
        targets.update(matches)

    if not targets:
        raise SystemExit("choose a target with --all, --forest-id, or --forest-name")
    return dict(sorted(targets.items(), key=lambda item: item[1]))


def build_headers(session: Session) -> dict[str, str]:
    cookie_header = "; ".join(f"{k}={v}" for k, v in session.cookies.items())
    return {
        "Accept": "application/json, text/javascript, */*; q=0.01",
        "Accept-Language": "ko-KR,ko;q=0.9",
        "Content-Type": "application/json; charset=UTF-8",
        "Cookie": cookie_header,
        "Origin": "https://www.foresttrip.go.kr",
        "Referer": RSRVT_PAGE,
        "User-Agent": session.user_agent,
        "X-CSRF-Token": session.csrf,
        "X-Requested-With": "XMLHttpRequest",
    }


def fetch_one(
    *,
    session: Session,
    forest_id: str,
    category: str,
    today: str,
    last_day: str,
) -> tuple[str, str, list[dict[str, Any]] | None, str | None]:
    payload = {
        "insttId": forest_id,
        "upperGoodsClsscCd": category,
        "srchDate": today,
        "lastDay": last_day,
        "inqurSctin": "02",
    }
    body = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        POST_URL,
        data=body,
        headers=build_headers(session),
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            if response.status != 200:
                return forest_id, category, None, f"http_{response.status}"
            data = json.loads(response.read().decode("utf-8"))
            if isinstance(data, list):
                return forest_id, category, data, None
            return forest_id, category, None, "unexpected_payload"
    except urllib.error.HTTPError as exc:
        return forest_id, category, None, f"http_{exc.code}"
    except Exception as exc:
        return forest_id, category, None, str(exc)


def is_available(row: dict[str, Any]) -> bool:
    count_value = row.get("rsrvtCnt")
    if count_value is None:
        return False
    try:
        reserved_count = int(count_value)
    except ValueError:
        return False
    return row.get("rsrvtAvail") == "Y" and reserved_count == 0


def is_reserve_room(row: dict[str, Any]) -> bool:
    return RESERVE_ROOM_MARKER in (row.get("goodsNm") or "")


def normalize_row(row: dict[str, Any], forests: dict[str, str]) -> dict[str, Any]:
    instt_id = str(row.get("insttId") or "")
    return {
        "forest_id": instt_id,
        "forest": forests.get(instt_id, row.get("insttNm") or instt_id),
        "use_dt": row.get("useDt") or "",
        "day": row.get("dywkDtTpcd"),
        "name": row.get("goodsNm") or "",
        "area": row.get("insttArea"),
        "capacity": row.get("mxmmAccptCnt"),
        "category": row.get("goodsClsscNm"),
        "region": row.get("insttAreaNm"),
        "waiting_possible": row.get("wtngPssblYn"),
    }


def collect_results(
    *,
    session: Session,
    targets: dict[str, str],
    categories: tuple[str, ...],
    dates: tuple[str, ...] | None,
    week_range: int | None,
    concurrency: int,
) -> dict[str, Any]:
    now = datetime.now()
    today = now.strftime("%Y%m%d")
    last_day = (
        max(dates)
        if dates
        else (now + timedelta(weeks=week_range or DEFAULT_WEEK_RANGE)).strftime("%Y%m%d")
    )
    date_filter = set(dates) if dates else None
    failures: list[dict[str, str]] = []
    rows: list[dict[str, Any]] = []

    jobs = [
        (forest_id, category)
        for forest_id in targets
        for category in categories
    ]
    with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, concurrency)) as pool:
        futures = [
            pool.submit(
                fetch_one,
                session=session,
                forest_id=forest_id,
                category=category,
                today=today,
                last_day=last_day,
            )
            for forest_id, category in jobs
        ]
        for future in concurrent.futures.as_completed(futures):
            forest_id, category, data, error = future.result()
            if error is not None or data is None:
                failures.append({"forest_id": forest_id, "category": category, "error": error or "unknown"})
                continue
            for row in data:
                if not is_available(row):
                    continue
                if is_reserve_room(row):
                    continue
                use_dt = row.get("useDt") or ""
                if use_dt < today or use_dt > last_day:
                    continue
                normalized = normalize_row(row, session.forests)
                normalized["source_category"] = category
                if date_filter is not None and normalized["use_dt"] not in date_filter:
                    continue
                rows.append(normalized)

    seen: set[tuple[str, str, str, str]] = set()
    deduped: list[dict[str, Any]] = []
    for row in rows:
        key = (row["forest_id"], row["use_dt"], row["source_category"], row["name"])
        if key in seen:
            continue
        seen.add(key)
        deduped.append(row)
    rows = deduped

    grouped: dict[str, dict[str, list[dict[str, Any]]]] = {}
    for row in sorted(rows, key=lambda item: (item["forest"], item["use_dt"], item["name"])):
        grouped.setdefault(row["forest"], {}).setdefault(row["use_dt"], []).append(row)

    return {
        "forests_scanned": len(targets),
        "filter_hits": len(rows),
        "fetch_failures": len(failures),
        "failures": failures[:20],
        "concurrency": concurrency,
        "date_range": {"from": today, "to": last_day},
        "results": [
            {
                "forest": forest_name,
                "dates": [
                    {"use_dt": use_dt, "rooms": rooms}
                    for use_dt, rooms in sorted(rows_by_date.items())
                ],
            }
            for forest_name, rows_by_date in sorted(grouped.items())
        ],
    }


def print_text(payload: dict[str, Any]) -> None:
    print("=== ForestTrip Vacancy Lookup ===")
    print(
        f"filter_hits: {payload['filter_hits']}   "
        f"fetch_failures: {payload['fetch_failures']}   "
        f"forests_scanned: {payload['forests_scanned']}"
    )
    if not payload["results"]:
        print("(no available rooms at lookup time)")
        return
    for forest in payload["results"]:
        print(f"\n{forest['forest']}")
        for date_group in forest["dates"]:
            rooms = date_group["rooms"]
            print(f"  {date_group['use_dt']} - {len(rooms)} slot(s)")
            for room in rooms[:8]:
                capacity = room["capacity"] if room["capacity"] is not None else "?"
                area = room["area"] if room["area"] is not None else "?"
                print(f"    - {room['name']} / {room['category']} / {area}sqm / max {capacity}")


def main() -> int:
    args = parse_args()
    if args.check_deps:
        check_dependencies()
        print("foresttrip-vacancy dependencies look ready")
        return 0
    session = get_session(args)
    targets = resolve_targets(args, session.forests)
    payload = collect_results(
        session=session,
        targets=targets,
        categories=args.categories,
        dates=args.dates,
        week_range=args.week_range,
        concurrency=args.concurrency,
    )

    if args.text and not args.json:
        print_text(payload)
    else:
        print(json.dumps(payload, ensure_ascii=False, indent=2))
    return 0 if payload["fetch_failures"] == 0 else 1


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

[
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260513",
    "dywkDtTpcd": "수",
    "goodsNm": "동백1",
    "goodsId": "GID-A1",
    "insttArea": "50㎡",
    "mxmmAccptCnt": "8",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260513",
    "dywkDtTpcd": "수",
    "goodsNm": "동백1",
    "goodsId": "GID-A2",
    "insttArea": "50㎡",
    "mxmmAccptCnt": "8",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260513",
    "dywkDtTpcd": "수",
    "goodsNm": "해송2",
    "goodsId": "GID-B",
    "insttArea": "50㎡",
    "mxmmAccptCnt": "8",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260513",
    "dywkDtTpcd": "수",
    "goodsNm": "고로쇠1",
    "goodsId": "GID-C",
    "insttArea": "60㎡",
    "mxmmAccptCnt": "10",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260513",
    "dywkDtTpcd": "수",
    "goodsNm": "(예비) 201호",
    "goodsId": "GID-RES1",
    "insttArea": "32㎡",
    "mxmmAccptCnt": "4",
    "goodsClsscNm": "휴양관",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260514",
    "dywkDtTpcd": "목",
    "goodsNm": "동백3",
    "goodsId": "GID-D",
    "insttArea": "50㎡",
    "mxmmAccptCnt": "8",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260516",
    "dywkDtTpcd": "토",
    "goodsNm": "동백1",
    "goodsId": "GID-A1",
    "insttArea": "50㎡",
    "mxmmAccptCnt": "8",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 1
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260516",
    "dywkDtTpcd": "토",
    "goodsNm": "해송2",
    "goodsId": "GID-B",
    "insttArea": "50㎡",
    "mxmmAccptCnt": "8",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 1
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260516",
    "dywkDtTpcd": "토",
    "goodsNm": "(예비) 201호",
    "goodsId": "GID-RES1",
    "insttArea": "32㎡",
    "mxmmAccptCnt": "4",
    "goodsClsscNm": "휴양관",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260517",
    "dywkDtTpcd": "일",
    "goodsNm": "중산막2",
    "goodsId": "GID-E",
    "insttArea": "32㎡",
    "mxmmAccptCnt": "4",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030059",
    "insttNm": "거제자연휴양림",
    "useDt": "20260517",
    "dywkDtTpcd": "일",
    "goodsNm": "동백3",
    "goodsId": "GID-F",
    "insttArea": "50㎡",
    "mxmmAccptCnt": "8",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  }
]

[
  {
    "insttId": "ID02030072",
    "insttNm": "구재봉자연휴양림",
    "useDt": "20260516",
    "dywkDtTpcd": "토",
    "goodsNm": "206호 쑥부쟁이방",
    "goodsId": "GID-G1",
    "insttArea": "37㎡",
    "mxmmAccptCnt": "8",
    "goodsClsscNm": "숲속휴양관",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030072",
    "insttNm": "구재봉자연휴양림",
    "useDt": "20260516",
    "dywkDtTpcd": "토",
    "goodsNm": "201호 배꽃방(예비용)",
    "goodsId": "GID-G2",
    "insttArea": "30㎡",
    "mxmmAccptCnt": "6",
    "goodsClsscNm": "숲속휴양관",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030072",
    "insttNm": "구재봉자연휴양림",
    "useDt": "20260516",
    "dywkDtTpcd": "토",
    "goodsNm": "편백나무2호(예비용)",
    "goodsId": "GID-G3",
    "insttArea": "28㎡",
    "mxmmAccptCnt": "6",
    "goodsClsscNm": "숲속의집",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  },
  {
    "insttId": "ID02030072",
    "insttNm": "구재봉자연휴양림",
    "useDt": "20260516",
    "dywkDtTpcd": "토",
    "goodsNm": "은행나무방(예비용)",
    "goodsId": "GID-G4",
    "insttArea": "22㎡",
    "mxmmAccptCnt": "4",
    "goodsClsscNm": "트리하우스",
    "insttAreaNm": null,
    "wtngPssblYn": "Y",
    "rsrvtAvail": "Y",
    "rsrvtCnt": 0
  }
]

import importlib.util
import io
import json
import sys
import unittest
from contextlib import redirect_stdout
from datetime import datetime
from pathlib import Path
from unittest import mock


SCRIPT_DIR = Path(__file__).resolve().parent
HELPER_PATH = SCRIPT_DIR.parent / "scripts" / "run_foresttrip_vacancy.py"
FIXTURES_DIR = SCRIPT_DIR / "fixtures"


def load_helper():
    spec = importlib.util.spec_from_file_location("run_foresttrip_vacancy", HELPER_PATH)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"cannot load helper from {HELPER_PATH}")
    module = importlib.util.module_from_spec(spec)
    sys.modules["run_foresttrip_vacancy"] = module
    spec.loader.exec_module(module)
    return module


helper = load_helper()


def load_fixture(name):
    return json.loads((FIXTURES_DIR / name).read_text(encoding="utf-8"))


GEOJE_ROWS = load_fixture("geoje_window.json")
GUJAEBONG_ROWS = load_fixture("gujaebong_window.json")

GEOJE_FOREST_ID = "ID02030059"
GEOJE_FOREST_NAME = "[공립](거제시)거제자연휴양림"
GUJAEBONG_FOREST_ID = "ID02030072"
GUJAEBONG_FOREST_NAME = "[공립](하동군)구재봉자연휴양림"

FIXED_NOW = datetime(2026, 5, 12, 0, 0, 0)


def make_session(forests):
    return helper.Session(
        cookies={},
        csrf="dummy-csrf",
        user_agent="test-ua",
        forests=forests,
        expires_at=FIXED_NOW.timestamp() + 3600,
    )


def stub_fetch(rows):
    def _stub(*, forest_id, category, **_):
        matched = [r for r in rows if r.get("insttId") == forest_id]
        return forest_id, category, matched, None
    return _stub


def run_collect(session, targets, rows, *, dates=None, week_range=None, categories=("01",)):
    with mock.patch.object(helper, "fetch_one", side_effect=stub_fetch(rows)):
        with mock.patch.object(helper, "datetime", wraps=datetime) as mock_dt:
            mock_dt.now.return_value = FIXED_NOW
            return helper.collect_results(
                session=session,
                targets=targets,
                categories=categories,
                dates=tuple(dates) if dates else None,
                week_range=week_range,
                concurrency=1,
            )


class IsReserveRoomTest(unittest.TestCase):
    def test_parens_with_suffix(self):
        self.assertTrue(helper.is_reserve_room({"goodsNm": "201호 배꽃방(예비용)"}))

    def test_parens_prefix(self):
        self.assertTrue(helper.is_reserve_room({"goodsNm": "(예비) 201호"}))

    def test_predicate_with_simple_suffix(self):
        self.assertTrue(helper.is_reserve_room({"goodsNm": "편백나무2호(예비용)"}))

    def test_normal_room_passes(self):
        self.assertFalse(helper.is_reserve_room({"goodsNm": "동백1"}))

    def test_empty_name(self):
        self.assertFalse(helper.is_reserve_room({"goodsNm": ""}))

    def test_missing_name_key(self):
        self.assertFalse(helper.is_reserve_room({}))


class IsAvailableTest(unittest.TestCase):
    def test_y_and_zero_count(self):
        self.assertTrue(helper.is_available({"rsrvtAvail": "Y", "rsrvtCnt": 0}))

    def test_y_and_string_zero_count(self):
        self.assertTrue(helper.is_available({"rsrvtAvail": "Y", "rsrvtCnt": "0"}))

    def test_y_but_already_booked(self):
        self.assertFalse(helper.is_available({"rsrvtAvail": "Y", "rsrvtCnt": 1}))

    def test_y_but_string_booked_count(self):
        self.assertFalse(helper.is_available({"rsrvtAvail": "Y", "rsrvtCnt": "1"}))

    def test_not_available(self):
        self.assertFalse(helper.is_available({"rsrvtAvail": "N", "rsrvtCnt": 0}))


class CollectResultsFilterTest(unittest.TestCase):
    def setUp(self):
        self.session = make_session({GEOJE_FOREST_ID: GEOJE_FOREST_NAME})
        self.targets = {GEOJE_FOREST_ID: GEOJE_FOREST_NAME}

    def test_geoje_5_13_three_unique_rooms_after_dedup_and_reserve_filter(self):
        payload = run_collect(self.session, self.targets, GEOJE_ROWS, dates=["20260513"])
        self.assertEqual(payload["filter_hits"], 3)
        names = {
            room["name"]
            for forest in payload["results"]
            for date in forest["dates"]
            for room in date["rooms"]
        }
        self.assertEqual(names, {"동백1", "해송2", "고로쇠1"})

    def test_geoje_5_16_returns_zero_when_only_reserved_or_booked(self):
        payload = run_collect(self.session, self.targets, GEOJE_ROWS, dates=["20260516"])
        self.assertEqual(payload["filter_hits"], 0)
        self.assertEqual(payload["results"], [])

    def test_geoje_5_17_two_rooms(self):
        payload = run_collect(self.session, self.targets, GEOJE_ROWS, dates=["20260517"])
        self.assertEqual(payload["filter_hits"], 2)
        names = {
            room["name"]
            for forest in payload["results"]
            for date in forest["dates"]
            for room in date["rooms"]
        }
        self.assertEqual(names, {"중산막2", "동백3"})

    def test_dates_outside_request_filtered_out(self):
        payload = run_collect(self.session, self.targets, GEOJE_ROWS, dates=["20260513"])
        observed_dates = {
            room["use_dt"]
            for forest in payload["results"]
            for date in forest["dates"]
            for room in date["rooms"]
        }
        self.assertEqual(observed_dates, {"20260513"})

    def test_reserve_rooms_excluded_across_all_dates(self):
        payload = run_collect(
            self.session, self.targets, GEOJE_ROWS,
            dates=["20260513", "20260516", "20260517"],
        )
        for forest in payload["results"]:
            for date in forest["dates"]:
                for room in date["rooms"]:
                    self.assertNotIn("예비", room["name"])

    def test_dedup_collapses_duplicate_room_with_different_goods_id(self):
        payload = run_collect(self.session, self.targets, GEOJE_ROWS, dates=["20260513"])
        donbaek_count = sum(
            1
            for forest in payload["results"]
            for date in forest["dates"]
            for room in date["rooms"]
            if room["name"] == "동백1"
        )
        self.assertEqual(donbaek_count, 1)

    def test_dedup_keeps_same_room_name_from_distinct_categories(self):
        rows_by_category = {
            "01": [
                {
                    "insttId": GEOJE_FOREST_ID,
                    "insttNm": GEOJE_FOREST_NAME,
                    "useDt": "20260513",
                    "goodsNm": "같은이름",
                    "goodsClsscNm": "숙박",
                    "rsrvtAvail": "Y",
                    "rsrvtCnt": 0,
                }
            ],
            "02": [
                {
                    "insttId": GEOJE_FOREST_ID,
                    "insttNm": GEOJE_FOREST_NAME,
                    "useDt": "20260513",
                    "goodsNm": "같은이름",
                    "goodsClsscNm": "야영",
                    "rsrvtAvail": "Y",
                    "rsrvtCnt": 0,
                }
            ],
        }

        def fetch_category(*, forest_id, category, **_):
            return forest_id, category, rows_by_category[category], None

        with mock.patch.object(helper, "fetch_one", side_effect=fetch_category):
            with mock.patch.object(helper, "datetime", wraps=datetime) as mock_dt:
                mock_dt.now.return_value = FIXED_NOW
                payload = helper.collect_results(
                    session=self.session,
                    targets=self.targets,
                    categories=("01", "02"),
                    dates=("20260513",),
                    week_range=None,
                    concurrency=1,
                )

        self.assertEqual(payload["filter_hits"], 2)
        observed = [
            (room["name"], room["category"])
            for forest in payload["results"]
            for date in forest["dates"]
            for room in date["rooms"]
        ]
        self.assertEqual(observed, [("같은이름", "숙박"), ("같은이름", "야영")])


class StrictUseDtGateTest(unittest.TestCase):
    """Bug 1 regression: API returns 5-day window even when single-day requested."""

    def test_useDt_before_today_blocked_even_if_available(self):
        past_row = dict(GEOJE_ROWS[0])
        past_row["useDt"] = "20260101"
        rows = [past_row]
        session = make_session({GEOJE_FOREST_ID: GEOJE_FOREST_NAME})
        payload = run_collect(
            session, {GEOJE_FOREST_ID: GEOJE_FOREST_NAME}, rows,
            week_range=1,
        )
        self.assertEqual(payload["filter_hits"], 0)

    def test_useDt_after_last_day_blocked(self):
        far_future = dict(GEOJE_ROWS[0])
        far_future["useDt"] = "20300101"
        rows = [far_future]
        session = make_session({GEOJE_FOREST_ID: GEOJE_FOREST_NAME})
        payload = run_collect(
            session, {GEOJE_FOREST_ID: GEOJE_FOREST_NAME}, rows,
            week_range=1,
        )
        self.assertEqual(payload["filter_hits"], 0)


class PrintTextTest(unittest.TestCase):
    """print_text is the user-facing output path — guard against format regressions."""

    def test_renders_forest_and_rooms(self):
        session = make_session({GEOJE_FOREST_ID: GEOJE_FOREST_NAME})
        payload = run_collect(
            session, {GEOJE_FOREST_ID: GEOJE_FOREST_NAME}, GEOJE_ROWS,
            dates=["20260513"],
        )
        buffer = io.StringIO()
        with redirect_stdout(buffer):
            helper.print_text(payload)
        output = buffer.getvalue()
        self.assertIn(GEOJE_FOREST_NAME, output)
        self.assertIn("20260513", output)
        self.assertIn("slot(s)", output)
        self.assertIn("동백1", output)

    def test_empty_results_message(self):
        session = make_session({GEOJE_FOREST_ID: GEOJE_FOREST_NAME})
        payload = run_collect(
            session, {GEOJE_FOREST_ID: GEOJE_FOREST_NAME}, GEOJE_ROWS,
            dates=["20260516"],
        )
        buffer = io.StringIO()
        with redirect_stdout(buffer):
            helper.print_text(payload)
        self.assertIn("(no available rooms at lookup time)", buffer.getvalue())


class MainOutputTest(unittest.TestCase):
    def run_main(self, argv, payload):
        session = make_session({GEOJE_FOREST_ID: GEOJE_FOREST_NAME})
        buffer = io.StringIO()
        with mock.patch.object(sys, "argv", ["run_foresttrip_vacancy.py", *argv]):
            with mock.patch.object(helper, "get_session", return_value=session):
                with mock.patch.object(helper, "resolve_targets", return_value={GEOJE_FOREST_ID: GEOJE_FOREST_NAME}):
                    with mock.patch.object(helper, "collect_results", return_value=payload):
                        with redirect_stdout(buffer):
                            exit_code = helper.main()
        return exit_code, buffer.getvalue()

    def test_main_text_output_returns_success_when_fetches_succeed(self):
        payload = run_collect(
            make_session({GEOJE_FOREST_ID: GEOJE_FOREST_NAME}),
            {GEOJE_FOREST_ID: GEOJE_FOREST_NAME},
            GEOJE_ROWS,
            dates=["20260513"],
        )
        exit_code, output = self.run_main(["--forest-id", GEOJE_FOREST_ID, "--text"], payload)

        self.assertEqual(exit_code, 0)
        self.assertIn("ForestTrip Vacancy Lookup", output)
        self.assertIn("filter_hits: 3", output)
        self.assertIn("동백1", output)

    def test_main_json_output_reports_failure_and_returns_nonzero(self):
        payload = {
            "forests_scanned": 1,
            "filter_hits": 0,
            "fetch_failures": 1,
            "failures": [{"forest_id": GEOJE_FOREST_ID, "category": "01", "error": "http_401"}],
            "concurrency": 1,
            "date_range": {"from": "20260512", "to": "20260513"},
            "results": [],
        }
        exit_code, output = self.run_main(["--forest-id", GEOJE_FOREST_ID, "--json"], payload)

        self.assertEqual(exit_code, 1)
        rendered = json.loads(output)
        self.assertEqual(rendered["fetch_failures"], 1)
        self.assertEqual(rendered["failures"][0]["error"], "http_401")


class GroundTruthTest(unittest.TestCase):
    """Anchored to user-verified counts from foresttrip.go.kr on 2026-05-12.
    Fixtures are simplified; tests assert the per-(forest, date) shape matches."""

    def test_gujaebong_5_16_one_room_named_쑥부쟁이방(self):
        session = make_session({GUJAEBONG_FOREST_ID: GUJAEBONG_FOREST_NAME})
        payload = run_collect(
            session, {GUJAEBONG_FOREST_ID: GUJAEBONG_FOREST_NAME}, GUJAEBONG_ROWS,
            dates=["20260516"],
        )
        self.assertEqual(payload["filter_hits"], 1)
        names = [
            room["name"]
            for forest in payload["results"]
            for date in forest["dates"]
            for room in date["rooms"]
        ]
        self.assertEqual(names, ["206호 쑥부쟁이방"])


if __name__ == "__main__":
    unittest.main()