import asyncio import configparser import json import os import sys import time import httpx from loguru import logger from cookie import get_cookie # code by wlt233 | for LoveLive! Series AsiaTour 2024 # 2024.11.13 | v0.3 # ref: https://www.ticketlink.co.kr/global/zh/product/51390 # log logger.add("./data/log/{time}.log") # seat data COOKIE, USERNAME, PASSWORD, HEADLESS, PROXY, SEATS_SPEC = "", "", "", "", "", "" SEATS_IGNORE = [] COOKIE_TIME = time.time() with open("./data/able_d1.json", "r") as f: able_d1 = json.load(f) with open("./data/able_d2.json", "r") as f: able_d2 = json.load(f) coordinates = [ (1, 1), (1, 2), (1, 3), (1, 4), (1, 5), (1, 6), (1, 7), (2, 1), (2, 2), (2, 3), (2, 4), (2, 5), (2, 6), (2, 7), (3, 2), (3, 3), (3, 4), (3, 5), (3, 6), (3, 7), (4, 2), (4, 3), (4, 4), (4, 5), (4, 6), (4, 7), (5, 2), (5, 3), (5, 4), (5, 5), (5, 6), (5, 7), (6, 2), (6, 3), (6, 4), (6, 5), (6, 6), (6, 7), (7, 1), (7, 2), (7, 3), (7, 4), (7, 5), (7, 6), (7, 7), (8, 2), (8, 3), (8, 4), (8, 5), ] json_data = [] for x, y in coordinates: json_data.append({ 'virtualX': str(x), 'virtualY': str(y), }) def get_seat(attr_or_seatid, day=None): if not day or day == 1: for k, l in able_d1.items(): for seat in l: if seat["mapInfo"] == attr_or_seatid: return 1, seat elif str(seat["logicalSeatId"]) == attr_or_seatid: return 1, seat if not day or day == 2: for k, l in able_d2.items(): for seat in l: if seat["mapInfo"] == attr_or_seatid: return 2, seat elif str(seat["logicalSeatId"]) == attr_or_seatid: return 2, seat logger.error(f"没有找到座位：{attr_or_seatid}") return None, None # web wrapper async def _async_req(method, url: str, proxy=None, *args, **kwargs): logger.info(f"[async {method}] {url}") # logger.info(f"{args} {kwargs}") async with httpx.AsyncClient(proxies=proxy, verify=False) as client: resp = await client.request(method, url, timeout=None, *args, **kwargs) return resp async def async_post_json(url: str, proxy=None, *args, **kwargs): resp = await _async_req("post", url, proxy, *args, **kwargs) try: return resp.json() except Exception as e: logger.error(resp.content) raise e # soldout check async def get_sold(): global COOKIE headers = { 'Cookie': COOKIE, 'Accept': 'application/json, text/javascript, */*; q=0.01', 'Accept-Language': 'zh-CN,zh;q=0.9,ja;q=0.8', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', 'Content-Type': 'application/json;charset=UTF-8', 'Origin': 'https://www.ticketlink.co.kr', 'Pragma': 'no-cache', 'Referer': 'https://www.ticketlink.co.kr/global/zh/reserve/plan/schedule/1740993756', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36', 'sec-ch-ua': '"Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', } try: resp_d1 = await async_post_json( 'https://www.ticketlink.co.kr/global/zh/api/V2/plan/552446605/seat-soldout/area', #d1 proxy=PROXY, headers=headers, json=json_data, #cookies=cookies ) resp_d2 = await async_post_json( 'https://www.ticketlink.co.kr/global/zh/api/V2/plan/1740993756/seat-soldout/area', #d2 proxy=PROXY, headers=headers, json=json_data, #cookies=cookies ) sold_d1 = resp_d1["data"] sold_d2 = resp_d2["data"] return sold_d1, sold_d2 except: logger.error("座位情况检查失败！请尝试清空配置里的 cookie 并重启！") return None, None async def search_seats(seats): sold_d1, sold_d2 = await get_sold() if sold_d1 and sold_d2: for k, l in able_d1.items(): for seat in l: seat_id = str(seat["logicalSeatId"]) if not "FLOOR" in seat['mapInfo'] or sold_d1[seat_id]: continue if int(seat["mapInfo"].split()[3][:-1]) >= 8: continue if not seat_id in SEATS_IGNORE: logger.success(f"发现内场票：d1 {seat['mapInfo']} {seat_id}") seats[1].append(seat) else: logger.warning(f"发现内场票：d1 {seat['mapInfo']} 在忽略列表中") for k, l in able_d2.items(): for seat in l: seat_id = str(seat["logicalSeatId"]) if not "FLOOR" in seat['mapInfo'] or sold_d2[seat_id]: continue if int(seat["mapInfo"].split()[3][:-1]) >= 8: continue if not seat_id in SEATS_IGNORE: logger.success(f"发现内场票：d2 {seat['mapInfo']} {seat_id}") seats[2].append(seat) else: logger.warning(f"发现内场票：d2 {seat['mapInfo']} 在忽略列表中") # lock ticket def parse_seat(seat): if seat["gradeId"] == 102831: productGradeName = 'VIP석' if seat["gradeId"] == 102832: productGradeName = 'R석' if seat["gradeId"] == 102833: productGradeName = 'S석' return { 'logicalSeatId': seat["logicalSeatId"], 'allotmentCode': 'TKL', 'seatAttribute': seat["mapInfo"], 'sortSeatAttribute': seat["sortMapInfo"], 'productGradeId': seat["gradeId"], 'orderNum': seat["orderNum"], 'productGradeName': productGradeName, 'blockId': seat["blockId"], 'area': { 'virtualX': seat["area"]["virtualX"], 'virtualY': seat["area"]["virtualY"], }, 'st': int(time.time() * 1000), } async def lock_ticket(day, seats): global COOKIE, COOKIE_TIME if not seats: return if len(seats) > 4: seats = seats[:4] day_code = 552446605 if day == 1 else 1740993756 headers = { 'Cookie': COOKIE, 'Accept': 'application/json, text/javascript, */*; q=0.01', 'Accept-Language': 'zh-CN,zh;q=0.9,ja;q=0.8', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', 'Content-Type': 'application/json;charset=UTF-8', 'Origin': 'https://www.ticketlink.co.kr', 'Pragma': 'no-cache', 'Referer': f'https://www.ticketlink.co.kr/global/en/reserve/plan/schedule/{day_code}', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36', 'sec-ch-ua': '"Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', } json_data = { 'scheduleId': day_code, 'memberNo': 0, 'code': 'TKL', 'totalCnt': len(seats), 'seats': [ parse_seat(seat) for seat in seats ], 'zones': [], 'auto': None, 'pt': int(time.time() * 1000), 'nbt': int(time.time() * 1000), } try: resp = await async_post_json( f'https://www.ticketlink.co.kr/global/en/api/V2/plan/occupy/schedules/{day_code}/', proxy=PROXY, headers=headers, json=json_data ) if resp["success"]: print("\x07") logger.success(f"锁票成功！支付链接：https://www.ticketlink.co.kr/global/en/reserve/key/{resp['data']}/price") seats.clear() else: msg = f"锁票失败！msg:{resp['result']['message']}" if "이전에 진행된 예매가 있습니다" in msg: msg += (f" 有正在进行的预售，请访问以下链接初始化。" f"https://www.ticketlink.co.kr/global/en/reserve/plan/schedule/{day_code}?loadPrevious=true") logger.critical(msg) else: logger.error(msg) except: logger.critical(f"锁票失败！可能是 cookie 过期了！正在尝试更新 cookie！") if time.time() - COOKIE_TIME > 60: update_cookie() # cookie def update_cookie(): global COOKIE, USERNAME, PASSWORD, HEADLESS, COOKIE_TIME for i in range(5): COOKIE = get_cookie(USERNAME, PASSWORD, HEADLESS) if COOKIE: config = configparser.ConfigParser(interpolation=None) config.read("./conf.ini", encoding="utf8") config.set("conf", "cookie", COOKIE) with open("./conf.ini", 'w', encoding="utf8") as f: config.write(f) logger.info("cookie 更新成功！") COOKIE_TIME = time.time() return logger.critical("登陆失败，无法获取 cookie！") os.system("pause") sys.exit(0) # main loop state machine async def loop(interval1, interval2, seats): global SEATS_SPEC while True: if not (seats[1] or seats[2]): t = time.time() while True: await search_seats(seats) if seats[1] or seats[2]: break else: logger.error("没有搜索到内场座位...") if time.time() - t > 1200: update_cookie() t = time.time() else: await asyncio.sleep(interval1 / 1000) else: event_loop = asyncio.get_event_loop() t = time.time() while True: event_loop.create_task(lock_ticket(1, seats[1])) event_loop.create_task(lock_ticket(2, seats[2])) if not SEATS_SPEC and time.time() - t > 30: seats[1].clear() seats[2].clear() break if not (seats[1] or seats[2]): break else: await asyncio.sleep(interval2 / 1000) if __name__ == "__main__": config = configparser.ConfigParser(allow_no_value=True, delimiters=('=', ':')) config.read("./conf.ini", encoding="utf8") USERNAME = config.get("conf", "username") PASSWORD = config.get("conf", "password") HEADLESS = True if config.get("conf", "headless") == "1" else False COOKIE = config.get("conf", "cookie", raw=True) if not COOKIE: logger.info("正在尝试获取 cookie...") if not USERNAME or not PASSWORD: logger.critical("请在配置中填写用户名与密码！") os.system("pause") sys.exit(0) update_cookie() PROXY = config.get("conf", "proxy") or None interval1 = int(config.get("conf", "interval1") or 10000) interval2 = int(config.get("conf", "interval2") or 1000) SEATS_SPEC = config.get("seat", "seats") seats_list = SEATS_SPEC.split(",") if SEATS_SPEC else [] ignore_list = config.get("seat", "ignore").split(",") logger.info(f"cookie = {COOKIE[:50]}...") logger.info(f"proxy = {PROXY}") logger.info(f"interval1 = {interval1} ms (search)") logger.info(f"interval2 = {interval2} ms (lock)") logger.info(f"seats = {seats_list}") seats = { 1: [], 2: [] } for seat_str in seats_list: if " " in seat_str: daystr, attr = seat_str.split(maxsplit=1) day = 1 if "1" in daystr else 2 day, seat = get_seat(attr, day) else: day, seat = get_seat(seat_str) if day and seat: seats[day].append(seat) logger.info(f"已找到锁票位置 d{day} {seat['mapInfo']} {seat['logicalSeatId']}") if not (seats[1] or seats[2]): logger.error(f"没有找到配置的锁票座位，开始自动搜索内场座位...") for seat_str in ignore_list: if " " in seat_str: daystr, attr = seat_str.split(maxsplit=1) day = 1 if "1" in daystr else 2 day, seat = get_seat(attr, day) else: day, seat = get_seat(seat_str) if day and seat: SEATS_IGNORE.append(str(seat['logicalSeatId'])) logger.info(f"已找到忽略位置 d{day} {seat['mapInfo']} {seat['logicalSeatId']}") asyncio.run(loop(interval1, interval2, seats))