Skip to content

Commit

Permalink
Feat: update all
Browse files Browse the repository at this point in the history
  • Loading branch information
@ThinkerWen
ThinkerWen committed Nov 19, 2024
1 parent a0d4188 commit 9e01a60
Show file tree
Hide file tree
Showing 6 changed files with 144 additions and 125 deletions.
2 changes: 1 addition & 1 deletion Monitor_DM.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ def __init__(self, perform: dict) -> None:
self.seat_url = DM.get_seat_url()
self.request = self.do_request()
self.show_info = {
"platform": 0,
"platform": "大麦",
"seat_info": list(),
"session_info": list(),
"show_id": perform.get('show_id'),
Expand Down
56 changes: 31 additions & 25 deletions Monitor_FWD.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,49 +9,55 @@

class FWD(Monitor):

show_start = False

def __init__(self, perform: dict) -> None:
super().__init__()
self.show_info = {
"platform": "纷玩岛",
"seat_info": list(),
"session_info": list(),
"show_id": perform.get('show_id'),
"show_name": perform.get('show_name')
}
logging.info(f"纷玩岛 {perform.get('show_name')} 开始加载")
self.performId = perform.get('show_id')
self.show_info = dict()
self.get_show_infos()
logging.info(f"纷玩岛 {self.show_info.get('show_name')} 加载成功")

def get_show_infos(self):
response = self.request(f'https://api.livelab.com.cn/performance/app/project/get_project_info?project_id={self.performId}&v=1695113662390&retry=false')
self.show_info["show_id"] = str(json.loads(response.text).get("data").get("projectId"))
self.show_info["show_name"] = json.loads(response.text).get("data").get("projectName")
self.show_info["seat_info"] = list()
self.show_info["platform"] = 2
response = self.request(f'https://api.livelab.com.cn/performance/app/project/get_performs?project_id={self.performId}&v=1694683437294&retry=false')
show_id = self.show_info.get('show_id')
response = self.request(f'https://api.livelab.com.cn/performance/app/project/get_performs?project_id={show_id}&v=1694683437294&retry=false')
show_info = json.loads(response.text)
for session_info in show_info.get("data").get("performInfos"):
session = session_info.get("performInfo")[0]
session_id = session.get("id")
session_name = session.get("name")
for seat in session.get("seatPlans"):
seat_info = dict()
seat_info["session_id"] = session_id
seat_info["session_name"] = session_name
seat_info["seat_plan_id"] = seat.get("seatPlanId")
seat_info["seat_plan_name"] = seat.get("seatPlanName")
self.show_info["seat_info"].append(seat_info)
self.show_info["seat_info"].append({
"session_id": session_id,
"session_name": session_name,
"seat_plan_id": seat.get("seatPlanId"),
"seat_plan_name": seat.get("seatPlanName"),
})

def monitor(self) -> list:
logging.info(f"纷玩岛 {self.show_info.get('show_name')} 监控中")
can_buy_list = list()
response = self.request(f'https://api.livelab.com.cn/performance/app/project/countDown?projectId={self.performId}')
if json.loads(response.text).get("data") > 0:
return can_buy_list
response = self.request(f'https://api.livelab.com.cn/performance/app/project/get_performs?project_id={self.performId}&v=1694683437294&retry=false')
show_info = json.loads(response.text)
can_buy_list = []
show_id = self.show_info.get('show_id')
if not self.show_start:
response = self.request(f'https://api.livelab.com.cn/performance/app/project/countDown?projectId={show_id}')
if response.json().get("data", 0) > 0:
return can_buy_list
self.show_start = True
response = self.request(f'https://api.livelab.com.cn/performance/app/project/get_performs?project_id={show_id}&v=1694683437294&retry=false')
show_info = response.json()
if show_info.get("code") != 10000:
return can_buy_list
for session_info in show_info.get("data").get("performInfos"):
for seat in session_info.get("performInfo")[0].get("seatPlans"):
if seat.get("display") != 1:
continue
can_buy_list.append(str(seat.get("seatPlanId")))
for session_info in show_info.get("data", {}).get("performInfos", []):
seat_plans = session_info.get("performInfo", [])[0].get("seatPlans", [])
can_buy_list.extend(
str(seat.get("seatPlanId")) for seat in seat_plans if seat.get("display") == 1
)
return can_buy_list

def request(self, url: str) -> Response:
Expand Down
111 changes: 54 additions & 57 deletions Monitor_MY.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,79 +9,76 @@

class MY(Monitor):

token = str()

def __init__(self, perform: dict) -> None:
super().__init__()
logging.info(f"猫眼 {perform.get('show_name')} 开始加载")
file = open("config.json", "r", encoding="utf-8")
self.token = json.load(file).get("token").get("my")
file.close()
self.performId = perform.get('show_id')
self.show_info = dict()
self.show_info = {
"platform": "猫眼",
"seat_info": list(),
"session_info": list(),
"show_id": perform.get('show_id'),
"show_name": perform.get('show_name')
}
logging.info(f"猫眼 {perform.get('show_name')} 开始加载")
self.get_show_infos()
logging.info(f"猫眼 {self.show_info.get('show_name')} 加载成功")

def get_show_infos(self):
response = self.request(f'https://wx.maoyan.com/maoyansh/myshow/ajax/v2/performance/{self.performId}', self.performId)
self.show_info["show_id"] = json.loads(response.text).get("data").get("performanceId")
self.show_info["show_name"] = json.loads(response.text).get("data").get("name")
self.show_info["seat_info"] = list()
self.show_info["platform"] = 1
response = self.request(f'https://wx.maoyan.com/maoyansh/myshow/ajax/v2/performance/{self.performId}/shows/0', self.performId)
show_info = json.loads(response.text)
for session in show_info.get("data"):
show_id = self.show_info.get('show_id')
response = requests.post(f"https://wx.maoyan.com/my/odea/project/shows?token={self.token}&clientPlatform=2", json={"projectId":show_id}, headers=MY.headers(), proxies=self._proxy)
show_info = response.json().get("data")
for session in show_info.get("showListVO"):
session_id = session.get("showId")
session_name = session.get("name")
response = self.request(f'https://wx.maoyan.com/maoyansh/myshow/ajax/v2/show/{session_id}/tickets?sellChannel=7&performanceId={self.performId}&cityId=1', self.performId)
show_session_info = json.loads(response.text)
for seat in show_session_info.get("data"):
seat_info = dict()
seat_info["session_id"] = session_id
seat_info["session_name"] = session_name
seat_info["seat_plan_id"] = seat.get("ticketClassId")
seat_info["seat_plan_name"] = str(int(seat.get("ticketPrice")))
self.show_info["seat_info"].append(seat_info)
session_name = session.get("showName")
self.show_info["session_info"].append({
"session_id": session_id,
"session_name": session_name,
})
response = self.request(f"https://wx.maoyan.com/my/odea/show/tickets?token={self.token}&showId={session_id}&projectId={show_id}&clientPlatform=2")
show_session_info = response.json().get("data")
for seat in show_session_info.get("ticketsVO"):
self.show_info["seat_info"].append({
"session_id": session_id,
"session_name": session_name,
"seat_plan_id": seat.get("ticketId"),
"seat_plan_name": str(int(seat.get("ticketPriceVO").get("ticketPrice"))),
})

def monitor(self) -> list:
logging.info(f"猫眼 {self.show_info.get('show_name')} 监控中")
can_buy_list = list()
response = self.request(f'https://wx.maoyan.com/maoyansh/myshow/ajax/v2/performance/{self.performId}/shows/0', self.performId)
show_info = json.loads(response.text)
if show_info.get("code") != 200:
return can_buy_list
for show in show_info.get("data"):
show_id = show.get("showId")
if show.get("soldOut"):
continue
response = self.request(f'https://wx.maoyan.com/maoyansh/myshow/ajax/v2/show/{show_id}/tickets?sellChannel=7&performanceId={self.performId}&cityId=1', self.performId)
show_session_info = json.loads(response.text)
if show_session_info.get("code") == 200:
for session in show_session_info.get("data"):
seat = session.get("salesPlanVO")
salesPlanId = seat.get("salesPlanId")
if not seat.get("currentAmount"):
continue
can_buy_list.append(salesPlanId)
can_buy_list = []
show_id = self.show_info.get('show_id')
for session in self.show_info.get("session_info", []):
session_id = session.get("session_id")
response = self.request(f"https://wx.maoyan.com/my/odea/show/tickets?token={self.token}&showId={session_id}&projectId={show_id}&clientPlatform=2")
tickets = response.json().get("data", {}).get("ticketsVO", [])
can_buy_list.extend(ticket.get("ticketId") for ticket in tickets if ticket.get("remainingStock"))
return can_buy_list

def request(self, url: str, perform_id: str) -> Response:
def request(self, url: str) -> Response:
return requests.get(
url=url,
headers={
'xweb_xhr': '1',
'Accept': '*/*',
'Host': 'wx.maoyan.com',
'X-Channel-ID': '70001',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Dest': 'empty',
'version': 'wallet-v4.5.11',
'X-Requested-With': 'wxapp',
'Accept-Language': 'zh-CN,zh',
'Sec-Fetch-Site': 'cross-site',
'x-wxa-referer': 'pages/search/index',
'Content-Type': 'multipart/form-data',
'x-wxa-page': 'pages/show/detail/index',
'Referer': 'https://servicewechat.com/wxdbb4c5f1b8ee7da1/1366/page-frame.html',
'x-wxa-query': f'%7B%22id%22%3A%22{perform_id}%22%2C%22utm_source%22%3A%22wxwallet_search%22%7D',
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.102 Safari/537.36 MicroMessenger/6.8.0(0x16080000) NetType/WIFI MiniProgramEnv/Mac MacWechat/WMPF XWEB/30817',
},
headers=MY.headers(),
proxies=super()._proxy,
verify=False,
timeout=10
)

@staticmethod
def headers():
return {
'Host': 'wx.maoyan.com',
'X-Channel-ID': '70001',
'version': 'wallet-v5.10.9',
'X-Requested-With': 'wxapp',
'content-type': 'application/json',
'x-wxa-referer': 'pages/show/detail/v2/index',
'x-wxa-page': 'pages/showsubs/ticket-level/v2/index',
'Referer': 'https://servicewechat.com/wxdbb4c5f1b8ee7da1/1557/page-frame.html',
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 17_5_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.53(0x18003531) NetType/WIFI Language/zh_CN',
}
71 changes: 41 additions & 30 deletions Monitor_PXQ.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,51 +9,62 @@

class PXQ(Monitor):

show_start = False

def __init__(self, perform: dict) -> None:
super().__init__()
self.show_info = {
"platform": "票星球",
"seat_info": list(),
"session_info": list(),
"show_id": perform.get('show_id'),
"show_name": perform.get('show_name')
}
logging.info(f"票星球 {perform.get('show_name')} 开始加载")
self.performId = perform.get('show_id')
self.show_info = dict()
self.get_show_infos()
logging.info(f"票星球 {self.show_info.get('show_name')} 加载成功")

def get_show_infos(self):
response = self.request(f"https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/{self.performId}/sessions_static_data")
self.show_info["show_id"] = json.loads(response.text).get("data").get("showId")
self.show_info["show_name"] = json.loads(response.text).get("data").get("showName")
self.show_info["seat_info"] = list()
self.show_info["platform"] = 3
show_id = self.show_info.get('show_id')
response = self.request(f"https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/{show_id}/sessions_static_data")
show_info = json.loads(response.text)
for session in show_info.get("data").get("sessionVOs"):
session_id = session.get("bizShowSessionId")
session_name = session.get("sessionName")
response = self.request(f'https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/{self.performId}/show_session/{session_id}/seat_plans_static_data')
self.show_info["session_info"].append({
"session_id": session_id,
"session_name": session_name,
})
response = self.request(f'https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/{show_id}/show_session/{session_id}/seat_plans_static_data')
for seat in response.json().get("data").get("seatPlans"):
seat_info = dict()
seat_info["session_id"] = session_id
seat_info["session_name"] = session_name
seat_info["seat_plan_id"] = seat.get("seatPlanId")
seat_info["seat_plan_name"] = seat.get("seatPlanName")
self.show_info["seat_info"].append(seat_info)
self.show_info["seat_info"].append({
"session_id": session_id,
"session_name": session_name,
"seat_plan_id": seat.get("seatPlanId"),
"seat_plan_name": seat.get("seatPlanName"),
})

def monitor(self) -> list:
logging.info(f"票星球 {self.show_info.get('show_name')} 监控中")
can_buy_list = list()
response = self.request(f'https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/{self.performId}/sessions_dynamic_data')
show_info = json.loads(response.text)
if show_info.get("statusCode") != 200:
return can_buy_list
for session in show_info.get("data").get("sessionVOs"):
session_id = session.get("bizShowSessionId")
if session.get("hasSessionSoldOut") or session.get("sessionSaleTimeCountdown", 0) > 0:
continue
response = self.request(f'https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/{self.performId}/show_session/{session_id}/seat_plans_dynamic_data').json()
if response.get("statusCode") != 200:
return can_buy_list
for seat in response.get("data").get("seatPlans"):
if seat.get("canBuyCount") <= 0:
continue
can_buy_list.append(seat.get("seatPlanId"))
can_buy_list = []
show_id = self.show_info.get('show_id')

if not self.show_start:
response = self.request( f'https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/{show_id}/sessions_dynamic_data')
show_info = response.json()
if show_info.get("statusCode") == 200:
for session in show_info.get("data", {}).get("sessionVOs", []):
if session.get("sessionSaleTimeCountdown", 0) > 0:
return can_buy_list
self.show_start = True
for session in self.show_info.get("session_info"):
session_id = session.get("session_id")
response = self.request(f'https://m.piaoxingqiu.com/cyy_gatewayapi/show/pub/v3/show/{show_id}/show_session/{session_id}/seat_plans_dynamic_data')
if response.json().get("statusCode") == 200:
can_buy_list.extend(
seat.get("seatPlanId") for seat in response.json().get("data", {}).get("seatPlans", [])
if seat.get("canBuyCount", 0) > 0
)
return can_buy_list

def request(self, url: str) -> Response:
Expand Down
27 changes: 16 additions & 11 deletions config.json
View file
Original file line number Diff line number Diff line change
@@ -1,27 +1,32 @@
{
"proxy": null,
"monitor_list": [{
"token": {
"my": "登陆小程序后抓包获取token"
},
"monitor_list": [
{
"show_id": "844180471181",
"show_name": "【北京】Jessie J 演唱会-北京站",
"platform": 0,
"deadline": "2024-12-01 00:00:00"
},
{
"show_id": "302945",
"show_name": "周杰伦2024“嘉年华”世界巡回演唱会-杭州站",
"show_id": "360282",
"show_name": "【北京】Jessie J 演唱会-北京站",
"platform": 1,
"deadline": "2024-04-21 00:00:00"
"deadline": "2024-12-01 00:00:00"
},
{
"show_id": "744",
"show_name": "【上海】邓紫棋I AM GLORIA巡回演唱会-上海站",
"show_id": "3191333934",
"show_name": "【深圳】陈奕迅FEAR and DREAMS世界巡回演唱会－深圳站",
"platform": 2,
"deadline": "2024-05-26 00:00:00"
},{
"show_id": "65ec0632cfc3b200019ea854",
"show_name": "【北京】于文文「魔方视界」演唱会-北京站",
"deadline": "2024-12-15 00:00:00"
},
{
"show_id": "67107aebae5e8300012f5444",
"show_name": "【北京】Jessie J 演唱会-北京站",
"platform": 3,
"deadline": "2024-04-27 00:00:00"
"deadline": "2024-12-01 00:00:00"
}
]
}
2 changes: 1 addition & 1 deletion start.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ def loop_monitor(monitor: Union[DM, MY, FWD, PXQ], show: dict) -> None:
while datetime.strptime(show.get("deadline"), "%Y-%m-%d %H:%M:%S") > datetime.now():
try:
if monitor.monitor():
info = f"平台{show.get('platform')} {show.get('show_name')} 已回流，请及时购票！"
info = f"{monitor.show_info.get('platform')} {show.get('show_name')} 已回流，请及时购票！"
logging.info(info)
monitor.bark_alert(info)
except Exception as e:
Expand Down

0 comments on commit 9e01a60

Please sign in to comment.