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

Let intercity booking helper create temporary seat holds

Browse source 
Extend the Tmoney intercity helper from read-only timetable lookup to the browser-equivalent seat-stage and temporary hold flow, saving the official card-information page and cancel/back fields while still avoiding card submission.

Constraint: readPcpySats.do creates a live sats_Pcpy_Id hold, so abandoned test holds must be released with the official pcpyCanc=C back flow.

Rejected: Automating final payment | card submission is irreversible and remains a manual user action.

Confidence: high

Scope-risk: narrow

Directive: Treat holds as short-lived; hand off immediately and cancel abandoned holds.

Tested: python3 -m py_compile intercity-bus-booking/scripts/intercity_bus_search.py ~/.agents/skills/intercity-bus-booking/scripts/intercity_bus_search.py; live --hold-first-seat for 동서울→속초 20260520 produced sats_Pcpy_Id and card-info page; posted cancel/back fields and verified timetable remained 24/28; ./scripts/validate-skills.sh; node --test scripts/skill-docs.test.js; npm run lint

Not-tested: card info entry, final payment, mixed passenger hold payloads
This commit is contained in:
Jeffrey (Dongkyu) Kim 2026-05-13 15:43:26 +09:00
commit 34127550fa
4 changed files with 231 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

18
docs/features/intercity-bus-booking.md
View file

@ -41,7 +41,23 @@ python3 intercity-bus-booking/scripts/intercity_bus_search.py \
--date 20260520
```
이 helper는 쿠키 세션을 시작하고 공식 배차 조회 POST를 수행한 뒤 출발시각, 운수사, 등급, 요금, 잔여/총 좌석을 JSON으로 출력한다. 임시 좌석 선점이나 결제 단계는 수행하지 않는다.
이 helper는 쿠키 세션을 시작하고 공식 배차 조회 POST를 수행한 뒤 출발시각, 운수사, 등급, 요금, 잔여/총 좌석을 JSON으로 출력한다. 기본은 read-only이며, `--hold-seat` 또는 `--hold-first-seat`를 주면 좌석/요금 단계에 진입해 `readPcpySats.do`로 임시 좌석 선점을 만들고 공식 카드정보 입력 HTML과 cancel/back 필드를 저장한다. 결제 정보 입력·제출은 수행하지 않는다.
### 임시 선점 예시
```bash
python3 intercity-bus-booking/scripts/intercity_bus_search.py \
--depart-code 0511601 \
--arrive-code 2482701 \
--depart-name 동서울 \
--arrive-name 속초 \
--date 20260520 \
--select-index 1 \
--hold-first-seat \
--output-dir /tmp/tmoney-hold
```
성공 조건은 JSON의 `hold.success=true`, `hold.hold_id` 존재, 저장된 HTML에 `카드정보 입력` 표시가 있는 것이다. 라이브 응답 페이지에는 정확한 만료 카운트다운 문구가 노출되지 않았으므로, 선점 후 결제는 즉시 진행하게 안내하고 방치된 선점은 저장된 cancel/back 필드로 해제한다.
## 주의할 점

20
intercity-bus-booking/SKILL.md
View file

@ -128,7 +128,25 @@ python3 intercity-bus-booking/scripts/intercity_bus_search.py \
--date 20260520
```
The helper starts a cookie-backed session, posts the browser-required timetable fields, parses `readSasFeeInf(...)`, and prints JSON with departure time, company, class, fares, and remaining/total seats. It intentionally does not create temporary holds or submit payment data.
The helper starts a cookie-backed session, posts the browser-required timetable fields, parses `readSasFeeInf(...)`, and prints JSON with departure time, company, class, fares, and remaining/total seats. By default it is read-only. With `--hold-seat <seatNo>` or `--hold-first-seat`, it enters `readSatsFee.do`, posts `readPcpySats.do`, and saves the official Tmoney card-information HTML page plus cancel/back fields. It still never submits card data or final payment.
### Temporary Hold Helper
To create a temporary hold and save the official card-information page:
```bash
python3 intercity-bus-booking/scripts/intercity_bus_search.py \
--depart-code 0511601 \
--arrive-code 2482701 \
--depart-name 동서울 \
--arrive-name 속초 \
--date 20260520 \
--select-index 1 \
--hold-first-seat \
--output-dir /tmp/tmoney-hold
```
Success requires `hold.success=true`, a `sats_Pcpy_Id`, and the saved page containing `카드정보 입력`. The saved cancel fields can be posted back to `/otck/readSatsFee.do` with `pcpyCanc=C` to abandon the hold. Live probes did not expose an exact countdown on the card-information page; treat the hold as short-lived and have the user complete payment immediately.
## Checkout-Entry Link Helper

3
intercity-bus-booking/references/tmoney-intercity-http-flow.md
View file

@ -104,6 +104,8 @@ Observed success markers:
sats_Pcpy_Id
```
Re-verified on 2026-05-13 with 동서울 -> 속초, 2026-05-20, 06:05 금강고속 우등, seat 1. `readPcpySats.do` returned `카드정보 입력` and `sats_Pcpy_Id=SP...`. Posting the resulting cancel/back fields with `pcpyCanc=C` to `/otck/readSatsFee.do` returned to the seat-selection page and subsequent timetable lookup still showed 24/28 seats.
### Cancellation / Back Flow
A POST back to `/otck/readSatsFee.do` with `pcpyCanc=C` and the hold fields returned to seat selection and appeared to release the temporary hold in testing.
@ -120,4 +122,5 @@ A POST back to `/otck/readSatsFee.do` with `pcpyCanc=C` and the hold fields retu
- CAPTCHA was not observed in the tested flow.
- A generic `errorCont` response usually means the posted form contract is incomplete, not necessarily that the route is unavailable; first verify `bef_Aft_Dvs` and `req_Rec_Num`.
- Payment/card-info submission is separate and should not be automated without explicit confirmation.
- The live card-information page did not expose an exact countdown/expiry text in probes. Treat temporary holds as short-lived: hand off immediately, and post the cancel/back fields for abandoned holds.
- Terminal codes are Tmoney-specific and must not be mixed with KOBUS codes.

226
intercity-bus-booking/scripts/intercity_bus_search.py
View file

@ -1,8 +1,9 @@
#!/usr/bin/env python3
"""Search Tmoney intercity-bus timetables through the official read-only flow.
"""Search and optionally hold Tmoney intercity-bus seats through official flows.
This helper intentionally stops at timetable parsing. It does not create seat holds,
submit card data, or perform payment.
Default mode is read-only timetable parsing. With --hold-seat, the helper performs
Tmoney's temporary seat-hold POST and saves the official card-information page.
It never submits card fields or final payment.
"""
from __future__ import annotations
@ -13,14 +14,18 @@ import json
import re
import ssl
import sys
import tempfile
import urllib.parse
import urllib.request
from dataclasses import dataclass, asdict
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterable
BASE_URL = "https://intercitybus.tmoney.co.kr"
ENTRY_PATH = "/otck/trmlInfEnty.do"
TIMETABLE_PATH = "/otck/readAlcnList.do"
SEAT_STAGE_PATH = "/otck/readSatsFee.do"
HOLD_PATH = "/otck/readPcpySats.do"
DEFAULT_UA = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125 Safari/537.36"
@ -30,6 +35,10 @@ ROW_RE = re.compile(r"<tr>\s*(.*?)readSasFeeInf\((.*?)\).*?</tr>", re.DOTALL | r
TD_WRAP_RE = re.compile(r'<div class="td_wrap1">(.*?)</div>', re.DOTALL | re.IGNORECASE)
TAG_RE = re.compile(r"<[^>]+>")
ARG_RE = re.compile(r"'((?:\\'|[^'])*)'")
FORM_RE = re.compile(r"<form\b([^>]*)>(.*?)</form>", re.DOTALL | re.IGNORECASE)
INPUT_RE = re.compile(r"<input\b([^>]+)>", re.DOTALL | re.IGNORECASE)
ATTR_RE = re.compile(r"([\w:-]+)=[\"']([^\"']*)[\"']")
SEAT_RE = re.compile(r"<li([^>]*)>\s*<a[^>]*>.*?<span>(\d+)</span>", re.DOTALL | re.IGNORECASE)
@dataclass
@ -46,6 +55,17 @@ class Schedule:
raw_args: list[str]
@dataclass
class HoldResult:
success: bool
hold_id: str | None
seat: str
card_page_path: str | None
cancel_fields_path: str | None
markers: dict[str, int]
failure_message: str | None = None
def _ssl_context() -> ssl.SSLContext:
# Tmoney has historically required curl -k in probes on some machines.
# Keep this helper resilient while limiting it to the official host.
@ -58,8 +78,11 @@ def _strip(value: str) -> str:
return html.unescape(value).replace("\xa0", " ").strip()
def _attrs(fragment: str) -> dict[str, str]:
return {k.lower(): html.unescape(v) for k, v in ATTR_RE.findall(fragment)}
def _open(opener: urllib.request.OpenerDirector, request: urllib.request.Request, timeout: int) -> str:
# urllib opener.open does not accept context; HTTPS context must be installed in handler.
with opener.open(request, timeout=timeout) as response:
charset = response.headers.get_content_charset() or "utf-8"
return response.read().decode(charset, errors="replace")
@ -73,6 +96,17 @@ def build_opener() -> urllib.request.OpenerDirector:
)
def _request(url: str, data: list[tuple[str, str]] | dict[str, str] | None = None, referer: str | None = None) -> urllib.request.Request:
headers = {"User-Agent": DEFAULT_UA}
if referer:
headers["Referer"] = referer
if data is None:
return urllib.request.Request(url, headers=headers, method="GET")
headers["Content-Type"] = "application/x-www-form-urlencoded"
encoded = urllib.parse.urlencode(data).encode("utf-8")
return urllib.request.Request(url, data=encoded, headers=headers, method="POST")
def search_timetable(
depart_code: str,
arrive_code: str,
@ -85,14 +119,10 @@ def search_timetable(
children: int = 0,
veterans: int = 0,
timeout: int = 20,
) -> tuple[str, list[Schedule]]:
opener = build_opener()
entry_req = urllib.request.Request(
f"{BASE_URL}{ENTRY_PATH}",
headers={"User-Agent": DEFAULT_UA},
method="GET",
)
_open(opener, entry_req, timeout)
opener: urllib.request.OpenerDirector | None = None,
) -> tuple[urllib.request.OpenerDirector, str, list[Schedule]]:
opener = opener or build_opener()
_open(opener, _request(f"{BASE_URL}{ENTRY_PATH}"), timeout)
fields = {
"depr_Trml_Cd": depart_code,
@ -110,23 +140,8 @@ def search_timetable(
"bef_Aft_Dvs": "D",
"req_Rec_Num": "10",
}
req = _post_opener_request(f"{BASE_URL}{TIMETABLE_PATH}", fields)
body = _open(opener, req, timeout)
return body, parse_schedules(body)
def _post_opener_request(url: str, data: dict[str, str]) -> urllib.request.Request:
encoded = urllib.parse.urlencode(data).encode("utf-8")
return urllib.request.Request(
url,
data=encoded,
headers={
"User-Agent": DEFAULT_UA,
"Referer": f"{BASE_URL}{ENTRY_PATH}",
"Content-Type": "application/x-www-form-urlencoded",
},
method="POST",
)
body = _open(opener, _request(f"{BASE_URL}{TIMETABLE_PATH}", fields, f"{BASE_URL}{ENTRY_PATH}"), timeout)
return opener, body, parse_schedules(body)
def parse_schedules(body: str) -> list[Schedule]:
@ -162,8 +177,125 @@ def parse_schedules(body: str) -> list[Schedule]:
return schedules
def _seat_stage_fields(schedule: Schedule, search_time: str) -> dict[str, str]:
a = schedule.raw_args
if len(a) < 21:
raise ValueError("schedule raw_args does not contain the expected readSasFeeInf payload")
return {
"atl_Depr_Dt_S1": a[2],
"atl_Depr_Time_S1": search_time,
"rot_Id": a[0],
"rot_Sqno": a[1],
"alcn_Dt": a[2],
"alcn_Sqno": a[3],
"depr_Trml_Cd": a[4],
"arvl_Trml_Cd": a[5],
"depr_Trml_Nm": a[6],
"arvl_Trml_Nm": a[7],
"depr_Time": a[8],
"bus_Cacm_Cd": a[9],
"bus_Cls_Cd": a[10],
"bus_Cacm_Nm": a[11],
"bus_Cls_Nm": a[12],
"ig": a[13],
"im": a[14],
"ic": a[15],
"rmn_Scnt": a[16],
"sats_Num": a[17],
"atl_Depr_Dt": a[18],
"atl_Depr_Time": a[19],
"dc_Psb_Yn": a[20],
}
def _form_fields(body: str, form_id: str) -> list[tuple[str, str]]:
for attrs_text, form_body in FORM_RE.findall(body):
attrs = _attrs(attrs_text)
if attrs.get("id") == form_id or attrs.get("name") == form_id:
fields: list[tuple[str, str]] = []
for input_text in INPUT_RE.findall(form_body):
input_attrs = _attrs(input_text)
name = input_attrs.get("name")
if name:
fields.append((name, input_attrs.get("value", "")))
return fields
return []
def _available_seats(seat_stage_body: str) -> list[str]:
seats: list[str] = []
for li_attrs, seat_no in SEAT_RE.findall(seat_stage_body):
classes = _attrs(li_attrs).get("class", "")
if "disabled" not in classes.split():
seats.append(seat_no)
return seats
def hold_seat(
opener: urllib.request.OpenerDirector,
schedule: Schedule,
search_time: str,
seat: str | None,
output_dir: Path,
timeout: int = 20,
) -> tuple[str, list[str], HoldResult]:
seat_stage_body = _open(
opener,
_request(f"{BASE_URL}{SEAT_STAGE_PATH}", _seat_stage_fields(schedule, search_time), f"{BASE_URL}{TIMETABLE_PATH}"),
timeout,
)
available = _available_seats(seat_stage_body)
selected = seat or (available[0] if available else "")
if not selected:
return seat_stage_body, available, HoldResult(False, None, "", None, None, {}, "No selectable seat was found")
if selected not in available:
return seat_stage_body, available, HoldResult(False, None, selected, None, None, {}, f"Seat {selected} is not selectable")
fields = _form_fields(seat_stage_body, "readPcpySats")
if not fields:
return seat_stage_body, available, HoldResult(False, None, selected, None, None, {}, "No readPcpySats form found")
# Mirror pcpySats() in /js/tckmrs/readSatsInfo.js for a normal adult-only hold.
field_map = dict(fields)
fields.extend(
[
("pcpy_Num", "1"),
("sats_No", selected),
("rtrp_Depr_Dt", ""),
("bus_Tck_Knd_Cd", field_map.get("ig_Knd_Cd", "IG00")),
("cty_Bus_Dc_Knd_Cd", "Z"),
("dcrt_Dvs_Cd", "0"),
]
)
hold_body = _open(opener, _request(f"{BASE_URL}{HOLD_PATH}", fields, f"{BASE_URL}{SEAT_STAGE_PATH}"), timeout)
markers = {k: hold_body.count(k) for k in ["카드정보 입력", "sats_Pcpy_Id", "이미 발매된 좌석", "발행을 실패", "errorCont"]}
hold_ids = re.findall(r'name=["\']sats_Pcpy_Id["\'][^>]*value=["\']([^"\']+)', hold_body)
success = bool(hold_ids and markers["카드정보 입력"] and not markers["errorCont"])
output_dir.mkdir(parents=True, exist_ok=True)
card_path = output_dir / "tmoney-intercity-card-info.html"
card_path.write_text(hold_body)
cancel_fields = _form_fields(hold_body, "alcnInfo") or _form_fields(hold_body, "onwayInfo")
cancel_path = output_dir / "tmoney-intercity-cancel-fields.txt"
if cancel_fields:
cancel_path.write_text("\n".join(f"{k}={v}" for k, v in cancel_fields))
else:
cancel_path = None # type: ignore[assignment]
failure = None if success else _strip(hold_body[hold_body.find("[처리결과]") : hold_body.find("[처리결과]") + 500]) or "Hold did not reach card-information page"
return seat_stage_body, available, HoldResult(
success=success,
hold_id=hold_ids[0] if hold_ids else None,
seat=selected,
card_page_path=str(card_path),
cancel_fields_path=str(cancel_path) if cancel_path else None,
markers=markers,
failure_message=failure,
)
def main(argv: Iterable[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Search Tmoney intercity-bus timetable")
parser = argparse.ArgumentParser(description="Search Tmoney intercity-bus timetable and optionally create a temporary seat hold")
parser.add_argument("--depart-code", required=True)
parser.add_argument("--arrive-code", required=True)
parser.add_argument("--depart-name", required=True)
@ -176,14 +308,22 @@ def main(argv: Iterable[str] | None = None) -> int:
parser.add_argument("--veterans", type=int, default=0)
parser.add_argument("--timeout", type=int, default=20)
parser.add_argument("--limit", type=int, default=20)
parser.add_argument("--select-index", type=int, default=1, help="1-based schedule index for --hold-seat")
parser.add_argument("--hold-seat", help="Temporarily hold this seat number and save the official card-info page")
parser.add_argument("--hold-first-seat", action="store_true", help="Hold the first selectable seat for the selected schedule")
parser.add_argument("--output-dir", help="Directory for saved hold/card page files; defaults to a temp directory")
args = parser.parse_args(argv)
if not re.fullmatch(r"\d{8}", args.date):
parser.error("--date must be YYYYMMDD")
if not re.fullmatch(r"\d{6}", args.time):
parser.error("--time must be HHMMSS")
if args.students or args.children or args.veterans:
parser.error("seat holding currently supports adult-only payloads; use search mode for mixed passenger counts")
if args.select_index < 1:
parser.error("--select-index must be 1 or greater")
body, schedules = search_timetable(
opener, body, schedules = search_timetable(
depart_code=args.depart_code,
arrive_code=args.arrive_code,
depart_name=args.depart_name,
@ -196,7 +336,7 @@ def main(argv: Iterable[str] | None = None) -> int:
veterans=args.veterans,
timeout=args.timeout,
)
result = {
result: dict[str, object] = {
"route": {
"depart_code": args.depart_code,
"arrive_code": args.arrive_code,
@ -215,8 +355,26 @@ def main(argv: Iterable[str] | None = None) -> int:
"or whether Tmoney returned its generic error page."
)
result["error_page_marker_count"] = body.count("errorCont")
print(json.dumps(result, ensure_ascii=False, indent=2))
return 2
if args.hold_seat or args.hold_first_seat:
if args.select_index > len(schedules):
parser.error(f"--select-index {args.select_index} exceeds schedule count {len(schedules)}")
output_dir = Path(args.output_dir) if args.output_dir else Path(tempfile.mkdtemp(prefix="tmoney-intercity-hold-"))
_, available, hold = hold_seat(opener, schedules[args.select_index - 1], args.time, args.hold_seat, output_dir, args.timeout)
result["selected_schedule"] = asdict(schedules[args.select_index - 1])
result["available_seats"] = available
result["hold"] = asdict(hold)
result["payment_window_note"] = (
"The live card-information page did not expose an exact countdown/expiry text in probes. "
"Treat the hold as short-lived and complete payment immediately; use the saved cancel fields to release abandoned holds."
)
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if hold.success else 3
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if schedules else 2
return 0
if __name__ == "__main__":