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  # type: ignore

# code by wlt233 | for LoveLive! Series AsiaTour 2024
# 2024.11.10 | v0.1
# ref: https://www.ticketlink.co.kr/global/zh/product/51390

# log
logger.add("./data/log/{time}.log")


# seat data
COOKIE, USERNAME, PASSWORD, HEADLESS = "", "", "", ""
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),
    })



# web
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
async def get_sold(proxy):
    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"',
    }
    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



# 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(proxy, 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"]:
            logger.success(f"锁票成功！支付链接：\nhttps://www.ticketlink.co.kr/global/en/reserve/key/{resp['data']}/price")
            os.system("pause")
            sys.exit(0)
        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)
                os.system("pause")
                sys.exit(0)
            else:
                logger.error(msg)
    except:
        logger.critical(f"锁票失败！可能是 cookie 过期了！正在尝试更新 cookie！")
        update_cookie()


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



async def loop(proxy, interval, seats):
    event_loop = asyncio.get_event_loop()
    while True:
        event_loop.create_task(lock_ticket(proxy, 1, seats[1]))
        event_loop.create_task(lock_ticket(proxy, 2, seats[2]))
        # await lock_ticket(proxy, 1, seats[1])
        # await lock_ticket(proxy, 2, seats[2])
        await asyncio.sleep(interval / 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
    interval = int(config.get("conf", "interval") or 100)
    seats_attr = config.get("seat", "seats").split(",")
    logger.info(f"cookie = {COOKIE[:50]}...")
    logger.info(f"proxy = {proxy}")
    logger.info(f"interval = {interval} ms")
    logger.info(f"seats = {seats_attr}")
    
    seats = { 1: [], 2: [] }
    for attr in seats_attr:
        day, a = attr.split(maxsplit=1)
        if "1" in day:
            for k, l in able_d1.items():
                for s in l:
                    if s["mapInfo"] == a:
                        seats[1].append(s)
                        logger.info(f"已找到位置 d1 {a}，尝试锁票")
                        break
        if "2" in day:
            for k, l in able_d2.items():
                for s in l:
                    if s["mapInfo"] == a:
                        seats[2].append(s)
                        logger.info(f"已找到位置 d2 {a}，尝试锁票")
                        break
    if not seats[1] and not seats[2]:
        logger.error(f"没有找到座位，请检检查座位参数！")
        os.system("pause")
        sys.exit(0)

    asyncio.run(loop(proxy, interval, seats))