""" bilibili_api.video 视频相关操作 注意，同时存在 page_index 和 cid 的参数，两者至少提供一个。 """ import re import os import json import struct import asyncio import logging import datetime from enum import Enum from inspect import isfunction from functools import cmp_to_key from dataclasses import dataclass from typing import Any, List, Union, Optional import httpx import aiohttp from . import settings from .utils.aid_bvid_transformer import bvid2aid, aid2bvid from .utils.utils import get_api from .utils.AsyncEvent import AsyncEvent from .utils.credential import Credential from .utils.BytesReader import BytesReader from .utils.danmaku import Danmaku, SpecialDanmaku from .utils.network import get_aiohttp_session, Api, get_session from .exceptions import ( ArgsException, NetworkException, ResponseException, DanmakuClosedException, ) API = get_api("video") def get_cid_info_sync(cid: int): """ 获取 cid 信息 (对应的视频，具体分 P 序号，up 等) Returns: dict: 调用 https://hd.biliplus.com 的 API 返回的结果 """ api = API["info"]["cid_info"] params = {"cid": cid} return Api(**api).update_params(**params).result_sync async def get_cid_info(cid: int): """ 获取 cid 信息 (对应的视频，具体分 P 序号，up 等) Returns: dict: 调用 https://hd.biliplus.com 的 API 返回的结果 """ api = API["info"]["cid_info"] params = {"cid": cid} return await Api(**api).update_params(**params).result class DanmakuOperatorType(Enum): """ 弹幕操作枚举 + DELETE - 删除弹幕 + PROTECT - 保护弹幕 + UNPROTECT - 取消保护弹幕 """ DELETE = 1 PROTECT = 2 UNPROTECT = 3 class VideoAppealReasonType: """ 视频投诉原因枚举 注意: 每一项均为函数，部分项有参数，没有参数的函数无需调用函数，直接传入即可，有参数的函数请调用结果之后传入。 - ILLEGAL(): 违法违禁 - PRON(): 色情 - VULGAR(): 低俗 - GAMBLED_SCAMS(): 赌博诈骗 - VIOLENT(): 血腥暴力 - PERSONAL_ATTACK(): 人身攻击 - PLAGIARISM(bvid: str): 与站内其他视频撞车 - BAD_FOR_YOUNGS(): 青少年不良信息 - CLICKBAIT(): 不良封面/标题 - POLITICAL_RUMORS(): 涉政谣言 - SOCIAL_RUMORS(): 涉社会事件谣言 - COVID_RUMORS(): 疫情谣言 - UNREAL_EVENT(): 虚假不实消息 - OTHER(): 有其他问题 - LEAD_WAR(): 引战 - CANNOT_CHARGE(): 不能参加充电 - UNREAL_COPYRIGHT(source: str): 转载/自制类型错误 """ ILLEGAL = lambda: 2 PRON = lambda: 3 VULGAR = lambda: 4 GAMBLED_SCAMS = lambda: 5 VIOLENT = lambda: 6 PERSONAL_ATTACK = lambda: 7 BAD_FOR_YOUNGS = lambda: 10000 CLICKBAIT = lambda: 10013 POLITICAL_RUMORS = lambda: 10014 SOCIAL_RUMORS = lambda: 10015 COVID_RUMORS = lambda: 10016 UNREAL_EVENT = lambda: 10017 OTHER = lambda: 1 LEAD_WAR = lambda: 9 CANNOT_CHARGE = lambda: 10 @staticmethod def PLAGIARISM(bvid: str): """ 与站内其他视频撞车 Args: bvid (str): 撞车对象 """ return {"tid": 8, "撞车对象": bvid} @staticmethod def UNREAL_COPYRIGHT(source: str): """ 转载/自制类型错误 Args: source (str): 原创视频出处 """ return {"tid": 52, "出处": source} class Video: """ 视频类，各种对视频的操作均在里面。 """ def __init__( self, bvid: Union[None, str] = None, aid: Union[None, int] = None, credential: Union[None, Credential] = None, ): """ Args: bvid (str | None, optional) : BV 号. bvid 和 aid 必须提供其中之一。 aid (int | None, optional) : AV 号. bvid 和 aid 必须提供其中之一。 credential (Credential | None, optional): Credential 类. Defaults to None. """ # ID 检查 if bvid is not None: self.set_bvid(bvid) elif aid is not None: self.set_aid(aid) else: # 未提供任一 ID raise ArgsException("请至少提供 bvid 和 aid 中的其中一个参数。") # 未提供 credential 时初始化该类 self.credential: Credential = Credential() if credential is None else credential # 用于存储视频信息，避免接口依赖视频信息时重复调用 self.__info: Union[dict, None] = None def set_bvid(self, bvid: str) -> None: """ 设置 bvid。 Args: bvid (str): 要设置的 bvid。 """ # 检查 bvid 是否有效 if not re.search("^BV[a-zA-Z0-9]{10}$", bvid): raise ArgsException("bvid 提供错误，必须是以 BV 开头的纯字母和数字组成的 12 位字符串（大小写敏感）。") self.__bvid = bvid self.__aid = bvid2aid(bvid) def get_bvid(self) -> str: """ 获取 BVID。 Returns: str: BVID。 """ return self.__bvid def set_aid(self, aid: int) -> None: """ 设置 aid。 Args: aid (int): AV 号。 """ if aid <= 0: raise ArgsException("aid 不能小于或等于 0。") self.__aid = aid self.__bvid = aid2bvid(aid) def get_aid(self) -> int: """ 获取 AID。 Returns: int: aid。 """ return self.__aid async def get_info(self) -> dict: """ 获取视频信息。 Returns: dict: 调用 API 返回的结果。 """ api = API["info"]["info"] params = {"bvid": self.get_bvid(), "aid": self.get_aid()} resp = ( await Api(**api, credential=self.credential).update_params(**params).result ) # 存入 self.__info 中以备后续调用 self.__info = resp return resp async def get_detail(self) -> dict: """ 获取视频详细信息 Returns: dict: 调用 API 返回的结果。 """ api = API["info"]["detail"] params = { "bvid": self.get_bvid(), "aid": self.get_aid(), "need_operation_card": 0, "need_elec": 0, } return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def __get_info_cached(self) -> dict: """ 获取视频信息，如果已获取过则使用之前获取的信息，没有则重新获取。 Returns: dict: 调用 API 返回的结果。 """ if self.__info is None: return await self.get_info() return self.__info # get_stat 403/404 https://github.com/SocialSisterYi/bilibili-API-collect/issues/797 等恢复 # async def get_stat(self) -> dict: # """ # 获取视频统计数据（播放量，点赞数等）。 # Returns: # dict: 调用 API 返回的结果。 # """ # api = API["info"]["stat"] # params = {"bvid": self.get_bvid(), "aid": self.get_aid()} # return await Api(**api, credential=self.credential).update_params(**params).result async def get_up_mid(self) -> int: """ 获取视频 up 主的 mid。 Returns: int: up_mid """ info = await self.__get_info_cached() return info["owner"]["mid"] async def get_tags( self, page_index: Union[int, None] = 0, cid: Union[int, None] = None ) -> List[dict]: """ 获取视频标签。 Args: page_index (int | None): 分 P 序号. Defaults to 0. cid (int | None): 分 P 编码. Defaults to None. Returns: List[dict]: 调用 API 返回的结果。 """ if cid == None: if page_index == None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.get_cid(page_index=page_index) api = API["info"]["tags"] params = {"bvid": self.get_bvid(), "aid": self.get_aid(), "cid": cid} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def get_chargers(self) -> dict: """ 获取视频充电用户。 Returns: dict: 调用 API 返回的结果。 """ info = await self.__get_info_cached() mid = info["owner"]["mid"] api = API["info"]["chargers"] params = {"aid": self.get_aid(), "bvid": self.get_bvid(), "mid": mid} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def get_pages(self) -> List[dict]: """ 获取分 P 信息。 Returns: dict: 调用 API 返回的结果。 """ api = API["info"]["pages"] params = {"aid": self.get_aid(), "bvid": self.get_bvid()} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def __get_cid_by_index(self, page_index: int) -> int: """ 根据分 p 号获取 cid。 Args: page_index (int): 分 P 号，从 0 开始。 Returns: int: cid 分 P 的唯一 ID。 """ if page_index < 0: raise ArgsException("分 p 号必须大于或等于 0。") info = await self.__get_info_cached() pages = info["pages"] if len(pages) <= page_index: raise ArgsException("不存在该分 p。") page = pages[page_index] cid = page["cid"] return cid async def get_video_snapshot( self, cid: Union[int, None] = None, json_index: bool = False, pvideo: bool = True, ) -> dict: """ 获取视频快照(视频各个时间段的截图拼图) Args: cid(int): 分 P CID(可选) json_index(bool): json 数组截取时间表 True 为需要，False 不需要 pvideo(bool): 是否只获取预览 Returns: dict: 调用 API 返回的结果,数据中 Url 没有 http 头 """ params: dict[str, Any] = {"aid": self.get_aid()} if pvideo: api = API["info"]["video_snapshot_pvideo"] else: params["bvid"] = self.get_bvid() if json_index: params["index"] = 1 if cid: params["cid"] = cid api = API["info"]["video_snapshot"] return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def get_cid(self, page_index: int) -> int: """ 获取稿件 cid Args: page_index(int): 分 P Returns: int: cid """ return await self.__get_cid_by_index(page_index) async def get_download_url( self, page_index: Union[int, None] = None, cid: Union[int, None] = None, html5: bool = False, ) -> dict: """ 获取视频下载信息。 返回结果可以传入 `VideoDownloadURLDataDetecter` 进行解析。 page_index 和 cid 至少提供其中一个，其中 cid 优先级最高 Args: page_index (int | None, optional) : 分 P 号，从 0 开始。Defaults to None cid (int | None, optional) : 分 P 的 ID。Defaults to None html5 (bool, optional): 是否以 html5 平台访问，这样子能直接在网页中播放，但是链接少。 Returns: dict: 调用 API 返回的结果。 """ if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) api = API["info"]["playurl"] if html5: params = { "avid": self.get_aid(), "cid": cid, "qn": "127", "otype": "json", "fnval": 4048, "fourk": 1, "platform": "html5", "high_quality": "1", } else: params = { "avid": self.get_aid(), "cid": cid, "qn": "127", "otype": "json", "fnval": 4048, "fourk": 1, } result = ( await Api(**api, credential=self.credential).update_params(**params).result ) result.update({"is_html5": True} if html5 else {}) return result async def get_related(self) -> dict: """ 获取相关视频信息。 Returns: dict: 调用 API 返回的结果。 """ api = API["info"]["related"] params = {"aid": self.get_aid(), "bvid": self.get_bvid()} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def get_relation(self) -> dict: """ 获取用户与视频的关系 Returns: dict: 调用 API 返回的结果。 """ self.credential.raise_for_no_sessdata() api = API["info"]["relation"] params = {"aid": self.get_aid(), "bvid": self.get_bvid()} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def has_liked(self) -> bool: """ 视频是否点赞过。 Returns: bool: 视频是否点赞过。 """ self.credential.raise_for_no_sessdata() api = API["info"]["has_liked"] params = {"bvid": self.get_bvid(), "aid": self.get_aid()} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def get_pay_coins(self) -> int: """ 获取视频已投币数量。 Returns: int: 视频已投币数量。 """ self.credential.raise_for_no_sessdata() api = API["info"]["get_pay_coins"] params = {"bvid": self.get_bvid(), "aid": self.get_aid()} return ( await Api(**api, credential=self.credential).update_params(**params).result )["multiply"] async def has_favoured(self) -> bool: """ 是否已收藏。 Returns: bool: 视频是否已收藏。 """ self.credential.raise_for_no_sessdata() api = API["info"]["has_favoured"] params = {"bvid": self.get_bvid(), "aid": self.get_aid()} return ( await Api(**api, credential=self.credential).update_params(**params).result )["favoured"] async def is_forbid_note(self) -> bool: """ 是否禁止笔记。 Returns: bool: 是否禁止笔记。 """ api = API["info"]["is_forbid"] params = {"aid": self.get_aid()} return ( await Api(**api, credential=self.credential).update_params(**params).result )["forbid_note_entrance"] async def get_private_notes_list(self) -> list: """ 获取稿件私有笔记列表。 Returns: list: note_Ids。 """ self.credential.raise_for_no_sessdata() api = API["info"]["private_notes"] params = {"oid": self.get_aid(), "oid_type": 0} return ( await Api(**api, credential=self.credential).update_params(**params).result )["noteIds"] async def get_public_notes_list(self, pn: int, ps: int) -> dict: """ 获取稿件公开笔记列表。 Args: pn (int): 页码 ps (int): 每页项数 Returns: dict: 调用 API 返回的结果。 """ api = API["info"]["public_notes"] params = {"oid": self.get_aid(), "oid_type": 0, "pn": pn, "ps": ps} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def get_ai_conclusion(self, cid: Optional[int] = None, page_index: Optional[int] = None, up_mid: Optional[int] = None) -> dict: """ 获取稿件 AI 总结结果。 cid 和 page_index 至少提供其中一个，其中 cid 优先级最高 Args: cid (Optional, int): 分 P 的 cid。 page_index (Optional, int): 分 P 号，从 0 开始。 up_mid (Optional, int): up 主的 mid。 Returns: dict: 调用 API 返回的结果。 """ if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) api = API["info"]["ai_conclusion"] params = {"aid": self.get_aid(), "bvid": self.get_bvid(), "cid": cid, "up_mid": await self.get_up_mid() if up_mid is None else up_mid} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def get_danmaku_view( self, page_index: Union[int, None] = None, cid: Union[int, None] = None ) -> dict: """ 获取弹幕设置、特殊弹幕、弹幕数量、弹幕分段等信息。 Args: page_index (int, optional): 分 P 号，从 0 开始。Defaults to None cid (int, optional): 分 P 的 ID。Defaults to None Returns: dict: 调用 API 返回的结果。 """ if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) session = get_session() api = API["danmaku"]["view"] config = {} config["url"] = api["url"] config["params"] = {"type": 1, "oid": cid, "pid": self.get_aid()} config["cookies"] = self.credential.get_cookies() config["headers"] = { "Referer": "https://www.bilibili.com", "User-Agent": "Mozilla/5.0", } try: resp = await session.get(**config) except Exception as e: raise NetworkException(-1, str(e)) resp_data = resp.read() json_data = {} reader = BytesReader(resp_data) # 解析二进制数据流 def read_dm_seg(stream: bytes): reader_ = BytesReader(stream) data = {} while not reader_.has_end(): t = reader_.varint() >> 3 if t == 1: data["page_size"] = reader_.varint() elif t == 2: data["total"] = reader_.varint() else: continue return data def read_flag(stream: bytes): reader_ = BytesReader(stream) data = {} while not reader_.has_end(): t = reader_.varint() >> 3 if t == 1: data["rec_flag"] = reader_.varint() elif t == 2: data["rec_text"] = reader_.string() elif t == 3: data["rec_switch"] = reader_.varint() else: continue return data def read_command_danmakus(stream: bytes): reader_ = BytesReader(stream) data = {} while not reader_.has_end(): t = reader_.varint() >> 3 if t == 1: data["id"] = reader_.varint() elif t == 2: data["oid"] = reader_.varint() elif t == 3: data["mid"] = reader_.varint() elif t == 4: data["commend"] = reader_.string() elif t == 5: data["content"] = reader_.string() elif t == 6: data["progress"] = reader_.varint() elif t == 7: data["ctime"] = reader_.string() elif t == 8: data["mtime"] = reader_.string() elif t == 9: data["extra"] = json.loads(reader_.string()) elif t == 10: data["id_str"] = reader_.string() else: continue return data def read_settings(stream: bytes): reader_ = BytesReader(stream) data = {} while not reader_.has_end(): t = reader_.varint() >> 3 if t == 1: data["dm_switch"] = reader_.bool() elif t == 2: data["ai_switch"] = reader_.bool() elif t == 3: data["ai_level"] = reader_.varint() elif t == 4: data["enable_top"] = reader_.bool() elif t == 5: data["enable_scroll"] = reader_.bool() elif t == 6: data["enable_bottom"] = reader_.bool() elif t == 7: data["enable_color"] = reader_.bool() elif t == 8: data["enable_special"] = reader_.bool() elif t == 9: data["prevent_shade"] = reader_.bool() elif t == 10: data["dmask"] = reader_.bool() elif t == 11: data["opacity"] = reader_.float(True) elif t == 12: data["dm_area"] = reader_.varint() elif t == 13: data["speed_plus"] = reader_.float(True) elif t == 14: data["font_size"] = reader_.float(True) elif t == 15: data["screen_sync"] = reader_.bool() elif t == 16: data["speed_sync"] = reader_.bool() elif t == 17: data["font_family"] = reader_.string() elif t == 18: data["bold"] = reader_.bool() elif t == 19: data["font_border"] = reader_.varint() elif t == 20: data["draw_type"] = reader_.string() else: continue return data def read_image_danmakus(string: bytes): image_list = [] reader_ = BytesReader(string) while not reader_.has_end(): type_ = reader_.varint() >> 3 if type_ == 1: details_dict: dict[Any, Any] = {"texts": []} img_details = reader_.bytes_string() reader_details = BytesReader(img_details) while not reader_details.has_end(): type_details = reader_details.varint() >> 3 if type_details == 1: details_dict["texts"].append(reader_details.string()) elif type_details == 2: details_dict["image"] = reader_details.string() elif type_details == 3: id_string = reader_details.bytes_string() id_reader = BytesReader(id_string) while not id_reader.has_end(): type_id = id_reader.varint() >> 3 if type_id == 2: details_dict["id"] = id_reader.varint() else: raise ResponseException("解析响应数据错误") image_list.append(details_dict) else: raise ResponseException("解析响应数据错误") return image_list while not reader.has_end(): type_ = reader.varint() >> 3 if type_ == 1: json_data["state"] = reader.varint() elif type_ == 2: json_data["text"] = reader.string() elif type_ == 3: json_data["text_side"] = reader.string() elif type_ == 4: json_data["dm_seg"] = read_dm_seg(reader.bytes_string()) elif type_ == 5: json_data["flag"] = read_flag(reader.bytes_string()) elif type_ == 6: if "special_dms" not in json_data: json_data["special_dms"] = [] json_data["special_dms"].append(reader.string()) elif type_ == 7: json_data["check_box"] = reader.bool() elif type_ == 8: json_data["count"] = reader.varint() elif type_ == 9: if "command_dms" not in json_data: json_data["command_dms"] = [] json_data["command_dms"].append( read_command_danmakus(reader.bytes_string()) ) elif type_ == 10: json_data["dm_setting"] = read_settings(reader.bytes_string()) elif type_ == 12: json_data["image_dms"] = read_image_danmakus(reader.bytes_string()) else: continue return json_data async def get_danmakus( self, page_index: int = 0, date: Union[datetime.date, None] = None, cid: Union[int, None] = None, from_seg: Union[int, None] = None, to_seg: Union[int, None] = None, ) -> List[Danmaku]: """ 获取弹幕。 Args: page_index (int, optional): 分 P 号，从 0 开始。Defaults to None date (datetime.Date | None, optional): 指定日期后为获取历史弹幕，精确到年月日。Defaults to None. cid (int | None, optional): 分 P 的 ID。Defaults to None from_seg (int, optional): 从第几段开始(0 开始编号，None 为从第一段开始，一段 6 分钟). Defaults to None. to_seg (int, optional): 到第几段结束(0 开始编号，None 为到最后一段，包含编号的段，一段 6 分钟). Defaults to None. 注意： - 1. 段数可以使用 `get_danmaku_view()["dm_seg"]["total"]` 查询。 - 2. `from_seg` 和 `to_seg` 仅对 `date == None` 的时候有效果。 - 3. 例：取前 `12` 分钟的弹幕：`from_seg=0, to_seg=1` Returns: List[Danmaku]: Danmaku 类的列表。 """ if date is not None: self.credential.raise_for_no_sessdata() if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) session = get_session() aid = self.get_aid() params: dict[str, Any] = {"oid": cid, "type": 1, "pid": aid} if date is not None: # 获取历史弹幕 api = API["danmaku"]["get_history_danmaku"] params["date"] = date.strftime("%Y-%m-%d") params["type"] = 1 from_seg = to_seg = 0 else: api = API["danmaku"]["get_danmaku"] if from_seg == None: from_seg = 0 if to_seg == None: view = await self.get_danmaku_view(cid=cid) to_seg = view["dm_seg"]["total"] - 1 danmakus = [] for seg in range(from_seg, to_seg + 1): if date is None: # 仅当获取当前弹幕时需要该参数 params["segment_index"] = seg + 1 config = {} config["url"] = api["url"] config["params"] = params config["headers"] = { "Referer": "https://www.bilibili.com", "User-Agent": "Mozilla/5.0", } config["cookies"] = self.credential.get_cookies() try: req = await session.get(**config) except Exception as e: raise NetworkException(-1, str(e)) if "content-type" not in req.headers.keys(): break else: content_type = req.headers["content-type"] if content_type != "application/octet-stream": raise ResponseException("返回数据类型错误：") # 解析二进制流数据 data = req.read() if data == b"\x10\x01": # 视频弹幕被关闭 raise DanmakuClosedException() reader = BytesReader(data) while not reader.has_end(): type_ = reader.varint() >> 3 if type_ != 1: if type_ == 4: reader.bytes_string() # 什么鬼？我用 protoc 解析出乱码！ elif type_ == 5: # 大会员专属颜色 reader.varint() reader.varint() reader.varint() reader.bytes_string() elif type_ == 13: # ??? continue else: raise ResponseException("解析响应数据错误") dm = Danmaku("") dm_pack_data = reader.bytes_string() dm_reader = BytesReader(dm_pack_data) while not dm_reader.has_end(): data_type = dm_reader.varint() >> 3 if data_type == 1: dm.id_ = dm_reader.varint() elif data_type == 2: dm.dm_time = dm_reader.varint() / 1000 elif data_type == 3: dm.mode = dm_reader.varint() elif data_type == 4: dm.font_size = dm_reader.varint() elif data_type == 5: color = dm_reader.varint() if color != 60001: color = hex(color)[2:] else: color = "special" dm.color = color elif data_type == 6: dm.crc32_id = dm_reader.string() elif data_type == 7: dm.text = dm_reader.string() elif data_type == 8: dm.send_time = dm_reader.varint() elif data_type == 9: dm.weight = dm_reader.varint() elif data_type == 10: dm.action = str(dm_reader.string()) elif data_type == 11: dm.pool = dm_reader.varint() elif data_type == 12: dm.id_str = dm_reader.string() elif data_type == 13: dm.attr = dm_reader.varint() elif data_type == 14: dm.uid = dm_reader.varint() elif data_type == 15: dm_reader.varint() elif data_type == 20: dm_reader.bytes_string() elif data_type == 21: dm_reader.bytes_string() elif data_type == 22: dm_reader.bytes_string() elif data_type == 25: dm_reader.varint() elif data_type == 26: dm_reader.varint() else: break danmakus.append(dm) return danmakus async def get_special_dms( self, page_index: int = 0, cid: Union[int, None] = None ) -> List[SpecialDanmaku]: """ 获取特殊弹幕 Args: page_index (int, optional) : 分 P 号. Defaults to 0. cid (int | None, optional): 分 P id. Defaults to None. Returns: List[SpecialDanmaku]: 调用接口解析后的结果 """ if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) view = await self.get_danmaku_view(cid=cid) special_dms = view["special_dms"][0] if settings.proxy != "": sess = httpx.AsyncClient(proxies={"all://": settings.proxy}) else: sess = httpx.AsyncClient() dm_content = await sess.get(special_dms, cookies=self.credential.get_cookies()) dm_content.raise_for_status() reader = BytesReader(dm_content.content) dms: List[SpecialDanmaku] = [] while not reader.has_end(): spec_dm = SpecialDanmaku("") type_ = reader.varint() >> 3 if type_ == 1: reader_ = BytesReader(reader.bytes_string()) while not reader_.has_end(): type__ = reader_.varint() >> 3 if type__ == 1: spec_dm.id_ = reader_.varint() elif type__ == 3: spec_dm.mode = reader_.varint() elif type__ == 4: reader_.varint() elif type__ == 5: reader_.varint() elif type__ == 6: reader_.string() elif type__ == 7: spec_dm.content = reader_.string() elif type__ == 8: reader_.varint() elif type__ == 11: spec_dm.pool = reader_.varint() elif type__ == 12: spec_dm.id_str = reader_.string() else: continue else: continue dms.append(spec_dm) return dms async def get_history_danmaku_index( self, page_index: Union[int, None] = None, date: Union[datetime.date, None] = None, cid: Union[int, None] = None, ) -> Union[None, List[str]]: """ 获取特定月份存在历史弹幕的日期。 Args: page_index (int | None, optional): 分 P 号，从 0 开始。Defaults to None date (datetime.date | None): 精确到年月. Defaults to None。 cid (int | None, optional): 分 P 的 ID。Defaults to None Returns: None | List[str]: 调用 API 返回的结果。不存在时为 None。 """ if date is None: raise ArgsException("请提供 date 参数") self.credential.raise_for_no_sessdata() if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) api = API["danmaku"]["get_history_danmaku_index"] params = {"oid": cid, "month": date.strftime("%Y-%m"), "type": 1} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def has_liked_danmakus( self, page_index: Union[int, None] = None, ids: Union[List[int], None] = None, cid: Union[int, None] = None, ) -> dict: """ 是否已点赞弹幕。 Args: page_index (int | None, optional): 分 P 号，从 0 开始。Defaults to None ids (List[int] | None): 要查询的弹幕 ID 列表。 cid (int | None, optional): 分 P 的 ID。Defaults to None Returns: dict: 调用 API 返回的结果。 """ if ids is None or len(ids) == 0: raise ArgsException("请提供 ids 参数并至少有一个元素") self.credential.raise_for_no_sessdata() if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) api = API["danmaku"]["has_liked_danmaku"] params = {"oid": cid, "ids": ",".join(ids)} # type: ignore return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def send_danmaku( self, page_index: Union[int, None] = None, danmaku: Union[Danmaku, None] = None, cid: Union[int, None] = None, ) -> dict: """ 发送弹幕。 Args: page_index (int | None, optional): 分 P 号，从 0 开始。Defaults to None danmaku (Danmaku | None) : Danmaku 类。 cid (int | None, optional): 分 P 的 ID。Defaults to None Returns: dict: 调用 API 返回的结果。 """ if danmaku is None: raise ArgsException("请提供 danmaku 参数") self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) api = API["danmaku"]["send_danmaku"] if danmaku.is_sub: pool = 1 else: pool = 0 data = { "type": 1, "oid": cid, "msg": danmaku.text, "aid": self.get_aid(), "bvid": self.get_bvid(), "progress": int(danmaku.dm_time * 1000), "color": int(danmaku.color, 16), "fontsize": danmaku.font_size, "pool": pool, "mode": danmaku.mode, "plat": 1, } return await Api(**api, credential=self.credential).update_data(**data).result async def get_danmaku_xml( self, page_index: Union[int, None] = None, cid: Union[int, None] = None ) -> str: """ 获取所有弹幕的 xml 源文件（非装填） Args: page_index (int, optional) : 分 P 序号. Defaults to 0. cid (int | None, optional): cid. Defaults to None. Return: xml 文件源 """ if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) url = f"https://comment.bilibili.com/{cid}.xml" sess = get_session() config: dict[Any, Any] = {"url": url} # 代理 if settings.proxy: config["proxies"] = {"all://", settings.proxy} resp = await sess.get(**config) return resp.content.decode("utf-8") async def like_danmaku( self, page_index: Union[int, None] = None, dmid: Union[int, None] = None, status: Union[bool, None] = True, cid: Union[int, None] = None, ) -> dict: """ 点赞弹幕。 Args: page_index (int | None, optional) : 分 P 号，从 0 开始。Defaults to None dmid (int | None) : 弹幕 ID。 status (bool | None, optional): 点赞状态。Defaults to True cid (int | None, optional) : 分 P 的 ID。Defaults to None Returns: dict: 调用 API 返回的结果。 """ if dmid is None: raise ArgsException("请提供 dmid 参数") self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) api = API["danmaku"]["like_danmaku"] data = { "dmid": dmid, "oid": cid, "op": 1 if status else 2, "platform": "web_player", } return await Api(**api, credential=self.credential).update_data(**data).result async def get_online(self, cid: Optional[int] = None, page_index: Optional[int] = 0) -> dict: """ 获取实时在线人数 Returns: dict: 调用 API 返回的结果。 """ api = API["info"]["online"] params = {"aid": self.get_aid(), "bvid": self.get_bvid(), "cid": cid if cid is not None else await self.get_cid(page_index=page_index)} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def operate_danmaku( self, page_index: Union[int, None] = None, dmids: Union[List[int], None] = None, cid: Union[int, None] = None, type_: Union[DanmakuOperatorType, None] = None, ) -> dict: """ 操作弹幕 Args: page_index (int | None, optional) : 分 P 号，从 0 开始。Defaults to None dmids (List[int] | None) : 弹幕 ID 列表。 cid (int | None, optional) : 分 P 的 ID。Defaults to None type_ (DanmakuOperatorType | None): 操作类型 Returns: dict: 调用 API 返回的结果。 """ if dmids is None or len(dmids) == 0: raise ArgsException("请提供 dmid 参数") if type_ is None: raise ArgsException("请提供 type_ 参数") self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) api = API["danmaku"]["edit_danmaku"] data = { "type": 1, "dmids": ",".join(map(lambda x: str(x), dmids)), "oid": cid, "state": type_.value, } return await Api(**api, credential=self.credential).update_data(**data).result async def like(self, status: bool = True) -> dict: """ 点赞视频。 Args: status (bool, optional): 点赞状态。Defaults to True. Returns: dict: 调用 API 返回的结果。 """ self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() api = API["operate"]["like"] data = {"aid": self.get_aid(), "like": 1 if status else 2} return await Api(**api, credential=self.credential).update_data(**data).result async def pay_coin(self, num: int = 1, like: bool = False) -> dict: """ 投币。 Args: num (int, optional) : 硬币数量，为 1 ~ 2 个。Defaults to 1. like (bool, optional): 是否同时点赞。Defaults to False. Returns: dict: 调用 API 返回的结果。 """ self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() if num not in (1, 2): raise ArgsException("投币数量只能是 1 ~ 2 个。") api = API["operate"]["coin"] data = { "aid": self.get_aid(), "bvid": self.get_bvid(), "multiply": num, "like": 1 if like else 0, } return await Api(**api, credential=self.credential).update_data(**data).result async def share(self) -> int: """ 分享视频 Returns: int: 当前分享数 """ api = API["operate"]["share"] data = { "bvid": self.get_bvid(), "aid": self.get_aid(), "csrf": self.credential.bili_jct, } return await Api(**api, credential=self.credential).update_data(**data).result async def triple(self) -> dict: """ 给阿婆主送上一键三连 Returns: dict: 调用 API 返回的结果 """ api = API["operate"]["yjsl"] data = {"bvid": self.get_bvid(), "aid": self.get_aid()} return await Api(**api, credential=self.credential).update_data(**data).result async def add_tag(self, name: str) -> dict: """ 添加标签。 Args: name (str): 标签名字。 Returns: dict: 调用 API 返回的结果。会返回标签 ID。 """ self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() api = API["operate"]["add_tag"] data = {"aid": self.get_aid(), "bvid": self.get_bvid(), "tag_name": name} return await Api(**api, credential=self.credential).update_data(**data).result async def delete_tag(self, tag_id: int) -> dict: """ 删除标签。 Args: tag_id (int): 标签 ID。 Returns: dict: 调用 API 返回的结果。 """ self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() api = API["operate"]["del_tag"] data = {"tag_id": tag_id, "aid": self.get_aid(), "bvid": self.get_bvid()} return await Api(**api, credential=self.credential).update_data(**data).result async def appeal(self, reason: Any, detail: str): """ 投诉稿件 Args: reason (Any): 投诉类型。传入 VideoAppealReasonType 中的项目即可。 detail (str): 详情信息。 Returns: dict: 调用 API 返回的结果 """ api = API["operate"]["appeal"] data = {"aid": self.get_aid(), "desc": detail} if isfunction(reason): reason = reason() if isinstance(reason, int): reason = {"tid": reason} data.update(reason) # XXX: 暂不支持上传附件 return await Api(**api, credential=self.credential).update_data(**data).result async def set_favorite( self, add_media_ids: List[int] = [], del_media_ids: List[int] = [] ) -> dict: """ 设置视频收藏状况。 Args: add_media_ids (List[int], optional): 要添加到的收藏夹 ID. Defaults to []. del_media_ids (List[int], optional): 要移出的收藏夹 ID. Defaults to []. Returns: dict: 调用 API 返回结果。 """ if len(add_media_ids) + len(del_media_ids) == 0: raise ArgsException("对收藏夹无修改。请至少提供 add_media_ids 和 del_media_ids 中的其中一个。") self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() api = API["operate"]["favorite"] data = { "rid": self.get_aid(), "type": 2, "add_media_ids": ",".join(map(lambda x: str(x), add_media_ids)), "del_media_ids": ",".join(map(lambda x: str(x), del_media_ids)), } return await Api(**api, credential=self.credential).update_data(**data).result async def get_subtitle( self, cid: Union[int, None] = None, ) -> dict: """ 获取视频上一次播放的记录，字幕和地区信息。需要分集的 cid, 返回数据中含有json字幕的链接 Args: cid (int | None): 分 P ID,从视频信息中获取 Returns: 调用 API 返回的结果 """ if cid is None: raise ArgsException("需要 cid") return (await self.get_player_info(cid=cid)).get("subtitle") async def get_player_info( self, cid: Union[int, None] = None, epid: Union[int, None] = None, ) -> dict: """ 获取字幕信息 Args: cid (int | None): 分 P ID,从视频信息中获取 epid (int | None): 番剧分集 ID,从番剧信息中获取 Returns: 调用 API 返回的结果 """ if cid is None: raise ArgsException("需要 cid") api = API["info"]["get_player_info"] params = { "aid": self.get_aid(), "cid": cid, } if epid: params["epid"] = epid return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def submit_subtitle( self, lan: str, data: dict, submit: bool, sign: bool, page_index: Union[int, None] = None, cid: Union[int, None] = None, ) -> dict: """ 上传字幕 字幕数据 data 参考： ```json { "font_size": "float: 字体大小，默认 0.4", "font_color": "str: 字体颜色，默认 \"#FFFFFF\"", "background_alpha": "float: 背景不透明度，默认 0.5", "background_color": "str: 背景颜色，默认 \"#9C27B0\"", "Stroke": "str: 描边，目前作用未知，默认为 \"none\"", "body": [ { "from": "int: 字幕开始时间（秒）", "to": "int: 字幕结束时间（秒）", "location": "int: 字幕位置，默认为 2", "content": "str: 字幕内容" } ] } ``` Args: lan (str) : 字幕语言代码，参考 https://s1.hdslb.com/bfs/subtitle/subtitle_lan.json data (dict) : 字幕数据 submit (bool) : 是否提交，不提交为草稿 sign (bool) : 是否署名 page_index (int | None, optional): 分 P 索引. Defaults to None. cid (int | None, optional): 分 P id. Defaults to None. Returns: dict: API 调用返回结果 """ if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() api = API["operate"]["submit_subtitle"] # lan check，应该是这里面的语言代码 with open( os.path.join(os.path.dirname(__file__), "data/subtitle_lan.json"), encoding="utf-8", ) as f: subtitle_lans = json.load(f) for lan in subtitle_lans: if lan["lan"] == lan: break else: raise ArgsException("lan 参数错误，请参见 https://s1.hdslb.com/bfs/subtitle/subtitle_lan.json") payload = { "type": 1, "oid": cid, "lan": lan, "data": json.dumps(data), "submit": submit, "sign": sign, "bvid": self.get_bvid(), } return await Api(**api, credential=self.credential).update_data(**payload).result async def get_danmaku_snapshot(self) -> dict: """ 获取弹幕快照 Returns: 调用 API 返回的结果 """ api = API["danmaku"]["snapshot"] params = {"aid": self.get_aid()} return ( await Api(**api, credential=self.credential).update_params(**params).result ) async def recall_danmaku( self, page_index: Union[int, None] = None, dmid: int = 0, cid: Union[int, None] = None, ) -> dict: """ 撤回弹幕 Args: page_index(int | None, optional): 分 P 号 dmid(int) : 弹幕 id cid(int | None, optional) : 分 P 编码 Returns: 调用 API 返回的结果 """ if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() api = API["danmaku"]["recall"] data = {"dmid": dmid, "cid": cid} return await Api(**api, credential=self.credential).update_data(**data).result async def get_pbp( self, page_index: Union[int, None] = None, cid: Union[int, None] = None ) -> dict: """ 获取高能进度条 Args: page_index(int | None): 分 P 号 cid(int | None) : 分 P 编码 Returns: 调用 API 返回的结果 """ if cid is None: if page_index is None: raise ArgsException("page_index 和 cid 至少提供一个。") cid = await self.__get_cid_by_index(page_index) api = API["info"]["pbp"] params = {"cid": cid} session = get_session() return json.loads( ( await session.get( api["url"], params=params, cookies=self.credential.get_cookies() ) ).text ) async def add_to_toview(self) -> dict: """ 添加视频至稍后再看列表 Returns: 调用 API 返回的结果 """ self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() api = get_api("toview")["operate"]["add"] datas = { "aid": self.get_aid(), } return await Api(**api, credential=self.credential).update_data(**datas).result async def delete_from_toview(self) -> dict: """ 从稍后再看列表删除视频 Returns: 调用 API 返回的结果 """ self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() api = get_api("toview")["operate"]["del"] datas = {"viewed": "false", "aid": self.get_aid()} return await Api(**api, credential=self.credential).update_data(**datas).result class VideoOnlineMonitor(AsyncEvent): """ 视频在线人数实时监测。 示例代码： ```python import asyncio from bilibili_api import video # 实例化 r = video.VideoOnlineMonitor("BV1Bf4y1Q7QP") # 装饰器方法注册事件监听器 @r.on("ONLINE") async def handler(data): print(data) # 函数方法注册事件监听器 async def handler2(data): print(data) r.add_event_listener("ONLINE", handler2) asyncio.get_event_loop().run_until_complete(r.connect()) ``` Extends: AsyncEvent Events: ONLINE： 在线人数更新。 CallbackData: dict。 DANMAKU： 收到实时弹幕。 CallbackData: Danmaku。 DISCONNECTED： 正常断开连接。 CallbackData: None。 ERROR: 发生错误。 CallbackData: aiohttp.ClientWebSocketResponse。 CONNECTED: 成功连接。 CallbackData: None。 """ class Datapack(Enum): """ 数据包类型枚举。 + CLIENT_VERIFY : 客户端发送验证信息。 + SERVER_VERIFY : 服务端响应验证信息。 + CLIENT_HEARTBEAT: 客户端发送心跳包。 + SERVER_HEARTBEAT: 服务端响应心跳包。 + DANMAKU : 实时弹幕更新。 """ CLIENT_VERIFY = 0x7 SERVER_VERIFY = 0x8 CLIENT_HEARTBEAT = 0x2 SERVER_HEARTBEAT = 0x3 DANMAKU = 0x3E8 def __init__( self, bvid: Union[str, None] = None, aid: Union[int, None] = None, page_index: int = 0, credential: Union[Credential, None] = None, debug: bool = False, ): """ Args: bvid (str | None, optional) : BVID. Defaults to None. aid (int | None, optional) : AID. Defaults to None. page_index (int, optional) : 分 P 序号. Defaults to 0. credential (Credential | None, optional): Credential 类. Defaults to None. debug (bool, optional) : 调试模式，将输出更详细信息. Defaults to False. """ super().__init__() self.credential = credential self.__video = Video(bvid, aid, credential=credential) # 智能选择在 log 中展示的 ID。 id_showed = None if bvid is not None: id_showed = bvid else: id_showed = aid # logger 初始化 self.logger = logging.getLogger(f"VideoOnlineMonitor-{id_showed}") if not self.logger.handlers: handler = logging.StreamHandler() handler.setFormatter( logging.Formatter( "[" + str(id_showed) + "][%(asctime)s][%(levelname)s] %(message)s" ) ) self.logger.addHandler(handler) self.logger.setLevel(logging.INFO if not debug else logging.DEBUG) self.__page_index = page_index self.__tasks = [] async def connect(self): """ 连接服务器 """ await self.__main() async def disconnect(self): """ 断开服务器 """ self.logger.info("主动断开连接。") self.dispatch("DISCONNECTED") await self.__cancel_all_tasks() await self.__ws.close() async def __main(self): """ 入口。 """ # 获取分 P id pages = await self.__video.get_pages() if self.__page_index >= len(pages): raise ArgsException("不存在该分 P。") cid = pages[self.__page_index]["cid"] # 获取服务器信息 self.logger.debug(f"准备连接：{self.__video.get_bvid()}") self.logger.debug(f"获取服务器信息中...") api = API["info"]["video_online_broadcast_servers"] resp = await Api(**api, credential=self.credential).result uri = f"wss://{resp['domain']}:{resp['wss_port']}/sub" self.__heartbeat_interval = resp["heartbeat"] self.logger.debug(f"服务器信息获取成功，URI：{uri}") # 连接服务器 self.logger.debug("准备连接服务器...") session = get_aiohttp_session() async with session.ws_connect(uri) as ws: self.__ws = ws # 发送认证信息 self.logger.debug("服务器连接成功，准备发送认证信息...") verify_info = { "room_id": f"video://{self.__video.get_aid()}/{cid}", "platform": "web", "accepts": [1000, 1015], } verify_info = json.dumps(verify_info, separators=(",", ":")) await ws.send_bytes( self.__pack( VideoOnlineMonitor.Datapack.CLIENT_VERIFY, 1, verify_info.encode() ) ) # 循环接收消息 async for msg in ws: if msg.type == aiohttp.WSMsgType.BINARY: data = self.__unpack(msg.data) self.logger.debug(f"收到消息：{data}") await self.__handle_data(data) # type: ignore elif msg.type == aiohttp.WSMsgType.ERROR: self.logger.warning("连接被异常断开") await self.__cancel_all_tasks() self.dispatch("ERROR", msg) break async def __handle_data(self, data: List[dict]): """ 处理数据。 Args: data (List[dict]): 收到的数据（已解析好）。 """ for d in data: if d["type"] == VideoOnlineMonitor.Datapack.SERVER_VERIFY.value: # 服务器认证反馈。 if d["data"]["code"] == 0: # 创建心跳 Task heartbeat = asyncio.create_task(self.__heartbeat_task()) self.__tasks.append(heartbeat) self.logger.info("连接服务器并验证成功") elif d["type"] == VideoOnlineMonitor.Datapack.SERVER_HEARTBEAT.value: # 心跳包反馈，同时包含在线人数。 self.logger.debug(f'收到服务器心跳包反馈，编号：{d["number"]}') self.logger.info(f'实时观看人数：{d["data"]["data"]["room"]["online"]}') self.dispatch("ONLINE", d["data"]) elif d["type"] == VideoOnlineMonitor.Datapack.DANMAKU.value: # 实时弹幕。 info = d["data"][0].split(",") text = d["data"][1] if info[5] == "0": is_sub = False else: is_sub = True dm = Danmaku( dm_time=float(info[0]), send_time=int(info[4]), crc32_id=info[6], color=info[3], mode=info[1], font_size=info[2], is_sub=is_sub, text=text, ) self.logger.info(f"收到实时弹幕：{dm.text}") self.dispatch("DANMAKU", dm) else: # 未知类型数据包 self.logger.warning("收到未知的数据包类型，无法解析：" + json.dumps(d)) async def __heartbeat_task(self): """ 心跳 Task。 """ index = 2 while True: self.logger.debug(f"发送心跳包，编号：{index}") await self.__ws.send_bytes( self.__pack( VideoOnlineMonitor.Datapack.CLIENT_HEARTBEAT, index, b"[object Object]", ) ) index += 1 await asyncio.sleep(self.__heartbeat_interval) async def __cancel_all_tasks(self): """ 取消所有 Task。 """ for task in self.__tasks: task.cancel() @staticmethod def __pack(data_type: Datapack, number: int, data: bytes): """ 打包数据。 # 数据包格式： 16B 头部: | offset(bytes) | length(bytes) | type | description | | ------------- | ------------- | ---- | ------------------- | | 0 | 4 | I | 数据包长度 | | 4 | 4 | I | 固定 0x00120001 | | 8 | 4 | I | 数据包类型 | | 12 | 4 | I | 递增数据包编号 | | 16 | 2 | H | 固定 0x0000 | 之后是有效载荷。 # 数据包类型表： + 0x7 客户端发送认证信息 + 0x8 服务端回应认证结果 + 0x2 客户端发送心跳包，有效载荷：'[object Object]' + 0x3 服务端回应心跳包，会带上在线人数等信息，返回 JSON + 0x3e8 实时弹幕更新，返回列表，[0]弹幕信息，[1]弹幕文本 Args: data_type (VideoOnlineMonitor.DataType): 数据包类型枚举。 Returns: bytes: 打包好的数据。 """ packed_data = bytearray() packed_data += struct.pack(">I", 0x00120001) packed_data += struct.pack(">I", data_type.value) packed_data += struct.pack(">I", number) packed_data += struct.pack(">H", 0) packed_data += data packed_data = struct.pack(">I", len(packed_data) + 4) + packed_data return bytes(packed_data) @staticmethod def __unpack(data: bytes): """ 解包数据。 Args: data (bytes): 原始数据。 Returns: tuple(dict): 解包后的数据。 """ offset = 0 real_data = [] while offset < len(data): region_header = struct.unpack(">IIII", data[:16]) region_data = data[offset: offset + region_header[0]] real_data.append( { "type": region_header[2], "number": region_header[3], "data": json.loads( region_data[offset + 18: offset + 18 + (region_header[0] - 16)] ), } ) offset += region_header[0] return tuple(real_data) class VideoQuality(Enum): """ 视频的视频流分辨率枚举 - _360P: 流畅 360P - _480P: 清晰 480P - _720P: 高清 720P60 - _1080P: 高清 1080P - _1080P_PLUS: 高清 1080P 高码率 - _1080P_60: 高清 1080P 60 帧码率 - _4K: 超清 4K - HDR: 真彩 HDR - DOLBY: 杜比视界 - _8K: 超高清 8K """ _360P = 16 _480P = 32 _720P = 64 _1080P = 80 _1080P_PLUS = 112 _1080P_60 = 116 _4K = 120 HDR = 125 DOLBY = 126 _8K = 127 class VideoCodecs(Enum): """ 视频的视频流编码枚举 - HEV: HEVC(H.265) - AVC: AVC(H.264) - AV1: AV1 """ HEV = "hev" AVC = "avc" AV1 = "av01" class AudioQuality(Enum): """ 视频的音频流清晰度枚举 - _64K: 64K - _132K: 132K - _192K: 192K - HI_RES: Hi-Res 无损 - DOLBY: 杜比全景声 """ _64K = 30216 _132K = 30232 DOLBY = 30250 HI_RES = 30251 _192K = 30280 @dataclass class VideoStreamDownloadURL: """ (@dataclass) 视频流 URL 类 Attributes: url (str) : 视频流 url video_quality (VideoQuality): 视频流清晰度 video_codecs (VideoCodecs) : 视频流编码 """ url: str video_quality: VideoQuality video_codecs: VideoCodecs @dataclass class AudioStreamDownloadURL: """ (@dataclass) 音频流 URL 类 Attributes: url (str) : 音频流 url audio_quality (AudioQuality): 音频流清晰度 """ url: str audio_quality: AudioQuality @dataclass class FLVStreamDownloadURL: """ (@dataclass) FLV 视频流 Attributes: url (str): FLV 流 url """ url: str @dataclass class HTML5MP4DownloadURL: """ (@dataclass) 可供 HTML5 播放的 mp4 视频流 Attributes: url (str): HTML5 mp4 视频流 """ url: str @dataclass class EpisodeTryMP4DownloadURL: """ (@dataclass) 番剧/课程试看的 mp4 播放流 Attributes: url (str): 番剧试看的 mp4 播放流 """ url: str class VideoDownloadURLDataDetecter: """ `Video.get_download_url` 返回结果解析类。 在调用 `Video.get_download_url` 之后可以将代入 `VideoDownloadURLDataDetecter`，此类将一键解析。 目前支持: - 视频清晰度: 360P, 480P, 720P, 1080P, 1080P 高码率, 1080P 60 帧, 4K, HDR, 杜比视界, 8K - 视频编码: HEVC(H.265), AVC(H.264), AV1 - 音频清晰度: 64K, 132K, Hi-Res 无损音效, 杜比全景声, 192K - FLV 视频流 - 番剧/课程试看视频流 """ def __init__(self, data: dict): """ Args: data (dict): `Video.get_download_url` 返回的结果 """ self.__data = data def check_video_and_audio_stream(self) -> bool: """ 判断是否为音视频分离流 Returns: bool: 是否为音视频分离流 """ if "dash" in self.__data.keys(): return True return False def check_flv_stream(self) -> bool: """ 判断是否为 FLV 视频流 Returns: bool: 是否为 FLV 视频流 """ if "durl" in self.__data.keys(): if self.__data["format"].startswith("flv"): return True return False def check_html5_mp4_stream(self) -> bool: """ 判断是否为 HTML5 可播放的 mp4 视频流 Returns: bool: 是否为 HTML5 可播放的 mp4 视频流 """ if "durl" in self.__data.keys(): if self.__data["format"].startswith("mp4"): if self.__data.get("is_html5") == True: return True return False def check_episode_try_mp4_stream(self): """ 判断是否为番剧/课程试看的 mp4 视频流 Returns: bool: 是否为番剧试看的 mp4 视频流 """ if "durl" in self.__data.keys(): if self.__data["format"].startswith("mp4"): if self.__data.get("is_html5") != True: return True return False def detect_all(self): """ 解析并返回所有数据 Returns: List[VideoStreamDownloadURL | AudioStreamDownloadURL | FLVStreamDownloadURL | HTML5MP4DownloadURL | EpisodeTryMP4DownloadURL]: 所有的视频/音频流 """ return self.detect() def detect( self, video_max_quality: VideoQuality = VideoQuality._8K, audio_max_quality: AudioQuality = AudioQuality._192K, video_min_quality: VideoQuality = VideoQuality._360P, audio_min_quality: AudioQuality = AudioQuality._64K, video_accepted_qualities: List[VideoQuality] = [ item for _, item in VideoQuality.__dict__.items() if isinstance(item, VideoQuality) ], audio_accepted_qualities: List[AudioQuality] = [ item for _, item in AudioQuality.__dict__.items() if isinstance(item, AudioQuality) ], codecs: List[VideoCodecs] = [VideoCodecs.AV1, VideoCodecs.AVC, VideoCodecs.HEV], no_dolby_video: bool = False, no_dolby_audio: bool = False, no_hdr: bool = False, no_hires: bool = False, ) -> List[ Union[ VideoStreamDownloadURL, AudioStreamDownloadURL, FLVStreamDownloadURL, HTML5MP4DownloadURL, EpisodeTryMP4DownloadURL, ] ]: """ 解析数据 Args: **以下参数仅能在音视频流分离的情况下产生作用，flv / mp4 试看流 / html5 mp4 流下以下参数均没有作用** video_max_quality (VideoQuality, optional) : 设置提取的视频流清晰度最大值，设置此参数绝对不会禁止 HDR/杜比. Defaults to VideoQuality._8K. audio_max_quality (AudioQuality, optional) : 设置提取的音频流清晰度最大值. 设置此参数绝对不会禁止 Hi-Res/杜比. Defaults to AudioQuality._192K. video_min_quality (VideoQuality, optional) : 设置提取的视频流清晰度最小值，设置此参数绝对不会禁止 HDR/杜比. Defaults to VideoQuality._360P. audio_min_quality (AudioQuality, optional) : 设置提取的音频流清晰度最小值. 设置此参数绝对不会禁止 Hi-Res/杜比. Defaults to AudioQuality._64K. video_accepted_qualities(List[VideoQuality], optional): 设置允许的所有视频流清晰度. Defaults to ALL. audio_accepted_qualities(List[AudioQuality], optional): 设置允许的所有音频清晰度. Defaults to ALL. codecs (List[VideoCodecs], optional) : 设置所有允许提取出来的视频编码. 此项不会忽略 HDR/杜比. Defaults to ALL codecs. no_dolby_video (bool, optional) : 是否禁止提取杜比视界视频流. Defaults to False. no_dolby_audio (bool, optional) : 是否禁止提取杜比全景声音频流. Defaults to False. no_hdr (bool, optional) : 是否禁止提取 HDR 视频流. Defaults to False. no_hires (bool, optional) : 是否禁止提取 Hi-Res 音频流. Defaults to False. Returns: List[VideoStreamDownloadURL | AudioStreamDownloadURL | FLVStreamDownloadURL | HTML5MP4DownloadURL | EpisodeTryMP4DownloadURL]: 提取出来的视频/音频流 """ if "durl" in self.__data.keys(): if self.__data["format"].startswith("flv"): # FLV 视频流 return [FLVStreamDownloadURL(url=self.__data["durl"][0]["url"])] else: if self.check_html5_mp4_stream(): # HTML5 MP4 视频流 return [HTML5MP4DownloadURL(url=self.__data["durl"][0]["url"])] else: # 会员番剧试看 MP4 流 return [EpisodeTryMP4DownloadURL(url=self.__data["durl"][0]["url"])] else: # 正常情况 streams = [] videos_data = self.__data["dash"]["video"] audios_data = self.__data["dash"]["audio"] flac_data = self.__data["dash"]["flac"] dolby_data = self.__data["dash"]["dolby"] for video_data in videos_data: video_stream_url = video_data["baseUrl"] video_stream_quality = VideoQuality(video_data["id"]) if video_stream_quality == VideoQuality.HDR and no_hdr: continue if video_stream_quality == VideoQuality.DOLBY and no_dolby_video: continue if ( video_stream_quality != VideoQuality.DOLBY and video_stream_quality != VideoQuality.HDR and video_stream_quality.value > video_max_quality.value ): continue if ( video_stream_quality != VideoQuality.DOLBY and video_stream_quality != VideoQuality.HDR and video_stream_quality.value < video_min_quality.value ): continue if ( video_stream_quality != VideoQuality.DOLBY and video_stream_quality != VideoQuality.HDR and (not video_stream_quality in video_accepted_qualities) ): continue video_stream_codecs = None for val in VideoCodecs: if val.value in video_data["codecs"]: video_stream_codecs = val if (not video_stream_codecs in codecs) and ( video_stream_codecs != None ): continue video_stream = VideoStreamDownloadURL( url=video_stream_url, video_quality=video_stream_quality, video_codecs=video_stream_codecs, # type: ignore ) streams.append(video_stream) if audios_data: for audio_data in audios_data: audio_stream_url = audio_data["baseUrl"] audio_stream_quality = AudioQuality(audio_data["id"]) if audio_stream_quality.value > audio_max_quality.value: continue if audio_stream_quality.value < audio_min_quality.value: continue if not audio_stream_quality in audio_accepted_qualities: continue audio_stream = AudioStreamDownloadURL( url=audio_stream_url, audio_quality=audio_stream_quality ) streams.append(audio_stream) if flac_data and (not no_hires): if flac_data["audio"]: flac_stream_url = flac_data["audio"]["baseUrl"] flac_stream_quality = AudioQuality(flac_data["audio"]["id"]) flac_stream = AudioStreamDownloadURL( url=flac_stream_url, audio_quality=flac_stream_quality ) streams.append(flac_stream) if dolby_data and (not no_dolby_audio): if dolby_data["audio"]: dolby_stream_data = dolby_data["audio"][0] dolby_stream_url = dolby_stream_data["baseUrl"] dolby_stream_quality = AudioQuality(dolby_stream_data["id"]) dolby_stream = AudioStreamDownloadURL( url=dolby_stream_url, audio_quality=dolby_stream_quality ) streams.append(dolby_stream) return streams def detect_best_streams( self, video_max_quality: VideoQuality = VideoQuality._8K, audio_max_quality: AudioQuality = AudioQuality._192K, video_min_quality: VideoQuality = VideoQuality._360P, audio_min_quality: AudioQuality = AudioQuality._64K, video_accepted_qualities: List[VideoQuality] = [ item for _, item in VideoQuality.__dict__.items() if isinstance(item, VideoQuality) ], audio_accepted_qualities: List[AudioQuality] = [ item for _, item in AudioQuality.__dict__.items() if isinstance(item, AudioQuality) ], codecs: List[VideoCodecs] = [VideoCodecs.AV1, VideoCodecs.AVC, VideoCodecs.HEV], no_dolby_video: bool = False, no_dolby_audio: bool = False, no_hdr: bool = False, no_hires: bool = False, ) -> Union[ List[FLVStreamDownloadURL], List[HTML5MP4DownloadURL], List[EpisodeTryMP4DownloadURL], List[Union[VideoStreamDownloadURL, AudioStreamDownloadURL, None]], ]: """ 提取出分辨率、音质等信息最好的音视频流。 Args: **以下参数仅能在音视频流分离的情况下产生作用，flv / mp4 试看流 / html5 mp4 流下以下参数均没有作用** video_max_quality (VideoQuality) : 设置提取的视频流清晰度最大值，设置此参数绝对不会禁止 HDR/杜比. Defaults to VideoQuality._8K. audio_max_quality (AudioQuality) : 设置提取的音频流清晰度最大值. 设置此参数绝对不会禁止 Hi-Res/杜比. Defaults to AudioQuality._192K. video_min_quality (VideoQuality, optional) : 设置提取的视频流清晰度最小值，设置此参数绝对不会禁止 HDR/杜比. Defaults to VideoQuality._360P. audio_min_quality (AudioQuality, optional) : 设置提取的音频流清晰度最小值. 设置此参数绝对不会禁止 Hi-Res/杜比. Defaults to AudioQuality._64K. video_accepted_qualities(List[VideoQuality], optional): 设置允许的所有视频流清晰度. Defaults to ALL. audio_accepted_qualities(List[AudioQuality], optional): 设置允许的所有音频清晰度. Defaults to ALL. codecs (List[VideoCodecs]) : 设置所有允许提取出来的视频编码. 在数组中越靠前的编码选择优先级越高. 此项不会忽略 HDR/杜比. Defaults to [VideoCodecs.AV1, VideoCodecs.AVC, VideoCodecs.HEV]. no_dolby_video (bool) : 是否禁止提取杜比视界视频流. Defaults to False. no_dolby_audio (bool) : 是否禁止提取杜比全景声音频流. Defaults to False. no_hdr (bool) : 是否禁止提取 HDR 视频流. Defaults to False. no_hires (bool) : 是否禁止提取 Hi-Res 音频流. Defaults to False. Returns: List[VideoStreamDownloadURL | AudioStreamDownloadURL | FLVStreamDownloadURL | HTML5MP4DownloadURL | None]: FLV 视频流 / HTML5 MP4 视频流 / 番剧或课程试看 MP4 视频流返回 `[FLVStreamDownloadURL | HTML5MP4StreamDownloadURL | EpisodeTryMP4DownloadURL]`, 否则为 `[VideoStreamDownloadURL, AudioStreamDownloadURL]`, 如果未匹配上任何合适的流则对应的位置位 `None` """ if self.check_flv_stream(): return self.detect_all() # type: ignore elif self.check_html5_mp4_stream(): return self.detect_all() # type: ignore elif self.check_episode_try_mp4_stream(): return self.detect_all() # type: ignore else: data = self.detect( video_max_quality=video_max_quality, audio_max_quality=audio_max_quality, video_min_quality=video_min_quality, audio_min_quality=audio_min_quality, video_accepted_qualities=video_accepted_qualities, audio_accepted_qualities=audio_accepted_qualities, codecs=codecs, ) video_streams = [] audio_streams = [] for stream in data: if isinstance(stream, VideoStreamDownloadURL): video_streams.append(stream) if isinstance(stream, AudioStreamDownloadURL): audio_streams.append(stream) def video_stream_cmp( s1: VideoStreamDownloadURL, s2: VideoStreamDownloadURL ): # 杜比/HDR 优先 if s1.video_quality == VideoQuality.DOLBY and (not no_dolby_video): return 1 elif s2.video_quality == VideoQuality.DOLBY and (not no_dolby_video): return -1 elif s1.video_quality == VideoQuality.HDR and (not no_hdr): return 1 elif s2.video_quality == VideoQuality.HDR and (not no_hdr): return -1 if s1.video_quality.value != s2.video_quality.value: return s1.video_quality.value - s2.video_quality.value # Detect the high quality stream to the end. elif s1.video_codecs.value != s2.video_codecs.value: return codecs.index(s2.video_codecs) - codecs.index(s1.video_codecs) return -1 def audio_stream_cmp( s1: AudioStreamDownloadURL, s2: AudioStreamDownloadURL ): # 杜比/Hi-Res 优先 if s1.audio_quality == AudioQuality.DOLBY and (not no_dolby_audio): return 1 if s2.audio_quality == AudioQuality.DOLBY and (not no_dolby_audio): return -1 if s1.audio_quality == AudioQuality.HI_RES and (not no_hires): return 1 if s2.audio_quality == AudioQuality.HI_RES and (not no_hires): return -1 return s1.audio_quality.value - s2.audio_quality.value video_streams.sort(key=cmp_to_key(video_stream_cmp), reverse=True) audio_streams.sort(key=cmp_to_key(audio_stream_cmp), reverse=True) if len(video_streams) == 0: video_streams = [None] if len(audio_streams) == 0: audio_streams = [None] return [video_streams[0], audio_streams[0]]