Skip to content

Instantly share code, notes, and snippets.

@LifeHackerBee

LifeHackerBee/gist:17159a18c737a48ed35dbe2eb8c13850

Created July 20, 2025 01:01
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save LifeHackerBee/17159a18c737a48ed35dbe2eb8c13850 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save LifeHackerBee/17159a18c737a48ed35dbe2eb8c13850 to your computer and use it in GitHub Desktop.
Download ZIP
TicketMaster
Raw
gistfile1.txt
import base64
from PIL import Image
import pytesseract
from io import BytesIO
import requests
import time
import re
import threading
import json
from tqdm import tqdm
def get_captcha_base64():
url = "https://ticket.zxgbdjykf.mil.cn/japi/sw-trm-cloud/api/personalReserve/queryImgCode"
headers = {
"Authorizationc": "xxx",
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.60(0x18003c31) NetType/4G Language/zh_CN",
"Referer": "https://servicewechat.com/xxxx/6/page-frame.html"
}
payload = {"type": 0}
resp = requests.post(url, headers=headers, json=payload, timeout=10)
resp.raise_for_status()
resp_json = resp.json()
base64_img = resp_json.get('data', '').strip()
return base64_img
def get_captcha():
base64_img = get_captcha_base64()
result = recognize_captcha_from_base64(base64_img)
return result
def recognize_captcha_from_base64(base64_str: str) -> str:
# 1. 解码 base64 为二进制
img_bytes = base64.b64decode(base64_str)
# 2. 用 PIL 打开图片并转为灰度
img = Image.open(BytesIO(img_bytes)).convert("L")
# 3. 用 pytesseract 识别
config = '--psm 7 -c tessedit_char_whitelist=0123456789'
text = pytesseract.image_to_string(img, config=config)
# 只保留数字，并取前5位
digits = re.findall(r'\d', text)
code = ''.join(digits)[:5]
print(f"识别验证码: {code}")
return code
def get_available_intervals(companyInfoId, reserve_date):
url = "https://ticket.zxgbdjykf.mil.cn/japi/sw-trm-cloud/api/personalReserve/queryPersonal"
headers = {
"Appid": "xxxx",
"Content-Type": "application/json",
"Token": "xxxx",
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.60(0x18003c31) NetType/WIFI Language/zh_CN",
"Referer": "https://servicewechat.com/xxxx/6/page-frame.html",
"Accept": "application/json"
}
payload = {
"systemId": "4ea641273b8246f38c79c66863af321c",
"companyInfoId": companyInfoId
}
try:
resp = requests.post(url, headers=headers, json=payload, timeout=10)
resp.raise_for_status()
data = resp.json().get("data", [])
for day in data:
if day.get("dayTime") == reserve_date:
return [
(item["intervalValue"], item["surplusId"])
for item in day.get("surplusList", [])
if item.get("surplusCount", 0) > 0
]
except Exception as e:
print(f"获取余票信息失败: {e}")
return []
def send_bark_msg(msg: str):
bark_url = "https://api.day.app/xxxx/GetTicket"
try:
resp = requests.get(bark_url, params={"body": msg}, timeout=5)
print(f"Bark推送: {msg}，返回: {resp.status_code}")
except Exception as e:
print(f"Bark推送失败: {e}")
def send_bark_msg_init():
bark_url = "https://api.day.app/xxxx/TestTicket"
try:
resp = requests.get(bark_url, timeout=5)
print(f"Bark推送: 初始化成功，返回: {resp.status_code}")
except Exception as e:
print(f"Bark推送失败: {e}")
def submit_ticket(
reserve_date,
reserve_time,
company_info_id,
order_personal_list,
img_code,
cert_info,
expire_time,
names="",
authorizationc=None
):
url = "https://ticket.zxgbdjykf.mil.cn/japi/sw-trm-cloud/api/personalReserve/saveForPrepareSubmit"
headers = {
"Authorizationc": authorizationc or "xxxx",
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.60(0x18003c31) NetType/WIFI Language/zh_CN",
"Referer": "https://servicewechat.com/xxxx/6/page-frame.html"
}
payload = {
"reserveDate": reserve_date,
"reserveTime": reserve_time,
"companyInfoId": company_info_id,
"orderPersonalList": order_personal_list,
"imgCode": img_code,
"reserveFrom": "TRM0305",
"certInfo": cert_info,
"expireTime": expire_time,
"names": names
}
resp = requests.post(url, headers=headers, json=payload, timeout=10)
resp.raise_for_status()
return resp.json()
if __name__ == "__main__":
# 示例：给定payload字符串，解析后用submit_ticket发送
payload_str = '''
{"reserveDate":"2025-07-06","reserveTime":"13:30-14:30","companyInfoId":"b8346b3e9ea99c5727e0993619414zyj","orderPersonalList":[{"visitorName":"xxx","documentType":"xxx","documentNumber":"xxx","phoneNumber":"xxx","customerContactId":"xxx"},{"visitorName":"xxx","documentType":"xxx","documentNumber":"xxx","phoneNumber":"xxx","customerContactId":"xxx"}],"imgCode":"1231","reserveFrom":"xxx","certInfo":{"visitorName":"xxx","documentNumber":"xxx","phoneNumber":"xxx"},"expireTime":"xxx","names":""}
'''
payload = json.loads(payload_str)
attempts = 0
with tqdm(desc="抢票尝试次数", unit="次") as pbar:
while True:
result = submit_ticket(
reserve_date=payload["reserveDate"],
reserve_time=payload["reserveTime"],
company_info_id=payload["companyInfoId"],
order_personal_list=payload["orderPersonalList"],
img_code=get_captcha(),
cert_info=payload["certInfo"],
expire_time=payload["expireTime"],
names=payload.get("names", "")
)
attempts += 1
pbar.update(1)
if result.get('code') == 200:
send_bark_msg(f" 订票返回: code={result.get('code')} msg={result.get('msg')} 全部: {result}")
print(f"抢票成功，总共尝试了 {attempts} 次！")
break
print(result)
time.sleep(0.5)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment