""" bilibili_api.interactive_video 互动视频相关操作 """ # pylint: skip-file import os import copy import enum import json import time import shutil import zipfile from urllib import parse from random import randint as rand from asyncio import CancelledError, create_task from typing import List, Tuple, Union, Callable, Coroutine import httpx from . import settings from .video import Video from .utils.utils import get_api from .utils.AsyncEvent import AsyncEvent from .utils.credential import Credential from .utils.network import Api API = get_api("interactive_video") class InteractiveButtonAlign(enum.Enum): """ 按钮的文字在按钮中的位置 ``` text ----- |xxx|----o (TEXT_LEFT) ----- ----- o----|xxx| (TEXT_RIGHT) ----- ---------- |XXXXXXXX| (DEFAULT) ---------- ``` - DEFAULT - TEXT_UP - TEXT_RIGHT - TEXT_DOWN - TEXT_LEFT """ DEFAULT = 0 TEXT_UP = 1 TEXT_RIGHT = 2 TEXT_DOWN = 3 TEXT_LEFT = 4 class InteractiveNodeJumpingType(enum.Enum): """ 对下一节点的跳转的方式 - ASK : 选择 - DEFAULT: 跳转到默认节点 - READY : 选择(只有一个选择) """ READY = 1 DEFAULT = 0 ASK = 2 class InteractiveVariable: """ 互动节点的变量 """ def __init__( self, name: str, var_id: str, var_value: int, show: bool = False, random: bool = False, ): """ Args: name (str) : 变量名 var_id (str) : 变量 id var_value (int) : 变量的值 show (bool) : 是否显示 random (bool) : 是否为随机值(1-100) """ self.__var_id = var_id self.__var_value = var_value self.__name = name self.__is_show = show self.__random = random def get_id(self) -> str: return self.__var_id def refresh_value(self) -> None: """ 刷新变量数值 """ if self.is_random(): self.__var_value = int(rand(0, 100)) def get_value(self) -> int: return self.__var_value def is_show(self) -> bool: return self.__is_show def is_random(self) -> bool: return self.__random def get_name(self) -> str: return self.__name def __str__(self): return f"{self.__name} {self.__var_value}" class InteractiveButton: """ 互动视频节点按钮类 """ def __init__( self, text: str, x: int, y: int, align: Union[InteractiveButtonAlign, int] = InteractiveButtonAlign.DEFAULT, ): """ Args: text (str) : 文字 x (int) : x 轴 y (int) : y 轴 align (InteractiveButtonAlign | int): 按钮的文字在按钮中的位置 """ self.__text = text self.__pos = (x, y) if isinstance(align, InteractiveButtonAlign): align = align.value self.__align = align def get_text(self) -> str: return self.__text def get_align(self) -> int: return self.__align # type: ignore def get_pos(self) -> Tuple[int, int]: return self.__pos def __str__(self): return f"{self.__text} {self.__pos}" class InteractiveJumpingCondition: """ 节点跳转的公式，只有公式成立才会跳转 """ def __init__(self, var: List[InteractiveVariable] = [], condition: str = "True"): """ Args: var (List[InteractiveVariable]): 所有变量 condition (str) : 公式 """ self.__vars = var self.__command = condition def get_result(self) -> bool: """ 计算公式获得结果 Returns: bool: 是否成立 """ if self.__command == "": return True command = copy.copy(self.__command) for var in self.__vars: var_name = var.get_id() var_value = var.get_value() command = command.replace(var_name, str(var_value)) command = command.replace("&&", " and ") command = command.replace("||", " or ") command = command.replace("!", " not ") command = command.replace("===", "==") command = command.replace("!==", "!=") command = command.replace("true", "True") command = command.replace("false", "False") return eval(command) def __str__(self): return f"{self.__command}" class InteractiveJumpingCommand: """ 节点跳转对变量的操作 """ def __init__(self, var: List[InteractiveVariable] = [], command: str = ""): """ Args: var (List[InteractiveVariable]): 所有变量 command (str) : 公式 """ self.__vars = var self.__command = command def run_command(self) -> List["InteractiveVariable"]: """ 执行操作 Returns: List[InteractiveVariable] """ if self.__command == "": return self.__vars for code in self.__command.split(";"): var_name_ = code.split("=")[0] var_new_value = code.split("=")[1] for var in self.__vars: var_name = var.get_id() var_value = var.get_value() var_new_value = var_new_value.replace(var_name, str(var_value)) var_new_value_calc = eval(var_new_value) for var in self.__vars: if var.get_id() == var_name_: var._InteractiveVariable__var_value = var_new_value_calc # type: ignore return self.__vars class InteractiveNode: """ 互动视频节点类 """ def __init__( self, video: "InteractiveVideo", node_id: int, cid: int, vars: List[InteractiveVariable], button: Union[InteractiveButton, None] = None, condition: InteractiveJumpingCondition = InteractiveJumpingCondition(), native_command: InteractiveJumpingCommand = InteractiveJumpingCommand(), is_default: bool = False, ): """ Args: video (InteractiveVideo) : 视频类 node_id (int) : 节点 id cid (int) : CID vars (List[InteractiveVariable]) : 变量 button (InteractiveButton) : 对应的按钮 condition (InteractiveJumpingCondition): 跳转公式 native_command (InteractiveJumpingCommand) : 跳转时变量操作 is_default (bool) : 是不是默认的跳转的节点 """ self.__parent = video self.__id = node_id self.__cid = cid self.__button = button self.__jumping_command = condition self.__is_default = is_default self.__vars = vars self.__command = native_command self.__vars = self.__command.run_command() async def get_vars(self) -> List[InteractiveVariable]: """ 获取节点的所有变量 Returns: List[InteractiveVariable]: 节点的所有变量 """ return self.__vars async def get_children(self) -> List["InteractiveNode"]: """ 获取节点的所有子节点 Returns: List[InteractiveNode]: 所有子节点 """ edge_info = await self.__parent.get_edge_info(self.__id) nodes = [] if edge_info["edges"].get("questions") == None: return [] for node in edge_info["edges"]["questions"][0]["choices"]: node_id = node["id"] node_cid = node["cid"] if "text_align" in node.keys(): text_align = node["text_align"] else: text_align = 0 if "option" in node.keys(): node_button = InteractiveButton( node["option"], node.get("x"), node.get("y"), text_align ) else: node_button = None node_condition = InteractiveJumpingCondition( await self.get_vars(), node["condition"] ) node_command = InteractiveJumpingCommand( await self.get_vars(), node["native_action"] ) if "is_default" in node.keys(): node_is_default = node["is_default"] else: node_is_default = False node_vars = copy.deepcopy(await self.get_vars()) nodes.append( InteractiveNode( self.__parent, node_id, node_cid, node_vars, node_button, node_condition, node_command, node_is_default, ) ) return nodes def is_default(self) -> bool: return self.__is_default async def get_jumping_type(self) -> int: """ 获取子节点跳转方式 (参考 InteractiveNodeJumpingType) """ edge_info = await self.__parent.get_edge_info(self.__id) return edge_info["edges"]["questions"][0]["type"] def get_node_id(self) -> int: return self.__id def get_cid(self) -> int: return self.__cid def get_self_button(self) -> "InteractiveButton": if self.__button == None: return InteractiveButton("", -1, -1) return self.__button def get_jumping_condition(self) -> "InteractiveJumpingCondition": return self.__jumping_command def get_video(self) -> "InteractiveVideo": return self.__parent async def get_info(self) -> dict: """ 获取节点的简介 Returns: dict: 调用 API 返回的结果 """ return await self.__parent.get_edge_info(self.__id) def __str__(self): return f"{self.get_node_id()}" class InteractiveGraph: """ 情节树类 """ def __init__(self, video: "InteractiveVideo", skin: dict, root_cid: int): """ Args: video (InteractiveVideo): 互动视频类 skin (dict) : 样式 root_cid (int) : 根节点 CID """ self.__parent = video self.__skin = skin self.__node = InteractiveNode(self.__parent, 1, root_cid, []) def get_video(self) -> "InteractiveVideo": return self.__parent def get_skin(self) -> dict: return self.__skin async def get_root_node(self) -> "InteractiveNode": """ 获取根节点 Returns: InteractiveNode: 根节点 """ edge_info = await self.__parent.get_edge_info(None) if "hidden_vars" in edge_info.keys(): node_vars = edge_info["hidden_vars"] else: node_vars = [] var_list = [] for var in node_vars: var_value = var["value"] var_name = var["name"] var_show = var["is_show"] var_id = var["id_v2"] if var["type"] == 2: random = True else: random = False var_list.append( InteractiveVariable(var_name, var_id, var_value, var_show, random) ) self.__node._InteractiveNode__command = InteractiveJumpingCommand( # type: ignore var_list ) self.__node._InteractiveNode__vars = var_list # type: ignore return self.__node async def get_children(self) -> List["InteractiveNode"]: """ 获取子节点 Returns: List[InteractiveNode]: 子节点 """ return await self.__node.get_children() class InteractiveVideo(Video): """ 互动视频类 """ def __init__(self, bvid=None, aid=None, credential=None): super().__init__(bvid, aid, credential) async def up_get_ivideo_pages(self) -> dict: """ 获取交互视频的分 P 信息。up 主需要拥有视频所有权。 Returns: dict: 调用 API 返回的结果 """ credential = self.credential if self.credential else Credential() api = API["info"]["videolist"] params = {"bvid": self.get_bvid()} return await Api(**api, credential=credential).update_params(**params).result async def up_submit_story_tree(self, story_tree: str) -> dict: """ 上传交互视频的情节树。up 主需要拥有视频所有权。 Args: story_tree (str): 情节树的描述，参考 bilibili_storytree.StoryGraph, 需要 Serialize 这个结构 Returns: dict: 调用 API 返回的结果 """ credential = self.credential if self.credential else Credential() api = API["operate"]["savestory"] form_data = {"preview": "0", "data": story_tree, "csrf": credential.bili_jct} headers = { "User-Agent": "Mozilla/5.0", "Referer": "https://member.bilibili.com", "Content-Encoding": "gzip, deflate, br", "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json, text/plain, */*", } data = parse.urlencode(form_data) return ( await Api(**api, credential=credential, no_csrf=True) .update_data(**data) .update_headers(**headers) .result ) async def get_graph_version(self) -> int: """ 获取剧情图版本号，仅供 `get_edge_info()` 使用。 Returns: int: 剧情图版本号 """ # 取得初始顶点 cid bvid = self.get_bvid() credential = self.credential if self.credential else Credential() v = Video(bvid=bvid, credential=credential) page_list = await v.get_pages() cid = page_list[0]["cid"] # 获取剧情图版本号 url = "https://api.bilibili.com/x/player/v2" params = {"bvid": bvid, "cid": cid} resp = ( await Api(method="GET", url=url, credential=credential) .update_params(**params) .result ) return resp["interaction"]["graph_version"] async def get_edge_info(self, edge_id: Union[int, None] = None): """ 获取剧情图节点信息 Args: edge_id (int, optional) : 节点 ID，为 None 时获取根节点信息. Defaults to None. Returns: dict: 调用 API 返回的结果 """ bvid = self.get_bvid() credential = self.credential if self.credential is not None else Credential() api = API["info"]["edge_info"] params = {"bvid": bvid, "graph_version": (await self.get_graph_version())} if edge_id is not None: params["edge_id"] = edge_id return await Api(**api, credential=credential).update_params(**params).result async def mark_score(self, score: int = 5): """ 为互动视频打分 Args: score (int): 互动视频分数. Defaults to 5. Returns: dict: 调用 API 返回的结果 """ self.credential.raise_for_no_sessdata() self.credential.raise_for_no_bili_jct() api = API["operate"]["mark_score"] data = {"mark": score, "bvid": self.get_bvid()} return await Api(**api, credential=self.credential).update_data(**data).result async def get_cid(self) -> int: """ 获取稿件 cid """ return await super().get_cid(0) async def get_graph(self): """ 获取稿件情节树 Returns: InteractiveGraph: 情节树 """ edge_info = await self.get_edge_info(1) cid = await self.get_cid() return InteractiveGraph(self, edge_info["edges"]["skin"], cid) class InteractiveVideoDownloaderEvents(enum.Enum): """ 互动视频下载器事件枚举 | event | meaning | IVI mode | NODE_VIDEOS mode | DOT_GRAPH mode | NO_PACKAGING mode | Is Built-In downloader event | | ----- | ------- | -------- | ---------------- | -------------- | ----------------- | ------------------------- | | START | 开始下载 | [x] | [x] | [x] | [x] | [ ] | | GET | 获取到节点信息 | [x] | [x] | [x] | [x] | [ ] | | PREPARE_DOWNLOAD | 准备下载单个节点 | [x] | [x] | [ ] | [x] | [ ] | | DOWNLOAD_START | 开始下载单个文件 | Unknown | Unknown | [ ] | Unknown | [x] | | DOWNLOAD_PART | 文件分块部分完成 | Unknown | Unknown | [ ] | Unknown | [x] | | DOWNLOAD_SUCCESS | 完成下载 | Unknown | Unknown | [ ] | Unknown | [x] | | PACKAGING | 正在打包 | [x] | [ ] | [ ] | [ ] | [ ] | | SUCCESS | 下载成功 | [x] | [x] | [x] | [x] | [ ] | | ABORTED | 用户暂停 | [x] | [x] | [x] | [x] | [ ] | | FAILED | 下载失败 | [x] | [x] | [x] | [x] | [ ] | """ START = "START" GET = "GET" DOWNLOAD_START = "DOWNLOAD_START" DOWNLOAD_PART = "DOWNLOAD_PART" DOWNLOAD_SUCCESS = "DOWNLOAD_SUCCESS" PACKAGING = "PACKAGING" SUCCESS = "SUCCESS" ABORTED = "ABORTED" FAILED = "FAILED" class InteractiveVideoDownloaderMode(enum.Enum): """ 互动视频下载模式 - IVI: 下载可播放的 ivi 文件 - NODE_VIDEOS: 下载所有节点的所有视频并存放在某个文件夹，每一个节点的视频命名为 `{节点 id} {节点标题 (自动去除敏感字符)}.mp4` - DOT_GRAPH: 下载 dot 格式的情节树图表 - NO_PACKAGING: 前面按照 ivi 文件下载步骤进行下载，但是最终不会打包成为 ivi 文件，所有文件将存放于一个文件夹中。互动视频数据将存放在一个文件夹中，里面的文件命名/含义与拆包后的 ivi 文件完全相同。 """ IVI = "ivi" NODE_VIDEOS = "videos" DOT_GRAPH = "dot" NO_PACKAGING = "no_pack" class InteractiveVideoDownloader(AsyncEvent): """ 互动视频下载类 """ def __init__( self, video: InteractiveVideo, out: str = "", self_download_func: Union[Coroutine, None] = None, downloader_mode: InteractiveVideoDownloaderMode = InteractiveVideoDownloaderMode.IVI, ): """ Args: video (InteractiveVideo) : 互动视频类 out (str) : 输出文件地址 (如果模式为 NODE_VIDEOS/NO_PACKAGING 则此参数表示所有节点视频的存放目录) self_download_func (Coroutine | None) : 自定义下载函数（需 async 函数） downloader_mode (InteractiveVideoDownloaderMode): 下载模式 `self_download_func` 函数应接受两个参数（第一个是下载 URL，第二个是输出地址（精确至文件名）） """ super().__init__() self.__video = video if self_download_func == None: self.__download_func = self.__download else: self.__download_func = self_download_func self.__task = None self.__out = out self.__mode = downloader_mode async def __download(self, url: str, out: str) -> None: resp = httpx.get( url, headers={ "User-Agent": "Mozilla/5.0", "Referer": "https://www.bilibili.com", }, proxies={"all://": settings.proxy}, stream=True, ) resp.raise_for_status() if os.path.exists(out): os.remove(out) parent = os.path.dirname(out) if not os.path.exists(parent): os.mkdir(parent) self.dispatch("DOWNLOAD_START", {"url": url, "out": out}) all_length = int(resp.headers["Content-Length"]) parts = all_length // 1024 + (1 if all_length % 1024 != 0 else 0) cnt = 0 start_time = time.perf_counter() with open(out, "wb") as f: for chunk in resp.iter_bytes(1024): cnt += 1 self.dispatch( "DOWNLOAD_PART", { "done": cnt, "total": parts, "time": int(time.perf_counter() - start_time), }, ) f.write(chunk) self.dispatch("DOWNLOAD_SUCCESS") async def __main(self) -> None: # 初始化 self.dispatch("START") if self.__out == "": self.__out = self.__video.get_bvid() + ".ivi" if self.__out.endswith(".ivi"): self.__out = self.__out.rstrip(".ivi") if os.path.exists(self.__out + ".ivi"): os.remove(self.__out + ".ivi") tmp_dir_name = self.__out + ".tmp" if not os.path.exists(tmp_dir_name): os.mkdir(tmp_dir_name) def createEdge(edge_id: int): """ 创建节点信息到 edges_info """ edges_info[edge_id] = { "title": None, "cid": None, "button": None, "condition": None, "jump_type": None, "is_default": None, "command": None, "sub": [], } def var2dict(var: InteractiveVariable): return { "name": var.get_name(), "id": var.get_id(), "value": var.get_value(), "show": var.is_show(), "random": var.is_random(), } # 存储顶点信息 edges_info = {} # 使用队列来遍历剧情图，初始为 None 是为了从初始顶点开始 queue: List[InteractiveNode] = [ await (await self.__video.get_graph()).get_root_node() ] # 设置初始顶点 n = await (await self.__video.get_graph()).get_root_node() if n.get_node_id() not in edges_info: createEdge(n.get_node_id()) edges_info[n.get_node_id()]["cid"] = n.get_cid() edges_info[n.get_node_id()]["button"] = { "text": n.get_self_button().get_text(), "align": n.get_self_button().get_align(), "pos": (n.get_self_button().get_pos()), } edges_info[n.get_node_id()]["vars"] = [ var2dict(var) for var in (await n.get_vars()) ] edges_info[n.get_node_id()]["condition"] = (n.get_jumping_condition()._InteractiveJumpingCondition__command,) # type: ignore edges_info[n.get_node_id()]["jump_type"] = 0 edges_info[n.get_node_id()]["is_default"] = True edges_info[n.get_node_id()]["command"] = n._InteractiveNode__command._InteractiveJumpingCommand__command # type: ignore while queue: # 出队 now_node = queue.pop() if ( now_node.get_node_id() in edges_info and edges_info[now_node.get_node_id()]["title"] is not None and edges_info[now_node.get_node_id()]["cid"] is not None ): # 该情况为已获取到所有信息，说明是跳转到之前已处理的顶点，不作处理 continue # 获取顶点信息，最大重试 3 次 retry = 3 while True: try: node = await now_node.get_info() subs = await now_node.get_children() self.dispatch( "GET", {"title": node["title"], "node_id": now_node.get_node_id()}, ) break except Exception as e: retry -= 1 if retry < 0: raise e # 检查节顶点是否在 edges_info 中，本次步骤得到 title 信息 if node["edge_id"] not in edges_info: # 不在，新建 createEdge(node["edge_id"]) # 设置 title edges_info[node["edge_id"]]["title"] = node["title"] # 无可达顶点，即不能再往下走了，类似树的叶子节点 if "questions" not in node["edges"]: continue # 遍历所有可达顶点 for n in subs: # 该步骤获取顶点的 cid（视频分 P 的 ID） if n.get_node_id() not in edges_info: createEdge(n.get_node_id()) edges_info[n.get_node_id()]["cid"] = n.get_cid() edges_info[n.get_node_id()]["button"] = { "text": n.get_self_button().get_text(), "align": n.get_self_button().get_align(), "pos": n.get_self_button().get_pos(), } def var2dict(var: InteractiveVariable): return { "name": var.get_name(), "id": var.get_id(), "value": var.get_value(), "show": var.is_show(), "random": var.is_random(), } edges_info[n.get_node_id()]["condition"] = n.get_jumping_condition()._InteractiveJumpingCondition__command # type: ignore edges_info[n.get_node_id()][ "jump_type" ] = await now_node.get_jumping_type() edges_info[n.get_node_id()]["is_default"] = n.is_default() edges_info[n.get_node_id()]["command"] = n._InteractiveNode__command._InteractiveJumpingCommand__command # type: ignore edges_info[now_node.get_node_id()]["sub"] = [ n.get_node_id() for n in subs ] # 所有可达顶点 ID 入队 queue.insert(0, n) json.dump( edges_info, open(tmp_dir_name + "/ivideo.json", "w+", encoding="utf-8"), indent=2, ) json.dump( { "bvid": self.__video.get_bvid(), "title": (await self.__video.get_info())["title"], }, open(tmp_dir_name + "/bilivideo.json", "w+", encoding="utf-8"), indent=2, ) cid_set = set() for key, item in edges_info.items(): cid = item["cid"] if not cid in cid_set: self.dispatch("PREPARE_DOWNLOAD", {"cid": item["cid"]}) cid_set.add(cid) url = await self.__video.get_download_url(cid=cid, html5=True) await self.__download_func( url["durl"][0]["url"], tmp_dir_name + "/" + str(cid) + ".mp4", ) # type: ignore root_cid = await self.__video.get_cid() if not root_cid in cid_set: self.dispatch("PREPARE_DOWNLOAD", {"cid": root_cid}) cid = await self.__video.get_cid() url = await self.__video.get_download_url(cid=cid, html5=True) title = (await self.__video.get_info())["title"] await self.__download_func( url["durl"][0]["url"], tmp_dir_name + "/" + str(cid) + ".mp4" ) # type: ignore self.dispatch("PACKAGING") zip = zipfile.ZipFile( open(self.__out + ".ivi", "wb+"), mode="w", compression=zipfile.ZIP_DEFLATED ) # outFullName为压缩文件的完整路径 for path, dirnames, filenames in os.walk(tmp_dir_name): # 去掉目标跟路径，只对目标文件夹下边的文件及文件夹进行压缩 fpath = path.replace(tmp_dir_name, "") for filename in filenames: zip.write(os.path.join(path, filename), os.path.join(fpath, filename)) zip.close() shutil.rmtree(tmp_dir_name) self.dispatch("SUCCESS") async def __node_videos_main(self) -> None: # 初始化 self.dispatch("START") tmp_dir_name = self.__out if not os.path.exists(tmp_dir_name): os.mkdir(tmp_dir_name) def createEdge(edge_id: int): """ 创建节点信息到 edges_info """ edges_info[edge_id] = { "title": None, "cid": None, "button": None, "condition": None, "jump_type": None, "is_default": None, "command": None, "sub": [], } def var2dict(var: InteractiveVariable): return { "name": var.get_name(), "id": var.get_id(), "value": var.get_value(), "show": var.is_show(), "random": var.is_random(), } # 存储顶点信息 edges_info = {} # 使用队列来遍历剧情图，初始为 None 是为了从初始顶点开始 queue: List[InteractiveNode] = [ await (await self.__video.get_graph()).get_root_node() ] # 设置初始顶点 n = await (await self.__video.get_graph()).get_root_node() if n.get_node_id() not in edges_info: createEdge(n.get_node_id()) edges_info[n.get_node_id()]["cid"] = n.get_cid() edges_info[n.get_node_id()]["button"] = { "text": n.get_self_button().get_text(), "align": n.get_self_button().get_align(), "pos": (n.get_self_button().get_pos()), } edges_info[n.get_node_id()]["vars"] = [ var2dict(var) for var in (await n.get_vars()) ] edges_info[n.get_node_id()]["condition"] = (n.get_jumping_condition()._InteractiveJumpingCondition__command,) # type: ignore edges_info[n.get_node_id()]["jump_type"] = 0 edges_info[n.get_node_id()]["is_default"] = True edges_info[n.get_node_id()]["command"] = n._InteractiveNode__command._InteractiveJumpingCommand__command # type: ignore while queue: # 出队 now_node = queue.pop() if ( now_node.get_node_id() in edges_info and edges_info[now_node.get_node_id()]["title"] is not None and edges_info[now_node.get_node_id()]["cid"] is not None ): # 该情况为已获取到所有信息，说明是跳转到之前已处理的顶点，不作处理 continue # 获取顶点信息，最大重试 3 次 retry = 3 while True: try: node = await now_node.get_info() subs = await now_node.get_children() self.dispatch( "GET", {"title": node["title"], "node_id": now_node.get_node_id()}, ) break except Exception as e: retry -= 1 if retry < 0: raise e # 检查节顶点是否在 edges_info 中，本次步骤得到 title 信息 if node["edge_id"] not in edges_info: # 不在，新建 createEdge(node["edge_id"]) # 设置 title edges_info[node["edge_id"]]["title"] = node["title"] # 无可达顶点，即不能再往下走了，类似树的叶子节点 if "questions" not in node["edges"]: continue # 遍历所有可达顶点 for n in subs: # 该步骤获取顶点的 cid（视频分 P 的 ID） if n.get_node_id() not in edges_info: createEdge(n.get_node_id()) edges_info[n.get_node_id()]["cid"] = n.get_cid() edges_info[n.get_node_id()]["button"] = { "text": n.get_self_button().get_text(), "align": n.get_self_button().get_align(), "pos": n.get_self_button().get_pos(), } def var2dict(var: InteractiveVariable): return { "name": var.get_name(), "id": var.get_id(), "value": var.get_value(), "show": var.is_show(), "random": var.is_random(), } edges_info[n.get_node_id()]["condition"] = n.get_jumping_condition()._InteractiveJumpingCondition__command # type: ignore edges_info[n.get_node_id()][ "jump_type" ] = await now_node.get_jumping_type() edges_info[n.get_node_id()]["is_default"] = n.is_default() edges_info[n.get_node_id()]["command"] = n._InteractiveNode__command._InteractiveJumpingCommand__command # type: ignore edges_info[now_node.get_node_id()]["sub"] = [ n.get_node_id() for n in subs ] # 所有可达顶点 ID 入队 queue.insert(0, n) cid_set = set() for key, item in edges_info.items(): cid = item["cid"] if not cid in cid_set: self.dispatch("PREPARE_DOWNLOAD", {"cid": item["cid"]}) cid_set.add(cid) url = await self.__video.get_download_url(cid=cid, html5=True) await self.__download_func( url["durl"][0]["url"], tmp_dir_name + "/" + str(key) + " " + item["title"] + ".mp4", ) # type: ignore root_cid = await self.__video.get_cid() if not root_cid in cid_set: self.dispatch("PREPARE_DOWNLOAD", {"cid": root_cid}) cid = await self.__video.get_cid() url = await self.__video.get_download_url(cid=cid, html5=True) title = (await self.__video.get_info())["title"] await self.__download_func( url["durl"][0]["url"], tmp_dir_name + "/1 " + title + ".mp4" ) # type: ignore self.dispatch("SUCCESS") async def __dot_graph_main(self) -> None: self.dispatch("START") if not self.__out.endswith(".dot"): self.__out += ".dot" class node_info: node_id: int subs: List[int] cid: int title: str def __eq__(self, info: "node_info"): self.subs.sort() info.subs.sort() return ( (info.subs == self.subs) and (info.title == self.title) and (info.cid == self.cid) ) def __lt__(self, info: "node_info"): return self.cid < info.cid def __gt__(self, info: "node_info"): return self.cid > info.cid fetched_nodes_info: List[node_info] = [] node_info_dict = {} scripts = [] graph = await self.__video.get_graph() queue: List[InteractiveNode] = [await graph.get_root_node()] while queue: queue_backup = copy.copy(queue) queue = [] for cur_node in queue_backup: cur_node_info = await cur_node.get_info() cur_node_children = await cur_node.get_children() self.dispatch( "GET", { "title": cur_node_info["title"], "node_id": cur_node.get_node_id(), }, ) cur_node_info_class = node_info() cur_node_info_class.node_id = cur_node.get_node_id() cur_node_info_class.cid = cur_node.get_cid() cur_node_info_class.subs = [n.get_node_id() for n in cur_node_children] cur_node_info_class.title = cur_node_info["title"] back_to_pre = False back_to_node_title = -1 for fetched_info in fetched_nodes_info: if fetched_info == cur_node_info_class: back_to_pre = True back_to_node_title = fetched_info.title if not back_to_pre: node_info_dict[cur_node.get_node_id()] = cur_node_info_class for cur_node_child in cur_node_children: script_label = "" if cur_node_child.get_jumping_condition()._InteractiveJumpingCondition__command != "": # type: ignore script_label = script_label + "Condition: [" + cur_node_child.get_jumping_condition()._InteractiveJumpingCondition__command + "]" # type: ignore if cur_node_child._InteractiveNode__command._InteractiveJumpingCommand__command != "": # type: ignore script_label = script_label + "\nNative Command: [" + cur_node_child._InteractiveNode__command._InteractiveJumpingCommand__command + "]" # type: ignore elif cur_node_child._InteractiveNode__command._InteractiveJumpingCommand__command != "": # type: ignore script_label = script_label + "\nNative Command: [" + cur_node_child._InteractiveNode__command._InteractiveJumpingCommand__command + "]" # type: ignore scripts.append( { "from": cur_node.get_node_id(), "to": cur_node_child.get_node_id(), "label": script_label, } ) queue.append(cur_node_child) fetched_nodes_info.append(cur_node_info_class) else: node_info_dict[cur_node.get_node_id()] = f"跳转至 {back_to_node_title}" graph_content = "digraph {\nfontname=FangSong\nnode [fontname=FangSong]\n" for script in scripts: graph_content += f'\t{script["from"]} -> {script["to"]}' if script["label"] != "": graph_content += f' [label="{script["label"]}"]\n' else: graph_content += "\n" for node_info_key, node_info_item in node_info_dict.items(): if isinstance(node_info_item, node_info): graph_content += f'\t{node_info_key} [label="{node_info_item.title}"]\n' else: graph_content += f'\t{node_info_key} [label="{node_info_item}"]\n' vars_string = "Variables: " for var in await (await graph.get_root_node()).get_vars(): var_attribute = "" if var.is_random(): var_attribute = "Random" else: if var.is_show(): var_attribute = "Normal" else: var_attribute = "Hide" vars_string += f"[{var.get_id()} -> {var.get_name()} = {var.get_value()}, {var_attribute}]\n" graph_content += f'\tlabel="{vars_string}"' graph_content += "}" with open(self.__out, "w+", encoding="utf-8") as dot_file: dot_file.write(graph_content) self.dispatch("SUCCESS") async def __no_packaging_main(self) -> None: # 初始化 self.dispatch("START") tmp_dir_name = self.__out if not os.path.exists(tmp_dir_name): os.mkdir(tmp_dir_name) def createEdge(edge_id: int): """ 创建节点信息到 edges_info """ edges_info[edge_id] = { "title": None, "cid": None, "button": None, "condition": None, "jump_type": None, "is_default": None, "command": None, "sub": [], } def var2dict(var: InteractiveVariable): return { "name": var.get_name(), "id": var.get_id(), "value": var.get_value(), "show": var.is_show(), "random": var.is_random(), } # 存储顶点信息 edges_info = {} # 使用队列来遍历剧情图，初始为 None 是为了从初始顶点开始 queue: List[InteractiveNode] = [ await (await self.__video.get_graph()).get_root_node() ] # 设置初始顶点 n = await (await self.__video.get_graph()).get_root_node() if n.get_node_id() not in edges_info: createEdge(n.get_node_id()) edges_info[n.get_node_id()]["cid"] = n.get_cid() edges_info[n.get_node_id()]["button"] = { "text": n.get_self_button().get_text(), "align": n.get_self_button().get_align(), "pos": (n.get_self_button().get_pos()), } edges_info[n.get_node_id()]["vars"] = [ var2dict(var) for var in (await n.get_vars()) ] edges_info[n.get_node_id()]["condition"] = (n.get_jumping_condition()._InteractiveJumpingCondition__command,) # type: ignore edges_info[n.get_node_id()]["jump_type"] = 0 edges_info[n.get_node_id()]["is_default"] = True edges_info[n.get_node_id()]["command"] = n._InteractiveNode__command._InteractiveJumpingCommand__command # type: ignore while queue: # 出队 now_node = queue.pop() if ( now_node.get_node_id() in edges_info and edges_info[now_node.get_node_id()]["title"] is not None and edges_info[now_node.get_node_id()]["cid"] is not None ): # 该情况为已获取到所有信息，说明是跳转到之前已处理的顶点，不作处理 continue # 获取顶点信息，最大重试 3 次 retry = 3 while True: try: node = await now_node.get_info() subs = await now_node.get_children() self.dispatch( "GET", {"title": node["title"], "node_id": now_node.get_node_id()}, ) break except Exception as e: retry -= 1 if retry < 0: raise e # 检查节顶点是否在 edges_info 中，本次步骤得到 title 信息 if node["edge_id"] not in edges_info: # 不在，新建 createEdge(node["edge_id"]) # 设置 title edges_info[node["edge_id"]]["title"] = node["title"] # 无可达顶点，即不能再往下走了，类似树的叶子节点 if "questions" not in node["edges"]: continue # 遍历所有可达顶点 for n in subs: # 该步骤获取顶点的 cid（视频分 P 的 ID） if n.get_node_id() not in edges_info: createEdge(n.get_node_id()) edges_info[n.get_node_id()]["cid"] = n.get_cid() edges_info[n.get_node_id()]["button"] = { "text": n.get_self_button().get_text(), "align": n.get_self_button().get_align(), "pos": n.get_self_button().get_pos(), } def var2dict(var: InteractiveVariable): return { "name": var.get_name(), "id": var.get_id(), "value": var.get_value(), "show": var.is_show(), "random": var.is_random(), } edges_info[n.get_node_id()]["condition"] = n.get_jumping_condition()._InteractiveJumpingCondition__command # type: ignore edges_info[n.get_node_id()][ "jump_type" ] = await now_node.get_jumping_type() edges_info[n.get_node_id()]["is_default"] = n.is_default() edges_info[n.get_node_id()]["command"] = n._InteractiveNode__command._InteractiveJumpingCommand__command # type: ignore edges_info[now_node.get_node_id()]["sub"] = [ n.get_node_id() for n in subs ] # 所有可达顶点 ID 入队 queue.insert(0, n) json.dump( edges_info, open(tmp_dir_name + "/ivideo.json", "w+", encoding="utf-8"), indent=2, ) json.dump( { "bvid": self.__video.get_bvid(), "title": (await self.__video.get_info())["title"], }, open(tmp_dir_name + "/bilivideo.json", "w+", encoding="utf-8"), indent=2, ) cid_set = set() for key, item in edges_info.items(): cid = item["cid"] if not cid in cid_set: self.dispatch("PREPARE_DOWNLOAD", {"cid": item["cid"]}) cid_set.add(cid) url = await self.__video.get_download_url(cid=cid, html5=True) await self.__download_func( url["durl"][0]["url"], tmp_dir_name + "/" + str(key) + " " + item["title"] + ".mp4", ) # type: ignore root_cid = await self.__video.get_cid() if not root_cid in cid_set: self.dispatch("PREPARE_DOWNLOAD", {"cid": root_cid}) cid = await self.__video.get_cid() url = await self.__video.get_download_url(cid=cid, html5=True) title = (await self.__video.get_info())["title"] await self.__download_func( url["durl"][0]["url"], tmp_dir_name + "/1 " + title + ".mp4" ) # type: ignore self.dispatch("SUCCESS") async def start(self) -> None: """ 开始下载 """ if self.__mode.value == "ivi": task = create_task(self.__main()) elif self.__mode.value == "dot": task = create_task(self.__dot_graph_main()) elif self.__mode.value == "no_pack": task = create_task(self.__no_packaging_main()) else: task = create_task(self.__node_videos_main()) self.__task = task try: result = await task self.__task = None return result except CancelledError: # 忽略 task 取消异常 pass except Exception as e: self.dispatch("FAILED", {"err": e}) raise e async def abort(self) -> None: """ 中断下载 """ if self.__task: self.__task.cancel("用户手动取消") self.dispatch("ABORTED", None) def get_ivi_file_meta(path: str) -> dict: """ 获取 ivi 文件信息 Args: path (str): 文件地址 Returns: dict: 文件信息 """ ivi = zipfile.ZipFile(open(path, "rb")) info = ivi.open("bilivideo.json").read() return json.loads(info)