mirror of
https://github.com/NomaDamas/k-skill.git
synced 2026-06-24 02:04:11 +00:00
d31157cba3
Prefer N-card selection by list index so full card numbers are not echoed through JSON output or required in shell history. Keep direct card-number input as a compatibility escape hatch with an explicit warning.\n\nConstraint: PR #231 scope is limited to ktx-booking docs, helper, and tests.\nRejected: Require users to copy full card numbers from ncard-list | exposes sensitive identifiers in logs and shell history.\nConfidence: high\nScope-risk: narrow\nDirective: Keep N-card list outputs masked; prefer index-based selection for future reservation flows.\nTested: python3 -m py_compile scripts/ktx_booking.py scripts/test_ktx_booking.py; PYTHONPATH=scripts python3 -m unittest scripts.test_ktx_booking; npm run lint; npm run typecheck; npm test\nNot-tested: Live Korail N-card reservation; requires real user credentials and owned N-card.
941 lines
34 KiB
Python
Executable file
941 lines
34 KiB
Python
Executable file
#!/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())
|