mirror of
https://github.com/NomaDamas/k-skill.git
synced 2026-06-24 02:04:11 +00:00
83079cd4c8
- 공개 endpoint (YES24 axPerfDay/PlayTime/RemainSeat, 인터파크 playSeq/REMAINSEAT) 만 단일 HTTP 호출 - httpx only, CloakBrowser/Playwright 없음, 로그인·시크릿·쿠키 없음 - 예매·결제·좌석 선택·자동화 의도적 제외 (공연법 §4조의2 매크로 부정구매 형사처벌) - 20 unit test (mocked httpx) + validate-skills.sh PASS - README + docs/features 가이드 추가
404 lines
14 KiB
Python
404 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""ticket-availability — YES24 / 인터파크 공연 일정 + 잔여석 조회 CLI.
|
||
|
||
조회 전용. 예매·결제·로그인 자동화 없음.
|
||
공연법 §4조의2 (매크로 입장권 부정구매·판매 금지) 비적용.
|
||
|
||
Usage:
|
||
ticket-availability schedule <url>
|
||
ticket-availability seats <url> [--all-dates]
|
||
ticket-availability health
|
||
|
||
Supported URLs:
|
||
YES24: https://ticket.yes24.com/Perf/<perf_id>
|
||
https://ticket.yes24.com/New/Perf/Detail/View/<perf_id>
|
||
yes24:<perf_id>
|
||
인터파크: https://tickets.interpark.com/goods/<goods_code>
|
||
interpark:<goods_code>
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
# ── URL Parsing ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def parse_url(url: str) -> tuple[str, str]:
|
||
"""Return (platform, id). Accepts full URL or `platform:id` shorthand."""
|
||
if url.startswith("yes24:"):
|
||
return "yes24", url[6:]
|
||
if url.startswith("interpark:"):
|
||
return "interpark", url[10:]
|
||
|
||
m = re.search(
|
||
r"yes24\.com/(?:[Nn]ew/)?[Pp]erf/(?:[Dd]etail/)?(?:[Vv]iew/)?(\d+)", url
|
||
)
|
||
if m:
|
||
return "yes24", m.group(1)
|
||
|
||
m = re.search(r"interpark\.com/goods/(\d+)", url, re.IGNORECASE)
|
||
if m:
|
||
return "interpark", m.group(1)
|
||
|
||
if re.fullmatch(r"\d+", url):
|
||
raise ValueError(
|
||
f"플랫폼을 명시하세요: yes24:{url} 또는 interpark:{url}"
|
||
)
|
||
|
||
raise ValueError(f"URL을 인식할 수 없습니다: {url}")
|
||
|
||
|
||
def _fmt_date(d: str) -> str:
|
||
if d and len(d) == 8 and d.isdigit():
|
||
return f"{d[:4]}-{d[4:6]}-{d[6:]}"
|
||
return d
|
||
|
||
|
||
def _fmt_time(t: str) -> str:
|
||
if t and len(t) == 4 and t.isdigit():
|
||
return f"{t[:2]}:{t[2:]}"
|
||
return t
|
||
|
||
|
||
# ── HTTP Setup ────────────────────────────────────────────────────────────────
|
||
|
||
UA = (
|
||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
||
"Chrome/124.0.0.0 Safari/537.36"
|
||
)
|
||
|
||
HEADERS_YES24 = {
|
||
"User-Agent": UA,
|
||
"Referer": "https://ticket.yes24.com/",
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Accept": "*/*",
|
||
"X-Requested-With": "XMLHttpRequest",
|
||
}
|
||
|
||
HEADERS_INTERPARK = {
|
||
"User-Agent": UA,
|
||
"Referer": "https://tickets.interpark.com/",
|
||
"Accept": "application/json",
|
||
}
|
||
|
||
YES24_BASE = "https://ticket.yes24.com"
|
||
INTERPARK_BASE = "https://api-ticketfront.interpark.com"
|
||
|
||
|
||
# ── YES24 Client ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
class Yes24Client:
|
||
def __init__(self) -> None:
|
||
self.http = httpx.Client(
|
||
headers=HEADERS_YES24, timeout=20, follow_redirects=True
|
||
)
|
||
|
||
def _dates(self, perf_id: str, month_count: int) -> list[str]:
|
||
now = datetime.now()
|
||
months: list[str] = []
|
||
for delta in range(month_count):
|
||
month = now.month + delta
|
||
year = now.year + (month - 1) // 12
|
||
month = ((month - 1) % 12) + 1
|
||
months.append(f"{year:04d}-{month:02d}")
|
||
|
||
dates: list[str] = []
|
||
cutoff = now.strftime("%Y%m%d")
|
||
for month_str in months:
|
||
r = self.http.post(
|
||
f"{YES24_BASE}/New/Perf/Sale/Ajax/axPerfDay.aspx",
|
||
data={
|
||
"pGetMode": "days",
|
||
"pIdPerf": perf_id,
|
||
"pPerfMonth": month_str,
|
||
"pIdCode": "",
|
||
"pIsMania": "0",
|
||
},
|
||
)
|
||
r.raise_for_status()
|
||
text = r.text.strip().strip(",")
|
||
if not text:
|
||
continue
|
||
for raw in text.split(","):
|
||
d = raw.strip()
|
||
if not d:
|
||
continue
|
||
normalized = d.replace("-", "")
|
||
if normalized >= cutoff:
|
||
dates.append(normalized)
|
||
return sorted(set(dates))
|
||
|
||
def get_dates(self, perf_id: str) -> list[str]:
|
||
"""Available dates within ~3 weeks (fast)."""
|
||
return self._dates(perf_id, month_count=3)
|
||
|
||
def get_all_dates(self, perf_id: str) -> list[str]:
|
||
"""Available dates across 6 months (full schedule)."""
|
||
return self._dates(perf_id, month_count=6)
|
||
|
||
def get_slots(self, perf_id: str, perf_day: str) -> list[dict]:
|
||
r = self.http.post(
|
||
f"{YES24_BASE}/NEw/Perf/Detail/Ajax/axPerfPlayTime.aspx",
|
||
data={"IdPerf": perf_id, "PerfDay": perf_day},
|
||
)
|
||
r.raise_for_status()
|
||
html = r.text
|
||
slots: list[dict] = []
|
||
seen: set[str] = set()
|
||
for m in re.finditer(r"idTime='(\d+)'", html):
|
||
id_time = m.group(1)
|
||
if id_time in seen:
|
||
continue
|
||
seen.add(id_time)
|
||
ctx_start = max(0, m.start() - 200)
|
||
ctx = html[ctx_start : m.end() + 200]
|
||
time_m = re.search(r"(\d{1,2}:\d{2}|\d[회]|[12]\d{3}회)", ctx)
|
||
label = time_m.group(0) if time_m else id_time
|
||
slots.append({"idTime": id_time, "label": label})
|
||
return slots
|
||
|
||
def get_seats(self, id_time: str) -> list[dict]:
|
||
r = self.http.post(
|
||
f"{YES24_BASE}/New/Perf/Detail/Ajax/axPerfRemainSeat.aspx",
|
||
data={"Type": "calendar", "IdTime": id_time, "IdLock": "0"},
|
||
)
|
||
r.raise_for_status()
|
||
html = r.text
|
||
seats: list[dict] = []
|
||
for m in re.finditer(
|
||
r"<dt>([^<]+)</dt>\s*<dd>([^<]*)<span[^>]*>\(잔여:(\d+)석\)</span>",
|
||
html,
|
||
):
|
||
seats.append(
|
||
{
|
||
"grade": m.group(1).strip(),
|
||
"price": m.group(2).strip().rstrip(",").strip(),
|
||
"remain": int(m.group(3)),
|
||
}
|
||
)
|
||
if not seats:
|
||
for i, m in enumerate(re.finditer(r"\(잔여:(\d+)석\)", html)):
|
||
seats.append({"grade": f"좌석{i+1}", "price": "", "remain": int(m.group(1))})
|
||
return seats
|
||
|
||
def schedule(self, perf_id: str, all_dates: bool) -> list[dict]:
|
||
"""Schedule = dates × slots flattened. No seat lookup."""
|
||
dates = self.get_all_dates(perf_id) if all_dates else self.get_dates(perf_id)
|
||
out: list[dict] = []
|
||
for d in dates:
|
||
for slot in self.get_slots(perf_id, d):
|
||
out.append(
|
||
{
|
||
"date": _fmt_date(d),
|
||
"time_label": slot["label"],
|
||
"id_time": slot["idTime"],
|
||
}
|
||
)
|
||
return out
|
||
|
||
def all_seats(self, perf_id: str, all_dates: bool) -> dict:
|
||
result: dict = {}
|
||
dates = self.get_all_dates(perf_id) if all_dates else self.get_dates(perf_id)
|
||
for d in dates:
|
||
for slot in self.get_slots(perf_id, d):
|
||
seats = self.get_seats(slot["idTime"])
|
||
key = f"{_fmt_date(d)}|{slot['label']}"
|
||
result[key] = {
|
||
"date": _fmt_date(d),
|
||
"time_label": slot["label"],
|
||
"id_time": slot["idTime"],
|
||
"seats": seats,
|
||
}
|
||
time.sleep(0.4)
|
||
return result
|
||
|
||
|
||
# ── Interpark Client ──────────────────────────────────────────────────────────
|
||
|
||
|
||
class InterparkClient:
|
||
def __init__(self) -> None:
|
||
self.http = httpx.Client(
|
||
headers=HEADERS_INTERPARK, timeout=20, follow_redirects=True
|
||
)
|
||
|
||
def get_schedule(self, goods_code: str) -> list[dict]:
|
||
now = datetime.now()
|
||
r = self.http.get(
|
||
f"{INTERPARK_BASE}/v1/goods/{goods_code}/playSeq",
|
||
params={
|
||
"goodsCode": goods_code,
|
||
"isBookableDate": "true",
|
||
"page": "1",
|
||
"pageSize": "200",
|
||
"startDate": now.strftime("%Y%m%d"),
|
||
"endDate": f"{now.year + 1}{now.month:02d}{now.day:02d}",
|
||
},
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
if isinstance(data, list):
|
||
return data
|
||
return data.get("response", {}).get("data") or data.get("data") or []
|
||
|
||
def get_seats(self, goods_code: str, play_seq: str) -> list[dict]:
|
||
r = self.http.get(
|
||
f"{INTERPARK_BASE}/v1/goods/{goods_code}/playSeq/PlaySeq/{play_seq}/REMAINSEAT"
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
if isinstance(data, dict):
|
||
return (
|
||
data.get("remainSeat")
|
||
or (data.get("data") or {}).get("remainSeat")
|
||
or data.get("response", {}).get("remainSeat")
|
||
or []
|
||
)
|
||
return []
|
||
|
||
def schedule(self, goods_code: str) -> list[dict]:
|
||
out: list[dict] = []
|
||
for item in self.get_schedule(goods_code):
|
||
out.append(
|
||
{
|
||
"date": _fmt_date(item.get("playDate", "")),
|
||
"time": _fmt_time(item.get("playTime", "")),
|
||
"play_seq": item.get("playSeq", ""),
|
||
}
|
||
)
|
||
return out
|
||
|
||
def all_seats(self, goods_code: str) -> dict:
|
||
result: dict = {}
|
||
for item in self.get_schedule(goods_code):
|
||
seq = item.get("playSeq", "")
|
||
if not seq:
|
||
continue
|
||
seats_raw = self.get_seats(goods_code, seq)
|
||
normalized = [
|
||
{
|
||
"grade": s.get("seatGradeName", s.get("seatGrade", "")),
|
||
"remain": int(s.get("remainCnt", 0)),
|
||
}
|
||
for s in seats_raw
|
||
]
|
||
key = f"{_fmt_date(item.get('playDate', ''))}|{_fmt_time(item.get('playTime', ''))}|{seq}"
|
||
result[key] = {
|
||
"date": _fmt_date(item.get("playDate", "")),
|
||
"time": _fmt_time(item.get("playTime", "")),
|
||
"play_seq": seq,
|
||
"seats": normalized,
|
||
}
|
||
time.sleep(0.3)
|
||
return result
|
||
|
||
|
||
# ── CLI ───────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _dump(obj: Any, compact: bool) -> str:
|
||
if compact:
|
||
return json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
|
||
return json.dumps(obj, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def cmd_schedule(args: argparse.Namespace) -> int:
|
||
platform, pid = parse_url(args.url)
|
||
if platform == "yes24":
|
||
out = Yes24Client().schedule(pid, all_dates=args.all_dates)
|
||
else:
|
||
out = InterparkClient().schedule(pid)
|
||
print(_dump({"platform": platform, "id": pid, "schedule": out}, args.compact))
|
||
return 0
|
||
|
||
|
||
def cmd_seats(args: argparse.Namespace) -> int:
|
||
platform, pid = parse_url(args.url)
|
||
if platform == "yes24":
|
||
out = Yes24Client().all_seats(pid, all_dates=args.all_dates)
|
||
else:
|
||
out = InterparkClient().all_seats(pid)
|
||
print(_dump({"platform": platform, "id": pid, "seats": out}, args.compact))
|
||
return 0
|
||
|
||
|
||
def cmd_health(args: argparse.Namespace) -> int:
|
||
results: dict = {}
|
||
for name, url in [
|
||
("yes24",
|
||
f"{YES24_BASE}/New/Perf/Sale/Ajax/axPerfDay.aspx"),
|
||
("interpark",
|
||
f"{INTERPARK_BASE}/v1/goods/00000000/playSeq"),
|
||
]:
|
||
try:
|
||
if name == "yes24":
|
||
r = httpx.post(url, headers=HEADERS_YES24,
|
||
data={"pGetMode": "days", "pIdPerf": "0",
|
||
"pPerfMonth": "2000-01", "pIdCode": "",
|
||
"pIsMania": "0"}, timeout=10)
|
||
else:
|
||
r = httpx.get(url, headers=HEADERS_INTERPARK,
|
||
params={"goodsCode": "00000000",
|
||
"isBookableDate": "true",
|
||
"page": "1", "pageSize": "1",
|
||
"startDate": "20000101",
|
||
"endDate": "20000102"},
|
||
timeout=10)
|
||
results[name] = {"status": r.status_code, "ok": r.status_code < 500}
|
||
except Exception as e:
|
||
results[name] = {"status": 0, "ok": False, "error": str(e)}
|
||
print(_dump(results, args.compact))
|
||
return 0 if all(v.get("ok") for v in results.values()) else 1
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(
|
||
prog="ticket-availability",
|
||
description="YES24 / 인터파크 공연 일정 + 잔여석 조회 (조회 전용)",
|
||
)
|
||
sub = parser.add_subparsers(dest="cmd", required=True)
|
||
|
||
def _common(p: argparse.ArgumentParser) -> None:
|
||
p.add_argument("--compact", action="store_true",
|
||
help="One-line JSON (기본: 들여쓰기 출력)")
|
||
|
||
p_sch = sub.add_parser("schedule", help="공연 일정 조회")
|
||
p_sch.add_argument("url", help="공연 URL 또는 platform:id")
|
||
p_sch.add_argument("--all-dates", action="store_true",
|
||
help="YES24 — 6개월 전체 (기본: 3주)")
|
||
_common(p_sch)
|
||
p_sch.set_defaults(func=cmd_schedule)
|
||
|
||
p_st = sub.add_parser("seats", help="등급별 잔여석 조회 (전 일정)")
|
||
p_st.add_argument("url", help="공연 URL 또는 platform:id")
|
||
p_st.add_argument("--all-dates", action="store_true",
|
||
help="YES24 — 6개월 전체 (기본: 3주)")
|
||
_common(p_st)
|
||
p_st.set_defaults(func=cmd_seats)
|
||
|
||
p_h = sub.add_parser("health", help="API endpoint reachability check")
|
||
_common(p_h)
|
||
p_h.set_defaults(func=cmd_health)
|
||
|
||
args = parser.parse_args(argv)
|
||
try:
|
||
return args.func(args)
|
||
except ValueError as e:
|
||
print(f"error: {e}", file=sys.stderr)
|
||
return 2
|
||
except httpx.HTTPError as e:
|
||
print(f"http error: {e}", file=sys.stderr)
|
||
return 3
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|