ticketlink_script/ticket.py

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_IGNORE = []
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 "FLOOR" in seat['mapInfo'] and not sold_d1[seat_id]:
                    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 "FLOOR" in seat['mapInfo'] and not sold_d2[seat_id]:
                    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
    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！")
        update_cookie()




# cookie
def update_cookie():
    global COOKIE, USERNAME, PASSWORD, HEADLESS
    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 更新成功！")
            return
    logger.critical("登陆失败，无法获取 cookie！")
    os.system("pause")
    sys.exit(0)




# main loop state machine
async def loop(interval1, interval2, seats):
    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()
            while True:
                event_loop.create_task(lock_ticket(1, seats[1]))
                event_loop.create_task(lock_ticket(2, seats[2]))
                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_str = config.get("seat", "seats")
    seats_list = seats_str.split(",") if seats_str 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))