from __future__ import annotations
import argparse
import json
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Iterable
USER_AGENT = "Mozilla/5.0 (compatible; SoopClient/3.0)"
PLAYER_LIVE_API_URL = "https://live.sooplive.com/afreeca/player_live_api.php"
EMOTICON_API_URL = "https://live.sooplive.com/api/signature_emoticon_api.php"
VOD_DETAIL_API_URL = "https://api.m.sooplive.co.kr/station/video/a/view"
@dataclass(frozen=True)
class ProbeRequest:
name: str
method: str
url: str
form: dict[str, str] | None = None
@dataclass(frozen=True)
class ProbeResponse:
status: int
reason: str
final_url: str
headers: list[tuple[str, str]]
body: bytes
elapsed_ms: int
def live_status_request(streamer_id: str) -> ProbeRequest:
query = urllib.parse.urlencode({"bjid": streamer_id})
return ProbeRequest(
name="live-status",
method="POST",
url=f"{PLAYER_LIVE_API_URL}?{query}",
form={"bid": streamer_id},
)
def station_request(streamer_id: str) -> ProbeRequest:
quoted_id = urllib.parse.quote(streamer_id, safe="")
return ProbeRequest(
name="station",
method="GET",
url=f"https://chapi.sooplive.co.kr/api/{quoted_id}/station",
)
def emoticon_request(streamer_id: str) -> ProbeRequest:
return ProbeRequest(
name="emoticon",
method="POST",
url=EMOTICON_API_URL,
form={"szBjId": streamer_id, "work": "list", "v": "tier"},
)
def vod_list_request(streamer_id: str, page: int) -> ProbeRequest:
quoted_id = urllib.parse.quote(streamer_id, safe="")
query = urllib.parse.urlencode(
{
"page": page,
"per_page": 60,
"orderby": "reg_date",
"field": "title,contents",
"created": "false",
}
)
return ProbeRequest(
name="vod-list",
method="GET",
url=f"https://chapi.sooplive.co.kr/api/{quoted_id}/vods/review?{query}",
)
def vod_detail_request(vod_id: int) -> ProbeRequest:
return ProbeRequest(
name="vod-detail",
method="POST",
url=VOD_DETAIL_API_URL,
form={"nTitleNo": str(vod_id)},
)
def vod_chat_request(chat_url: str, start_time: int) -> ProbeRequest:
separator = "&" if "?" in chat_url else "?"
return ProbeRequest(
name="vod-chat",
method="GET",
url=f"{chat_url}{separator}{urllib.parse.urlencode({'startTime': start_time})}",
)
def build_url_request(probe: ProbeRequest) -> urllib.request.Request:
headers = {"User-Agent": USER_AGENT}
data = None
if probe.form is not None:
data = urllib.parse.urlencode(probe.form).encode("utf-8")
headers["Content-Type"] = "application/x-www-form-urlencoded"
return urllib.request.Request(
probe.url,
data=data,
headers=headers,
method=probe.method,
)
def send(probe: ProbeRequest, timeout: float) -> ProbeResponse:
request = build_url_request(probe)
started = time.monotonic()
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
body = response.read()
elapsed_ms = int((time.monotonic() - started) * 1000)
return ProbeResponse(
status=response.status,
reason=response.reason,
final_url=response.url,
headers=list(response.headers.items()),
body=body,
elapsed_ms=elapsed_ms,
)
except urllib.error.HTTPError as error:
body = error.read()
elapsed_ms = int((time.monotonic() - started) * 1000)
return ProbeResponse(
status=error.code,
reason=error.reason,
final_url=error.url,
headers=list(error.headers.items()),
body=body,
elapsed_ms=elapsed_ms,
)
def decode_body(body: bytes) -> str:
return body.decode("utf-8", errors="replace")
def format_body(body: bytes, raw: bool) -> str:
text = decode_body(body)
if raw:
return text
try:
parsed = json.loads(text)
except json.JSONDecodeError:
return text
return json.dumps(parsed, ensure_ascii=False, indent=2)
def truncate(text: str, limit: int | None) -> str:
if limit is None or len(text) <= limit:
return text
omitted = len(text) - limit
return f"{text[:limit]}\n... truncated {omitted} chars; rerun with --full"
def print_probe(
probe: ProbeRequest,
response: ProbeResponse,
*,
raw: bool,
full: bool,
show_response_headers: bool,
) -> None:
print(f"## {probe.name}")
print()
print("Request:")
print(f" {probe.method} {probe.url}")
print(f" User-Agent: {USER_AGENT}")
if probe.form is not None:
print(" Content-Type: application/x-www-form-urlencoded")
print(f" Form: {json.dumps(probe.form, ensure_ascii=False)}")
print()
print("Response:")
print(f" Status: {response.status} {response.reason}")
print(f" Final URL: {response.final_url}")
print(f" Elapsed: {response.elapsed_ms} ms")
print(f" Body bytes: {len(response.body)}")
if show_response_headers:
print(" Headers:")
for key, value in response.headers:
print(f" {key}: {value}")
print()
print("Body:")
body = format_body(response.body, raw=raw)
print(truncate(body, None if full else 20_000))
print()
def run_probes(args: argparse.Namespace, probes: Iterable[ProbeRequest]) -> int:
exit_code = 0
for index, probe in enumerate(probes):
if index > 0:
print("=" * 80)
print()
try:
response = send(probe, timeout=args.timeout)
except urllib.error.URLError as error:
print(f"## {probe.name}", file=sys.stderr)
print(f"request failed: {error}", file=sys.stderr)
exit_code = 1
continue
print_probe(
probe,
response,
raw=args.raw,
full=args.full,
show_response_headers=args.show_response_headers,
)
if response.status < 200 or response.status >= 300:
exit_code = 1
return exit_code
def positive_int(value: str) -> int:
parsed = int(value)
if parsed < 0:
raise argparse.ArgumentTypeError("must be zero or greater")
return parsed
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Probe SOOP HTTP APIs used by soup-sdk and print raw responses."
)
subparsers = parser.add_subparsers(dest="command", required=True)
def add_common_flags(subparser: argparse.ArgumentParser) -> None:
subparser.add_argument(
"--timeout", type=float, default=15.0, help="request timeout seconds"
)
subparser.add_argument(
"--raw", action="store_true", help="print JSON bodies without formatting"
)
subparser.add_argument(
"--full", action="store_true", help="do not truncate large bodies"
)
subparser.add_argument(
"--show-response-headers",
action="store_true",
help="include all response headers",
)
live = subparsers.add_parser("live-status", help="POST player_live_api.php")
live.add_argument("streamer_id")
add_common_flags(live)
station = subparsers.add_parser("station", help="GET station summary")
station.add_argument("streamer_id")
add_common_flags(station)
emoticon = subparsers.add_parser("emoticon", help="POST signature emoticon API")
emoticon.add_argument("streamer_id")
add_common_flags(emoticon)
vod_list = subparsers.add_parser("vod-list", help="GET review VOD list")
vod_list.add_argument("streamer_id")
vod_list.add_argument("--page", type=positive_int, default=1)
add_common_flags(vod_list)
vod_detail = subparsers.add_parser("vod-detail", help="POST VOD detail API")
vod_detail.add_argument("vod_id", type=positive_int)
add_common_flags(vod_detail)
vod_chat = subparsers.add_parser("vod-chat", help="GET VOD chat XML chunk")
vod_chat.add_argument("chat_url")
vod_chat.add_argument("--start-time", type=positive_int, default=0)
add_common_flags(vod_chat)
all_probe = subparsers.add_parser(
"all", help="run live-status, station, emoticon, and vod-list"
)
all_probe.add_argument("streamer_id")
all_probe.add_argument("--page", type=positive_int, default=1)
add_common_flags(all_probe)
return parser
def main() -> int:
parser = build_parser()
args = parser.parse_args()
if args.command == "live-status":
probes = [live_status_request(args.streamer_id)]
elif args.command == "station":
probes = [station_request(args.streamer_id)]
elif args.command == "emoticon":
probes = [emoticon_request(args.streamer_id)]
elif args.command == "vod-list":
probes = [vod_list_request(args.streamer_id, args.page)]
elif args.command == "vod-detail":
probes = [vod_detail_request(args.vod_id)]
elif args.command == "vod-chat":
probes = [vod_chat_request(args.chat_url, args.start_time)]
elif args.command == "all":
probes = [
live_status_request(args.streamer_id),
station_request(args.streamer_id),
emoticon_request(args.streamer_id),
vod_list_request(args.streamer_id, args.page),
]
else:
parser.error(f"unknown command: {args.command}")
return run_probes(args, probes)
if __name__ == "__main__":
raise SystemExit(main())