基于requests库和多线程的teket.jp抢票脚本

基于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()

五、成果

现在知道为什么要用多线程了吧，哈哈哈哈！！！