bilibili_api/show.py

"""
bilibili_api.show

展出相关
"""
import json
import random
import time
from dataclasses import dataclass, field
from typing import List

from .utils.credential import Credential
from .utils.network import Api
from .utils.utils import get_api, get_deviceid

API = get_api("show")


@dataclass
class Ticket:
    """
    票对象

    id (int): 场次id

    price (float): 价格(RMB)

    desc (str): 描述

    sale_start (str): 开售开始时间

    sale_end (str): 开售结束时间
    """
    id: int
    price: int
    desc: str
    sale_start: str
    sale_end: str


@dataclass
class Session:
    """
    场次对象

    id (int): 场次id

    start_time (int): 场次开始时间戳

    formatted_time (str): 格式化start_time后的时间格式: YYYY-MM-DD dddd

    ticket_list (list[Ticket]): 存放Ticket对象的list
    """
    id: int
    start_time: int
    formatted_time: str
    ticket_list: List[Ticket] = field(default_factory=list)


@dataclass
class BuyerInfo:
    """
    购买人信息

    id (int): 信息序号

    uid (int): 用户 ID

    account_channel (str): 默认为空

    personal_id (str): 身份证号

    name (str): 姓名

    id_card_front (str): 未知

    id_card_back (str): 未知

    is_default (bool): 是否为默认信息

    tel (str): 电话号码

    error_code (str): 错误代码

    id_type (int): 默认 0

    verify_status (int): 认证状态

    accountId (int): 用户 ID

    isBuyerInfoVerified (bool): 默认为 True

    isBuyerValid (bool): 默认为 True
    """
    id: int
    uid: int
    account_channel: str
    personal_id: str
    name: str
    id_card_front: str
    id_card_back: str
    is_default: int
    tel: str
    error_code: str
    id_type: int
    verify_status: int
    accountId: int
    isBuyerInfoVerified: bool = True
    isBuyerValid: bool = True


async def get_project_info(project_id: int) -> dict:
    """
    返回项目全部信息

    Args:
        project_id (int): 项目id

    Returns:
        dict: 调用 API 返回的结果
    """
    api = API["info"]["get"]
    return await Api(**api).update_params(id=project_id).result


async def get_available_sessions(project_id: int) -> List[Session]:
    """
    返回该项目的所有可用场次

    Args:
        project_id (int): 项目id

    Returns:
        list[Session]: 存放场次对象的list
    """
    rtn_list = []
    project_info = await get_project_info(project_id)
    for v in project_info["screen_list"]:
        sess_obj = Session(id=v["id"], start_time=v["start_time"], formatted_time=v["name"])
        for t in v["ticket_list"]:
            sess_obj.ticket_list.append(
                Ticket(
                    id=t["id"],
                    price=t["price"],
                    desc=t["desc"],
                    sale_start=t["sale_start"],
                    sale_end=t["sale_end"],
                )
            )
        rtn_list.append(sess_obj)
    return rtn_list


async def get_all_buyer_info(credential: Credential) -> dict:
    """
    返回账号的全部身份信息

    Args:
        credential (Credential): 登录凭证

    Returns:
        dict: 调用 API 返回的结果
    """
    credential.raise_for_no_sessdata()
    api = API["info"]["buyer_info"]
    return await Api(**api, credential=credential).result


async def get_all_buyer_info_obj(credential: Credential) -> List[BuyerInfo]:
    """
    以BuyerInfo对象返回账号的全部身份信息

    Args:
        credential (Credential): 登录凭证

    Returns:
        list[BuyerInfo]: BuyerInfo对象列表
    """
    res = await get_all_buyer_info(credential)
    return [BuyerInfo(**v) for v in res["list"]]


def generate_clickPosition() -> dict:
    """
    生成虚假的点击事件

    Returns:
        dict: 点击坐标和时间
    """
    # 生成随机的 x 和 y 坐标，以下范围大概是1920x1080屏幕下可能的坐标
    x = random.randint(1320, 1330)
    y = random.randint(880, 890)
    # 生成随机的起始时间和结束时间（或当前时间）
    origin_timestamp = int(time.time() * 1000)
    # 添加一些随机时间差 (5s ~ 10s)
    now_timestamp = origin_timestamp + random.randint(5000, 10000)
    return {
        "x": x,
        "y": y,
        "origin": origin_timestamp,
        "now": now_timestamp
    }


@dataclass
class OrderTicket:
    """
    购票类

    Args:
        credential (Credential): Credential 对象

        target_buyer (BuyerInfo): 购票人

        project_id (int): 展出id

        session (Session): Session 对象

        ticket (Ticket): Ticket 对象
    """
    credential: Credential
    target_buyer: BuyerInfo
    project_id: int
    session: Session
    ticket: Ticket

    async def _get_create_order_payload(self) -> dict:
        """
        获取 create order API 所需的载荷

        Returns:
            dict: payload
        """
        res = await self.get_token()
        header = {
            "count": 1,
            "order_type": 1,
            "pay_money": self.ticket.price,
            "project_id": self.project_id,
            "screen_id": self.session.id,
            "sku_id": self.ticket.id,
            "timestamp": int(time.time() * 1000),
            "token": res["token"],
            "deviceId": get_deviceid('', True),
            "clickPosition": json.dumps(generate_clickPosition()),
        }
        info = await get_project_info(self.project_id)
        info = info["performance_desc"]["list"]
        for element in info:
            if element["module"] == "base_info":
                info = element
                break
        for detail in info["details"]:
            content = detail["content"]
            if "一人一证" in content or "一单一证" in content:
                header.update({
                    "buyer_info": json.dumps([self.target_buyer.__dict__])
                })
                return header
        header.update({
            "buyer": self.target_buyer.name,
            "tel": self.target_buyer.tel,
        })
        return header

    async def get_token(self):
        """
        获取购票Token

        Returns:
            dict: 调用 API 返回的结果
        """
        self.credential.raise_for_no_sessdata()
        api = API["info"]["token"]
        payload = {
            "count": "1",
            "order_type": 1,
            "project_id": self.project_id,
            "screen_id": self.session.id,
            "sku_id": self.ticket.id
        }
        return await Api(**api, credential=self.credential).update_data(**payload).result

    async def create_order(self):
        """
        创建购买订单

        Returns:
            dict: 调用 API 返回的结果
        """
        payload = await self._get_create_order_payload()
        api = API["operate"]["order"]
        return await Api(**api, credential=self.credential).update_params(project_id=self.project_id).update_data(
            **payload).result