live-hub-py/core/huya/rpc_requests.py

"""HTTP/WSS 共用的虎牙 RPC 请求构造。

两个传输客户端仍各自处理同步/异步发送与响应解码；这里只保存业务参数到
TAF 请求结构体的唯一映射，防止字段默认值在两条链路中发生漂移。
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from typing import Any

from .activity_structs import (
    GetActPrizeDetailReq,
    GetActPrizeDetailResp,
    GetActTaskDetailReq,
    GetActTaskDetailResp,
    GetActUserTaskDetailReq,
    GetActUserTaskDetailResp,
    GetUserPrizeRecordsReq,
    GetUserPrizeRecordsResp,
    GetUserScoreReq,
    GetUserScoreResp,
    ScoreExchangePrizeReq,
    ScoreExchangePrizeResp,
)
from .shop_structs import (
    CreateOrderAccountParam,
    CreateOrderExtraParam,
    CreateOrderItemReqV6,
    CreateOrderPromotionParam,
    CreateOrderReqV6,
    CreateOrderRspV6,
    GetGoodsInfoReqV5,
    GoodsInfoRsp,
    PlayMallGetMyPromotionReq,
    PlayMallQueryAllocationReq,
    PromotionGoodsKey,
    shop_spu_number,
)


@dataclass(frozen=True)
class RpcRequest:
    service: str
    method: str
    request: Any
    response_type: Any = None


def activity_score_request(user: Any, sid: int) -> RpcRequest:
    request = GetUserScoreReq()
    request.userId = user
    request.sid = int(sid or 0)
    return RpcRequest("webActUI", "getUserScore", request, GetUserScoreResp)


def activity_records_request(user: Any, sid: int) -> RpcRequest:
    request = GetUserPrizeRecordsReq()
    request.userId = user
    request.sid = int(sid or 0)
    return RpcRequest(
        "webActUI",
        "getUserPrizeRecords",
        request,
        GetUserPrizeRecordsResp,
    )


def activity_exchange_request(user: Any, sid: int, pid: int) -> RpcRequest:
    request = ScoreExchangePrizeReq()
    request.userId = user
    request.sid = int(sid or 0)
    request.pid = int(pid or 0)
    return RpcRequest("webActUI", "scoreExchangePrize", request, ScoreExchangePrizeResp)


def activity_prize_detail_request(user: Any, sid: int, pid: int) -> RpcRequest:
    request = GetActPrizeDetailReq()
    request.userId = user
    request.sid = int(sid or 0)
    request.pid = int(pid or 0)
    return RpcRequest(
        "webActUI",
        "getActPrizeDetail",
        request,
        GetActPrizeDetailResp,
    )


def activity_task_detail_request(user: Any, act_id: int) -> RpcRequest:
    request = GetActTaskDetailReq()
    request.userId = user
    request.actId = int(act_id or 0)
    return RpcRequest("webActUI", "getActTaskDetail", request, GetActTaskDetailResp)


def activity_user_task_detail_request(user: Any, act_id: int) -> RpcRequest:
    request = GetActUserTaskDetailReq()
    request.userId = user
    request.actId = int(act_id or 0)
    return RpcRequest(
        "webActUI",
        "getActUserTaskDetail",
        request,
        GetActUserTaskDetailResp,
    )


def goods_info_request(
    user: Any,
    shop_app: Any,
    *,
    pid: int,
    spu_id: str,
    sku_id: int = 0,
    game_id: str = "",
) -> RpcRequest:
    request = GetGoodsInfoReqV5()
    request.userId = user
    request.shopAppInfo = shop_app
    request.pid = pid
    request.spuId = spu_id
    request.skuId = sku_id
    request.gameId = game_id
    return RpcRequest("shopMiddleUI", "getGoodsInfoV5", request, GoodsInfoRsp)


def play_mall_allocation_request(
    user: Any,
    app_info: Any,
    *,
    page_code: str,
    scene: int,
    page_id: int = 0,
    waterfall_tab_id: int | str | None = None,
) -> RpcRequest:
    request = PlayMallQueryAllocationReq()
    request.userId = user
    request.appInfo = app_info
    request.appInfo.scene = int(scene or 0)
    request.pageCode = page_code
    if waterfall_tab_id not in (None, ""):
        request.blank7 = json.dumps(
            {"waterfallTabId": str(waterfall_tab_id)},
            separators=(",", ":"),
        )
        request.flag9 = int(page_id or 0)
        request.flags11 = []
    return RpcRequest("PlayMallUI", "queryAllocation", request)


def play_mall_promotion_request(
    user: Any,
    app_info: Any,
    *,
    spu_id: str,
    sku_id: int = 0,
    price: int = 0,
    supplier_id: int = 0,
    game_id: int = 0,
) -> RpcRequest:
    spu_num = shop_spu_number(spu_id)
    request = PlayMallGetMyPromotionReq()
    request.userId = user
    request.appInfo = app_info
    goods = PromotionGoodsKey()
    goods.supplierId = int(supplier_id or 0)
    goods.spuNum = spu_num
    goods.skuId = int(sku_id or 0)
    goods.price = int(price or 0)
    goods.gameId = int(game_id or 0)
    goods.spuNum2 = spu_num
    request.goods = goods
    return RpcRequest("PlayMallUI", "getMyPromotion", request)


def create_order_v6_request(
    user: Any,
    shop_app: Any,
    *,
    pid: int,
    items: list[dict],
    source_id: str,
    game_id: str = "0",
    biz_params: str,
) -> RpcRequest:
    request = CreateOrderReqV6()
    request.userId = user
    request.shopAppInfo = shop_app
    request.items = []
    for item in items:
        row = CreateOrderItemReqV6()
        row.spuId = str(item.get("spu_id") or "")
        row.skuId = int(item.get("sku_id") or 0)
        row.itemCount = int(item.get("item_count") or 1)
        row.goodsType = int(item.get("goods_type") or 0)
        row.virtualType = int(item.get("virtual_type") or 0)
        request.items.append(row)
    request.pid = int(pid or 0)
    request.gameId = str(game_id or "0")
    request.src = 2
    request.orderType = 6
    extra = CreateOrderExtraParam()
    extra.bizParams = biz_params
    request.extraParam = extra
    request.sourceId = source_id
    request.env = {"user_click_seq": "[]"}
    request.orderScene = 3
    request.promotionParam = CreateOrderPromotionParam()
    request.accountParam = CreateOrderAccountParam()
    return RpcRequest("shopMiddleUI", "createOrderV6", request, CreateOrderRspV6)