#!/usr/bin/env python3 from __future__ import annotations import argparse import base64 import json import os import random import re import string import sys import time from functools import reduce try: from Crypto.Cipher import AES from Crypto.Util.Padding import pad except ModuleNotFoundError as exc: AES = None pad = None _CRYPTO_IMPORT_ERROR = exc else: _CRYPTO_IMPORT_ERROR = None try: from korail2 import ( AdultPassenger, ChildPassenger, Korail, KorailError, NeedToLoginError, NoResultsError, Passenger, ReserveOption, SeniorPassenger, SoldOutError, ToddlerPassenger, TrainType, ) import korail2.korail2 as korail_mod except ModuleNotFoundError as exc: _KORAIL_IMPORT_ERROR = exc class KorailError(Exception): pass class NeedToLoginError(KorailError): pass class NoResultsError(KorailError): pass class SoldOutError(KorailError): pass class Passenger: def __init__(self, count: int = 1): self.count = count @staticmethod def reduce(passengers): return passengers def get_dict(self, _: int) -> dict[str, str]: return {} class AdultPassenger(Passenger): pass class ChildPassenger(Passenger): pass class ToddlerPassenger(Passenger): pass class SeniorPassenger(Passenger): pass class ReserveOption: GENERAL_FIRST = "GENERAL_FIRST" GENERAL_ONLY = "GENERAL_ONLY" SPECIAL_FIRST = "SPECIAL_FIRST" SPECIAL_ONLY = "SPECIAL_ONLY" class TrainType: # Fallback constants used only when korail2 is missing so module # import succeeds and ensure_runtime_dependencies() can surface # the install message. Values mirror upstream korail2.TrainType. KTX = "100" KTX_SANCHEON = "100" ITX_SAEMAEUL = "101" SAEMAEUL = "101" MUGUNGHWA = "102" NURIRO = "102" TONGGUEN = "103" ITX_CHEONGCHUN = "104" AIRPORT = "105" ALL = "109" class Korail: def __init__(self, *args, **kwargs): raise ModuleNotFoundError("korail2") class _FallbackKorailModule: EMAIL_REGEX = re.compile(r".+@.+") PHONE_NUMBER_REGEX = re.compile(r"^\d+$") korail_mod = _FallbackKorailModule() else: _KORAIL_IMPORT_ERROR = None try: from korail2 import NCardPassenger _NCARD_AVAILABLE = True except ImportError: _NCARD_AVAILABLE = False class NCardPassenger(AdultPassenger): def __init__(self, count=1, card_no='', card='', card_pw='', discount_type='153'): AdultPassenger.__init__(self, count) self.card_no = card_no self.card = card self.card_pw = card_pw self.discount_type = discount_type DEFAULT_USER_AGENT = "Dalvik/2.1.0 (Linux; U; Android 13; SM-S928N Build/UP1A.231005.007)" DYNAPATH_PATHS = [ "/classes/com.korail.mobile.certification.TicketReservation", "/classes/com.korail.mobile.nonMember.NonMemTicket", "/classes/com.korail.mobile.seatMovie.ScheduleView", "/classes/com.korail.mobile.seatMovie.ScheduleViewSpecial", "/classes/com.korail.mobile.trn.prcFare.do", "/classes/com.korail.mobile.login.Login", ] RESERVE_OPTION_MAP = { "general-first": ReserveOption.GENERAL_FIRST, "general-only": ReserveOption.GENERAL_ONLY, "special-first": ReserveOption.SPECIAL_FIRST, "special-only": ReserveOption.SPECIAL_ONLY, } TRAIN_TYPE_MAP = { "ktx": TrainType.KTX, # 100 — KTX/KTX-산천 "itx-saemaeul": TrainType.ITX_SAEMAEUL, # 101 — ITX-새마을 "mugunghwa": TrainType.MUGUNGHWA, # 102 — 무궁화호 "nuriro": TrainType.NURIRO, # 102 — 누리로 "tonggeun": TrainType.TONGGUEN, # 103 — 통근열차 "itx-cheongchun": TrainType.ITX_CHEONGCHUN, # 104 — ITX-청춘 "airport": TrainType.AIRPORT, # 105 — 공항직통 "all": TrainType.ALL, # 109 — 전체 } TRAIN_ID_PREFIX = "ktx:v1:" TRAIN_ID_INVALID_MESSAGE = "train_id is invalid; rerun search and copy a fresh train_id" TRAIN_ID_STALE_MESSAGE = "train_id no longer matches any current search result; rerun search and choose a fresh train_id" TRAIN_ID_FIELDS = ( "train_no", "dep_date", "dep_time", "arr_date", "arr_time", "run_date", "train_group", "dep_code", "arr_code", ) def ensure_runtime_dependencies() -> None: missing: list[str] = [] if _KORAIL_IMPORT_ERROR is not None: missing.append("korail2") if _CRYPTO_IMPORT_ERROR is not None: missing.append("pycryptodome") if missing: install_command = f"python3 -m pip install {' '.join(missing)}" raise SystemExit( "scripts/ktx_booking.py requires additional Python packages " f"({', '.join(missing)}). Install them before running this helper: {install_command}" ) class DynaPathMasterEngine: APP_ID = "com.korail.talk" AS_VALUE = "%5B38ff229cb34c7dda8e28220a2d750cce%5D" DEVICE_MODEL = "SM-S928N" OS_TYPE = "Android" SDK_VERSION = "v1" def __init__(self) -> None: self.table = "3FE9jgRD4KdCyuawklqGJYmvfMn15P7US8XbxeLQtWT6OicBAopINs2Vh0HZrz" self.i8 = 161 self.i9 = 30 self.i10 = 2 self.app_start_ts = str(int(time.time() * 1000)) def string2xa1s(self, data: str) -> list[int]: result: list[int] = [] idx = 0 while idx < len(data): codepoint = ord(data[idx]) idx += 1 if codepoint < 128: result.append(codepoint) elif codepoint < 2048: result.append(128 | ((codepoint >> 7) & 15)) result.append(codepoint & 127) elif codepoint >= 262144: result.append(160) result.append((codepoint >> 14) & 127) result.append((codepoint >> 7) & 127) result.append(codepoint & 127) elif (63488 & codepoint) != 55296: result.append(((codepoint >> 14) & 15) | 144) result.append((codepoint >> 7) & 127) result.append(codepoint & 127) return result def make_key(self, key: str) -> int: total = 0 for char in key: codepoint = ord(char) bit = 32768 for _ in range(16): if bit & codepoint: break bit >>= 1 total = (total * (bit << 1)) + codepoint return total def internal_char(self, base_table: str, remainder: int, current: str) -> str: seen = 0 for char in base_table: if char in current: continue if seen == remainder: return char seen += 1 return " " def make_encode_table(self, number: int, encode_size: int, base_table: str) -> str: chars = "" temp = number for index in range(encode_size): divisor = encode_size - index remainder = temp % divisor chars += self.internal_char(base_table, remainder, chars) temp //= divisor return chars def encode_normal_be(self, data: str, table: str) -> str: values = self.string2xa1s(data) output: list[str] = [] digits = [0] * (self.i10 + 1) idx = 0 tail = len(values) % self.i10 body_size = len(values) - tail while idx < body_size: value = 0 for _ in range(self.i10): value = (value * self.i8) + values[idx] idx += 1 for digit_index in range(self.i10 + 1): digits[digit_index] = value % self.i9 value //= self.i9 for digit_index in range(self.i10, -1, -1): output.append(table[digits[digit_index]]) if tail > 0: value = 0 for _ in range(tail): value = (value * self.i8) + values[idx] idx += 1 for digit_index in range(tail + 1): digits[digit_index] = value % self.i9 value //= self.i9 while tail >= 0: output.append(table[digits[tail]]) tail -= 1 return "".join(output) def generate_token(self, device_id: str, timestamp_ms: int, nonce: str) -> str: plaintext = ( f"ai={self.APP_ID}&di={device_id}&as={self.AS_VALUE}&su=false&dbg=false&emu=false&hk=false" f"&it={self.app_start_ts}&ts={timestamp_ms}&rt=0&os=13&dm={self.DEVICE_MODEL}&st={self.OS_TYPE}&sv={self.SDK_VERSION}" ) dyn_key = f"v1+{nonce}+{timestamp_ms}" key_encoded = self.encode_normal_be(dyn_key, self.table) table = self.make_encode_table(self.make_key(dyn_key), self.i9, self.table) body_encoded = self.encode_normal_be(plaintext, table) return f"bEeEP{self.table[len(key_encoded)]}{key_encoded}{body_encoded}" class PatchedKorail(Korail): _device = "AD" _version = "250601002" _sid_key = b"2485dd54d9deaa36" _device_id = "558a4f02041657ea" def __init__(self, korail_id: str, korail_pw: str, auto_login: bool = True, want_feedback: bool = False): import requests self._session = requests.session() self._session.headers.update({"User-Agent": DEFAULT_USER_AGENT}) self._engine = DynaPathMasterEngine() super().__init__(korail_id, korail_pw, auto_login=False, want_feedback=want_feedback) self._session.headers.update({"User-Agent": DEFAULT_USER_AGENT}) if auto_login: self.login(korail_id, korail_pw) def _generate_sid(self, timestamp_ms: int) -> str: ensure_runtime_dependencies() plaintext = f"{self._device}{timestamp_ms}".encode("utf-8") cipher = AES.new(self._sid_key, AES.MODE_CBC, iv=self._sid_key) return base64.b64encode(cipher.encrypt(pad(plaintext, 16))).decode("utf-8") + "\n" def _auth_headers_and_sid(self, url: str) -> tuple[dict[str, str], str | None]: headers: dict[str, str] = {} sid = None if any(path in url for path in DYNAPATH_PATHS): timestamp_ms = int(time.time() * 1000) nonce = "".join(random.choices(string.ascii_uppercase + string.digits, k=4)) headers["x-dynapath-m-token"] = self._engine.generate_token(self._device_id, timestamp_ms, nonce) sid = self._generate_sid(timestamp_ms) return headers, sid def login(self, korail_id: str | None = None, korail_pw: str | None = None) -> bool: if korail_id is None: korail_id = self.korail_id else: self.korail_id = korail_id if korail_pw is None: korail_pw = self.korail_pw else: self.korail_pw = korail_pw if korail_mod.EMAIL_REGEX.match(korail_id): input_flag = "5" elif korail_mod.PHONE_NUMBER_REGEX.match(korail_id): input_flag = "4" else: input_flag = "2" headers, sid = self._auth_headers_and_sid(korail_mod.KORAIL_LOGIN) payload = { "Device": self._device, "Version": self._version, "txtInputFlg": input_flag, "txtMemberNo": korail_id, "txtPwd": self._Korail__enc_password(korail_pw), "idx": self._idx, } if sid: payload["Sid"] = sid response = self._session.post(korail_mod.KORAIL_LOGIN, data=payload, headers=headers) data = json.loads(response.text) if data["strResult"] == "SUCC" and data.get("strMbCrdNo") is not None: self._key = data["Key"] self.membership_number = data["strMbCrdNo"] self.name = data["strCustNm"] self.email = data["strEmailAdr"] self.logined = True return True self.logined = False return False 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, ): kst_now = korail_mod.datetime.now(korail_mod.timezone.utc) + korail_mod.timedelta(hours=9) if date is None: date = kst_now.strftime("%Y%m%d") if time_value is None: time_value = kst_now.strftime("%H%M%S") if passengers is None: passengers = [AdultPassenger()] passengers = Passenger.reduce(passengers) adult_count = reduce(lambda total, passenger: total + passenger.count, [p for p in passengers if isinstance(p, AdultPassenger)], 0) child_count = reduce(lambda total, passenger: total + passenger.count, [p for p in passengers if isinstance(p, ChildPassenger)], 0) toddler_count = reduce( lambda total, passenger: total + passenger.count, [p for p in passengers if isinstance(p, ToddlerPassenger)], 0, ) senior_count = reduce(lambda total, passenger: total + passenger.count, [p for p in passengers if isinstance(p, SeniorPassenger)], 0) headers, sid = self._auth_headers_and_sid(korail_mod.KORAIL_SEARCH_SCHEDULE) payload = { "Device": self._device, "radJobId": "1", "selGoTrain": train_type, "txtCardPsgCnt": "0", "txtGdNo": "", "txtGoAbrdDt": date, "txtGoEnd": arr, "txtGoHour": time_value, "txtGoStart": dep, "txtJobDv": "", "txtMenuId": "11", "txtPsgFlg_1": adult_count, "txtPsgFlg_2": child_count, "txtPsgFlg_8": toddler_count, "txtPsgFlg_3": senior_count, "txtPsgFlg_4": "0", "txtPsgFlg_5": "0", "txtSeatAttCd_2": "000", "txtSeatAttCd_3": "000", "txtSeatAttCd_4": "015", "txtTrnGpCd": train_type, "Version": self._version, } if sid: payload["Sid"] = sid 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] 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: raise NoResultsError() return trains def reserve(self, train, passengers=None, option=ReserveOption.GENERAL_FIRST, try_waiting=False): reserving_seat = True try: if not train.has_seat(): raise SoldOutError() if option == ReserveOption.GENERAL_ONLY: if train.has_general_seat(): seat_type = "1" else: raise SoldOutError() elif option == ReserveOption.SPECIAL_ONLY: if train.has_special_seat(): seat_type = "2" else: raise SoldOutError() elif option == ReserveOption.GENERAL_FIRST: seat_type = "1" if train.has_general_seat() else "2" elif option == ReserveOption.SPECIAL_FIRST: seat_type = "2" if train.has_special_seat() else "1" else: raise ValueError(f"unsupported reserve option: {option}") except SoldOutError: if try_waiting and option != ReserveOption.SPECIAL_ONLY and train.has_general_waiting_list(): reserving_seat = False seat_type = "1" else: raise if passengers is None: passengers = [AdultPassenger()] passengers = Passenger.reduce(passengers) passenger_count = reduce(lambda total, passenger: total + passenger.count, passengers, 0) headers, sid = self._auth_headers_and_sid(korail_mod.KORAIL_TICKETRESERVATION) payload = { "Device": self._device, "Version": self._version, "Key": self._key, "txtGdNo": "", "txtJobId": "1101" if reserving_seat else "1102", "txtTotPsgCnt": passenger_count, "txtSeatAttCd1": "000", "txtSeatAttCd2": "000", "txtSeatAttCd3": "000", "txtSeatAttCd4": "015", "txtSeatAttCd5": "000", "hidFreeFlg": "N", "txtStndFlg": "N", "txtMenuId": "11", "txtSrcarCnt": "0", "txtJrnyCnt": "1", "txtJrnySqno1": "001", "txtJrnyTpCd1": "11", "txtDptDt1": train.dep_date, "txtDptRsStnCd1": train.dep_code, "txtDptTm1": train.dep_time, "txtArvRsStnCd1": train.arr_code, "txtTrnNo1": train.train_no, "txtRunDt1": train.run_date, "txtTrnClsfCd1": train.train_type, "txtPsrmClCd1": seat_type, "txtTrnGpCd1": train.train_group, "txtChgFlg1": "", "txtJrnySqno2": "", "txtJrnyTpCd2": "", "txtDptDt2": "", "txtDptRsStnCd2": "", "txtDptTm2": "", "txtArvRsStnCd2": "", "txtTrnNo2": "", "txtRunDt2": "", "txtTrnClsfCd2": "", "txtPsrmClCd2": "", "txtChgFlg2": "", } if sid: payload["Sid"] = sid for index, passenger in enumerate(passengers, start=1): payload.update(passenger.get_dict(index)) response = self._session.get(korail_mod.KORAIL_TICKETRESERVATION, params=payload, headers=headers) data = json.loads(response.text) if self._result_check(data): reservation_id = data["h_pnr_no"] matches = [reservation for reservation in self.reservations() if reservation.rsv_id == reservation_id] if len(matches) == 1: return matches[0] raise KorailError(f"reservation {reservation_id} was created but could not be reloaded") def reservations(self): payload = {"Device": self._device, "Version": self._version, "Key": self._key} response = self._session.get(korail_mod.KORAIL_MYRESERVATIONLIST, params=payload) data = json.loads(response.text) try: if self._result_check(data): return [ korail_mod.Reservation(train_info) for journey in data["jrny_infos"]["jrny_info"] for train_info in journey["train_infos"]["train_info"] ] except NoResultsError: return [] return [] def cancel(self, reservation): assert isinstance(reservation, korail_mod.Reservation) payload = { "Device": self._device, "Version": self._version, "Key": self._key, "txtPnrNo": reservation.rsv_id, "txtJrnySqno": reservation.journey_no, "txtJrnyCnt": reservation.journey_cnt, "hidRsvChgNo": reservation.rsv_chg_no, } response = self._session.get(korail_mod.KORAIL_CANCEL, params=payload) data = json.loads(response.text) if self._result_check(data): return True return False def parse_passengers(args: argparse.Namespace) -> list[Passenger]: passengers: list[Passenger] = [] if args.adults: passengers.append(AdultPassenger(args.adults)) if args.children: passengers.append(ChildPassenger(args.children)) if args.toddlers: passengers.append(ToddlerPassenger(args.toddlers)) if args.seniors: passengers.append(SeniorPassenger(args.seniors)) if not passengers: passengers.append(AdultPassenger()) return passengers def build_train_id_payload(train) -> dict[str, str]: return {field: getattr(train, field) for field in TRAIN_ID_FIELDS} def build_train_id(train) -> str: payload = json.dumps(build_train_id_payload(train), ensure_ascii=False, separators=(",", ":")).encode("utf-8") encoded = base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=") return f"{TRAIN_ID_PREFIX}{encoded}" def parse_train_id(train_id: str) -> dict[str, str]: if not train_id.startswith(TRAIN_ID_PREFIX): raise SystemExit("train_id must start with ktx:v1:") encoded = train_id.removeprefix(TRAIN_ID_PREFIX) padded = encoded + ("=" * ((4 - len(encoded) % 4) % 4)) try: payload = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8")) except (ValueError, json.JSONDecodeError, UnicodeDecodeError) as exc: raise SystemExit(TRAIN_ID_INVALID_MESSAGE) from exc if not isinstance(payload, dict): raise SystemExit(TRAIN_ID_INVALID_MESSAGE) invalid_fields = [field for field in TRAIN_ID_FIELDS if not isinstance(payload.get(field), str) or not payload[field]] if invalid_fields: raise SystemExit(TRAIN_ID_INVALID_MESSAGE) return {field: payload[field] for field in TRAIN_ID_FIELDS} def find_train_by_id(trains, train_id: str): expected = parse_train_id(train_id) for train in trains: if build_train_id_payload(train) == expected: return train return None def normalize_train(train, index: int) -> dict[str, object]: return { "index": index, "train_id": build_train_id(train), "train_no": train.train_no, "train_type": train.train_type_name, "dep_name": train.dep_name, "dep_date": train.dep_date, "dep_time": train.dep_time, "arr_name": train.arr_name, "arr_date": train.arr_date, "arr_time": train.arr_time, "has_general_seat": train.has_general_seat(), "has_special_seat": train.has_special_seat(), "has_waiting_list": train.has_waiting_list(), "description": str(train), } def mask_identifier(value: object, visible: int = 4) -> str: text = str(value or "") if not text: return "" if len(text) <= visible: return "*" * len(text) return f"{'*' * (len(text) - visible)}{text[-visible:]}" def normalize_reservation(reservation) -> dict[str, object]: return { "reservation_id": reservation.rsv_id, "train_no": reservation.train_no, "train_type": reservation.train_type_name, "dep_name": reservation.dep_name, "dep_date": reservation.dep_date, "dep_time": reservation.dep_time, "arr_name": reservation.arr_name, "arr_date": reservation.arr_date, "arr_time": reservation.arr_time, "seat_count": reservation.seat_no_count, "price": reservation.price, "buy_limit_date": reservation.buy_limit_date, "buy_limit_time": reservation.buy_limit_time, "journey_no": reservation.journey_no, "journey_cnt": reservation.journey_cnt, "rsv_chg_no": reservation.rsv_chg_no, "description": str(reservation), } def print_json(payload: dict[str, object]) -> None: print(json.dumps(payload, ensure_ascii=False, indent=2)) def build_client() -> PatchedKorail: ensure_runtime_dependencies() korail_id = os.environ.get("KSKILL_KTX_ID") korail_pw = os.environ.get("KSKILL_KTX_PASSWORD") if not korail_id or not korail_pw: raise SystemExit( "이 작업에는 KSKILL_KTX_ID, KSKILL_KTX_PASSWORD 환경변수가 필요합니다. " "환경변수가 설정되어 있지 않으면 ~/.config/k-skill/secrets.env 에 추가하거나 " "에이전트의 secret vault에서 주입해 주세요." ) client = PatchedKorail(korail_id, korail_pw) if not client.logined: raise NeedToLoginError() return client def command_search(args: argparse.Namespace) -> None: client = build_client() passengers = parse_passengers(args) trains = client.search_train( args.dep, args.arr, args.date, args.time, train_type=TRAIN_TYPE_MAP[args.train_type], passengers=passengers, include_no_seats=args.include_no_seats, include_waiting_list=args.include_waiting_list, ) visible_trains = trains[: args.limit] print_json({ "count": len(visible_trains), "trains": [normalize_train(train, index) for index, train in enumerate(visible_trains, start=1)], }) def ensure_ncard_available() -> None: if not _NCARD_AVAILABLE: raise SystemExit( "N카드 기능을 사용하려면 korail2-ncard 패키지가 필요합니다: " "pip install korail2-ncard pycryptodome" ) def resolve_ncard_no(client: PatchedKorail, ncard_index: int | None, ncard_no: str | None) -> str | None: if ncard_index is None and not ncard_no: return None ensure_ncard_available() if ncard_index is None: return ncard_no ncards = client.owned_ncards() if not ncards: raise SystemExit("보유한 N카드가 없습니다.") if ncard_index < 1 or ncard_index > len(ncards): raise SystemExit(f"ncard-index는 1~{len(ncards)} 사이여야 합니다.") selected = ncards[ncard_index - 1] selected_no = getattr(selected, "discount_card_no", None) if not selected_no: raise SystemExit("선택한 N카드에서 카드 번호를 확인할 수 없습니다.") return selected_no def command_reserve(args: argparse.Namespace) -> None: client = build_client() ncard_no = resolve_ncard_no( client, getattr(args, "ncard_index", None), getattr(args, "ncard_no", None), ) if ncard_no: passengers = [NCardPassenger(card_no=ncard_no)] else: passengers = parse_passengers(args) include_waiting_list = args.include_waiting_list or args.try_waiting trains = client.search_train( args.dep, args.arr, args.date, args.time, train_type=TRAIN_TYPE_MAP[args.train_type], passengers=passengers, include_no_seats=args.include_no_seats, include_waiting_list=include_waiting_list, ) selected_train = find_train_by_id(trains, args.train_id) if selected_train is None: raise SystemExit(TRAIN_ID_STALE_MESSAGE) reservation = client.reserve( selected_train, passengers=passengers, option=RESERVE_OPTION_MAP[args.seat_option], try_waiting=args.try_waiting, ) print_json({"reservation": normalize_reservation(reservation)}) def command_reservations(_: argparse.Namespace) -> None: client = build_client() reservations = client.reservations() print_json({ "count": len(reservations), "reservations": [normalize_reservation(reservation) for reservation in reservations], }) def normalize_ncard(ncard, index: int) -> dict[str, object]: return { "index": index, "card_no": mask_identifier(getattr(ncard, "discount_card_no", "")), "card_no_masked": True, "ticket_kind": ncard.ticket_kind_name or "", "dep_name": ncard.dep_name or "", "arr_name": ncard.arr_name or "", "valid": ncard.valid or "", "description": str(ncard), } def normalize_ncard_train(train, index: int) -> dict[str, object]: base = normalize_train(train, index) base["price"] = getattr(train, "price", None) base["discount_name"] = getattr(train, "discount_name", None) base["general_remaining_seats"] = getattr(train, "general_remaining_seats", None) base["standing_remaining_seats"] = getattr(train, "standing_remaining_seats", None) return base def command_ncard_list(args: argparse.Namespace) -> None: ensure_ncard_available() client = build_client() ncards = client.owned_ncards() print_json({ "count": len(ncards), "ncards": [normalize_ncard(ncard, index) for index, ncard in enumerate(ncards, start=1)], }) def command_ncard_search(args: argparse.Namespace) -> None: ensure_ncard_available() client = build_client() ncards = client.owned_ncards() if not ncards: raise SystemExit("보유한 N카드가 없습니다.") if args.ncard_index < 1 or args.ncard_index > len(ncards): raise SystemExit(f"ncard-index는 1~{len(ncards)} 사이여야 합니다.") ncard = ncards[args.ncard_index - 1] trains = client.search_owned_ncard_trains( ncard, dep=args.dep, arr=args.arr, date=args.date, time=args.time, train_type=TRAIN_TYPE_MAP[args.train_type], ) visible_trains = trains[: args.limit] print_json({ "count": len(visible_trains), "ncard": normalize_ncard(ncard, args.ncard_index), "trains": [normalize_ncard_train(train, index) for index, train in enumerate(visible_trains, start=1)], }) def command_cancel(args: argparse.Namespace) -> None: client = build_client() reservations = client.reservations() match = next((reservation for reservation in reservations if reservation.rsv_id == args.reservation_id), None) if match is None: raise SystemExit(f"reservation {args.reservation_id} not found") client.cancel(match) print_json({"cancelled": True, "reservation_id": args.reservation_id}) def add_common_trip_args(parser: argparse.ArgumentParser) -> None: parser.add_argument("dep", help="출발역") parser.add_argument("arr", help="도착역") parser.add_argument("date", help="출발일 YYYYMMDD") parser.add_argument("time", help="희망 시작 시각 HHMMSS") parser.add_argument("--adults", type=int, default=1, help="성인 수") parser.add_argument("--children", type=int, default=0, help="어린이 수") parser.add_argument("--toddlers", type=int, default=0, help="유아 수") parser.add_argument("--seniors", type=int, default=0, help="경로 수") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Patched KTX/Korail booking helper for k-skill") subparsers = parser.add_subparsers(dest="command", required=True) search_parser = subparsers.add_parser("search", help="KTX/Korail 열차를 조회합니다") add_common_trip_args(search_parser) search_parser.add_argument("--limit", type=int, default=5, help="출력할 최대 열차 수") search_parser.add_argument( "--train-type", choices=sorted(TRAIN_TYPE_MAP), default="ktx", help="조회할 열차 종류 (기본 ktx). ITX-청춘 노선은 itx-cheongchun, 무궁화는 mugunghwa, 전체는 all 사용", ) search_parser.add_argument("--include-no-seats", action="store_true", help="매진 열차도 포함") search_parser.add_argument("--include-waiting-list", action="store_true", help="예약 대기 가능 열차도 포함") search_parser.set_defaults(func=command_search) 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") reserve_parser.add_argument("--seat-option", choices=sorted(RESERVE_OPTION_MAP), default="general-first") reserve_parser.add_argument( "--train-type", choices=sorted(TRAIN_TYPE_MAP), default="ktx", help="재조회할 열차 종류 — search 단계에서 사용한 값과 동일하게 지정 (기본 ktx)", ) reserve_parser.add_argument("--include-no-seats", action="store_true", help="검색 시 매진 열차도 포함") reserve_parser.add_argument("--include-waiting-list", action="store_true", help="검색 시 예약대기 열차도 포함") reserve_parser.add_argument( "--try-waiting", action="store_true", help="좌석이 없으면 예약대기를 시도 (reserve 재조회 시 예약대기 열차 자동 포함)", ) reserve_parser.add_argument( "--ncard-index", type=int, metavar="N", default=None, help="ncard-list 결과의 N카드 순번 (권장). 지정하면 N카드 할인 승객으로 예약", ) reserve_parser.add_argument( "--ncard-no", metavar="CARD_NO", default=None, help="N카드 번호 직접 입력 (비권장: 셸 히스토리에 남을 수 있음)", ) reserve_parser.set_defaults(func=command_reserve) ncard_list_parser = subparsers.add_parser("ncard-list", help="보유한 N카드 목록을 조회합니다") ncard_list_parser.set_defaults(func=command_ncard_list) ncard_search_parser = subparsers.add_parser("ncard-search", help="N카드 할인 열차를 조회합니다") ncard_search_parser.add_argument("dep", help="출발역") ncard_search_parser.add_argument("arr", help="도착역") ncard_search_parser.add_argument("date", help="출발일 YYYYMMDD") ncard_search_parser.add_argument("time", help="희망 시작 시각 HHMMSS") ncard_search_parser.add_argument( "--ncard-index", type=int, required=True, metavar="N", help="ncard-list 결과의 N카드 순번 (1부터)", ) ncard_search_parser.add_argument("--limit", type=int, default=5, help="출력할 최대 열차 수") ncard_search_parser.add_argument( "--train-type", choices=sorted(TRAIN_TYPE_MAP), default="ktx", help="조회할 열차 종류 (기본 ktx)", ) ncard_search_parser.set_defaults(func=command_ncard_search) reservations_parser = subparsers.add_parser("reservations", help="현재 예약 목록을 조회합니다") reservations_parser.set_defaults(func=command_reservations) cancel_parser = subparsers.add_parser("cancel", help="예약번호로 예약을 취소합니다") cancel_parser.add_argument("reservation_id", help="취소할 예약번호") cancel_parser.set_defaults(func=command_cancel) return parser def main() -> int: parser = build_parser() args = parser.parse_args() try: args.func(args) except (KorailError, NeedToLoginError, NoResultsError, SoldOutError) as exc: print(str(exc), file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())