Forked from LifeHackerBee/gist:17159a18c737a48ed35dbe2eb8c13850
Created
July 22, 2025 06:37
-
-
Save jiang-yuan/d52f3e204e56c8dc11c47e54a4b93cff to your computer and use it in GitHub Desktop.
TicketMaster
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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