mirrors/k-skill
2
0
Fork 0
mirror of https://github.com/NomaDamas/k-skill.git synced 2026-06-24 02:04:11 +00:00
Code Issues Projects Releases Packages Wiki Activity

Expose KTX seat-level lookup before reservation

Browse source 
Add a seats command to inspect residual seats by car, normalize seat metadata, and surface power-outlet hints while preserving the existing search and reserve selector flow. Update skill docs so seat-number and good-seat requests route through the detailed lookup before reservation.

Constraint: Korail reservation remains separate from seat inspection; the new flow must not select, hold, or pay for seats.

Rejected: Reusing train-level has_general_seat flags for good-seat requests | those flags cannot answer car number, seat label, or outlet availability.

Confidence: high

Scope-risk: moderate

Directive: Keep search, seats, and reserve on the same train_type and stable train_id path so stale results fail explicitly.

Tested: PYTHONPATH=.:scripts python3 -m unittest scripts.test_ktx_booking; python3 -m py_compile scripts/ktx_booking.py scripts/test_ktx_booking.py; node --check scripts/skill-docs.test.js

Not-tested: npm run ci is blocked locally because gitignored .agents/ and .venv directories are treated as missing SKILL.md by scripts/validate-skills.sh.
This commit is contained in:
iamiks 2026-05-28 22:12:52 +09:00
commit 8420792a82
5 changed files with 651 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -23,7 +23,7 @@ Claude Code, Codex, OpenCode, OpenClaw/ClawHub 등 각종 코딩 에이전트
| 할 수 있는 일 | 스킬 이름 | 설명 | 사용자 로그인 | 문서 |
| --- | --- | --- | --- | --- |
| SRT 예매 | `srt-booking` | SRT 열차 조회, 예약, 예약 확인, 취소 | 필요 | [SRT 예매 가이드](docs/features/srt-booking.md) |
| KTX 예매 | `ktx-booking` | KTX/Korail 열차 조회, 예약, 예약 확인, 취소 | 필요 | [KTX 예매 가이드](docs/features/ktx-booking.md) |
| KTX 예매 | `ktx-booking` | KTX/Korail 열차 조회, 호차별 좌석번호·콘센트 좌석 확인, 예약, 예약 확인, 취소 | 필요 | [KTX 예매 가이드](docs/features/ktx-booking.md) |
| 고속버스 예매 | `express-bus-booking` | KOBUS 고속버스 배차·좌석·요금 조회와 결제 직전 handoff(결제는 수동) | 불필요 | [고속버스 예매 가이드](docs/features/express-bus-booking.md) |
| 시외버스 예매 | `intercity-bus-booking` | 티머니 시외버스 배차·좌석·요금 조회와 결제 직전 handoff(결제는 수동) | 불필요 | [시외버스 예매 가이드](docs/features/intercity-bus-booking.md) |
| 자연휴양림 빈 객실 조회 | `foresttrip-vacancy` | 공식 숲나들e 자연휴양림 예약 가능 객실 조회 자동화 (예약/결제 제외) | 필요 | [자연휴양림 빈 객실 조회 가이드](docs/features/foresttrip-vacancy.md) |

43
docs/features/ktx-booking.md
View file

@ -4,6 +4,8 @@
- KTX/Korail 열차 조회
- 좌석 가능 여부 확인
- 호차별 남은 좌석번호 확인
- 콘센트 꿀팁 좌석 필터링
- 예약 진행
- 예약 내역 확인
- 예약 취소
@ -35,6 +37,7 @@
- 희망 시작 시각: `HHMMSS`
- 인원 수와 승객 유형
- 좌석 선호
- 좌석 상세 조건: 객실 등급, 호차 번호, 남은 좌석만 보기, 콘센트 좌석 우선
- 조회 결과에서 복사한 `train_id`
## 왜 helper 를 쓰는가
@ -54,8 +57,9 @@
2. `KSKILL_KTX_ID`, `KSKILL_KTX_PASSWORD` 가 없으면 credential resolution order에 따라 확보한다.
3. helper 로 먼저 열차를 조회한다.
4. 후보 열차의 `index`, `train_id`, 출발/도착 시각, KTX 여부, 좌석 여부를 보여준다.
5. 대상 열차가 명확할 때만 예약한다.
6. 예약 확인/취소는 대상 예약을 다시 식별한 뒤 진행한다.
5. 사용자가 좌석번호, 호차별 잔여석, 콘센트 꿀팁 좌석을 물으면 `seats` 로 상세 좌석을 먼저 확인한다.
6. 대상 열차가 명확할 때만 예약한다.
7. 예약 확인/취소는 대상 예약을 다시 식별한 뒤 진행한다.
## 예시
@ -69,6 +73,41 @@ python3 scripts/ktx_booking.py search 서울 부산 20260328 090000 --limit 5
응답 JSON 의 `train_id` 는 검색 시점의 정확한 열차를 가리키는 stable selector 다. 예약할 때는 이 값을 그대로 복사해서 쓴다. 같은 열차가 더 이상 조회되지 않으면 helper 가 실패하고 새로 조회하게 만든다.
상세 좌석 확인:
```bash
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id>
```
남은 좌석번호만 확인:
```bash
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --available-only
```
특정 호차의 남은 좌석만 확인:
```bash
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --car-no 5 --available-only
```
콘센트 꿀팁 좌석부터 확인:
```bash
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --available-only --power-only
```
특실 좌석을 확인하려면 `--room special`, KTX 외 열차를 조회했다면 `search` 와 같은 `--train-type` 을 함께 넘긴다.
```bash
python3 scripts/ktx_booking.py seats 남춘천 용산 20260503 150000 \
--train-id <train_id> \
--train-type itx-cheongchun \
--available-only
```
`seats` 응답은 호차별 `remaining_seats`, `available_seats`, 좌석별 순방향/역방향, 창측/내측, 좌석 종류, 문 근처 여부, 콘센트 힌트를 JSON 으로 반환한다. 이 단계는 좌석을 선택하거나 선점하지 않고, 예약 전 확인만 한다.
예약:
```bash

69
ktx-booking/SKILL.md
View file

@ -1,6 +1,6 @@
---
name: ktx-booking
description: Search, reserve, inspect, and cancel KTX or Korail tickets in Korea with the korail2 + pycryptodome Python packages. Use when the user asks for KTX seats, Korail bookings, train changes, or reservation status.
description: Search, reserve, inspect, and cancel KTX or Korail tickets in Korea with the korail2 + pycryptodome Python packages. Use when the user asks for KTX seats, Korail bookings, train changes, reservation status, remaining seat numbers, car-by-car seats, or power-outlet/good-seat tips.
license: MIT
metadata:
category: travel
@ -12,7 +12,7 @@ metadata:
## What this skill does
`korail2` 위에 `scripts/ktx_booking.py` helper 를 얹어 KTX/Korail 조회, 예약, 예약 확인, 취소를 처리한다.
`korail2` 위에 `scripts/ktx_booking.py` helper 를 얹어 KTX/Korail 조회, 호차별 좌석번호 확인, 예약, 예약 확인, 취소를 처리한다.
최근 Korail 앱의 Dynapath anti-bot 체크 때문에 원본 `korail2` 0.4.0 예제만으로는 `MACRO ERROR` 가 날 수 있다. 이 스킬은 helper 가 `x-dynapath-m-token`, `Sid`, 최신 app version(`250601002`)을 붙여 실제 예매 흐름을 복구하는 것을 전제로 한다.
@ -22,6 +22,10 @@ metadata:
- "코레일 예약 확인해줘"
- "KTX 취소해줘"
- "오전 9시 이후 KTX 중 제일 빠른 거 잡아줘"
- "KTX 남은 좌석 번호 확인해줘"
- "이 열차 콘센트 있는 꿀팁 좌석부터 보여줘"
- "KTX 5호차 남은 자리만 봐줘"
- "예약하기 전에 호차별 좌석 확인해줘"
- "N카드로 할인 열차 찾아줘"
- "내 N카드 목록 보여줘"
- "N카드 할인 적용해서 예약해줘"
@ -59,6 +63,7 @@ metadata:
- 희망 시작 시각: `HHMMSS`
- 인원 수와 승객 유형
- 좌석 선호
- 좌석 상세 조건: 객실 등급, 호차 번호, 남은 좌석만 보기, 콘센트 꿀팁 좌석 우선
- 조회 결과에서 복사한 `train_id`
## Workflow
@ -108,7 +113,60 @@ python3 scripts/ktx_booking.py search 남춘천 용산 20260503 150000 --train-t
- 일반실/특실 가능 여부
- 예약 대기 가능 여부
### 4. Reserve only after the target train is unambiguous
### 4. Inspect detailed seats when the user asks for good seats
`search` 의 좌석 가능 여부는 열차 단위 플래그다. 사용자가 "남은 좌석 번호", "호차별 좌석", "콘센트", "꿀팁 좌석", "창측/순방향 자리", "예약 전에 자리 확인"처럼 구체적인 좌석을 물으면 예약 전에 `seats` 를 호출한다.
기본 상세 좌석 조회:
```bash
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id>
```
일반실/특실은 `--room` 으로 나눈다.
```bash
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --room special
```
남은 좌석번호만 보고 싶으면 `--available-only` 를 쓴다.
```bash
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --available-only
```
특정 호차만 확인하려면 `--car-no` 를 쓴다.
```bash
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --car-no 5 --available-only
```
콘센트 꿀팁 자리부터 확인하려면 `--power-only` 를 붙인다. 응답의 `power_outlet` 은 `direct`, `adjacent`, `none` 중 하나다.
```bash
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --available-only --power-only
```
`seats` 도 `search` 와 같은 `--train-type` 을 넘겨야 한다. ITX-청춘 등 KTX 외 열차를 조회했다면 상세 좌석 조회에도 같은 값을 사용한다.
```bash
python3 scripts/ktx_booking.py seats 남춘천 용산 20260503 150000 \
--train-id <train_id> \
--train-type itx-cheongchun \
--available-only
```
상세 좌석 응답을 보여줄 때는 사용자 의도에 맞춰 아래를 우선 요약한다.
- 호차별 `remaining_seats`, `available_seat_count`
- 남은 좌석 번호 (`available_seats`)
- 좌석별 `direction`, `position`, `seat_type`
- 콘센트 힌트 (`power_outlet`)
- 문 근처 여부 (`near_door`)
이 기능은 좌석을 선택/선점하지 않는다. 실제 예약은 다음 단계의 `reserve` 로만 진행한다.
### 5. Reserve only after the target train is unambiguous
조회 결과의 `train_id` 를 고른 뒤에만 예약한다. 이 값은 helper 가 열차 번호/운행일/시각/역 코드를 묶어 만든 stable selector 이므로, 재조회 시 같은 열차가 아직 있으면 그대로 잡고 없으면 실패한다.
@ -125,7 +183,7 @@ python3 scripts/ktx_booking.py reserve 남춘천 용산 20260503 150000 --train-
응답에는 예약번호, 운임, 구입기한이 포함된다. **결제는 자동화하지 않는다.**
좌석이 없을 때는 조회 단계에서 `--include-waiting-list` 를 켜고 예약 단계에서 `--try-waiting` 으로 예약 대기까지 시도할 수 있다.
### 4-1. N-card discounted reservation
### 5-1. N-card discounted reservation
N카드 할인을 적용하려면 먼저 보유 N카드 목록을 조회해 카드 번호를 확인한다.
@ -151,7 +209,7 @@ python3 scripts/ktx_booking.py reserve 대전 서울 20260512 100000 \
N카드 기능은 `korail2-ncard` 패키지가 필요하다. 없으면 해당 커맨드 실행 시 설치 안내가 출력된다.
### 5. Inspect or cancel
### 6. Inspect or cancel
취소는 대상 예약을 다시 조회해 식별한 뒤에만 진행한다.
@ -166,6 +224,7 @@ python3 scripts/ktx_booking.py cancel <reservation_id>
## Done when
- 조회면 열차 후보가 정리되어 있다
- 좌석 상세 확인이면 호차별 남은 좌석번호와 필요한 꿀팁 조건이 정리되어 있다
- 예약이면 예약 결과와 제한 시간이 확인되어 있다
- 취소면 어떤 예약을 취소했는지 남아 있다

269
scripts/ktx_booking.py
View file

@ -132,6 +132,8 @@ DYNAPATH_PATHS = [
"/classes/com.korail.mobile.trn.prcFare.do",
"/classes/com.korail.mobile.login.Login",
]
KORAIL_CARS_INFO = "https://smart.letskorail.com:443/classes/com.korail.mobile.research.TrainResearch"
KORAIL_CAR_DETAIL = "https://smart.letskorail.com:443/classes/com.korail.mobile.research.ResidualSeatsResearch.do"
RESERVE_OPTION_MAP = {
"general-first": ReserveOption.GENERAL_FIRST,
"general-only": ReserveOption.GENERAL_ONLY,
@ -163,6 +165,43 @@ TRAIN_ID_FIELDS = (
"arr_code",
)
PHONE_NUMBER_DIGITS_REGEX = re.compile(r"^01\d{8,9}$")
ROOM_CLASS_MAP = {
"general": "1",
"special": "2",
}
ROOM_CLASS_NAME = {
"1": "일반실",
"2": "특실",
}
SEAT_DIRECTION_NAME = {
"009": "순방향",
"010": "역방향",
}
SEAT_POSITION_NAME = {
"011": "1인",
"012": "창측",
"013": "내측",
}
SEAT_TYPE_NAME = {
"015": "일반석",
"018": "2층석",
"019": "유아동반석",
"021": "휠체어석",
"023": "4인동반석",
"027": "4인석",
"028": "전동휠체어석",
"032": "자전거",
"052": "대피도우미",
}
POWER_OUTLET_ROWS = {1, 3, 5, 7, 10, 12, 14, 15}
POWER_OUTLET_DIRECT_COLUMNS = {"A", "D"}
POWER_OUTLET_ADJACENT_COLUMNS = {"B", "C"}
def is_phone_login_id(korail_id: str) -> bool:
return bool(korail_mod.PHONE_NUMBER_REGEX.match(korail_id) or PHONE_NUMBER_DIGITS_REGEX.match(korail_id))
def ensure_runtime_dependencies() -> None:
missing: list[str] = []
@ -334,7 +373,7 @@ class PatchedKorail(Korail):
if korail_mod.EMAIL_REGEX.match(korail_id):
input_flag = "5"
elif korail_mod.PHONE_NUMBER_REGEX.match(korail_id):
elif is_phone_login_id(korail_id):
input_flag = "4"
else:
input_flag = "2"
@ -364,7 +403,7 @@ class PatchedKorail(Korail):
self.logined = False
return False
def search_train(
def search_train_details(
self,
dep: str,
arr: str,
@ -424,17 +463,98 @@ class PatchedKorail(Korail):
response = self._session.post(korail_mod.KORAIL_SEARCH_SCHEDULE, params=payload, headers=headers)
data = json.loads(response.text)
if self._result_check(data):
trains = [korail_mod.Train(info) for info in data["trn_infos"]["trn_info"]]
trains = [train for train in trains if train.dep_name == dep and train.arr_name == arr]
train_infos = data["trn_infos"]["trn_info"]
if isinstance(train_infos, dict):
train_infos = [train_infos]
details = [(korail_mod.Train(info), info) for info in train_infos]
details = [(train, info) for train, info in details if train.dep_name == dep and train.arr_name == arr]
filters = [lambda train: train.has_seat()]
if include_no_seats:
filters.append(lambda train: not train.has_seat())
if include_waiting_list:
filters.append(lambda train: train.has_waiting_list())
trains = [train for train in trains if any(check(train) for check in filters)]
if not trains:
details = [(train, info) for train, info in details if any(check(train) for check in filters)]
if not details:
raise NoResultsError()
return trains
return details
def search_train(
self,
dep: str,
arr: str,
date: str | None = None,
time_value: str | None = None,
train_type: str = TrainType.ALL,
passengers: list[Passenger] | None = None,
include_no_seats: bool = False,
include_waiting_list: bool = False,
):
return [
train
for train, _ in self.search_train_details(
dep,
arr,
date,
time_value,
train_type=train_type,
passengers=passengers,
include_no_seats=include_no_seats,
include_waiting_list=include_waiting_list,
)
]
def train_cars(self, raw_train: dict[str, object], passenger_count: int = 1, room_class: str = "1") -> list[dict[str, object]]:
payload = self._seat_lookup_payload(raw_train, passenger_count, room_class)
headers, sid = self._auth_headers_and_sid(KORAIL_CARS_INFO)
if sid:
payload["Sid"] = sid
response = self._session.post(KORAIL_CARS_INFO, data=payload, headers=headers)
data = json.loads(response.text)
if self._result_check(data):
cars = data.get("srcar_infos", {}).get("srcar_info", [])
if isinstance(cars, dict):
cars = [cars]
return cars
return []
def car_seats(
self,
raw_train: dict[str, object],
car_no: str,
passenger_count: int = 1,
room_class: str = "1",
) -> dict[str, object]:
payload = self._seat_lookup_payload(raw_train, passenger_count, room_class)
payload["txtSrcarNo"] = car_no
headers, sid = self._auth_headers_and_sid(KORAIL_CAR_DETAIL)
if sid:
payload["Sid"] = sid
response = self._session.post(KORAIL_CAR_DETAIL, data=payload, headers=headers)
data = json.loads(response.text)
if self._result_check(data):
return data
return {}
def _seat_lookup_payload(self, raw_train: dict[str, object], passenger_count: int, room_class: str) -> dict[str, object]:
return {
"Device": self._device,
"Version": self._version,
"Key": self._key,
"txtArvRsStnCd": raw_train.get("h_arv_rs_stn_cd", ""),
"txtArvStnRunOrdr": raw_train.get("h_arv_stn_run_ordr", ""),
"txtDptDt": raw_train.get("h_dpt_dt", ""),
"txtDptRsStnCd": raw_train.get("h_dpt_rs_stn_cd", ""),
"txtDptStnRunOrdr": raw_train.get("h_dpt_stn_run_ordr", ""),
"txtGdNo": "",
"txtMenuId": "11",
"txtPsrmClCd": room_class,
"txtRunDt": raw_train.get("h_run_dt", ""),
"txtSeatAttCd": "015",
"txtTotPsgCnt": str(passenger_count),
"txtTrnClsfCd": raw_train.get("h_trn_clsf_cd", ""),
"txtTrnGpCd": raw_train.get("h_trn_gp_cd", ""),
"txtTrnNo": raw_train.get("h_trn_no", ""),
}
def reserve(self, train, passengers=None, option=ReserveOption.GENERAL_FIRST, try_waiting=False):
reserving_seat = True
@ -609,6 +729,14 @@ def find_train_by_id(trains, train_id: str):
return None
def find_train_detail_by_id(details, train_id: str):
expected = parse_train_id(train_id)
for train, raw_train in details:
if build_train_id_payload(train) == expected:
return train, raw_train
return None
def normalize_train(train, index: int) -> dict[str, object]:
return {
"index": index,
@ -628,6 +756,49 @@ def normalize_train(train, index: int) -> dict[str, object]:
}
def parse_seat_label(seat_label: str) -> tuple[int | None, str]:
match = re.match(r"^(\d+)([A-Za-z])$", seat_label or "")
if not match:
return None, ""
return int(match.group(1)), match.group(2).upper()
def power_outlet_match(seat_label: str) -> str:
row, column = parse_seat_label(seat_label)
if row not in POWER_OUTLET_ROWS:
return "none"
if column in POWER_OUTLET_DIRECT_COLUMNS:
return "direct"
if column in POWER_OUTLET_ADJACENT_COLUMNS:
return "adjacent"
return "none"
def normalize_seat(raw_seat: dict[str, object]) -> dict[str, object]:
seat_label = str(raw_seat.get("h_con_seat_no", ""))
return {
"seat": seat_label,
"seat_no": str(raw_seat.get("h_seat_no", "")),
"available": raw_seat.get("h_sale_psb_flg") == "Y",
"direction": SEAT_DIRECTION_NAME.get(str(raw_seat.get("h_for_rev_dir_dv", "")), str(raw_seat.get("h_for_rev_dir_dv", ""))),
"position": SEAT_POSITION_NAME.get(str(raw_seat.get("h_sigl_win_in_dv", "")), str(raw_seat.get("h_sigl_win_in_dv", ""))),
"seat_type": SEAT_TYPE_NAME.get(str(raw_seat.get("h_dmd_seat_att", "")), str(raw_seat.get("h_dmd_seat_att", ""))),
"near_door": raw_seat.get("h_door_nbor_flg") == "Y",
"power_outlet": power_outlet_match(seat_label),
}
def normalize_car(raw_car: dict[str, object]) -> dict[str, object]:
return {
"car_no": int(str(raw_car.get("h_srcar_no", "0"))),
"car_no_raw": str(raw_car.get("h_srcar_no", "")),
"room_class": ROOM_CLASS_NAME.get(str(raw_car.get("h_psrm_cl_cd", "")), str(raw_car.get("h_psrm_cl_nm", ""))),
"room_class_code": str(raw_car.get("h_psrm_cl_cd", "")),
"total_seats": int(str(raw_car.get("h_seat_cnt", "0")) or "0"),
"remaining_seats": int(str(raw_car.get("h_rest_seat_cnt", "0")) or "0"),
}
def mask_identifier(value: object, visible: int = 4) -> str:
text = str(value or "")
if not text:
@ -699,6 +870,63 @@ def command_search(args: argparse.Namespace) -> None:
})
def command_seats(args: argparse.Namespace) -> None:
client = build_client()
passengers = parse_passengers(args)
passenger_count = sum(passenger.count for passenger in Passenger.reduce(passengers))
details = client.search_train_details(
args.dep,
args.arr,
args.date,
args.time,
train_type=TRAIN_TYPE_MAP[args.train_type],
passengers=passengers,
include_no_seats=True,
include_waiting_list=True,
)
match = find_train_detail_by_id(details, args.train_id)
if match is None:
raise SystemExit(TRAIN_ID_STALE_MESSAGE)
train, raw_train = match
room_class = ROOM_CLASS_MAP[args.room]
cars = [normalize_car(car) for car in client.train_cars(raw_train, passenger_count, room_class)]
if args.car_no is not None:
cars = [car for car in cars if car["car_no"] == args.car_no]
if not cars:
raise SystemExit(f"car_no {args.car_no} is not available for {args.room}")
car_payloads: list[dict[str, object]] = []
for car in cars:
raw = client.car_seats(raw_train, str(car["car_no_raw"]), passenger_count, room_class)
raw_seats = raw.get("seat_infos", {}).get("seat_info", [])
if isinstance(raw_seats, dict):
raw_seats = [raw_seats]
all_seats = [normalize_seat(seat) for seat in raw_seats if seat.get("h_con_seat_no") != "0A"]
available_seats = [seat for seat in all_seats if seat["available"]]
seats = all_seats
if args.available_only:
seats = available_seats
if args.power_only:
seats = [seat for seat in seats if seat["power_outlet"] != "none"]
seats = seats[: args.limit]
car_payload = dict(car)
car_payload["available_seat_count"] = len(available_seats)
car_payload["available_seats"] = [seat["seat"] for seat in available_seats]
car_payload["shown_seat_count"] = len(seats)
car_payload["seats"] = seats
car_payloads.append(car_payload)
print_json({
"train": normalize_train(train, 1),
"room": args.room,
"passenger_count": passenger_count,
"available_only": args.available_only,
"power_only": args.power_only,
"cars": car_payloads,
})
def ensure_ncard_available() -> None:
if not _NCARD_AVAILABLE:
raise SystemExit(
@ -863,6 +1091,33 @@ def build_parser() -> argparse.ArgumentParser:
search_parser.add_argument("--include-waiting-list", action="store_true", help="예약 대기 가능 열차도 포함")
search_parser.set_defaults(func=command_search)
seats_parser = subparsers.add_parser("seats", help="조회 결과 중 하나의 호차별 좌석번호를 조회합니다")
add_common_trip_args(seats_parser)
seats_parser.add_argument("--train-id", required=True, help="search 결과에서 복사한 stable train_id")
seats_parser.add_argument(
"--room",
choices=sorted(ROOM_CLASS_MAP),
default="general",
help="좌석을 조회할 객실 등급 (기본 general)",
)
seats_parser.add_argument(
"--train-type",
choices=sorted(TRAIN_TYPE_MAP),
default="ktx",
help="재조회할 열차 종류 — search 단계에서 사용한 값과 동일하게 지정 (기본 ktx)",
)
seats_parser.add_argument("--car-no", type=int, default=None, help="특정 호차만 조회")
seats_parser.add_argument(
"--available-only",
"--remaining-only",
dest="available_only",
action="store_true",
help="예약 가능한/남은 좌석만 출력",
)
seats_parser.add_argument("--power-only", action="store_true", help="콘센트 꿀팁 좌석(direct/adjacent)만 출력")
seats_parser.add_argument("--limit", type=int, default=100, help="호차별 출력할 최대 좌석 수")
seats_parser.set_defaults(func=command_seats)
reserve_parser = subparsers.add_parser("reserve", help="조회 결과 중 하나를 예약합니다")
add_common_trip_args(reserve_parser)
reserve_parser.add_argument("--train-id", required=True, help="search 결과에서 복사한 stable train_id")

284
scripts/test_ktx_booking.py
View file

@ -100,12 +100,27 @@ class FakeNCard:
class FakeClient:
def __init__(self, trains, search_handler=None, ncards=None, ncard_trains=None):
def __init__(
self,
trains,
search_handler=None,
ncards=None,
ncard_trains=None,
train_details=None,
cars=None,
seats_by_car=None,
):
self._trains = trains
self._search_handler = search_handler
self._ncards = ncards or []
self._ncard_trains = ncard_trains or []
self._train_details = train_details
self._cars = cars or []
self._seats_by_car = seats_by_car or {}
self.search_calls = []
self.search_detail_calls = []
self.train_car_calls = []
self.car_seat_calls = []
self.reserved_train = None
self.reserved_passengers = None
@ -115,6 +130,29 @@ class FakeClient:
return list(self._search_handler(*args, **kwargs))
return list(self._trains)
def search_train_details(self, *args, **kwargs):
self.search_detail_calls.append(kwargs)
if self._train_details is not None:
return list(self._train_details)
return [(train, {}) for train in self._trains]
def train_cars(self, raw_train, passenger_count=1, room_class="1"):
self.train_car_calls.append({
"raw_train": raw_train,
"passenger_count": passenger_count,
"room_class": room_class,
})
return list(self._cars)
def car_seats(self, raw_train, car_no, passenger_count=1, room_class="1"):
self.car_seat_calls.append({
"raw_train": raw_train,
"car_no": car_no,
"passenger_count": passenger_count,
"room_class": room_class,
})
return {"seat_infos": {"seat_info": list(self._seats_by_car.get(car_no, []))}}
def reserve(self, train, **kwargs):
self.reserved_train = train
self.reserved_passengers = kwargs.get("passengers")
@ -173,6 +211,32 @@ class KtxBookingTests(unittest.TestCase):
self.assertEqual(args.train_id, "ktx:v1:test")
self.assertEqual(args.train_type, "ktx")
def test_build_parser_accepts_seats_filters(self):
args = ktx_booking.build_parser().parse_args([
"seats",
"서울",
"부산",
"20260328",
"090000",
"--train-id",
"ktx:v1:test",
"--room",
"special",
"--car-no",
"5",
"--available-only",
"--power-only",
"--limit",
"10",
])
self.assertEqual(args.train_id, "ktx:v1:test")
self.assertEqual(args.room, "special")
self.assertEqual(args.car_no, 5)
self.assertTrue(args.available_only)
self.assertTrue(args.power_only)
self.assertEqual(args.limit, 10)
def test_build_parser_defaults_search_train_type_to_ktx(self):
args = ktx_booking.build_parser().parse_args([
"search",
@ -207,8 +271,61 @@ class KtxBookingTests(unittest.TestCase):
"--train-type",
train_type,
])
seats_args = parser.parse_args([
"seats",
"서울",
"부산",
"20260328",
"090000",
"--train-id",
"ktx:v1:test",
"--train-type",
train_type,
])
self.assertEqual(search_args.train_type, train_type)
self.assertEqual(reserve_args.train_type, train_type)
self.assertEqual(seats_args.train_type, train_type)
def test_normalize_car_and_seat_maps_korail_codes(self):
car = ktx_booking.normalize_car({
"h_srcar_no": "05",
"h_psrm_cl_cd": "1",
"h_psrm_cl_nm": "ignored",
"h_seat_cnt": "48",
"h_rest_seat_cnt": "7",
})
seat = ktx_booking.normalize_seat({
"h_con_seat_no": "7A",
"h_seat_no": "007001",
"h_sale_psb_flg": "Y",
"h_for_rev_dir_dv": "009",
"h_sigl_win_in_dv": "012",
"h_dmd_seat_att": "015",
"h_door_nbor_flg": "Y",
})
self.assertEqual(car["car_no"], 5)
self.assertEqual(car["room_class"], "일반실")
self.assertEqual(car["remaining_seats"], 7)
self.assertEqual(seat["seat"], "7A")
self.assertTrue(seat["available"])
self.assertEqual(seat["direction"], "순방향")
self.assertEqual(seat["position"], "창측")
self.assertEqual(seat["seat_type"], "일반석")
self.assertTrue(seat["near_door"])
self.assertEqual(seat["power_outlet"], "direct")
def test_power_outlet_match_distinguishes_direct_adjacent_and_none(self):
self.assertEqual(ktx_booking.power_outlet_match("1A"), "direct")
self.assertEqual(ktx_booking.power_outlet_match("1B"), "adjacent")
self.assertEqual(ktx_booking.power_outlet_match("2A"), "none")
self.assertEqual(ktx_booking.power_outlet_match("bad"), "none")
def test_is_phone_login_id_accepts_digits_only_mobile_numbers(self):
self.assertTrue(ktx_booking.is_phone_login_id("01012345678"))
self.assertTrue(ktx_booking.is_phone_login_id("0101234567"))
self.assertFalse(ktx_booking.is_phone_login_id("1234567890"))
self.assertFalse(ktx_booking.is_phone_login_id("user@example.com"))
def test_command_search_replays_selected_train_type(self):
selected = FakeTrain(
@ -316,6 +433,171 @@ class KtxBookingTests(unittest.TestCase):
self.assertTrue(client.search_calls[-1]["include_waiting_list"])
self.assertIs(client.reserved_train, waiting_only)
def test_command_seats_returns_available_power_seats_for_selected_car(self):
selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="selected")
raw_train = {"h_trn_no": "009", "h_dpt_dt": "20260328"}
train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]
client = FakeClient(
[],
train_details=[(selected, raw_train)],
cars=[
{
"h_srcar_no": "04",
"h_psrm_cl_cd": "1",
"h_seat_cnt": "48",
"h_rest_seat_cnt": "9",
},
{
"h_srcar_no": "05",
"h_psrm_cl_cd": "1",
"h_seat_cnt": "48",
"h_rest_seat_cnt": "3",
},
],
seats_by_car={
"05": [
{
"h_con_seat_no": "1A",
"h_seat_no": "001001",
"h_sale_psb_flg": "Y",
"h_for_rev_dir_dv": "009",
"h_sigl_win_in_dv": "012",
"h_dmd_seat_att": "015",
},
{
"h_con_seat_no": "1B",
"h_seat_no": "001002",
"h_sale_psb_flg": "N",
"h_for_rev_dir_dv": "010",
"h_sigl_win_in_dv": "013",
"h_dmd_seat_att": "015",
},
{
"h_con_seat_no": "2A",
"h_seat_no": "002001",
"h_sale_psb_flg": "Y",
"h_for_rev_dir_dv": "009",
"h_sigl_win_in_dv": "012",
"h_dmd_seat_att": "015",
},
{
"h_con_seat_no": "0A",
"h_seat_no": "000000",
"h_sale_psb_flg": "Y",
},
],
},
)
args = argparse.Namespace(
dep="서울",
arr="부산",
date="20260328",
time="090000",
adults=2,
children=1,
toddlers=0,
seniors=0,
train_id=train_id,
room="general",
train_type="ktx",
car_no=5,
available_only=True,
power_only=True,
limit=10,
)
output = io.StringIO()
with patch.object(ktx_booking, "build_client", return_value=client):
with redirect_stdout(output):
ktx_booking.command_seats(args)
result = json.loads(output.getvalue())
self.assertEqual(result["room"], "general")
self.assertEqual(result["passenger_count"], 3)
self.assertTrue(result["available_only"])
self.assertTrue(result["power_only"])
self.assertEqual(len(result["cars"]), 1)
self.assertEqual(result["cars"][0]["car_no"], 5)
self.assertEqual(result["cars"][0]["remaining_seats"], 3)
self.assertEqual(result["cars"][0]["available_seat_count"], 2)
self.assertEqual(result["cars"][0]["available_seats"], ["1A", "2A"])
self.assertEqual(result["cars"][0]["shown_seat_count"], 1)
self.assertEqual(result["cars"][0]["seats"][0]["seat"], "1A")
self.assertEqual(result["cars"][0]["seats"][0]["power_outlet"], "direct")
self.assertEqual(client.search_detail_calls[-1]["train_type"], ktx_booking.TRAIN_TYPE_MAP["ktx"])
self.assertTrue(client.search_detail_calls[-1]["include_no_seats"])
self.assertTrue(client.search_detail_calls[-1]["include_waiting_list"])
self.assertEqual(client.train_car_calls[-1]["passenger_count"], 3)
self.assertEqual(client.train_car_calls[-1]["room_class"], "1")
self.assertEqual(client.car_seat_calls[-1]["car_no"], "05")
def test_command_seats_supports_special_room_and_stale_train_error(self):
selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="selected")
train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]
other = FakeTrain(train_no="011", dep_time="093000", arr_time="120000", label="other")
client = FakeClient(
[],
train_details=[(other, {"h_trn_no": "011"})],
cars=[{"h_srcar_no": "01", "h_psrm_cl_cd": "2", "h_seat_cnt": "30", "h_rest_seat_cnt": "1"}],
)
args = argparse.Namespace(
dep="서울",
arr="부산",
date="20260328",
time="090000",
adults=1,
children=0,
toddlers=0,
seniors=0,
train_id=train_id,
room="special",
train_type="ktx",
car_no=None,
available_only=False,
power_only=False,
limit=10,
)
with patch.object(ktx_booking, "build_client", return_value=client):
with self.assertRaises(SystemExit) as exc:
with redirect_stdout(io.StringIO()):
ktx_booking.command_seats(args)
self.assertIn("train_id", str(exc.exception))
def test_command_seats_fails_when_requested_car_is_not_available(self):
selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="selected")
train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]
client = FakeClient(
[],
train_details=[(selected, {"h_trn_no": "009"})],
cars=[{"h_srcar_no": "04", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "9"}],
)
args = argparse.Namespace(
dep="서울",
arr="부산",
date="20260328",
time="090000",
adults=1,
children=0,
toddlers=0,
seniors=0,
train_id=train_id,
room="general",
train_type="ktx",
car_no=5,
available_only=False,
power_only=False,
limit=10,
)
with patch.object(ktx_booking, "build_client", return_value=client):
with self.assertRaises(SystemExit) as exc:
with redirect_stdout(io.StringIO()):
ktx_booking.command_seats(args)
self.assertIn("car_no 5", str(exc.exception))
def test_build_parser_has_ncard_commands(self):
parser = ktx_booking.build_parser()
help_text = parser.format_help()