File size: 4,168 Bytes
0aee47a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 |
"""
bilibili_api.live_area
直播间分区相关操作。
"""
import os
import copy
import json
from enum import Enum
from typing import Dict, List, Tuple, Union
from .utils.utils import get_api
from .utils.network import Api
API = get_api("live-area")
class LiveRoomOrder(Enum):
"""
直播间排序方式
- RECOMMEND: 综合
- NEW: 最新
"""
RECOMMEND = ""
NEW = "live_time"
def get_area_info_by_id(id: int) -> Tuple[Union[dict, None], Union[dict, None]]:
"""
根据 id 获取分区信息。
Args:
id (int): 分区的 id。
Returns:
`Tuple[dict | None, dict | None]`: 第一个是主分区，第二个是子分区，没有时返回 None。
"""
with open(
os.path.join(os.path.dirname(__file__), "data/live_area.json"), encoding="utf8"
) as f:
channel = json.loads(f.read())
for main_ch in channel:
if "id" not in main_ch:
continue
if id == int(main_ch["id"]):
return main_ch, None
# 搜索子分区
if "list" in main_ch.keys():
for sub_ch in main_ch["list"]:
if "id" not in sub_ch:
continue
if str(id) == sub_ch["id"]:
return main_ch, sub_ch
else:
return None, None
def get_area_info_by_name(name: str) -> Tuple[Union[dict, None], Union[dict, None]]:
"""
根据频道名称获取频道信息。
Args:
name (str): 分区的名称。
Returns:
Tuple[dict | None, dict | None]: 第一个是主分区，第二个是子分区，没有时返回 None。
"""
with open(
os.path.join(os.path.dirname(__file__), "data/live_area.json"), encoding="utf8"
) as f:
channel = json.loads(f.read())
for main_ch in channel:
if name in main_ch["name"]:
return main_ch, None
if "list" in main_ch.keys():
for sub_ch in main_ch["list"]:
if name in sub_ch["name"]:
return main_ch, sub_ch
else:
return None, None
def get_area_list() -> List[Dict]:
"""
获取所有分区的数据
Returns:
List[dict]: 所有分区的数据
"""
with open(
os.path.join(os.path.dirname(__file__), "data/live_area.json"), encoding="utf8"
) as f:
channel = json.loads(f.read())
channel_list = []
for channel_big in channel:
channel_big_copy = copy.copy(channel_big)
channel_list.append(channel_big_copy)
if "list" in channel_big.keys():
channel_big_copy.pop("list")
for channel_sub in channel_big["list"]:
channel_sub_copy = copy.copy(channel_sub)
channel_sub_copy["father"] = channel_big_copy
channel_list.append(channel_sub_copy)
return channel_list
def get_area_list_sub() -> dict:
"""
获取所有分区的数据
含父子关系（即一层次只有主分区）
Returns:
dict: 所有分区的数据
"""
with open(
os.path.join(os.path.dirname(__file__), "data/live_area.json"), encoding="utf8"
) as f:
channel = json.loads(f.read())
return channel
async def get_list_by_area(
area_id: int, page: int = 1, order: LiveRoomOrder = LiveRoomOrder.RECOMMEND
) -> dict:
"""
根据分区获取直播间列表
Args:
area_id (int) : 分区 id
page (int) : 第几页. Defaults to 1.
order (LiveRoomOrder): 直播间排序方式. Defaults to LiveRoomOrder.RECOMMEND.
Returns:
dict: 调用 API 返回的结果
"""
api = API["info"]["list"]
params = {
"platform": "web",
"parent_area_id": get_area_info_by_id(area_id)[0]["id"],
"area_id": 0 if (get_area_info_by_id(area_id)[1] == None) else area_id,
"page": page,
"sort_type": order.value,
}
return await Api(**api).update_params(**params).result
|