以下內容僅供交流學習使用!!!
思路分析
前面的爬蟲筆記一步一步走過來我們的技術水平也有了較大的提升了,現在我們來進行一下票星球搶票實戰項目,實現票星球的自動搶票。
- 我們打開票星球的移動端頁面,分析每一步操作需要用到的接口。
- 先手動操作搶票找到搶票流程的每一個接口后,判斷接口請求要求,請求方式,以及是否需要傳遞什么參數。
- 接下來就是每個接口一步一步操作請求響應,然后成功返回200即可。
- 然后實現過程主要是創建一個類對象,配置需要用到的屬性,每一步操作封裝成每一個函數,最后用一個主函數運行。
- 當然,下面是代碼的大體內容,主要講解如何操作,需要全部代碼可以找一下我發布的資源下載。
示例代碼
首先我們要用到下面三個包,pip下載即可
#pip install requests fake_useragent datetime
#請求
import requests
#隨機UA
from fake_useragent import UserAgent
#時間
import datetime
然后我們創建一個類對象,初始化方法里面創建需要用到的屬性變量:
class PXQ:def __init__(self):# 輸入自己的tokenself.token = '這里要打開網頁開發者工具找到你自己的token'# 演唱會項目id,必填,最簡單就是在url上可以看到,下面這個是要操作搶購的演唱會的show_id# 或者也是打開網頁開發者工具中的網絡也是可以找到self.show_id = '665708481d06bc0001627d83'# 指定場次id,不指定則默認從第一場開始遍歷,查找同理self.session_id = '665708656a025300012ae72a'# 購票數量,自行設置要購買的數量,但一定要看購票須知,不要超過上限self.buy_count = 1# 指定觀演人,觀演人序號從0開始,人數需與票數保持一致self.audience_idx = [0]# audience_idx = [0]# 門票類型,不確定則可以不填,讓系統自行判斷。快遞送票:EXPRESS,電子票:E_TICKET/ID_CARD,現場取票:VENUE,電子票或現場取票:VENUE_E,目前只發現這四種,如有新發現可補充self.deliver_method = ''# 獲取想要購買的票價idself.seat_plan_id = ''self.session_id_exclude = [] # 被排除掉的場次self.price: int = 0# 搶購的票價名稱self.seatPlanName = ''self.threadsLists = []# 判斷是否對應價位有票數,默認是無self.flag = False# 設置搶票開始時間,自行設置開搶時間,分別是年月日時分秒self.startTime = datetime.datetime(2024, 7, 3, 2, 0, 0)# 隨機UAself.user_agent = UserAgent()
接下來就是操作流程每個請求接口設置,封裝成方法,依次是:
- 根據項目id獲取所有場次和在售狀態
# 根據項目id獲取所有場次和在售狀態def get_sessions(self, useragent):headers = {'User-Agent': useragent,'Content-Type': 'application/json'}url = "https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/" + self.show_id + "/sessions_dynamic_data"response = requests.get(url=url, headers=headers).json()if response["statusCode"] == 200:# 獲取場次的id# print(response)return response["data"]["sessionVOs"]else:print("get_sessions異常:" + str(response))return None
- 根據場次id獲取座位信息,箭頭后面表示函數返回數據類型
def get_seat_plans(self, useragent) -> list:headers = {'User-Agent': useragent,'Content-Type': 'application/json'}url = "https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/" + self.show_id + "/show_session/" + self.session_id + "/seat_plans_static_data"response = requests.get(url=url, headers=headers).json()if response["statusCode"] == 200:# 查看票價的id# print(response)return response["data"]["seatPlans"]else:raise Exception("get_seat_plans異常:" + str(response))
- 獲取座位余票
# 獲取座位余票def get_seat_count(self, useragent) -> list:headers = {'User-Agent': useragent,'Content-Type': 'application/json'}url = "https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/" + self.show_id + "/show_session/" + self.session_id + "/seat_plans_dynamic_data"response = requests.get(url=url, headers=headers).json()if response["statusCode"] == 200:return response["data"]["seatPlans"]else:raise Exception("get_seat_count異常:" + str(response))
- 獲取門票類型(快遞送票EXPRESS,電子票E_TICKET,現場取票VENUE,電子票或現場取票VENUE_E)
# 獲取門票類型(快遞送票EXPRESS,電子票E_TICKET,現場取票VENUE,電子票或現場取票VENUE_E)def get_deliver_method(self, useragent, qty: int) -> str:headers = {'User-Agent': useragent,'Content-Type': 'application/json','access-token': self.token}data = {"items": [{"skus": [{"seatPlanId": self.seat_plan_id, "sessionId": self.session_id, "showId": self.show_id, "skuId": self.seat_plan_id,"skuType": "SINGLE","ticketPrice": self.price, "qty": qty #數量}],"spu": {"id": self.show_id,"spuType": "SINGLE"}}]}url = "https://m.piaoxingqiu.com/cyy_gatewayapi/trade/buyer/order/v3/pre_order"response = requests.post(url=url, headers=headers, json=data).json()if response["statusCode"] == 200:# 這里的print可以去掉# print(response["data"]["supportDeliveries"][0])return response["data"]["supportDeliveries"][0]["name"]else:raise Exception("獲取門票類型異常:" + str(response))
- 獲取觀演人信息
# 獲取觀演人信息# def get_audiences() -> list | None:def get_audiences(self, useragent):headers = {'User-Agent': useragent,'Content-Type': 'application/json','access-token': self.token}url = "https://m.piaoxingqiu.com/cyy_gatewayapi/user/buyer/v3/user_audiences"response = requests.get(url=url, headers=headers).json()if response["statusCode"] == 200:# 可以獲取觀影人的信息查看是否正確return response["data"]else:print("get_audiences異常:" + str(response))return None
- 提交訂單需要做的判斷比較多,要判斷演出票是什么票型(快遞送票EXPRESS,電子票E_TICKET/ID_Card,現場取票VENUE,電子票或現場取票VENUE_E)
# 提交訂單(快遞送票EXPRESS,電子票E_TICKET/ID_Card,現場取票VENUE,電子票或現場取票VENUE_E)def create_order(self, useragent, qty: int, express_fee: int,receiver, cellphone, address_id, detail_address, location_city_id, audience_ids: list):headers = {'User-Agent': useragent,'Content-Type': 'application/json','access-token': self.token}if self.deliver_method == "EXPRESS":data = {"priceItemParam": [{"applyTickets": [],"priceItemName": "票款總額","priceItemVal": self.price * qty,"priceItemType": "TICKET_FEE","priceItemSpecies": "SEAT_PLAN","direction": "INCREASE","priceDisplay": "¥" + str(self.price * qty)},{"applyTickets": [],"priceItemName": "快遞費","priceItemVal": express_fee,"priceItemId": self.show_id,"priceItemSpecies": "SEAT_PLAN","priceItemType": "EXPRESS_FEE","direction": "INCREASE","priceDisplay": "¥" + str(express_fee)}],"items": [{"skus": [{"seatPlanId": self.seat_plan_id,"sessionId": self.session_id,"showId": self.show_id,"skuId": self.seat_plan_id,"skuType": "SINGLE","ticketPrice": self.price,"qty": qty,"deliverMethod": self.deliver_method}],"spu": {"id": self.show_id,"spuType": "SINGLE"}}],"contactParam": {"receiver": receiver, # 張三"cellphone": cellphone # 13812345678},"one2oneAudiences": [{"audienceId": i, "sessionId": self.session_id} for i in audience_ids],"addressParam": {"address": detail_address, # 星巴克咖啡門口"district": location_city_id[4:],"city": location_city_id[2:4],"province": location_city_id[0:2],"addressId": address_id}}elif self.deliver_method == "ID_CARD":data = {"priceItemParam": [{"applyTickets": [],"priceItemName": "票款總額","priceItemVal": self.price * qty,"priceItemType": "TICKET_FEE","priceItemSpecies": "SEAT_PLAN","direction": "INCREASE","priceDisplay": "¥" + str(self.price * qty),}],"items": [{"skus": [{"seatPlanId": self.seat_plan_id,"sessionId": self.session_id,"showId": self.show_id,"skuId": self.seat_plan_id,"skuType": "SINGLE","ticketPrice": self.price,"qty": qty,"deliverMethod": self.deliver_method,}],"spu": {"id": self.show_id, "spuType": "SINGLE"},}],"one2oneAudiences": [{"audienceId": i, "sessionId": self.session_id} for i in audience_ids],"many2OneAudience": {"audienceId": audience_ids[0],"sessionIds": [self.session_id],},}# 電子票已解決elif self.deliver_method == "E_TICKET":data = {"priceItemParam": [{"applyTickets": [],"priceItemName": "票款總額","priceItemVal": self.price * qty,"priceItemType": "TICKET_FEE","priceItemSpecies": "SEAT_PLAN","direction": "INCREASE","priceDisplay": "¥" + str(self.price * qty)}],"items": [{"skus": [{"seatPlanId": self.seat_plan_id,"sessionId": self.session_id,"showId": self.show_id,"skuId": self.seat_plan_id,"skuType": "SINGLE","ticketPrice": self.price,"qty": qty,"deliverMethod": self.deliver_method}],"spu": {"id": self.show_id,"spuType": "SINGLE"}}],"many2OneAudience": {"audienceId": audience_ids[0],"sessionIds": [self.session_id]}}elif self.deliver_method == "VENUE":data = {"priceItemParam": [{"applyTickets": [],"priceItemName": "票款總額","priceItemVal": self.price * qty,"priceItemType": "TICKET_FEE","priceItemSpecies": "SEAT_PLAN","direction": "INCREASE","priceDisplay": "¥" + str(self.price * qty)}],"items": [{"skus": [{"seatPlanId": self.seat_plan_id,"sessionId": self.session_id,"showId": self.show_id,"skuId": self.seat_plan_id,"skuType": "SINGLE","ticketPrice": self.price,"qty": qty,"deliverMethod": self.deliver_method}],"spu": {"id": self.show_id,"spuType": "SINGLE"}}],"one2oneAudiences": [{"audienceId": i, "sessionId": self.session_id} for i in audience_ids]}elif self.deliver_method == "VENUE_E":data = {"priceItemParam": [{"applyTickets": [],"priceItemName": "票款總額","priceItemVal": self.price * qty,"priceItemType": "TICKET_FEE","priceItemSpecies": "SEAT_PLAN","direction": "INCREASE","priceDisplay": "¥" + str(self.price * qty)}],"items": [{"skus": [{"seatPlanId": self.seat_plan_id,"sessionId": self.session_id,"showId": self.show_id,"skuId": self.seat_plan_id,"skuType": "SINGLE","ticketPrice": self.price,"qty": qty,"deliverMethod": self.deliver_method}],"spu": {"id": self.show_id,"spuType": "SINGLE"}}]}else:raise Exception("不支持的deliver_method:" + str(self.deliver_method))url = "https://m.piaoxingqiu.com/cyy_gatewayapi/trade/buyer/order/v3/create_order"response = requests.post(url=url, headers=headers, json=data).json()if response["statusCode"] == 200:print("下單成功!請盡快支付!")else:raise Exception("下單異常:" + str(response))
- 主函數,前面都是函數定義,主要是在主函數執行操作搶票。
- 我們前面已經知道哪個演唱會show_id,以及哪個場次session_id,現在要在主函數里面的seat_plan_id進行自行賦值,也可以開頭初始化方法里面賦值/默認為空,seat_plan_id可以通過前面get_seat_plans函數里的接口地址獲取。
# 主函數def main(self):# 創建隨機的useragentuseragent = self.user_agent.randomwhile True:try:# 獲取座位余票信息,默認從最低價開始seat_plans = self.get_seat_plans(useragent)# print(seat_plans)seat_count = self.get_seat_count(useragent)# print(seat_count)# 獲取想要購買的票價idself.seat_plan_id = '667ba34a090ca100015604df'# 提取出對應的票價for temp in seat_plans:if temp["seatPlanId"] == self.seat_plan_id:self.price = temp["originalPrice"]self.seatPlanName = temp["seatPlanName"]break# 判斷對應的票價id是否有剩余票數for i in seat_count:if i["seatPlanId"] == self.seat_plan_id:if i["canBuyCount"] > 0:self.flag = True# print(f'{self.seatPlanName}該價位剩余數量:可滿足您的購買需求')breakelse:self.flag = False# print(f'{self.seatPlanName}無剩余!')breakif self.flag:print(f'{self.seatPlanName}該價位剩余數量:可滿足您的購買需求')print('正在努力搶購中……')else:print(f'{self.seatPlanName}票價無剩余!,將繼續為你刷新搶購')continueif not self.deliver_method:self.deliver_method = self.get_deliver_method(useragent, self.buy_count)print("演唱票類型:" + self.deliver_method)if self.deliver_method == "VENUE_E":self.create_order(useragent, self.buy_count, 0, None, None, None, None, None, [])else:# 獲取觀演人信息audiences = self.get_audiences(useragent)if len(self.audience_idx) == 0:self.audience_idx = range(self.buy_count)audience_ids = [audiences[i]["id"] for i in self.audience_idx]if self.deliver_method == "EXPRESS":# 獲取默認收貨地址address = self.get_address(useragent)address_id = address["addressId"] # 地址idlocation_city_id = address["locationId"] # 460102receiver = address["username"] # 收件人cellphone = address["cellphone"] # 電話detail_address = address["detailAddress"] # 詳細地址# 獲取快遞費用express_fee = self.get_express_fee(useragent, self.buy_count, location_city_id)self.create_order(useragent, self.buy_count, express_fee["priceItemVal"], receiver,cellphone, address_id, detail_address, location_city_id, audience_ids)elif self.deliver_method == "VENUE" or self.deliver_method == "E_TICKET" or self.deliver_method == "ID_CARD":self.create_order(useragent, self.buy_count, 0, None, None, None, None, None, audience_ids)else:print("不支持的deliver_method:" + self.deliver_method)breakexcept Exception as e:print(e)
以上內容全是定義在一個類中,我們要使用那就要創建一個類對象,現在我們打開一個文件入口進行創建:
if __name__ == '__main__':# 創建對象pxq = PXQ()while True:now = datetime.datetime.now()# 判斷當前時間與設置的開搶時間,并進行倒計時,當當前時間大于開搶時間直接開搶if now < pxq.startTime:print(f"{int(pxq.startTime.timestamp()-now.timestamp())}秒后開搶", end="\r")else:pxq.main()break
執行前我們配置好show_id,session_id和seat_plan_id,運行起來根據你設置的搶購時間在等待搶購的:
獲取seat_plan_id的流程操作:
搶票成功會顯示: