instruction.md
# Foresttrip Vacancy
## What this skill does
숲나들e 공식 사이트(`https://foresttrip.go.kr/index.jsp`)에서 자연휴양림 예약 가능 객실을 날짜 기준으로 조회 자동화한다.
generic runtime에서 helper는 조회 전용이다. 돌쇠에서는 조회 결과를 공식 CloakBrowser 예약 흐름으로 이어가며, CAPTCHA·대기열 우회·공격적 반복 스나이핑은 여전히 하지 않는다.
## When to use
- "이번 주말 자연휴양림 빈 객실 있어?"
- "숲나들e 2026년 5월 4일 예약 가능한 곳 조회해줘"
- "자연휴양림 빈자리 전체 조회해줘"
- "관심 휴양림 중 예약 가능한 객실만 알려줘"
## When not to use
- 돌쇠가 아니며 예약 신청이나 결제까지 자동화해야 하는 경우
- 캡차를 풀거나 대기열을 우회해야 하는 경우
- 계정 정보를 채팅창에 직접 넣으려는 경우
- aggressive polling, 스나이핑, 반복 예약 시도가 필요한 경우
## Prerequisites
- Python 3.9+
- Playwright Chromium browser
```bash
python3 -m pip install playwright
python3 -m playwright install chromium
npx -y @nomadamas/k-skill@0 exec foresttrip-vacancy scripts/run_foresttrip_vacancy.py -- --check-deps
```
## Required environment variables
- `KSKILL_FORESTTRIP_ID`
- `KSKILL_FORESTTRIP_PASSWORD`
Optional:
- none
### Credential handling
- 돌쇠 credential mode에서는 `vault-run` capability를 사용하고, 없으면 `request_vault_credential`을 호출한다. ID/PW 원문을 채팅이나 shell에 넣지 않는다.
- 그 밖의 환경에서는 이미 주입된 환경변수 → host vault → `~/.config/k-skill/secrets.env` 순서로 사용한다.
- Generic helper 자체는 `KSKILL_FORESTTRIP_ID`, `KSKILL_FORESTTRIP_PASSWORD` 환경변수만 읽는다.
## Inputs
- 날짜: `YYYYMMDD`, 여러 날짜면 comma-separated `YYYYMMDD,YYYYMMDD`
- 조회 범위:
- `--all`: 전체 자연휴양림 조회
- `--forest-id`: 특정 `insttId` 조회
- `--forest-name`: 공식 휴양림명 부분 일치 조회
- 출력 형식:
- `--text`: 사람용 요약
- `--json`: 구조화 결과
- 선택 필터:
- `--categories 01`: 숙박
- `--categories 02`: 야영/캠핑
- `--categories 01,02`: 숙박 + 야영/캠핑
- 고급 실행 옵션:
- `--week-range N`: `--dates` 를 생략했을 때만 오늘부터 N주 범위를 조회
- `--concurrency N`: 병렬 조회 worker 수, 1-5 범위
- `--session-cache PATH`: 로그인 세션 캐시 경로 override
## Workflow
### 1. Ensure credentials are available
돌쇠 credential mode에서는 숲나들e capability를 사용하고, 없으면 `request_vault_credential`을 호출한다. generic fallback에서만 `KSKILL_FORESTTRIP_ID`, `KSKILL_FORESTTRIP_PASSWORD`를 확인한다.
시크릿이 없다는 이유로 대체 사이트, 캡차 우회, 비공식 예약 경로를 찾지 않는다.
### 2. Install runtime dependencies when missing
```bash
python3 -m pip install playwright
python3 -m playwright install chromium
```
### 3. Run a vacancy lookup
이 스킬의 helper를 통해 조회한다. Helper는 Playwright로 숲나들e에 로그인해 CSRF/cookie와 공식 휴양림 ID 목록을 얻은 뒤, 월별예약조회 JSON endpoint만 호출한다.
2026-04-29 확인 기준, 로그인 없이 월별예약조회 화면에 접근하면 `401 Unauthorized`가 반환되고, 조회 endpoint는 JSON 대신 안내 HTML을 반환한다. 따라서 현재 helper는 로그인 세션/CSRF 확보를 필수 전제로 둔다.
API는 `srchDate` 단일 일자만 요청해도 응답에 5일 윈도우를 포함할 수 있다. helper는 요청 범위(`today`–`last_day`) 밖 `useDt` 행을 자동 제거하므로 사용자에게는 요청한 날짜의 빈자리만 노출된다.
전체 자연휴양림에서 특정 날짜 조회:
```bash
npx -y @nomadamas/k-skill@0 exec foresttrip-vacancy scripts/run_foresttrip_vacancy.py -- --all --text --dates 20260504
```
JSON 출력:
```bash
npx -y @nomadamas/k-skill@0 exec foresttrip-vacancy scripts/run_foresttrip_vacancy.py -- --all --json --dates 20260504
```
캠핑/야영만 조회:
```bash
npx -y @nomadamas/k-skill@0 exec foresttrip-vacancy scripts/run_foresttrip_vacancy.py -- --all --text --dates 20260504 --categories 02
```
특정 휴양림명으로 조회:
```bash
npx -y @nomadamas/k-skill@0 exec foresttrip-vacancy scripts/run_foresttrip_vacancy.py -- --forest-name 유명산 --text --dates 20260504
```
### 4. Summarize results conservatively
응답은 아래 항목 중심으로 짧게 정리한다.
- 조회 날짜
- 조회 범위
- 예약 가능한 휴양림명
- 객실/시설명
- 숙박/야영 구분
- 정원 또는 수용 인원
- fetch failure가 있으면 실패 개수
결과가 없으면 "조회 시점 기준 예약 가능 객실 없음"이라고 말한다. 실제 예약 가능 여부는 숲나들e 화면에서 재확인될 수 있음을 덧붙인다.
`goodsNm`에 "예비"가 포함된 객실은 운영자가 보유하는 내부용 자리로, 사용자 예약 화면에는 노출되지 않는다. helper는 이 객실들을 결과에서 자동 제외한다. 같은 `(휴양림, 날짜, 객실명)` 조합의 중복 행도 dedup된다.
## Done when
- 요청 날짜와 조회 범위가 명확하다.
- read-only 월별예약조회 helper를 최소 1회 실행했다.
- 빈 객실이 있으면 날짜/휴양림/객실을 정리했다.
- 빈 객실이 없으면 없다고 명확히 말했다.
- 돌쇠의 예약 요청이면 공식 예약 흐름을 진행했고, 결제가 필요하면 `clarify` 승인 후 완료 상태를 확인했다.
- CAPTCHA/대기열 우회는 시도하지 않았다.
## Failure modes
- 로그인 실패: `KSKILL_FORESTTRIP_ID`, `KSKILL_FORESTTRIP_PASSWORD` 확인
- Playwright browser 미설치: `python3 -m playwright install chromium`
- fetch failure 일부 발생: 결과와 실패 개수를 함께 보고하고, 필요하면 `--refresh-session` 으로 1회 재조회
- 숲나들e 표면 변경: helper의 login/session bootstrap 또는 parser 점검 필요
- "(예비)" 객실이 결과에 안 나옴: 정상 동작이다. 사용자 예약 화면에 노출되지 않는 운영자 보유분이라 의도적으로 제외된다.
- 사용자 화면 객실 수와 helper 결과가 다름: 같은 객실의 중복 행이 dedup되었거나, 요청 범위 밖 `useDt`가 제거됐을 가능성이 높다. raw API 응답을 확인하려면 helper 로직을 우회해서 직접 호출 필요.
## Maintainer review notes
메인테이너가 이 스킬을 검토하기 위해 숲나들e 계정을 새로 만들 필요는 없다.
계정 없이 가능한 검증:
- `./scripts/validate-skills.sh`
- `python3 -m py_compile foresttrip-vacancy/scripts/run_foresttrip_vacancy.py`
- `npx -y @nomadamas/k-skill@0 exec foresttrip-vacancy scripts/run_foresttrip_vacancy.py -- --help`
- `npx -y @nomadamas/k-skill@0 exec foresttrip-vacancy scripts/run_foresttrip_vacancy.py -- --check-deps`
- `npm run ci`
실제 live smoke는 기여자 또는 이미 숲나들e 계정을 가진 사용자가 선택적으로 수행한다. PR에는 `forests_scanned`, `fetch_failures`, `filter_hits` 같은 비민감 요약만 남기고 계정 정보, 세션 쿠키, 개인 조회 세부 내역은 공유하지 않는다.
## Safety notes
- generic helper는 조회 전용이고, 돌쇠의 예약/결제는 CloakBrowser action path로 분리한다.
- 캡차 처리, 대기열 우회, 공격적인 반복 조회를 하지 않는다.
- 돌쇠에서는 vault action을 사용하고, generic fallback에서만 환경변수 또는 `~/.config/k-skill/secrets.env`를 사용한다.
references/DISCLAIMER.md
# DISCLAIMER — `foresttrip-vacancy`
## 비제휴 및 상표 사용
이 스킬은 이 문서에서 언급하는 상표권자나 서비스 운영사의 공식 기능 또는 공식 지원 도구가 아닙니다. 해당 권리자·운영사와 공식 제휴, 후원, 승인, 인증 또는 협업한 사실이 전혀 없습니다. 제3자 상표와 서비스명은 스킬의 기능, 조회 대상 또는 호환 대상을 정확히 설명하기 위해 필요한 범위에서만 사용하며, 각 상표의 권리는 해당 권리자에게 있습니다.
대법원 2005. 6. 10. 선고 [2005도1637 판결](https://www.law.go.kr/LSW/precInfoP.do?precSeq=83920)(소니용 리모컨 사건)은 타인의 표장을 자기 상품의 출처표시가 아니라 상품의 기능 또는 적용 기종을 설명하기 위해 사용하고, 실제 표시 태양도 상표 사용으로 인식될 수 없는 경우에는 상표권 침해가 아니라고 판시했습니다. 다만 이는 제3자 상표 사용을 포괄적으로 허용한다는 뜻이 아니며, [상표법 제2조](https://www.law.go.kr/법령/상표법/제2조), [제89조](https://www.law.go.kr/법령/상표법/제89조), [제90조](https://www.law.go.kr/법령/상표법/제90조), [제108조](https://www.law.go.kr/법령/상표법/제108조)와 실제 거래계의 인식에 따라 출처 혼동 여부를 개별적으로 판단해야 합니다.
## 공개정보에 대한 자동 접근
대법원 2022. 5. 12. 선고 [2021도1533 판결](https://www.law.go.kr/LSW/precInfoP.do?precSeq=221765)(야놀자 정보 수집 사건)은 서비스 제공자가 접근권한을 제한했는지를 보호조치·이용약관 등 객관적으로 드러난 사정을 종합해 판단해야 한다고 판시했습니다. 그 사건에서는 별도 보호조치 없이 일반 이용자가 접근할 수 있었고 이미 공개된 정보를 수집한 사정 등에 비추어 정보통신망 침입을 인정하지 않았습니다. 따라서 공개되어 누구나 얻을 수 있는 정보를 기계적으로 수집했다는 사정만으로 곧바로 [정보통신망법 제48조](https://www.law.go.kr/법령/정보통신망이용촉진및정보보호등에관한법률/제48조) 위반이 되는 것은 아닙니다.
그러나 위 판결은 크롤링을 일반적으로 허용한 판결이 아닙니다. 접근통제나 명시적 제한을 우회하는 행위, 데이터베이스의 전부·상당한 부분을 복제하거나 반복적·체계적으로 수집하여 같은 결과를 내는 행위는 별도로 판단됩니다. 특히 [저작권법 제93조](https://www.law.go.kr/법령/저작권법/제93조)는 데이터베이스제작자의 권리를 보호하며, 정보처리에 현실적 장애를 일으켜 영업을 방해하면 [형법 제314조 제2항](https://www.law.go.kr/법령/형법/제314조)에 따른 책임이 문제될 수 있습니다.
## 필수 사용 제한
1. 공개정보 자동 수집 기능은 반드시 개인의 정보 조회용으로만 사용합니다.
2. 조직적·체계적·대량 크롤링, 별도 데이터베이스 구축, 재판매 또는 원본 콘텐츠 재배포에 사용하지 않습니다.
3. 특정 브랜드나 업체의 정상적인 영업·서비스를 방해하지 않도록 요청 횟수와 범위를 최소화하며, 공격적 polling이나 병렬 대량 호출을 하지 않습니다.
4. 로그인·유료벽·CAPTCHA·접근통제·rate limit·IP 차단을 우회하거나 차단 회피를 위해 계정·IP·헤더를 변경하지 않습니다. 접근 거부나 차단 신호가 확인되면 즉시 중단합니다.
5. 적용되는 이용약관, robots 지침, API 조건, 저작권·데이터베이스권·개인정보 보호 의무를 별도로 준수합니다.
이 문서는 일반적인 위험 경계를 알리기 위한 것이며, 개별 행위의 적법성을 보증하거나 법률 자문을 제공하지 않습니다.
references/TRADEMARK-LEGAL-STATEMENT.md
# 상표 사용 법적 고지 — `foresttrip-vacancy`
이 스킬에서 `숲나들e` 및 `Foresttrip` 명칭은 국립 자연휴양림 숙박·야영 예약 조회의 **대상 공식 서비스**를 식별하기 위해 사용한다. k-skill의 출처나 운영주체를 해당 서비스 운영기관으로 표시하려는 사용이 아니다.
대법원 2005. 6. 10. 선고 [2005도1637 판결](https://www.law.go.kr/LSW/precInfoP.do?precSeq=83920)은 타인의 표장을 출처표시가 아니라 상품 기능 또는 적용 기종을 밝히기 위해 사용하고 상표 사용으로 인식될 수 없는 경우 침해가 아니라고 판시했다. [상표법 제2조](https://www.law.go.kr/법령/상표법/제2조), [제89조](https://www.law.go.kr/법령/상표법/제89조), [제90조](https://www.law.go.kr/법령/상표법/제90조), [제108조](https://www.law.go.kr/법령/상표법/제108조)와 대법원 [2011다18802](https://www.law.go.kr/LSW/precInfoP.do?precSeq=167457), [2019후10418](https://law.go.kr/LSW/precInfoP.do?mode=0&precSeq=230725) 판결에 따라 실제 거래계에서 출처표시로 기능하는지는 표시 태양과 사용 경위 등을 종합해 판단해야 한다.
따라서 이 명칭은 필요한 범위의 평문으로만 사용하고, 별도 근거 없이 로고, 공식·제휴·후원·인증·파트너 표현을 사용하지 않는다. 각 상표의 권리는 해당 권리자에게 있다. 이 문서는 저작권, 데이터베이스권, 부정경쟁방지법, 약관·계약, 개인정보 또는 접근 방식의 적법성을 판단하지 않는다.
전체 검토: [제3자 상표의 기능 설명·호환 대상 표시 검토](https://github.com/NomaDamas/k-skill/blob/dev/docs/legal/trademark-use-review.md)
scripts/run_foresttrip_vacancy.py
#!/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 = "예비"
LOGIN_FORM_SELECTOR = "#mmberId"
LOGIN_SUBMIT_SELECTOR = "input.loginBtn"
RESERVATION_SIDO_SELECTOR = "#srchSido"
RESERVATION_INSTT_SELECTOR = "#srchInstt"
SESSION_RETRY_HINT = (
"Fix the credentials, then rerun with --refresh-session so a stale cached "
"session is not reused."
)
@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 has_element(page: Any, selector: str) -> bool:
"""Return True when the selector resolves to at least one node on the page."""
return page.locator(selector).count() > 0
def login_failure_message() -> str:
return (
"foresttrip login failed: the login form is still present after submitting credentials. "
"Check KSKILL_FORESTTRIP_ID and KSKILL_FORESTTRIP_PASSWORD. "
"If CAPTCHA or additional authentication is required, complete it on the official "
"foresttrip.go.kr login screen and retry. " + SESSION_RETRY_HINT
)
def reservation_page_failure_message(page: Any) -> str:
missing = [
selector
for selector in (RESERVATION_SIDO_SELECTOR, RESERVATION_INSTT_SELECTOR)
if not has_element(page, selector)
]
return (
"foresttrip reservation page did not expose " + ", ".join(missing) + ". "
"The login session may have expired or 숲나들e changed the page markup. "
+ SESSION_RETRY_HINT
+ " If it persists, the helper's session bootstrap needs an update."
)
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(LOGIN_FORM_SELECTOR, forest_id)
page.fill("#gnrlMmberPssrd", forest_pw)
page.click(LOGIN_SUBMIT_SELECTOR)
page.wait_for_load_state("networkidle")
page.goto(RSRVT_PAGE)
page.wait_for_load_state("networkidle")
failure: str | None = None
if has_element(page, LOGIN_FORM_SELECTOR):
failure = login_failure_message()
elif not has_element(page, RESERVATION_SIDO_SELECTOR) or not has_element(
page, RESERVATION_INSTT_SELECTOR
):
failure = reservation_page_failure_message(page)
if failure is not None:
browser.close()
raise SystemExit(failure)
csrf_locator = page.locator('input[name="_csrf"]')
if csrf_locator.count() == 0:
browser.close()
raise SystemExit(
"foresttrip login appeared to succeed but the reservation page did not "
"expose a CSRF token. " + SESSION_RETRY_HINT
)
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())
skill.json
{
"name": "foresttrip-vacancy",
"description": "Look up available Korean national forest recreation lodging or camping slots on foresttrip.go.kr and, in Dolshoi, continue into the official booking flow with approval-gated payment.",
"profiles": [
"vault",
"browser",
"action:booking"
],
"frontmatter": "name: foresttrip-vacancy\ndescription: Look up available Korean national forest recreation lodging or camping slots on foresttrip.go.kr and, in Dolshoi, continue into the official booking flow with approval-gated payment.\nlicense: MIT\nmetadata:\n category: travel\n locale: ko-KR\n phase: v1.5"
}
SKILL.md
---
name: foresttrip-vacancy
description: Look up available Korean national forest recreation lodging or camping slots on foresttrip.go.kr and, in Dolshoi, continue into the official booking flow with approval-gated payment.
license: MIT
metadata:
category: travel
locale: ko-KR
phase: v1.5
---
# foresttrip-vacancy
<!-- k-skill:cli-stub — generated by scripts/generate-skill-stubs.js; edit skill.json / instruction.md instead -->
## Get the full instructions (required first step)
Run this and follow its output as the primary instructions for this skill:
```bash
npx -y @nomadamas/k-skill@0 instruct foresttrip-vacancy
```
The CLI detects the current runtime (Dolshoi vault/CloakBrowser vs generic) and prints only the applicable instructions, always up to date. Helper files bundled with the CLI are listed by:
```bash
npx -y @nomadamas/k-skill@0 files foresttrip-vacancy
```
Keep the CLI and every coding-agent skill install current (including Vercel Agent Skills) with:
```bash
npx -y @nomadamas/k-skill@0 update
```
If `npx` is unavailable, install Node.js 18+ or follow https://github.com/NomaDamas/k-skill#readme, or read the source instructions at https://github.com/NomaDamas/k-skill/blob/main/foresttrip-vacancy/instruction.md.
## Legal disclaimer (required)
This skill is not an official feature of, officially supported by, affiliated with, sponsored by, approved by, or developed in collaboration with any third-party trademark owner or service operator it identifies. Third-party names are used only to describe the skill's function, lookup target, or compatibility.
Any automated collection of publicly accessible information must be limited to personal, non-organizational lookup. Do not use this skill for systematic or bulk crawling, database building, access-control or block circumvention, or conduct that interferes with a third party's business or service.
Read the full Korean legal disclaimer, including the cited Korean Supreme Court precedents and statutory limits, before use:
```bash
npx -y @nomadamas/k-skill@0 read foresttrip-vacancy references/DISCLAIMER.md
```
## Hard rules even without the CLI
- Never execute payment, message/email delivery, final submission, cancellation, or public posting without the user's explicit approval immediately beforehand.
- Never ask for, print, or store plaintext credentials in chat, files, or shell arguments.
- Never bypass legal, physical-presence, CAPTCHA, identity-proofing, or electronic-signature boundaries.
tests/__init__.py
tests/fixtures/geoje_window.json
[
{
"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
}
]
tests/fixtures/gujaebong_window.json
[
{
"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
}
]
tests/test_foresttrip_session.py
"""Session bootstrap tests for foresttrip-vacancy (issue #699).
An invalid credential (or a stale/placeholder one) previously pushed the helper
onto the reservation page while still logged out, where `#srchSido` was absent
and `page.evaluate(...'#srchSido').options` raised an unhandled Playwright
`TypeError`. These tests drive `bootstrap_session()` through a fake Playwright
surface and assert the documented, actionable `SystemExit` paths instead.
"""
import importlib.util
import sys
import types
import unittest
from pathlib import Path
from unittest import mock
SCRIPT_DIR = Path(__file__).resolve().parent
HELPER_PATH = SCRIPT_DIR.parent / "scripts" / "run_foresttrip_vacancy.py"
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()
CSRF_SELECTOR = 'input[name="_csrf"]'
class FakePlaywrightError(Exception):
pass
class FakeLocator:
def __init__(self, page, selector):
self.page = page
self.selector = selector
def count(self):
return 1 if self.selector in self.page.present else 0
@property
def first(self):
return self
def get_attribute(self, name):
return self.page.attrs.get((self.selector, name))
class FakePage:
def __init__(
self,
*,
present=(),
csrf="csrf-token",
user_agent="test-ua",
sido_options=None,
instt_options=None,
):
self.present = set(present)
self.attrs = {(CSRF_SELECTOR, "value"): csrf}
self.user_agent = user_agent
self.sido_options = sido_options if sido_options is not None else []
self.instt_options = instt_options or {}
self.selected_sido = None
self.navigations = []
self.actions = []
def goto(self, url):
self.navigations.append(url)
def fill(self, selector, value):
self.actions.append(("fill", selector))
def click(self, selector):
self.actions.append(("click", selector))
def wait_for_load_state(self, state):
pass
def wait_for_timeout(self, ms):
pass
def locator(self, selector):
return FakeLocator(self, selector)
def select_option(self, selector, value=None):
self.selected_sido = value
@property
def context(self):
return types.SimpleNamespace(cookies=lambda: [{"name": "JSESSIONID", "value": "abc"}])
def evaluate(self, script):
if "#srchInstt" in script:
return list(self.instt_options.get(self.selected_sido, []))
if "#srchSido" in script:
return list(self.sido_options)
if "navigator.userAgent" in script:
return self.user_agent
raise AssertionError(f"unexpected evaluate script: {script}")
class FakeBrowser:
def __init__(self, page):
self.page = page
self.closed = False
def new_page(self):
return self.page
def close(self):
self.closed = True
def install_fake_playwright(page):
"""Patch sys.modules so the helper's lazy playwright import resolves to fakes."""
browser = FakeBrowser(page)
class FakePlaywright:
chromium = types.SimpleNamespace(launch=lambda **kwargs: browser)
class FakeManager:
def __enter__(self):
return FakePlaywright()
def __exit__(self, *args):
return False
module = types.ModuleType("playwright.sync_api")
module.sync_playwright = lambda: FakeManager()
module.Error = FakePlaywrightError
package = types.ModuleType("playwright")
package.sync_api = module
return mock.patch.dict(sys.modules, {"playwright": package, "playwright.sync_api": module}), browser
class BootstrapSessionTest(unittest.TestCase):
def bootstrap(self, page, **kwargs):
patcher, browser = install_fake_playwright(page)
with patcher:
try:
session = helper.bootstrap_session(forest_id="user", forest_pw="secret", **kwargs)
except SystemExit as exc:
return exc, browser, None
return None, browser, session
def test_login_failure_exits_with_actionable_message(self):
page = FakePage(present={helper.LOGIN_FORM_SELECTOR})
exc, browser, session = self.bootstrap(page)
self.assertIsInstance(exc, SystemExit)
self.assertIsNone(session)
self.assertTrue(browser.closed)
message = str(exc)
self.assertIn("login failed", message)
self.assertIn("KSKILL_FORESTTRIP_ID", message)
self.assertIn("KSKILL_FORESTTRIP_PASSWORD", message)
self.assertIn("CAPTCHA", message)
self.assertIn("--refresh-session", message)
def test_reservation_dom_missing_exits_without_type_error(self):
page = FakePage(present=set()) # login form gone, reservation selectors absent
exc, browser, session = self.bootstrap(page)
self.assertIsInstance(exc, SystemExit)
self.assertIsNone(session)
self.assertTrue(browser.closed)
message = str(exc)
self.assertIn(helper.RESERVATION_SIDO_SELECTOR, message)
self.assertIn(helper.RESERVATION_INSTT_SELECTOR, message)
self.assertIn("--refresh-session", message)
def test_missing_csrf_token_exits(self):
page = FakePage(
present={helper.RESERVATION_SIDO_SELECTOR, helper.RESERVATION_INSTT_SELECTOR},
)
exc, browser, session = self.bootstrap(page)
self.assertIsInstance(exc, SystemExit)
self.assertIn("CSRF token", str(exc))
self.assertTrue(browser.closed)
def test_empty_forest_list_exits(self):
page = FakePage(
present={helper.RESERVATION_SIDO_SELECTOR, helper.RESERVATION_INSTT_SELECTOR, CSRF_SELECTOR},
)
exc, browser, session = self.bootstrap(page)
self.assertIsInstance(exc, SystemExit)
self.assertIn("failed to extract forest list", str(exc))
def test_success_builds_session_with_forests(self):
page = FakePage(
present={helper.RESERVATION_SIDO_SELECTOR, helper.RESERVATION_INSTT_SELECTOR, CSRF_SELECTOR},
sido_options=[{"value": "01", "text": "서울"}],
instt_options={"01": [{"value": "ID02030059", "text": "[공립](거제시)거제자연휴양림"}]},
)
exc, browser, session = self.bootstrap(page)
self.assertIsNone(exc)
self.assertTrue(browser.closed)
self.assertIsNotNone(session)
self.assertEqual(session.csrf, "csrf-token")
self.assertEqual(session.cookies, {"JSESSIONID": "abc"})
self.assertEqual(session.forests, {"ID02030059": "[공립](거제시)거제자연휴양림"})
if __name__ == "__main__":
unittest.main()
tests/test_run_foresttrip_vacancy.py
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()