mirror of
https://github.com/NomaDamas/k-skill.git
synced 2026-06-24 02:04:11 +00:00
feat(srt-booking): SRT 좌석 확인과 탐색 우선순위 개선 (#305)
* feat(srt): 좌석 조회와 탐색 우선순위 추가 SRT search 결과의 stable train_id로 객차별 좌석을 조회하고, 특정 호차/좌석 확인과 탐색 우선순위 옵션을 제공한다. Constraint: SRT와 KTX는 별도 upstream 표면이므로 SRT HTML 파서와 테스트를 분리함 Rejected: KTX 좌석 helper 공유 | Korail API와 SRT 웹 좌석선택 HTML 계약이 달라 혼용하면 파서 안정성이 낮아짐 Confidence: medium Scope-risk: moderate Directive: SRT 좌석선택 HTML에서 노출되지 않는 속성은 추정하지 말고 명시적으로 처리할 것 Tested: PYTHONPATH=.:scripts python3 -m unittest scripts.test_srt_booking scripts.test_ktx_booking; python3 -m py_compile scripts/srt_booking.py scripts/srt_seats.py scripts/test_srt_booking.py Not-tested: 실제 예약 API에 우선순위 좌석 선택을 연결하는 흐름 * fix(srt): 좌석 조회 JSON 출력 안정화 SRT 대기열 메시지가 stdout에 섞여 seats JSON을 깨는 실제 표면 문제를 막고, 누락된 좌석 방향/위치 속성을 unknown으로 정규화한다. Constraint: issue #303 범위는 예약 부작용이 없는 좌석 조회 보조 흐름으로 제한됨 Rejected: 실제 예약 subcommand 추가 | 좌석 선점/예약은 외부 부작용이라 이번 acceptance criteria에 포함되지 않음 Confidence: high Scope-risk: narrow Directive: SRTrain upstream 출력이 추가되더라도 helper stdout은 JSON 전용으로 유지할 것 Tested: RED→GREEN in .omo/ulw-loop/evidence/srt-c002-red-green-tests.txt; live SRT tmux QA in .omo/ulw-loop/evidence/srt-c001-live-search-seats.txt; npm run ci in .omo/ulw-loop/evidence/srt-c003-regression-ci.txt Not-tested: 실제 예약/결제/취소 부작용 흐름 * test(srt): split seat helper regression coverage --------- Co-authored-by: Jeffrey (Dongkyu) Kim <vkehfdl1@gmail.com>
This commit is contained in:
parent
acc66861ea
commit
52dbfee064
8 changed files with 929 additions and 27 deletions
|
|
@ -4,6 +4,8 @@
|
|||
|
||||
- 수서 출발 SRT 열차 조회
|
||||
- 좌석 가능 여부 확인
|
||||
- 호차별 남은 좌석번호 확인
|
||||
- 특정 좌석 공석 여부 확인
|
||||
- 예약 진행
|
||||
- 예약 내역 확인
|
||||
- 예약 취소
|
||||
|
|
@ -35,33 +37,57 @@
|
|||
- 희망 시작 시각: `HHMMSS`
|
||||
- 인원 수
|
||||
- 좌석 선호
|
||||
- 좌석 상세 조건: 객실 등급, 호차 번호, 좌석 번호, 빈 좌석만 보기, 탐색 우선순위
|
||||
|
||||
## 기본 흐름
|
||||
|
||||
1. `SRTrain` 패키지가 없으면 다른 방법으로 우회하지 말고 먼저 전역 설치합니다.
|
||||
2. `KSKILL_SRT_ID`, `KSKILL_SRT_PASSWORD` 가 없으면 credential resolution order에 따라 확보합니다.
|
||||
3. 먼저 열차를 조회합니다.
|
||||
3. 먼저 helper 로 열차를 조회합니다.
|
||||
4. 후보 열차의 출발/도착 시각, 좌석 여부, 운임을 보여줍니다.
|
||||
5. 대상 열차가 명확할 때만 예약합니다.
|
||||
6. 예약 확인/취소는 예약을 다시 식별한 뒤 진행합니다.
|
||||
5. 사용자가 좌석번호, 호차별 잔여석, 특정 좌석 공석 여부를 물으면 `seats` 로 상세 좌석을 먼저 확인합니다.
|
||||
6. 대상 열차가 명확할 때만 예약합니다.
|
||||
7. 예약 확인/취소는 예약을 다시 식별한 뒤 진행합니다.
|
||||
|
||||
## 예시
|
||||
|
||||
```bash
|
||||
python3 - <<'PY'
|
||||
import os
|
||||
from SRT import SRT
|
||||
|
||||
srt = SRT(os.environ["KSKILL_SRT_ID"], os.environ["KSKILL_SRT_PASSWORD"])
|
||||
trains = srt.search_train("수서", "부산", "20260328", "080000", time_limit="120000")
|
||||
|
||||
for idx, train in enumerate(trains[:5], start=1):
|
||||
print(idx, train)
|
||||
PY
|
||||
python3 scripts/srt_booking.py search 수서 부산 20260328 080000 --time-limit 120000 --limit 5
|
||||
```
|
||||
|
||||
상세 좌석 확인:
|
||||
|
||||
```bash
|
||||
python3 scripts/srt_booking.py seats 수서 부산 20260328 080000 --train-id <train_id>
|
||||
```
|
||||
|
||||
특정 호차의 빈 좌석만 확인:
|
||||
|
||||
```bash
|
||||
python3 scripts/srt_booking.py seats 수서 부산 20260328 080000 --train-id <train_id> --car-no 5 --available-only
|
||||
```
|
||||
|
||||
특정 좌석이 비었는지 확인:
|
||||
|
||||
```bash
|
||||
python3 scripts/srt_booking.py seats 수서 부산 20260328 080000 --train-id <train_id> --car-no 5 --seat 11A
|
||||
```
|
||||
|
||||
탐색 순서 조정:
|
||||
|
||||
```bash
|
||||
python3 scripts/srt_booking.py seats 수서 부산 20260328 080000 \
|
||||
--train-id <train_id> \
|
||||
--car-priority center \
|
||||
--seat-priority window-forward \
|
||||
--available-only
|
||||
```
|
||||
|
||||
`seats` 응답은 호차별 `available_seat_count`, `available_seats`, 좌석별 순방향/역방향, 창측/내측, 특정 좌석 요청 시 `requested_seat_available` 을 JSON 으로 반환합니다. 이 단계는 좌석을 선택하거나 선점하지 않고, 예약 전 확인만 합니다.
|
||||
|
||||
## 주의할 점
|
||||
|
||||
- credential은 환경변수로 주입합니다.
|
||||
- 상세 좌석 확인은 SRT 웹 좌석선택 페이지의 공개 HTML을 조회 전용으로 파싱합니다.
|
||||
- 결제 완료까지 자동화하는 문서는 아닙니다.
|
||||
- 매진 시 공격적인 재시도 루프는 피합니다.
|
||||
|
|
|
|||
|
|
@ -11,10 +11,10 @@
|
|||
"build": "npm run build --workspaces --if-present",
|
||||
"build:manus-bundle": "node scripts/build-manus-bundle.js",
|
||||
"generate:plugin-manifest": "node scripts/generate-plugin-manifest.js",
|
||||
"lint": "node --check scripts/skill-docs.test.js scripts/korean_character_count.js scripts/test_korean_character_count.js scripts/korean_middle_korean.js scripts/test_korean_middle_korean.js scripts/build-manus-bundle.js scripts/test_build_manus_bundle.js scripts/workflow-actions.test.js scripts/generate-plugin-manifest.js scripts/test_generate_plugin_manifest.js && python3 -m py_compile scripts/k_skill_cleaner.py scripts/test_k_skill_cleaner.py corporate-registration-consulting/scripts/fill_official_hwp.py k-skill-cleaner/scripts/k_skill_cleaner.py scripts/fine_dust.py scripts/test_fine_dust.py scripts/ktx_booking.py scripts/test_ktx_booking.py scripts/sillok_search.py scripts/test_sillok_search.py scripts/korean_spell_check.py scripts/test_korean_spell_check.py scripts/patent_search.py scripts/test_patent_search.py scripts/mfds_drug_safety.py scripts/test_mfds_drug_safety.py scripts/nts_business_registration.py scripts/test_nts_business_registration.py scripts/mfds_food_safety.py scripts/test_mfds_food_safety.py scripts/zipcode_search.py scripts/test_zipcode_search.py scripts/subway_lost_property.py scripts/test_subway_lost_property.py scripts/geeknews_search.py scripts/test_geeknews_search.py nts-business-registration/scripts/nts_business_registration.py scripts/test_naver_blog_search.py scripts/test_korean_slang_writing.py scripts/kakaotalk_mac.py scripts/test_kakaotalk_mac.py scripts/test_coupang_partners_mcp_wrapper.py scripts/test_ohou_today_deal.py scripts/ticket_availability.py scripts/test_ticket_availability.py scripts/test_danawa_price_search.py ticket-availability/scripts/ticket_availability.py coupang-product-search/scripts/coupang_partners_mcp.py ohou-today-deal/scripts/ohou_today_deal.py kakaotalk-mac/scripts/kakaotalk_mac.py naver-blog-research/scripts/_naver_http.py naver-blog-research/scripts/naver_search.py naver-blog-research/scripts/naver_read.py naver-blog-research/scripts/naver_download_images.py korean-slang-writing/scripts/_slang_http.py korean-slang-writing/scripts/slang_search.py korean-slang-writing/scripts/slang_lookup.py korean-scholarship-search/scripts/scholarship_filter.py korean-scholarship-search/scripts/test_scholarship_filter.py korean-scholarship-search/scripts/university_search_plan.py seoul-bike/scripts/seoul_bike.py scripts/test_seoul_bike.py danawa-price-search/scripts/danawa_search.py kosis-stats/scripts/run_kosis_stats.py kosis-stats/tests/test_run_kosis_stats.py kstartup-search/scripts/run_kstartup.py kstartup-search/tests/test_run_kstartup.py intercity-bus-booking/scripts/intercity_bus_search.py daangn-used-goods-search/scripts/daangn_used_goods.py daangn-realty-search/scripts/daangn_realty.py daangn-jobs-search/scripts/daangn_jobs.py daangn-cars-search/scripts/daangn_cars.py foresttrip-vacancy/scripts/run_foresttrip_vacancy.py foresttrip-vacancy/tests/test_run_foresttrip_vacancy.py && npm run lint --workspaces --if-present && ./scripts/validate-skills.sh && node scripts/generate-plugin-manifest.js --check",
|
||||
"lint": "node --check scripts/skill-docs.test.js scripts/korean_character_count.js scripts/test_korean_character_count.js scripts/korean_middle_korean.js scripts/test_korean_middle_korean.js scripts/build-manus-bundle.js scripts/test_build_manus_bundle.js scripts/workflow-actions.test.js scripts/generate-plugin-manifest.js scripts/test_generate_plugin_manifest.js && python3 -m py_compile scripts/k_skill_cleaner.py scripts/test_k_skill_cleaner.py corporate-registration-consulting/scripts/fill_official_hwp.py k-skill-cleaner/scripts/k_skill_cleaner.py scripts/fine_dust.py scripts/test_fine_dust.py scripts/ktx_booking.py scripts/test_ktx_booking.py scripts/srt_booking.py scripts/srt_seats.py scripts/srt_booking_test_support.py scripts/test_srt_booking.py scripts/test_srt_seats.py scripts/sillok_search.py scripts/test_sillok_search.py scripts/korean_spell_check.py scripts/test_korean_spell_check.py scripts/patent_search.py scripts/test_patent_search.py scripts/mfds_drug_safety.py scripts/test_mfds_drug_safety.py scripts/nts_business_registration.py scripts/test_nts_business_registration.py scripts/mfds_food_safety.py scripts/test_mfds_food_safety.py scripts/zipcode_search.py scripts/test_zipcode_search.py scripts/subway_lost_property.py scripts/test_subway_lost_property.py scripts/geeknews_search.py scripts/test_geeknews_search.py nts-business-registration/scripts/nts_business_registration.py scripts/test_naver_blog_search.py scripts/test_korean_slang_writing.py scripts/kakaotalk_mac.py scripts/test_kakaotalk_mac.py scripts/test_coupang_partners_mcp_wrapper.py scripts/test_ohou_today_deal.py scripts/ticket_availability.py scripts/test_ticket_availability.py scripts/test_danawa_price_search.py ticket-availability/scripts/ticket_availability.py coupang-product-search/scripts/coupang_partners_mcp.py ohou-today-deal/scripts/ohou_today_deal.py kakaotalk-mac/scripts/kakaotalk_mac.py naver-blog-research/scripts/_naver_http.py naver-blog-research/scripts/naver_search.py naver-blog-research/scripts/naver_read.py naver-blog-research/scripts/naver_download_images.py korean-slang-writing/scripts/_slang_http.py korean-slang-writing/scripts/slang_search.py korean-slang-writing/scripts/slang_lookup.py korean-scholarship-search/scripts/scholarship_filter.py korean-scholarship-search/scripts/test_scholarship_filter.py korean-scholarship-search/scripts/university_search_plan.py seoul-bike/scripts/seoul_bike.py scripts/test_seoul_bike.py danawa-price-search/scripts/danawa_search.py kosis-stats/scripts/run_kosis_stats.py kosis-stats/tests/test_run_kosis_stats.py kstartup-search/scripts/run_kstartup.py kstartup-search/tests/test_run_kstartup.py intercity-bus-booking/scripts/intercity_bus_search.py daangn-used-goods-search/scripts/daangn_used_goods.py daangn-realty-search/scripts/daangn_realty.py daangn-jobs-search/scripts/daangn_jobs.py daangn-cars-search/scripts/daangn_cars.py foresttrip-vacancy/scripts/run_foresttrip_vacancy.py foresttrip-vacancy/tests/test_run_foresttrip_vacancy.py && npm run lint --workspaces --if-present && ./scripts/validate-skills.sh && node scripts/generate-plugin-manifest.js --check",
|
||||
"typecheck": "tsc --noEmit",
|
||||
"prepare:python-test-env": "python3 -m venv .cache/python-test-venv && ./.cache/python-test-venv/bin/python -m pip install --quiet beautifulsoup4",
|
||||
"test": "npm run prepare:python-test-env && node --test scripts/skill-docs.test.js scripts/test_korean_character_count.js scripts/test_korean_middle_korean.js scripts/test_build_manus_bundle.js scripts/workflow-actions.test.js scripts/test_generate_plugin_manifest.js && PYTHONPATH=.:scripts ./.cache/python-test-venv/bin/python -m unittest scripts.test_k_skill_cleaner scripts.test_fine_dust scripts.test_ktx_booking scripts.test_sillok_search scripts.test_korean_spell_check scripts.test_patent_search scripts.test_mfds_drug_safety scripts.test_nts_business_registration scripts.test_mfds_food_safety scripts.test_zipcode_search scripts.test_subway_lost_property scripts.test_geeknews_search scripts.test_naver_blog_search scripts.test_korean_slang_writing scripts.test_kakaotalk_mac scripts.test_coupang_partners_mcp_wrapper scripts.test_ohou_today_deal scripts.test_ticket_availability scripts.test_seoul_bike scripts.test_danawa_price_search && PYTHONPATH=.:scripts:korean-scholarship-search/scripts ./.cache/python-test-venv/bin/python -m unittest discover -s korean-scholarship-search/scripts -p 'test_scholarship_filter.py' && PYTHONPATH=.:scripts:kosis-stats/scripts ./.cache/python-test-venv/bin/python -m unittest discover -s kosis-stats/tests -p 'test_run_kosis_stats.py' && PYTHONPATH=.:scripts:kstartup-search/scripts ./.cache/python-test-venv/bin/python -m unittest discover -s kstartup-search/tests -p 'test_run_kstartup.py' && PYTHONPATH=.:foresttrip-vacancy/scripts ./.cache/python-test-venv/bin/python -m unittest discover -s foresttrip-vacancy/tests -p 'test_run_foresttrip_vacancy.py' && npm run test --workspaces --if-present && ./scripts/validate-skills.sh",
|
||||
"test": "npm run prepare:python-test-env && node --test scripts/skill-docs.test.js scripts/test_korean_character_count.js scripts/test_korean_middle_korean.js scripts/test_build_manus_bundle.js scripts/workflow-actions.test.js scripts/test_generate_plugin_manifest.js && PYTHONPATH=.:scripts ./.cache/python-test-venv/bin/python -m unittest scripts.test_k_skill_cleaner scripts.test_fine_dust scripts.test_ktx_booking scripts.test_srt_booking scripts.test_srt_seats scripts.test_sillok_search scripts.test_korean_spell_check scripts.test_patent_search scripts.test_mfds_drug_safety scripts.test_nts_business_registration scripts.test_mfds_food_safety scripts.test_zipcode_search scripts.test_subway_lost_property scripts.test_geeknews_search scripts.test_naver_blog_search scripts.test_korean_slang_writing scripts.test_kakaotalk_mac scripts.test_coupang_partners_mcp_wrapper scripts.test_ohou_today_deal scripts.test_ticket_availability scripts.test_seoul_bike scripts.test_danawa_price_search && PYTHONPATH=.:scripts:korean-scholarship-search/scripts ./.cache/python-test-venv/bin/python -m unittest discover -s korean-scholarship-search/scripts -p 'test_scholarship_filter.py' && PYTHONPATH=.:scripts:kosis-stats/scripts ./.cache/python-test-venv/bin/python -m unittest discover -s kosis-stats/tests -p 'test_run_kosis_stats.py' && PYTHONPATH=.:scripts:kstartup-search/scripts ./.cache/python-test-venv/bin/python -m unittest discover -s kstartup-search/tests -p 'test_run_kstartup.py' && PYTHONPATH=.:foresttrip-vacancy/scripts ./.cache/python-test-venv/bin/python -m unittest discover -s foresttrip-vacancy/tests -p 'test_run_foresttrip_vacancy.py' && npm run test --workspaces --if-present && ./scripts/validate-skills.sh",
|
||||
"pack:dry-run": "npm pack --workspace k-lotto --dry-run && npm pack --workspace daiso-product-search --dry-run && npm pack --workspace market-kurly-search --dry-run && npm pack --workspace kakao-bar-nearby --dry-run && npm pack --workspace cheap-gas-nearby --dry-run && npm pack --workspace public-restroom-nearby --dry-run && npm pack --workspace parking-lot-search --dry-run && npm pack --workspace court-auction-notice-search --dry-run && npm pack --workspace donation-place-search --dry-run && npm pack --workspace gongsijiga-search --dry-run && npm pack --workspace kbl-results --dry-run && npm pack --workspace kleague-results --dry-run && npm pack --workspace lck-analytics --dry-run && npm pack --workspace toss-securities --dry-run && npm pack --workspace hipass-receipt --dry-run && npm pack --workspace used-car-price-search --dry-run && npm pack --workspace k-skill-rhwp --dry-run && npm pack --workspace korean-marathon-schedule --dry-run && npm pack --workspace gangnamunni-clinic-search --dry-run && npm pack --workspace daishin-report-search --dry-run && npm pack --workspace sh-notice-search --dry-run && npm pack --workspace emergency-room-beds --dry-run && npm pack --workspace local-election-candidate-search --dry-run",
|
||||
"ci": "npm run lint && npm run typecheck && npm run test && npm run pack:dry-run",
|
||||
"version-packages": "changeset version",
|
||||
|
|
|
|||
272
scripts/srt_booking.py
Normal file
272
scripts/srt_booking.py
Normal file
|
|
@ -0,0 +1,272 @@
|
|||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import contextlib
|
||||
import io
|
||||
import importlib
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from types import ModuleType
|
||||
from typing import Protocol
|
||||
|
||||
from srt_seats import parse_cars, parse_seats, sort_cars_for_booking, sort_seats_for_booking
|
||||
|
||||
SRT_SEAT_ENDPOINT = "https://etk.srail.kr/hpg/hra/01/selectPassengerResearchList.do"
|
||||
TRAIN_ID_PREFIX = "srt:v1:"
|
||||
TRAIN_ID_FIELDS = (
|
||||
"train_number",
|
||||
"dep_date",
|
||||
"dep_time",
|
||||
"arr_date",
|
||||
"arr_time",
|
||||
"train_code",
|
||||
"dep_station_code",
|
||||
"arr_station_code",
|
||||
"dep_station_run_order",
|
||||
"arr_station_run_order",
|
||||
)
|
||||
ROOM_CODE = {"general": "1", "special": "2"}
|
||||
ROOM_NAME = {"general": "일반실", "special": "특실"}
|
||||
|
||||
|
||||
class SrtTrainLike(Protocol):
|
||||
train_number: str
|
||||
dep_date: str
|
||||
dep_time: str
|
||||
arr_date: str
|
||||
arr_time: str
|
||||
train_code: str
|
||||
train_name: str
|
||||
dep_station_code: str
|
||||
dep_station_name: str
|
||||
arr_station_code: str
|
||||
arr_station_name: str
|
||||
dep_station_run_order: str
|
||||
arr_station_run_order: str
|
||||
general_seat_state: str
|
||||
special_seat_state: str
|
||||
reserve_wait_possible_code: str
|
||||
|
||||
def general_seat_available(self) -> bool: ...
|
||||
|
||||
def special_seat_available(self) -> bool: ...
|
||||
|
||||
def reserve_standby_available(self) -> bool: ...
|
||||
|
||||
|
||||
class ResponseLike(Protocol):
|
||||
text: str
|
||||
|
||||
|
||||
class SessionLike(Protocol):
|
||||
def get(self, url: str, params: dict[str, str]) -> ResponseLike: ...
|
||||
|
||||
|
||||
class SrtClientLike(Protocol):
|
||||
_session: SessionLike
|
||||
|
||||
def search_train(
|
||||
self,
|
||||
dep: str,
|
||||
arr: str,
|
||||
date: str,
|
||||
time: str,
|
||||
time_limit: str | None = None,
|
||||
available_only: bool = True,
|
||||
) -> list[SrtTrainLike]: ...
|
||||
|
||||
|
||||
def load_srt_module() -> ModuleType:
|
||||
try:
|
||||
return importlib.import_module("SRT")
|
||||
except ModuleNotFoundError as exc:
|
||||
raise SystemExit("scripts/srt_booking.py requires SRTrain: python3 -m pip install SRTrain")
|
||||
|
||||
|
||||
def build_client(auto_login: bool = False) -> SrtClientLike:
|
||||
srt_module = load_srt_module()
|
||||
srt_id = os.environ.get("KSKILL_SRT_ID", "")
|
||||
srt_pw = os.environ.get("KSKILL_SRT_PASSWORD", "")
|
||||
return srt_module.SRT(srt_id, srt_pw, auto_login=auto_login)
|
||||
|
||||
|
||||
def train_id_payload(train: SrtTrainLike) -> dict[str, str]:
|
||||
return {field: getattr(train, field) for field in TRAIN_ID_FIELDS}
|
||||
|
||||
|
||||
def build_train_id(train: SrtTrainLike) -> str:
|
||||
raw = json.dumps(train_id_payload(train), ensure_ascii=False, separators=(",", ":")).encode()
|
||||
encoded = base64.urlsafe_b64encode(raw).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 srt:v1:")
|
||||
encoded = train_id.removeprefix(TRAIN_ID_PREFIX)
|
||||
padded = encoded + ("=" * ((4 - len(encoded) % 4) % 4))
|
||||
try:
|
||||
payload = json.loads(base64.urlsafe_b64decode(padded.encode()).decode())
|
||||
except (ValueError, json.JSONDecodeError, UnicodeDecodeError) as exc:
|
||||
raise SystemExit("train_id is invalid; rerun search and copy a fresh train_id") from exc
|
||||
if not isinstance(payload, dict):
|
||||
raise SystemExit("train_id is invalid; rerun search and copy a fresh train_id")
|
||||
if any(not isinstance(payload.get(field), str) or not payload[field] for field in TRAIN_ID_FIELDS):
|
||||
raise SystemExit("train_id is invalid; rerun search and copy a fresh train_id")
|
||||
return {field: payload[field] for field in TRAIN_ID_FIELDS}
|
||||
|
||||
|
||||
def find_train_by_id(trains: list[SrtTrainLike], train_id: str) -> SrtTrainLike | None:
|
||||
expected = parse_train_id(train_id)
|
||||
return next((train for train in trains if train_id_payload(train) == expected), None)
|
||||
|
||||
|
||||
def normalize_train(train: SrtTrainLike, index: int) -> dict[str, str | bool | int]:
|
||||
return {
|
||||
"index": index,
|
||||
"train_id": build_train_id(train),
|
||||
"train_no": train.train_number,
|
||||
"train_type": train.train_name,
|
||||
"dep_name": train.dep_station_name,
|
||||
"dep_date": train.dep_date,
|
||||
"dep_time": train.dep_time,
|
||||
"arr_name": train.arr_station_name,
|
||||
"arr_date": train.arr_date,
|
||||
"arr_time": train.arr_time,
|
||||
"has_general_seat": train.general_seat_available(),
|
||||
"has_special_seat": train.special_seat_available(),
|
||||
"has_waiting_list": train.reserve_standby_available(),
|
||||
}
|
||||
|
||||
|
||||
def seat_page_params(train: SrtTrainLike, room: str, car_no: int | None) -> dict[str, str]:
|
||||
return {
|
||||
"runDt1": train.dep_date,
|
||||
"dptDt1": train.dep_date,
|
||||
"dptTm1": train.dep_time,
|
||||
"trnNo1": f"{int(train.train_number):05d}",
|
||||
"trnGpCd1": "300",
|
||||
"dptRsStnCd1": train.dep_station_code,
|
||||
"arvRsStnCd1": train.arr_station_code,
|
||||
"dptStnRunOrdr1": train.dep_station_run_order,
|
||||
"arvStnRunOrdr1": train.arr_station_run_order,
|
||||
"seatAttCd1": "015",
|
||||
"psrmClCd1": ROOM_CODE[room],
|
||||
"index1": "0",
|
||||
"scarNo1": "" if car_no is None else f"{car_no:04d}",
|
||||
"chtnDvCd": "1",
|
||||
"jrnySqno": "001",
|
||||
"mode": "1",
|
||||
"psgNum": "1",
|
||||
"pageId": "",
|
||||
}
|
||||
|
||||
|
||||
def fetch_seat_page(client: SrtClientLike, train: SrtTrainLike, room: str, car_no: int | None) -> str:
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
response = client._session.get(SRT_SEAT_ENDPOINT, params=seat_page_params(train, room, car_no))
|
||||
return response.text
|
||||
|
||||
|
||||
def command_search(args: argparse.Namespace) -> None:
|
||||
client = build_client(auto_login=False)
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
trains = client.search_train(args.dep, args.arr, args.date, args.time, args.time_limit, args.available_only)
|
||||
print_json({"count": len(trains[: args.limit]), "trains": [normalize_train(train, index) for index, train in enumerate(trains[: args.limit], 1)]})
|
||||
|
||||
|
||||
def command_seats(args: argparse.Namespace) -> None:
|
||||
client = build_client(auto_login=False)
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
trains = client.search_train(args.dep, args.arr, args.date, args.time, args.time_limit, available_only=False)
|
||||
train = find_train_by_id(trains, args.train_id)
|
||||
if train is None:
|
||||
raise SystemExit("train_id no longer matches any current search result; rerun search and choose a fresh train_id")
|
||||
|
||||
initial_html = fetch_seat_page(client, train, args.room, args.car_no)
|
||||
cars = [car for car in parse_cars(initial_html) if car["room_class"] == ROOM_NAME[args.room]]
|
||||
if args.car_no is not None:
|
||||
cars = [car for car in cars if car["car_no"] == args.car_no]
|
||||
else:
|
||||
cars = [car for car in cars if car["available"]]
|
||||
if not cars:
|
||||
raise SystemExit(f"seat car data is unavailable for {args.room}; retry search or choose another train")
|
||||
|
||||
car_payloads: list[dict[str, object]] = []
|
||||
for car in sort_cars_for_booking(cars, args.car_priority):
|
||||
html = initial_html if args.car_no == car["car_no"] else fetch_seat_page(client, train, args.room, car["car_no"])
|
||||
seats = parse_seats(html)
|
||||
if args.seat:
|
||||
seats = [seat for seat in seats if seat["seat"] == args.seat]
|
||||
seats = sort_seats_for_booking(seats, args.seat_priority)
|
||||
if args.available_only:
|
||||
seats = [seat for seat in seats if seat["available"]]
|
||||
available_seats = [seat for seat in seats if seat["available"]]
|
||||
limited = seats[: args.limit]
|
||||
payload = dict(car)
|
||||
payload["available_seat_count"] = len(available_seats)
|
||||
payload["available_seats"] = [seat["seat"] for seat in available_seats]
|
||||
payload["shown_seat_count"] = len(limited)
|
||||
payload["seats"] = limited
|
||||
if args.seat:
|
||||
payload["requested_seat"] = args.seat
|
||||
payload["requested_seat_available"] = any(seat["available"] for seat in seats)
|
||||
car_payloads.append(payload)
|
||||
|
||||
print_json({
|
||||
"train": normalize_train(train, 1),
|
||||
"room": args.room,
|
||||
"available_only": args.available_only,
|
||||
"car_priority": args.car_priority,
|
||||
"seat_priority": args.seat_priority,
|
||||
"cars": car_payloads,
|
||||
})
|
||||
|
||||
|
||||
def print_json(payload: dict[str, object]) -> None:
|
||||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description="SRT search and seat lookup helper for k-skill")
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
search = subparsers.add_parser("search", help="SRT 열차를 조회합니다")
|
||||
add_trip_args(search)
|
||||
search.add_argument("--time-limit", default=None, help="조회 종료 시각 HHMMSS")
|
||||
search.add_argument("--available-only", action="store_true", default=False, help="예약 가능한 열차만 출력")
|
||||
search.add_argument("--limit", type=int, default=5, help="출력할 최대 열차 수")
|
||||
search.set_defaults(func=command_search)
|
||||
|
||||
seats = subparsers.add_parser("seats", help="SRT 호차별 좌석번호를 조회합니다")
|
||||
add_trip_args(seats)
|
||||
seats.add_argument("--train-id", required=True, help="search 결과에서 복사한 stable train_id")
|
||||
seats.add_argument("--time-limit", default=None, help="조회 종료 시각 HHMMSS")
|
||||
seats.add_argument("--room", choices=sorted(ROOM_CODE), default="general")
|
||||
seats.add_argument("--car-no", type=int, default=None, help="특정 호차만 조회")
|
||||
seats.add_argument("--seat", default=None, help="특정 좌석번호만 조회, 예: 6C")
|
||||
seats.add_argument("--available-only", action="store_true", help="빈 좌석만 출력")
|
||||
seats.add_argument("--car-priority", choices=("center", "low", "high"), default="center")
|
||||
seats.add_argument("--seat-priority", choices=("forward-window", "window-forward", "row-low"), default="forward-window")
|
||||
seats.add_argument("--limit", type=int, default=100, help="호차별 출력할 최대 좌석 수")
|
||||
seats.set_defaults(func=command_seats)
|
||||
return parser
|
||||
|
||||
|
||||
def add_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")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = build_parser().parse_args(argv)
|
||||
args.func(args)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
128
scripts/srt_booking_test_support.py
Normal file
128
scripts/srt_booking_test_support.py
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
from __future__ import annotations
|
||||
|
||||
|
||||
SEAT_HTML = "\n".join([
|
||||
'<li class="scar-01 off"><strong>일반실<br />1호차</strong></li>',
|
||||
'<li class="scar-04 on"><a href="#none" onclick="selectScarInfo(\'0004\'); return false;"><strong>일반실<br />4호차</strong></a></li>',
|
||||
'<li class="scar-05"><a href="#none" onclick="selectScarInfo(\'0005\'); return false;"><strong>일반실<br />5호차</strong></a></li>',
|
||||
'<li class="scar-03 off"><strong>특실<br />3호차</strong></li>',
|
||||
'<a href="#none" onclick="selectSeatInfo(this, \'23\', \'6C\'); return false;">6C<strong><em>(정방향, 내측)</em></strong></a>',
|
||||
'<a href="#none" onclick="selectSeatInfo(this, \'11\', \'3A\'); return false;">3A<strong><em>(역방향, 창측)</em></strong></a>',
|
||||
"<span>5C<strong><em>(정방향, 내측, 선택불가)</em></strong></span>",
|
||||
])
|
||||
|
||||
SPECIAL_SEAT_HTML = "\n".join([
|
||||
'<li class="scar-03 on"><a href="#none" onclick="selectScarInfo(\'0003\'); return false;"><strong>특실<br />3호차</strong></a></li>',
|
||||
'<li class="scar-05 off"><strong>일반실<br />5호차</strong></li>',
|
||||
'<a href="#none" onclick="selectSeatInfo(this, \'31\', \'1A\'); return false;">1A<strong><em>(정방향, 1인석)</em></strong></a>',
|
||||
"<span>2C<strong><em>(역방향, 내측, 선택불가)</em></strong></span>",
|
||||
])
|
||||
|
||||
|
||||
class FakeTrain:
|
||||
train_number = "313"
|
||||
dep_date = "20260610"
|
||||
dep_time = "080000"
|
||||
arr_date = "20260610"
|
||||
arr_time = "103400"
|
||||
train_code = "17"
|
||||
train_name = "SRT"
|
||||
dep_station_code = "0551"
|
||||
dep_station_name = "수서"
|
||||
arr_station_code = "0020"
|
||||
arr_station_name = "부산"
|
||||
dep_station_run_order = "000001"
|
||||
arr_station_run_order = "000007"
|
||||
general_seat_state = "예약가능"
|
||||
special_seat_state = "매진"
|
||||
reserve_wait_possible_code = "-2"
|
||||
|
||||
def general_seat_available(self) -> bool:
|
||||
return True
|
||||
|
||||
def special_seat_available(self) -> bool:
|
||||
return False
|
||||
|
||||
def reserve_standby_available(self) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, text: str) -> None:
|
||||
self.text = text
|
||||
|
||||
|
||||
class FakeSession:
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[dict[str, str]] = []
|
||||
|
||||
def get(self, _url: str, params: dict[str, str]) -> FakeResponse:
|
||||
self.calls.append(params)
|
||||
car = params["scarNo1"] or "0004"
|
||||
return FakeResponse(SEAT_HTML.replace("scar-04 on", f"scar-{car[-2:]} on"))
|
||||
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self, train: FakeTrain) -> None:
|
||||
self.train = train
|
||||
self._session = FakeSession()
|
||||
|
||||
def search_train(
|
||||
self,
|
||||
_dep: str,
|
||||
_arr: str,
|
||||
_date: str,
|
||||
_time: str,
|
||||
_time_limit: str | None = None,
|
||||
available_only: bool = True,
|
||||
) -> list[FakeTrain]:
|
||||
return [self.train]
|
||||
|
||||
|
||||
class NoisySession(FakeSession):
|
||||
def get(self, _url: str, params: dict[str, str]) -> FakeResponse:
|
||||
print("접속자가 많아 대기열에 들어갑니다.")
|
||||
return super().get(_url, params)
|
||||
|
||||
|
||||
class NoisyClient(FakeClient):
|
||||
def __init__(self, train: FakeTrain) -> None:
|
||||
self.train = train
|
||||
self._session = NoisySession()
|
||||
|
||||
def search_train(
|
||||
self,
|
||||
_dep: str,
|
||||
_arr: str,
|
||||
_date: str,
|
||||
_time: str,
|
||||
_time_limit: str | None = None,
|
||||
available_only: bool = True,
|
||||
) -> list[FakeTrain]:
|
||||
print("대기인원: 6명")
|
||||
return [self.train]
|
||||
|
||||
|
||||
class EmptyClient(FakeClient):
|
||||
def search_train(
|
||||
self,
|
||||
_dep: str,
|
||||
_arr: str,
|
||||
_date: str,
|
||||
_time: str,
|
||||
_time_limit: str | None = None,
|
||||
available_only: bool = True,
|
||||
) -> list[FakeTrain]:
|
||||
return []
|
||||
|
||||
|
||||
class SpecialSession(FakeSession):
|
||||
def get(self, _url: str, params: dict[str, str]) -> FakeResponse:
|
||||
self.calls.append(params)
|
||||
return FakeResponse(SPECIAL_SEAT_HTML)
|
||||
|
||||
|
||||
class SpecialClient(FakeClient):
|
||||
def __init__(self, train: FakeTrain) -> None:
|
||||
self.train = train
|
||||
self._session = SpecialSession()
|
||||
156
scripts/srt_seats.py
Normal file
156
scripts/srt_seats.py
Normal file
|
|
@ -0,0 +1,156 @@
|
|||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import TypedDict
|
||||
|
||||
|
||||
class SrtCar(TypedDict):
|
||||
car_no: int
|
||||
car_no_raw: str
|
||||
room_class: str
|
||||
available: bool
|
||||
current: bool
|
||||
|
||||
|
||||
class SrtSeat(TypedDict):
|
||||
seat: str
|
||||
seat_no: str
|
||||
available: bool
|
||||
direction: str
|
||||
position: str
|
||||
notes: list[str]
|
||||
|
||||
|
||||
CAR_RE = re.compile(
|
||||
r'<li class="scar-(?P<car>\d+)(?P<class>[^"]*)">(?P<body>.*?)</li>',
|
||||
re.DOTALL,
|
||||
)
|
||||
SEAT_LINK_RE = re.compile(
|
||||
r"<a[^>]+selectSeatInfo\(this,\s*'(?P<seat_no>[^']+)',\s*'(?P<seat>[^']+)'\)[^>]*>"
|
||||
r".*?<em>\((?P<detail>[^)]*)\)</em>",
|
||||
re.DOTALL,
|
||||
)
|
||||
SEAT_SPAN_RE = re.compile(
|
||||
r"<span>\s*(?P<seat>\d+[A-Z])\s*<strong><em>\((?P<detail>[^)]*)\)</em></strong></span>",
|
||||
re.DOTALL,
|
||||
)
|
||||
TAG_RE = re.compile(r"<[^>]+>")
|
||||
|
||||
|
||||
def strip_tags(value: str) -> str:
|
||||
return TAG_RE.sub(" ", value).replace("\xa0", " ").strip()
|
||||
|
||||
|
||||
def parse_detail(detail: str) -> tuple[str, str, list[str]]:
|
||||
parts = [part.strip() for part in detail.split(",")]
|
||||
direction = next((part for part in parts if part in {"정방향", "역방향"}), "unknown")
|
||||
position = next((part for part in parts if part in {"창측", "내측", "1인석"}), "unknown")
|
||||
notes = [part for part in parts if part not in {direction, position} and part]
|
||||
return direction, position, notes
|
||||
|
||||
|
||||
def parse_cars(html: str) -> list[SrtCar]:
|
||||
cars: list[SrtCar] = []
|
||||
for match in CAR_RE.finditer(html):
|
||||
body = match.group("body")
|
||||
text = strip_tags(body)
|
||||
room_class = "특실" if "특실" in text else "일반실"
|
||||
css_class = match.group("class")
|
||||
has_link = "selectScarInfo" in body
|
||||
cars.append(
|
||||
{
|
||||
"car_no": int(match.group("car")),
|
||||
"car_no_raw": f"{int(match.group('car')):04d}",
|
||||
"room_class": room_class,
|
||||
"available": has_link and "off" not in css_class.split(),
|
||||
"current": "on" in css_class.split(),
|
||||
}
|
||||
)
|
||||
return cars
|
||||
|
||||
|
||||
def parse_seats(html: str) -> list[SrtSeat]:
|
||||
seats: list[SrtSeat] = []
|
||||
seen: set[str] = set()
|
||||
for match in SEAT_LINK_RE.finditer(html):
|
||||
direction, position, notes = parse_detail(match.group("detail"))
|
||||
seat = match.group("seat")
|
||||
seen.add(seat)
|
||||
seats.append(
|
||||
{
|
||||
"seat": seat,
|
||||
"seat_no": match.group("seat_no"),
|
||||
"available": True,
|
||||
"direction": direction,
|
||||
"position": position,
|
||||
"notes": notes,
|
||||
}
|
||||
)
|
||||
for match in SEAT_SPAN_RE.finditer(html):
|
||||
seat = match.group("seat")
|
||||
if seat in seen:
|
||||
continue
|
||||
direction, position, notes = parse_detail(match.group("detail"))
|
||||
seats.append(
|
||||
{
|
||||
"seat": seat,
|
||||
"seat_no": "",
|
||||
"available": False,
|
||||
"direction": direction,
|
||||
"position": position,
|
||||
"notes": notes,
|
||||
}
|
||||
)
|
||||
return seats
|
||||
|
||||
|
||||
def parse_seat_label(seat_label: str) -> tuple[int | None, str]:
|
||||
match = re.match(r"^(\d+)([A-Z])$", seat_label)
|
||||
if match is None:
|
||||
return None, ""
|
||||
return int(match.group(1)), match.group(2)
|
||||
|
||||
|
||||
def car_center_priority(car: SrtCar, car_numbers: list[int]) -> tuple[float, int]:
|
||||
if not car_numbers:
|
||||
return (0.0, car["car_no"])
|
||||
center = (min(car_numbers) + max(car_numbers)) / 2
|
||||
return (abs(car["car_no"] - center), car["car_no"])
|
||||
|
||||
|
||||
def sort_cars_for_booking(cars: list[SrtCar], priority: str = "center") -> list[SrtCar]:
|
||||
match priority:
|
||||
case "center":
|
||||
car_numbers = [car["car_no"] for car in cars]
|
||||
return sorted(cars, key=lambda car: car_center_priority(car, car_numbers))
|
||||
case "low":
|
||||
return sorted(cars, key=lambda car: car["car_no"])
|
||||
case "high":
|
||||
return sorted(cars, key=lambda car: car["car_no"], reverse=True)
|
||||
case _:
|
||||
raise ValueError(f"unsupported car priority: {priority}")
|
||||
|
||||
|
||||
def seat_preference_key(seat: SrtSeat, priority: str = "forward-window") -> tuple[int, int, int, str]:
|
||||
row, column = parse_seat_label(seat["seat"])
|
||||
forward_rank = 0 if seat["direction"] == "정방향" else 1
|
||||
window_rank = 0 if seat["position"] in {"창측", "1인석"} else 1
|
||||
row_rank = 999 if row is None else row
|
||||
match priority:
|
||||
case "forward-window":
|
||||
return (forward_rank, window_rank, row_rank, column)
|
||||
case "window-forward":
|
||||
return (window_rank, forward_rank, row_rank, column)
|
||||
case "row-low":
|
||||
return (row_rank, forward_rank, window_rank, column)
|
||||
case _:
|
||||
raise ValueError(f"unsupported seat priority: {priority}")
|
||||
|
||||
|
||||
def sort_seats_for_booking(seats: list[SrtSeat], priority: str = "forward-window") -> list[SrtSeat]:
|
||||
return sorted(seats, key=lambda seat: seat_preference_key(seat, priority))
|
||||
|
||||
|
||||
sort_cars = sort_cars_for_booking
|
||||
sort_seats = sort_seats_for_booking
|
||||
209
scripts/test_srt_booking.py
Normal file
209
scripts/test_srt_booking.py
Normal file
|
|
@ -0,0 +1,209 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import io
|
||||
import json
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
from unittest.mock import patch
|
||||
|
||||
import srt_booking
|
||||
from srt_booking_test_support import EmptyClient, FakeClient, FakeTrain, NoisyClient, SpecialClient
|
||||
|
||||
|
||||
class SrtSeatTests(unittest.TestCase):
|
||||
def test_command_seats_outputs_available_seats_by_booking_preference(self) -> None:
|
||||
train = FakeTrain()
|
||||
train_id = srt_booking.build_train_id(train)
|
||||
client = FakeClient(train)
|
||||
args = argparse.Namespace(
|
||||
dep="수서",
|
||||
arr="부산",
|
||||
date="20260610",
|
||||
time="080000",
|
||||
time_limit=None,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
car_no=4,
|
||||
seat="6C",
|
||||
available_only=False,
|
||||
car_priority="center",
|
||||
seat_priority="forward-window",
|
||||
limit=10,
|
||||
)
|
||||
output = io.StringIO()
|
||||
|
||||
with patch.object(srt_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(output):
|
||||
srt_booking.command_seats(args)
|
||||
|
||||
result = json.loads(output.getvalue())
|
||||
car = result["cars"][0]
|
||||
self.assertEqual(car["car_no"], 4)
|
||||
self.assertTrue(car["requested_seat_available"])
|
||||
self.assertEqual(car["available_seats"], ["6C"])
|
||||
self.assertEqual(client._session.calls[-1]["scarNo1"], "0004")
|
||||
|
||||
def test_command_seats_filters_unavailable_when_available_only(self) -> None:
|
||||
train = FakeTrain()
|
||||
train_id = srt_booking.build_train_id(train)
|
||||
client = FakeClient(train)
|
||||
args = argparse.Namespace(
|
||||
dep="수서",
|
||||
arr="부산",
|
||||
date="20260610",
|
||||
time="080000",
|
||||
time_limit=None,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
car_no=4,
|
||||
seat=None,
|
||||
available_only=True,
|
||||
car_priority="center",
|
||||
seat_priority="forward-window",
|
||||
limit=10,
|
||||
)
|
||||
output = io.StringIO()
|
||||
|
||||
with patch.object(srt_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(output):
|
||||
srt_booking.command_seats(args)
|
||||
|
||||
result = json.loads(output.getvalue())
|
||||
shown_seats = result["cars"][0]["seats"]
|
||||
self.assertEqual([seat["seat"] for seat in shown_seats], ["6C", "3A"])
|
||||
self.assertTrue(all(seat["available"] for seat in shown_seats))
|
||||
|
||||
def test_command_seats_returns_special_room_cars(self) -> None:
|
||||
train = FakeTrain()
|
||||
train_id = srt_booking.build_train_id(train)
|
||||
client = SpecialClient(train)
|
||||
args = argparse.Namespace(
|
||||
dep="수서",
|
||||
arr="부산",
|
||||
date="20260610",
|
||||
time="080000",
|
||||
time_limit=None,
|
||||
train_id=train_id,
|
||||
room="special",
|
||||
car_no=3,
|
||||
seat=None,
|
||||
available_only=True,
|
||||
car_priority="center",
|
||||
seat_priority="window-forward",
|
||||
limit=10,
|
||||
)
|
||||
output = io.StringIO()
|
||||
|
||||
with patch.object(srt_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(output):
|
||||
srt_booking.command_seats(args)
|
||||
|
||||
result = json.loads(output.getvalue())
|
||||
self.assertEqual(result["room"], "special")
|
||||
self.assertEqual(result["cars"][0]["room_class"], "특실")
|
||||
self.assertEqual(result["cars"][0]["available_seats"], ["1A"])
|
||||
|
||||
def test_command_seats_fails_when_train_id_is_stale(self) -> None:
|
||||
train = FakeTrain()
|
||||
args = argparse.Namespace(
|
||||
dep="수서",
|
||||
arr="부산",
|
||||
date="20260610",
|
||||
time="080000",
|
||||
time_limit=None,
|
||||
train_id=srt_booking.build_train_id(train),
|
||||
room="general",
|
||||
car_no=4,
|
||||
seat=None,
|
||||
available_only=False,
|
||||
car_priority="center",
|
||||
seat_priority="forward-window",
|
||||
limit=10,
|
||||
)
|
||||
|
||||
with patch.object(srt_booking, "build_client", return_value=EmptyClient(train)):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
srt_booking.command_seats(args)
|
||||
|
||||
self.assertIn("train_id", str(exc.exception))
|
||||
|
||||
def test_command_seats_keeps_json_stdout_when_upstream_prints_queue_messages(self) -> None:
|
||||
train = FakeTrain()
|
||||
train_id = srt_booking.build_train_id(train)
|
||||
client = NoisyClient(train)
|
||||
args = argparse.Namespace(
|
||||
dep="수서",
|
||||
arr="부산",
|
||||
date="20260610",
|
||||
time="080000",
|
||||
time_limit=None,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
car_no=4,
|
||||
seat=None,
|
||||
available_only=True,
|
||||
car_priority="center",
|
||||
seat_priority="forward-window",
|
||||
limit=10,
|
||||
)
|
||||
output = io.StringIO()
|
||||
|
||||
with patch.object(srt_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(output):
|
||||
srt_booking.command_seats(args)
|
||||
|
||||
result = json.loads(output.getvalue())
|
||||
self.assertEqual(result["cars"][0]["available_seats"], ["6C", "3A"])
|
||||
|
||||
def test_command_seats_explores_middle_cars_first(self) -> None:
|
||||
train = FakeTrain()
|
||||
train_id = srt_booking.build_train_id(train)
|
||||
client = FakeClient(train)
|
||||
args = argparse.Namespace(
|
||||
dep="수서",
|
||||
arr="부산",
|
||||
date="20260610",
|
||||
time="080000",
|
||||
time_limit=None,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
car_no=None,
|
||||
seat=None,
|
||||
available_only=True,
|
||||
car_priority="center",
|
||||
seat_priority="forward-window",
|
||||
limit=10,
|
||||
)
|
||||
|
||||
with patch.object(srt_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(io.StringIO()):
|
||||
srt_booking.command_seats(args)
|
||||
|
||||
self.assertEqual([call["scarNo1"] for call in client._session.calls], ["", "0004", "0005"])
|
||||
|
||||
def test_build_parser_accepts_seats_filters(self) -> None:
|
||||
args = srt_booking.build_parser().parse_args([
|
||||
"seats",
|
||||
"수서",
|
||||
"부산",
|
||||
"20260610",
|
||||
"080000",
|
||||
"--train-id",
|
||||
"srt:v1:test",
|
||||
"--car-no",
|
||||
"5",
|
||||
"--seat",
|
||||
"11A",
|
||||
"--seat-priority",
|
||||
"window-forward",
|
||||
])
|
||||
|
||||
self.assertEqual(args.car_no, 5)
|
||||
self.assertEqual(args.seat, "11A")
|
||||
self.assertEqual(args.seat_priority, "window-forward")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
82
scripts/test_srt_seats.py
Normal file
82
scripts/test_srt_seats.py
Normal file
|
|
@ -0,0 +1,82 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
import srt_seats
|
||||
|
||||
|
||||
SEAT_HTML = "\n".join([
|
||||
'<li class="scar-01 off"><strong>일반실<br />1호차</strong></li>',
|
||||
'<li class="scar-04 on"><a href="#none" onclick="selectScarInfo(\'0004\'); return false;"><strong>일반실<br />4호차</strong></a></li>',
|
||||
'<li class="scar-05"><a href="#none" onclick="selectScarInfo(\'0005\'); return false;"><strong>일반실<br />5호차</strong></a></li>',
|
||||
'<li class="scar-03 off"><strong>특실<br />3호차</strong></li>',
|
||||
'<a href="#none" onclick="selectSeatInfo(this, \'23\', \'6C\'); return false;">6C<strong><em>(정방향, 내측)</em></strong></a>',
|
||||
'<a href="#none" onclick="selectSeatInfo(this, \'11\', \'3A\'); return false;">3A<strong><em>(역방향, 창측)</em></strong></a>',
|
||||
"<span>5C<strong><em>(정방향, 내측, 선택불가)</em></strong></span>",
|
||||
])
|
||||
|
||||
MISSING_DETAIL_HTML = "\n".join([
|
||||
'<a href="#none" onclick="selectSeatInfo(this, \'41\', \'9A\'); return false;">9A<strong><em>()</em></strong></a>',
|
||||
])
|
||||
|
||||
|
||||
class SrtSeatParserTests(unittest.TestCase):
|
||||
def test_normalize_car_and_seat_maps_srt_html(self) -> None:
|
||||
cars = srt_seats.parse_cars(SEAT_HTML)
|
||||
seats = srt_seats.parse_seats(SEAT_HTML)
|
||||
|
||||
self.assertEqual([car["car_no"] for car in cars if car["available"]], [4, 5])
|
||||
self.assertEqual(cars[1]["room_class"], "일반실")
|
||||
self.assertTrue(cars[1]["current"])
|
||||
self.assertEqual([seat["seat"] for seat in seats if seat["available"]], ["6C", "3A"])
|
||||
self.assertEqual([seat["seat"] for seat in seats if not seat["available"]], ["5C"])
|
||||
self.assertEqual(seats[0]["direction"], "정방향")
|
||||
self.assertEqual(seats[0]["position"], "내측")
|
||||
self.assertEqual(seats[2]["notes"], ["선택불가"])
|
||||
|
||||
def test_booking_priority_sorts_middle_cars_before_end_cars(self) -> None:
|
||||
cars: list[srt_seats.SrtCar] = [
|
||||
{"car_no": 1, "car_no_raw": "0001", "room_class": "일반실", "available": True, "current": False},
|
||||
{"car_no": 8, "car_no_raw": "0008", "room_class": "일반실", "available": True, "current": False},
|
||||
{"car_no": 2, "car_no_raw": "0002", "room_class": "일반실", "available": True, "current": False},
|
||||
{"car_no": 7, "car_no_raw": "0007", "room_class": "일반실", "available": True, "current": False},
|
||||
{"car_no": 3, "car_no_raw": "0003", "room_class": "일반실", "available": True, "current": False},
|
||||
{"car_no": 6, "car_no_raw": "0006", "room_class": "일반실", "available": True, "current": False},
|
||||
{"car_no": 4, "car_no_raw": "0004", "room_class": "일반실", "available": True, "current": False},
|
||||
{"car_no": 5, "car_no_raw": "0005", "room_class": "일반실", "available": True, "current": False},
|
||||
]
|
||||
|
||||
sorted_cars = srt_seats.sort_cars_for_booking(cars)
|
||||
|
||||
self.assertEqual([car["car_no"] for car in sorted_cars], [4, 5, 3, 6, 2, 7, 1, 8])
|
||||
|
||||
def test_booking_priority_sorts_forward_window_before_other_seats(self) -> None:
|
||||
seats: list[srt_seats.SrtSeat] = [
|
||||
{"seat": "3A", "seat_no": "11", "available": True, "direction": "역방향", "position": "창측", "notes": []},
|
||||
{"seat": "6C", "seat_no": "23", "available": True, "direction": "정방향", "position": "내측", "notes": []},
|
||||
{"seat": "2A", "seat_no": "7", "available": True, "direction": "정방향", "position": "창측", "notes": []},
|
||||
]
|
||||
|
||||
sorted_seats = srt_seats.sort_seats_for_booking(seats, "forward-window")
|
||||
|
||||
self.assertEqual([seat["seat"] for seat in sorted_seats], ["2A", "6C", "3A"])
|
||||
|
||||
def test_booking_priority_treats_single_seat_as_window_preference(self) -> None:
|
||||
seats: list[srt_seats.SrtSeat] = [
|
||||
{"seat": "1C", "seat_no": "3", "available": True, "direction": "정방향", "position": "내측", "notes": []},
|
||||
{"seat": "2A", "seat_no": "5", "available": True, "direction": "정방향", "position": "1인석", "notes": []},
|
||||
]
|
||||
|
||||
sorted_seats = srt_seats.sort_seats_for_booking(seats, "window-forward")
|
||||
|
||||
self.assertEqual([seat["seat"] for seat in sorted_seats], ["2A", "1C"])
|
||||
|
||||
def test_parse_seat_page_marks_missing_detail_attributes_unknown(self) -> None:
|
||||
seats = srt_seats.parse_seats(MISSING_DETAIL_HTML)
|
||||
|
||||
self.assertEqual(seats[0]["direction"], "unknown")
|
||||
self.assertEqual(seats[0]["position"], "unknown")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -12,12 +12,15 @@ metadata:
|
|||
|
||||
## What this skill does
|
||||
|
||||
`SRTrain` 위에서 SRT 좌석을 조회하고, 조건이 맞으면 예약과 취소까지 진행한다.
|
||||
`SRTrain` 위에 `scripts/srt_booking.py` helper 를 얹어 SRT 조회와 호차별 좌석번호 확인을 처리하고, 예약과 취소는 고정된 열차/예약을 다시 식별한 뒤 `SRTrain`으로 진행한다.
|
||||
|
||||
## When to use
|
||||
|
||||
- "수서에서 부산 가는 SRT 찾아줘"
|
||||
- "내일 오전 SRT 빈자리 있으면 잡아줘"
|
||||
- "SRT 5호차 빈 좌석 확인해줘"
|
||||
- "SRT 11A 좌석이 비었는지 봐줘"
|
||||
- "창측 좌석을 우선해서 SRT 빈자리 보여줘"
|
||||
- "예약 내역 확인해줘"
|
||||
- "이 SRT 예약 취소해줘"
|
||||
|
||||
|
|
@ -54,6 +57,7 @@ metadata:
|
|||
- 희망 시작 시각: `HHMMSS`
|
||||
- 인원 수와 승객 유형
|
||||
- 좌석 선호: 일반실 / 특실
|
||||
- 좌석 상세 조건: 객실 등급, 호차 번호, 좌석 번호, 빈 좌석만 보기, 탐색 우선순위
|
||||
|
||||
## Workflow
|
||||
|
||||
|
|
@ -73,19 +77,10 @@ python3 -m pip install SRTrain
|
|||
|
||||
### 2. Search first
|
||||
|
||||
먼저 조회해서 후보를 요약한다.
|
||||
먼저 helper 로 조회해서 후보를 요약한다.
|
||||
|
||||
```bash
|
||||
python3 - <<'PY'
|
||||
import os
|
||||
from SRT import SRT
|
||||
|
||||
srt = SRT(os.environ["KSKILL_SRT_ID"], os.environ["KSKILL_SRT_PASSWORD"])
|
||||
trains = srt.search_train("수서", "부산", "20260328", "080000", time_limit="120000")
|
||||
|
||||
for idx, train in enumerate(trains[:5], start=1):
|
||||
print(idx, train)
|
||||
PY
|
||||
python3 scripts/srt_booking.py search 수서 부산 20260328 080000 --time-limit 120000 --limit 5
|
||||
```
|
||||
|
||||
### 3. Summarize options before side effects
|
||||
|
|
@ -96,7 +91,38 @@ PY
|
|||
- 일반실/특실 가능 여부
|
||||
- 예상 운임
|
||||
|
||||
### 4. Reserve only after the train is fixed
|
||||
### 4. Inspect detailed seats when the user asks for seat numbers
|
||||
|
||||
`search` 의 좌석 가능 여부는 열차 단위 플래그다. 사용자가 "남은 좌석 번호", "호차별 좌석", "특정 좌석", "창측/순방향 자리", "예약 전에 자리 확인"처럼 구체적인 좌석을 물으면 예약 전에 `seats` 를 호출한다.
|
||||
|
||||
```bash
|
||||
python3 scripts/srt_booking.py seats 수서 부산 20260328 080000 --train-id <train_id>
|
||||
```
|
||||
|
||||
특정 호차의 빈 좌석만 확인하려면 `--car-no` 와 `--available-only` 를 쓴다.
|
||||
|
||||
```bash
|
||||
python3 scripts/srt_booking.py seats 수서 부산 20260328 080000 --train-id <train_id> --car-no 5 --available-only
|
||||
```
|
||||
|
||||
특정 좌석이 비었는지 확인하려면 `--seat` 를 붙인다.
|
||||
|
||||
```bash
|
||||
python3 scripts/srt_booking.py seats 수서 부산 20260328 080000 --train-id <train_id> --car-no 5 --seat 11A
|
||||
```
|
||||
|
||||
특정 호차를 지정하지 않으면 가운데 호차부터 탐색한다. `--car-priority center|low|high` 로 호차 탐색 순서를 바꾸고, `--seat-priority forward-window|window-forward|row-low` 로 좌석 정렬 우선순위를 바꾼다.
|
||||
|
||||
상세 좌석 응답을 보여줄 때는 아래를 우선 요약한다.
|
||||
|
||||
- 호차별 `available_seat_count`
|
||||
- 남은 좌석 번호 (`available_seats`)
|
||||
- 좌석별 `direction`, `position`
|
||||
- 특정 좌석 요청이면 `requested_seat_available`
|
||||
|
||||
이 기능은 좌석을 선택/선점하지 않는다. 실제 예약은 다음 단계에서만 진행한다.
|
||||
|
||||
### 5. Reserve only after the train is fixed
|
||||
|
||||
예약은 부작용이 있으므로 정확한 열차를 고른 뒤에만 진행한다.
|
||||
|
||||
|
|
@ -116,7 +142,7 @@ print(reservation)
|
|||
PY
|
||||
```
|
||||
|
||||
### 5. Inspect or cancel
|
||||
### 6. Inspect or cancel
|
||||
|
||||
취소 전에는 대상 예약을 다시 식별한다.
|
||||
|
||||
|
|
@ -134,6 +160,7 @@ PY
|
|||
## Done when
|
||||
|
||||
- 조회 요청이면 후보 열차가 정리되어 있다
|
||||
- 좌석 상세 확인이면 호차별 남은 좌석번호나 특정 좌석 공석 여부가 정리되어 있다
|
||||
- 예약 요청이면 예약 결과, 운임, 구입기한이 확인되어 있다
|
||||
- 취소 요청이면 어떤 예약을 취소했는지 명확하다
|
||||
|
||||
|
|
@ -141,10 +168,12 @@ PY
|
|||
|
||||
- 로그인 오류: 계정 정보나 SRT site policy 변경 가능성 확인
|
||||
- 매진: 다른 시간대나 좌석 타입으로 재조회
|
||||
- 좌석선택 페이지 형식 변경: helper 파서 업데이트 필요
|
||||
- 네트워크 오류: 짧게 재시도하되 aggressive polling은 피하기
|
||||
|
||||
## Notes
|
||||
|
||||
- 상세 좌석 확인은 SRT 웹 좌석선택 페이지를 조회 전용으로 읽는다
|
||||
- `SRTrain`은 SRT 전용 라이브러리라서 스킬 의도가 더 선명하다
|
||||
- 결제 완료까지는 자동화하지 않는다
|
||||
- 자동 재시도 루프는 계정 보호 차원에서 짧고 보수적으로 유지한다
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue