Fail early on incomplete KTX seat lookup context

Constraint: PR #296 review watchlist noted raw Korail mobile context can otherwise send empty residual-seat payload fields upstream.
Rejected: Keep empty-string defaults in residual-seat payloads | That hides stale or malformed search detail context until the anti-bot-sensitive endpoint is called.
Confidence: high
Scope-risk: narrow
Directive: Keep residual-seat payload requirements explicit if Korail changes the mobile h_* field contract.
Tested: PYTHONPATH=.:scripts PYTHONNOUSERSITE=1 python3 -m unittest discover -s scripts -p 'test_ktx_booking.py'; 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; node --test scripts/skill-docs.test.js; PYTHONPATH=.:scripts python3 scripts/ktx_booking.py seats --help
Not-tested: Live Korail authenticated seat lookup; credentials are not available in automation.
This commit is contained in:
Jeffrey (Dongkyu) Kim 2026-05-28 23:59:49 +09:00
commit 6a26a194c6
2 changed files with 50 additions and 11 deletions

View file

@ -136,6 +136,26 @@ DYNAPATH_PATHS = [
]
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"
SEAT_LOOKUP_FIELD_MAP = {
"h_arv_rs_stn_cd": "txtArvRsStnCd",
"h_arv_stn_run_ordr": "txtArvStnRunOrdr",
"h_dpt_dt": "txtDptDt",
"h_dpt_rs_stn_cd": "txtDptRsStnCd",
"h_dpt_stn_run_ordr": "txtDptStnRunOrdr",
"h_run_dt": "txtRunDt",
"h_trn_clsf_cd": "txtTrnClsfCd",
"h_trn_gp_cd": "txtTrnGpCd",
"h_trn_no": "txtTrnNo",
}
def make_korail_error(message: str, code: str = "KSKILL") -> KorailError:
try:
return KorailError(message, code)
except TypeError:
return KorailError(message)
RESERVE_OPTION_MAP = {
"general-first": ReserveOption.GENERAL_FIRST,
"general-only": ReserveOption.GENERAL_ONLY,
@ -538,25 +558,30 @@ class PatchedKorail(Korail):
return {}
def _seat_lookup_payload(self, raw_train: dict[str, object], passenger_count: int, room_class: str) -> dict[str, object]:
return {
missing_fields = [field for field in SEAT_LOOKUP_FIELD_MAP if not str(raw_train.get(field, ""))]
if missing_fields:
raise make_korail_error(
"seat lookup context missing "
+ ", ".join(missing_fields)
+ "; refresh train search details before requesting seats",
"KSKILL_SEAT_CONTEXT",
)
payload = {
payload_name: str(raw_train[field])
for field, payload_name in SEAT_LOOKUP_FIELD_MAP.items()
}
payload.update({
"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", ""),
}
})
return payload
def reserve(self, train, passengers=None, option=ReserveOption.GENERAL_FIRST, try_waiting=False):
reserving_seat = True

View file

@ -335,6 +335,20 @@ class KtxBookingTests(unittest.TestCase):
self.assertTrue(any(path in ktx_booking.KORAIL_CARS_INFO for path in ktx_booking.DYNAPATH_PATHS))
self.assertTrue(any(path in ktx_booking.KORAIL_CAR_DETAIL for path in ktx_booking.DYNAPATH_PATHS))
def test_seat_lookup_payload_rejects_missing_context_fields(self):
client = object.__new__(ktx_booking.PatchedKorail)
client._device = "AD"
client._version = "240906001"
client._key = "session-key"
with self.assertRaises(ktx_booking.KorailError) as exc:
client._seat_lookup_payload({"h_trn_no": "009", "h_dpt_dt": "20260328"}, 1, "1")
message = str(exc.exception)
self.assertIn("seat lookup context missing", message)
self.assertIn("h_arv_rs_stn_cd", message)
self.assertIn("h_trn_gp_cd", message)
def test_command_search_replays_selected_train_type(self):
selected = FakeTrain(
train_no="2080",