基于requests库和多线程的teket.jp抢票脚本
脚本实现了自动登录和购买票务的功能,有效地提高了购票的成功率
虽然脚本的执行效率很高,但是短时间内大量请求可能会导致IP被封禁
关于为什么要写这个脚本:纯纯工具人,谁懂啊家人们……
requestsライブラリとマルチスレッドを利用したteket.jpチケット予約スクリプト
このスクリプトはteket.jpに自動的にログインし、チケットを予約する機能を実現します、効率的にチケットの予約成功率を向上させます
スクリプトの実行効率は非常に高いですが、短時間で大量のリクエストを送るとIPがブロックされる可能性があります
スクリプトを書く理由:純粋なツール人間、誰が理解してくれるんだろう……家族のみんな……
一、脚本的主要功能
-
这个Python脚本的主要功能是自动登录teket.jp网站并购买票务
-
购票操作包括获取演出列表、确认选票和购买票务等步骤
-
注意:脚本目前只支持免费票!!!脚本不支持自动选座!!!
-
teket里的所有请求内容都怕大家找不到,开发人员直接写在html里面……
このPythonスクリプトの主な機能は、teket.jpウェブサイトに自動的にログインし、チケットを予約することです。
チケット予約操作には、パフォーマンスリストの取得、チケットの確認、チケットの予約などのステップが含まれます。
注意:スクリプトは現時点では無料チケットのみをサポートしています!!!スクリプトは自動的な座席選択をサポートしていません!!!
teketのすべてのリクエスト内容は、開発者が皆さんが見つけられないことを恐れて、直接htmlに書いています……
二、导入必要的库
脚本需要导入以下Python库:
import requests
import threading
import re
import json
import time
import random
from datetime import datetime
三、全局变量的作用
-
ticket_urls
为需要抢票的演出,required_num
每场演出抢票的场次数量 -
脚本使用了几个全局变量来控制请求的频率和重试的次数:
last_request_time
用来记录上次发送请求的时间,REQUEST_THRESHOLD
是两次请求之间的最小间隔时间,MAX_RETRIES
是每个请求的最大重试次数。
login_session
用于储存登录后的cookies,在特定的步骤会需要
web_url='https://teket.jp'
ticket_urls = [''] #ticket_urls = ['/238/24063','/238/24065','/238/24066']
#performance numbers
required_num = 3
# Headers used for HTTP requests
headers={
"Accept":"application/json, text/javascript, */*; q=0.01",
"Accept-Encoding":"gzip, deflate, br",
"Cache-Control":"no-cache",
"Dnt":"1",
"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.58"
}
# Create a global session and update its headers
global_session = requests.Session()
global_session.headers.update(headers)
# Create a login session and update its headers
login_session = requests.Session()
login_session.headers.update(headers)
# Set the maximum number of retries for a failed request
MAX_RETRIES = 5
last_request_time = 0
REQUEST_THRESHOLD = 0.5 # 全局阈值,单位为秒
四、函数的功能和实现方式
4.0发送HTTP请求
①safe_request()
safe_request()
是一个辅助函数,用于发送HTTP请求。
-
参数,包括session(用于发送请求的会话),method(HTTP方法,如GET或POST),url(请求的URL),以及一些其他的可选参数。这个函数的主要功能是根据提供的参数发送HTTP请求,并处理可能发生的错误。
-
函数使用了一个全局变量
last_request_time
来记录上次发送请求的时间。在发送请求之前,函数会检查上次请求的时间,如果距离上次请求的时间小于REQUEST_THRESHOLD
(另一个全局变量,表示两次请求之间的最小间隔时间),则函数会等待一段时间。这样做的目的是防止短时间内发送过多的请求,从而避免被服务器封禁。 -
如果请求失败,函数会根据全局变量
MAX_RETRIES
(表示最大的重试次数)的值进行重试。在每次重试之前,函数会等待一段随机的时间,这样做的目的是模拟人类的行为,防止服务器检测到异常。 -
这个函数的返回值是服务器的响应。如果请求成功,函数会打印一个消息并返回服务器的响应。如果请求失败,并且已经达到最大的重试次数,函数会抛出一个异常。
def safe_request(session, method, url, expected_status=200, **kwargs):
global last_request_time # 使用全局变量
for i in range(MAX_RETRIES):
try:
# 检查上次请求的时间,如果当前时间与上次请求时间的差值小于阈值,则等待
while time.time() - last_request_time < REQUEST_THRESHOLD:
time.sleep(0.05) # 每次等待0.1秒后再检查
response = session.request(method, url, **kwargs)
# 请求完成后,更新最后一次请求的时间
last_request_time = time.time()
if response.status_code != expected_status:
raise requests.RequestException(f"Unexpected status code: {response.status_code}")
else:
print(f"Successful request: {url}\\t Status code: {response.status_code}")
#if method == 'POST':
#time.sleep(random.uniform(0.25, 1.5))
return response
except (requests.RequestException, json.JSONDecodeError) as e:
print(f"Request error: {e}. Retrying ({i+1}/{MAX_RETRIES} {method} {url})")
time.sleep(random.uniform(2, 5)*required_num*i) # wait a bit before retrying
if i == MAX_RETRIES - 1: # if this was the last attempt
raise # re-raise the last exception
4.1登录与初始化
①init_cookies()
init_cookies()
函数用于初始化cookies,因为网站需要同意一个隐私协议
函数调用safe_request()
函数发送一个GET请求到网站的首页,然后设置"_ul_cookie_consent" cookie。
def init_cookies():
res = safe_request(global_session, "GET", url=web_url)
global_session.cookies.set("_ul_cookie_consent", "allow", domain=".teket.jp", path="/")
print("Cookie initialization successful.")
② login()
login()
函数用于登录teket.jp。
函数首先调用safe_request()
函数发送一个GET请求获取登录页面的HTML,然后从HTML中提取出_csrfToken和_Token字段,然后调用safe_request()
函数发送一个POST请求进行登录。
如果登录成功,函数会返回状态码302。
def login(email,password):
res = safe_request(login_session, "GET", url=web_url+"/"+"login")
login_session.cookies.set("_ul_cookie_consent", "allow", domain=".teket.jp", path="/")
pattern = r'<input type="hidden" name="_csrfToken" autocomplete="off" value="(.*?)"/>'
_csrfToken = re.findall(pattern, res.text)[0]
pattern = r'<input type="hidden" name="_Token\\[fields\\]" autocomplete="off" value="(.*?from_pin_code_input)"/>'
_Token_fields = re.findall(pattern, res.text)[0]
data={
"_csrfToken": _csrfToken,
"email": email,
"password": password,
"remember": "1",
"from_pin_code_input": "",
"_Token[fields]": _Token_fields,
"_Token[unlocked]": ""
}
pot = safe_request(login_session, "POST", url=web_url+"/"+"login", data=data, allow_redirects=False, expected_status=302)
# Return the status code to check if the login was successful
return pot.status_code
4.2获取演出场次
① get_performance()
get_performance()
函数用于获取演出列表。
函数首先调用safe_request()
函数发送一个GET请求获取演出页面的HTML,然后从HTML中提取出所有演出的链接和销售开始时间。
def get_performance(session, ticket_url):
res = safe_request(session, "GET", url=web_url+ticket_url, expected_status=200)
pattern = r'<a href="{}/(\\d+/order\\?ticket_type=.*?)" class="btn ">'.format(ticket_url)
matches = re.findall(pattern, res.text)
performance_list=matches
pattern = re.compile(r"(\\d{4}/\\d{1,2}/\\d{1,2})\\(\\w+\\) (\\d{1,2}:\\d{1,2})")
matches = re.findall(pattern, res.text)
try:
sales_start=matches[0]
except:
sales_start=None
return performance_list,sales_start
4.3购票与确认
①confirm_ticket()
confirm_ticket()
函数用于确认选票。
函数首先调用safe_request()
函数发送一个GET请求获取选票页面的HTML,然后从HTML中提取出_csrfToken和_Token字段,然后调用safe_request()
函数发送一个POST请求确认选票。
如果确认成功,函数会调用purchase_ticket()
函数进行购买票务。
def confirm_ticket(session, ticket_url,performance_ticket):
res = safe_request(session, "GET", url=web_url+ticket_url+"/"+performance_ticket, expected_status=200)
pattern = r'<input type="hidden" name="_csrfToken" autocomplete="off" value="(.*?)"/>'
_csrfToken = re.findall(pattern, res.text)[0]
pattern = r'<input type="hidden" name="_Token\\[fields\\]" autocomplete="off" value="(.*?)"/>'
_Token_fields = re.findall(pattern, res.text)[-1]
performance_id=performance_ticket.split('/')[0]
rres = safe_request(session, "GET", url=web_url+"/svg/tickets/"+ticket_url.split('/')[-1]+"/"+performance_id, expected_status=200)
rres_json=rres.json()
order_request=[{"id":str(rres_json["ticketList"][0]["id"]),"amountList":[{"id":rres_json["ticketList"][0]["amountList"][0]["id"],"num":1,"seatList":[]}]}]
data={
"_csrfToken": _csrfToken,
"order_request": order_request,
"_Token[fields]": _Token_fields,
"_Token[unlocked]": ""
}
data['order_request'] = json.dumps(data['order_request'])
#time.sleep(random.uniform(1, 3)*required_num) # wait a bit before retrying
pot = safe_request(session, "POST", url=web_url+ticket_url+"/"+performance_ticket, data=data, allow_redirects=False, expected_status=302)
#print(pot.headers)
if pot.status_code ==302:
print('confirm ticket success')
purchase_ticket(session, ticket_url,performance_ticket)
② purchase_ticket()
purchase_ticket()
函数用于购买票务。
函数首先将登录session的cookies加入到当前session中,然后调用safe_request()
函数发送一个GET请求获取购票页面的HTML,然后从HTML中提取出_csrfToken、_Token字段和order_id,然后发送几个POST请求进行购票。
如果购票成功,函数会打印出购票成功的消息。
def purchase_ticket(session, ticket_url,performance_ticket):
#将登录cookie加入
tktDataAutoLogin_cookie = None
for cookie in login_session.cookies:
if cookie.name == 'tktDataAutoLogin':
tktDataAutoLogin_cookie = cookie
break
session.cookies.set_cookie(tktDataAutoLogin_cookie)
performance_id=performance_ticket.split('/')[0]
#time.sleep(random.uniform(1, 3)*required_num) # wait a bit before retrying
res = safe_request(session, "GET", url=web_url+ticket_url+"/"+performance_id+"/order/purchase", expected_status=200)
pattern = r'<input type="hidden" name="_csrfToken" autocomplete="off" value="(.*?)"/>'
_csrfToken = re.findall(pattern, res.text)[0]
pattern = r'<input type="hidden" name="_Token\\[fields\\]" autocomplete="off" value="(.*?Aorder_id)"/>'
_Token_fields = re.findall(pattern, res.text)[0]
pattern = r'<input type="hidden" name="order_id" id="order-id" value="(\\d+)"/>'
order_id = re.findall(pattern, res.text)[0]
data={
"_csrfToken": _csrfToken,
"order_id": order_id,
"_Token[fields]": _Token_fields,
"_Token[unlocked]": ""
}
pot = safe_request(session, "POST", url="https://teket.jp/api/check-certification/mail", data=data, expected_status=200)
pattern = r'<input type="hidden" name="_Token\\[fields\\]" autocomplete="off" value="(.*?Aevent_id%7Cevent_performance_id%7Cgroup_id%7Corder_id)"/>'
_Token_fields = re.findall(pattern, res.text)[0]
data={
"_csrfToken": _csrfToken,
"order_id": order_id,
"event_id": ticket_url.split('/')[-1],
"group_id": ticket_url.split('/')[1],
"event_performance_id": performance_id,
"_Token[fields]": _Token_fields,
"_Token[unlocked]": ""
}
#time.sleep(random.uniform(1, 3)*required_num) # wait a bit before retrying
pot = safe_request(session, "POST", url="https://teket.jp/tickets/before-purchase", data=data, expected_status=200)
#time.sleep(random.uniform(1, 3)*required_num) # wait a bit before retrying
res = safe_request(session, "GET", url=web_url+ticket_url+"/"+performance_id+"/order/complete", expected_status=200)
print(f"Ticket purchased successfully: {web_url+ticket_url+"/"+performance_id+"/order/complete"}")
4.4抢票主函数
①ticket_process()
ticket_process()
函数是这个脚本的主要函数,用于处理一整个购票过程。
首先创建一个新的session,并更新其cookies。然后,它调用get_performance()
函数获取演出列表,然后对每个演出创建一个线程,每个线程会执行confirm_ticket()
函数确认选票并购买票务。
所有线程都完成后,函数结束。
def ticket_process(ticket_url):
session = requests.Session()
session.cookies.update(global_session.cookies)
performance_list = None
sales_start = None
while not performance_list:
performance_list,sales_start = get_performance(session, ticket_url)
if sales_start:
date_str = sales_start[0] + ' ' + sales_start[1] # 拼接日期和时间
date_obj = datetime.strptime(date_str, "%Y/%m/%d %H:%M") # 转换为datetime对象
# 计算现在和该日期的时间差
now = datetime.now()
diff = date_obj - now
diff_seconds = diff.total_seconds()
print(f"Time difference until {date_str} is {diff_seconds} seconds.")
if diff_seconds <-60:
time.sleep(2)
elif diff_seconds <60:
time.sleep(0.5)
else:
time.sleep(5)
start_index = len(performance_list) // 2 - required_num // 2
if start_index<0:
start_index=0
end_index = start_index + required_num
performance_tickets = performance_list[start_index:end_index]
child_threads = []
for performance_ticket in performance_tickets:
child_thread = threading.Thread(target=confirm_ticket, args=(session, ticket_url, performance_ticket))
child_threads.append(child_thread)
child_thread.start()
# 等待所有线程完成
for child_thread in child_threads:
child_thread.join()
②main()
if __name__=="__main__":
init_cookies()
#邮箱&密码
email=""
password=""
login_status = login(email,password)
if login_status == 302:
print("Login was successful.")
# Create a thread for each ticket link
threads = []
for ticket_url in ticket_urls:
thread = threading.Thread(target=ticket_process, args=(ticket_url,))
threads.append(thread)
thread.start()
# Wait for all threads to complete
for thread in threads:
thread.join()
五、成果
现在知道为什么要用多线程了吧,哈哈哈哈!!!
评论区