""" bilibili_api.manga 漫画相关操作 """ import datetime from enum import Enum from urllib.parse import urlparse from typing import Dict, List, Union, Optional import httpx from bilibili_api.utils.utils import get_api from bilibili_api.errors import ArgsException from bilibili_api.utils.picture import Picture from bilibili_api.utils.credential import Credential from bilibili_api.utils.network import HEADERS, Api API = get_api("manga") class MangaIndexFilter: """ 漫画索引筛选器类。 """ class Area(Enum): """ 漫画索引筛选器的地区枚举类。 - ALL: 全部 - CHINA: 大陆 - JAPAN: 日本 - SOUTHKOREA: 韩国 - OTHER: 其他 """ ALL = -1 CHINA = 1 JAPAN = 2 SOUTHKOREA = 6 OTHER = 5 class Order(Enum): """ 漫画索引筛选器的排序枚举类。 - HOT: 人气推荐 - UPDATE: 更新时间 - RELEASE_DATE: 上架时间 """ HOT = 0 UPDATE = 1 RELEASE_DATE = 3 class Status(Enum): """ 漫画索引筛选器的状态枚举类。 - ALL: 全部 - FINISHED: 完结 - UNFINISHED: 连载 """ ALL = -1 FINISHED = 1 UNFINISHED = 0 class Payment(Enum): """ 漫画索引筛选器的付费枚举类。 - ALL: 全部 - FREE: 免费 - PAID: 付费 - WILL_BE_FREE: 等就免费 """ ALL = -1 FREE = 1 PAID = 2 WILL_BE_FREE = 3 class Style(Enum): """ 漫画索引筛选器的风格枚举类。 - ALL: 全部 - WARM: 热血 - ANCIENT: 古风 - FANTASY: 玄幻 - IMAGING: 奇幻 - SUSPENSE: 悬疑 - CITY: 都市 - HISTORY: 历史 - WUXIA: 武侠仙侠 - GAME: 游戏竞技 - PARANORMAL: 悬疑灵异 - ALTERNATE: 架空 - YOUTH: 青春 - WEST_MAGIC: 西幻 - MORDEN: 现代 - POSITIVE: 正能量 - SCIENCE_FICTION: 科幻 """ ALL = -1 WARM = 999 ANCIENT = 997 FANTASY = 1016 IMAGING = 998 SUSPENSE = 1023 CITY = 1002 HISTORY = 1096 WUXIA = 1092 GAME = 1088 PARANORMAL = 1081 ALTERNATE = 1063 YOUTH = 1060 WEST_MAGIC = 1054 MORDEN = 1048 POSITIVE = 1028 SCIENCE_FICTION = 1027 class Manga: """ 漫画类 Attributes: credential (Credential): 凭据类。 """ def __init__(self, manga_id: int, credential: Optional[Credential] = None): """ Args: manga_id (int) : 漫画 id credential (Credential | None): 凭据类. Defaults to None. """ credential = credential if credential else Credential() self.__manga_id = manga_id self.credential = credential self.__info: Optional[Dict] = None def get_manga_id(self) -> int: return self.__manga_id async def get_info(self) -> dict: """ 获取漫画信息 Returns: dict: 调用 API 返回的结果 """ api = API["info"]["detail"] params = {"comic_id": self.__manga_id} return ( await Api( **api, credential=self.credential, no_csrf=( False if ( self.credential.has_sessdata() and self.credential.has_bili_jct() ) else True ), ) .update_params(**params) .result ) async def __get_info_cached(self) -> dict: """ 获取漫画信息,如果有缓存则使用缓存。 """ if self.__info == None: self.__info = await self.get_info() return self.__info async def get_episode_info( self, episode_count: Optional[Union[int, float]] = None, episode_id: Optional[int] = None, ) -> dict: """ 获取某一话的详细信息 Args: episode_count (int | float | None): 第几话. episode_id (int | None) : 对应的话的 id. 可以通过 `get_episode_id` 获取。 **注意:episode_count 和 episode_id 中必须提供一个参数。** Returns: dict: 对应的话的详细信息 """ info = await self.__get_info_cached() for ep in info["ep_list"]: if episode_count == None: if ep["id"] == episode_id: return ep elif episode_id == None: if ep["ord"] == episode_count: return ep else: raise ArgsException("episode_count 和 episode_id 中必须提供一个参数。") raise ArgsException("未找到对应的话") async def get_episode_id( self, episode_count: Optional[Union[int, float]] = None ) -> int: """ 获取某一话的 id Args: episode_count (int | float | None): 第几话. Returns: int: 对应的话的 id """ return (await self.get_episode_info(episode_count=episode_count))["id"] async def get_images_url( self, episode_count: Optional[Union[int, float]] = None, episode_id: Optional[int] = None, ) -> dict: """ 获取某一话的图片链接。(未经过处理,所有的链接无法直接访问) 获取的图片 url 请传入 `manga.manga_image_url_turn_to_Picture` 函数以转换为 `Picture` 类。 Args: episode_count (int | float | None): 第几话. episode_id (int | None) : 对应的话的 id. 可以通过 `get_episode_id` 获取。 **注意:episode_count 和 episode_id 中必须提供一个参数。** Returns: dict: 调用 API 返回的结果 """ if episode_id == None: if episode_count == None: raise ArgsException("episode_count 和 episode_id 中必须提供一个参数。") episode_id = await self.get_episode_id(episode_count) api = API["info"]["episode_images"] params = {"ep_id": episode_id} return ( await Api( **api, credential=self.credential, no_csrf=( False if ( self.credential.has_sessdata() and self.credential.has_bili_jct() ) else True ), ) .update_params(**params) .result ) async def get_images( self, episode_count: Optional[Union[int, float]] = None, episode_id: Optional[int] = None, ) -> List[Dict]: """ 获取某一话的所有图片 # 注意事项:此函数速度非常慢并且失败率高 Args: episode_count (int | float | None): 第几话. episode_id (int | None) : 对应的话的 id. 可以通过 `get_episode_id` 获取。 **注意:episode_count 和 episode_id 中必须提供一个参数。** Returns: List[Picture]: 所有的图片 """ data = await self.get_images_url( episode_count=episode_count, episode_id=episode_id ) pictures: List[Dict] = [] async def get_real_image_url(url: str) -> str: token_api = API["info"]["image_token"] datas = {"urls": f'["{url}"]'} token_data = ( await Api( **token_api, credential=self.credential, no_csrf=( False if ( self.credential.has_sessdata() and self.credential.has_bili_jct() ) else True ), ) .update_data(**datas) .result ) return token_data[0]["url"] + "?token=" + token_data[0]["token"] for img in data["images"]: url = await get_real_image_url(img["path"]) pictures.append( { "x": img["x"], "y": img["y"], "picture": Picture.from_content( (await httpx.AsyncClient().get(url, headers=HEADERS)).content, "jpg", ), } ) return pictures async def manga_image_url_turn_to_Picture( url: str, credential: Optional[Credential] = None ) -> Picture: """ 将 Manga.get_images_url 函数获得的图片 url 转换为 Picture 类。 Args: url (str) : 未经处理的漫画图片链接。 credential (Credential | None): 凭据类. Defaults to None. Returns: Picture: 图片类。 """ url = urlparse(url).path credential = credential if credential else Credential() async def get_real_image_url(url: str) -> str: token_api = API["info"]["image_token"] datas = {"urls": f'["{url}"]'} token_data = ( await Api( **token_api, credential=credential, no_csrf=( False if (credential.has_sessdata() and credential.has_bili_jct()) else True ), ) .update_data(**datas) .result ) return f'{token_data[0]["url"]}?token={token_data[0]["token"]}' url = await get_real_image_url(url) return await Picture.async_load_url(url) async def set_follow_manga( manga: Manga, status: bool = True, credential: Optional[Credential] = None ) -> dict: """ 设置追漫 Args: manga (Manga) : 漫画类。 status (bool) : 设置是否追漫。是为 True,否为 False。Defaults to True. credential (Credential): 凭据类。 """ if credential == None: if manga.credential.has_sessdata() and manga.credential.has_bili_jct(): credential = manga.credential else: credential = Credential() credential.raise_for_no_sessdata() credential.raise_for_no_bili_jct() if status == True: api = API["operate"]["add_favorite"] else: api = API["operate"]["del_favorite"] data = {"comic_ids": str(manga.get_manga_id())} return await Api(**api, credential=credential).update_data(**data).result async def get_raw_manga_index( area: MangaIndexFilter.Area = MangaIndexFilter.Area.ALL, order: MangaIndexFilter.Order = MangaIndexFilter.Order.HOT, status: MangaIndexFilter.Status = MangaIndexFilter.Status.ALL, payment: MangaIndexFilter.Payment = MangaIndexFilter.Payment.ALL, style: MangaIndexFilter.Style = MangaIndexFilter.Style.ALL, pn: int = 1, ps: int = 18, credential: Credential = None, ) -> list: """ 获取漫画索引 Args: area (MangaIndexFilter.Area) : 地区。Defaults to MangaIndexFilter.Area.ALL. order (MangaIndexFilter.Order) : 排序。Defaults to MangaIndexFilter.Order.HOT. status (MangaIndexFilter.Status) : 状态。Defaults to MangaIndexFilter.Status.ALL. payment (MangaIndexFilter.Payment): 支付。Defaults to MangaIndexFilter.Payment.ALL. style (MangaIndexFilter.Style) : 风格。Defaults to MangaIndexFilter.Style.ALL. pn (int) : 页码。Defaults to 1. ps (int) : 每页数量。Defaults to 18. credential (Credential) : 凭据类. Defaults to None. Returns: list: 调用 API 返回的结果 """ credential = credential if credential else Credential() api = API["info"]["index"] params = {"device": "pc", "platform": "web"} data = { "area_id": area.value, "order": order.value, "is_finish": status.value, "is_free": payment.value, "style_id": style.value, "page_num": pn, "page_size": ps, } return ( await Api(**api, credential=credential, no_csrf=True) .update_data(**data) .update_params(**params) .result ) async def get_manga_index( area: MangaIndexFilter.Area = MangaIndexFilter.Area.ALL, order: MangaIndexFilter.Order = MangaIndexFilter.Order.HOT, status: MangaIndexFilter.Status = MangaIndexFilter.Status.ALL, payment: MangaIndexFilter.Payment = MangaIndexFilter.Payment.ALL, style: MangaIndexFilter.Style = MangaIndexFilter.Style.ALL, pn: int = 1, ps: int = 18, credential: Credential = None, ) -> List[Manga]: """ 获取漫画索引 Args: area (MangaIndexFilter.Area) : 地区。Defaults to MangaIndexFilter.Area.ALL. order (MangaIndexFilter.Order) : 排序。Defaults to MangaIndexFilter.Order.HOT. status (MangaIndexFilter.Status) : 状态。Defaults to MangaIndexFilter.Status.ALL. payment (MangaIndexFilter.Payment): 支付。Defaults to MangaIndexFilter.Payment.ALL. style (MangaIndexFilter.Style) : 风格。Defaults to MangaIndexFilter.Style.ALL. pn (int) : 页码。Defaults to 1. ps (int) : 每页数量。Defaults to 18. credential (Credential) : 凭据类. Defaults to None. Returns: List[Manga]: 漫画索引 """ data = await get_raw_manga_index( area, order, status, payment, style, pn, ps, credential ) return [Manga(manga_data["season_id"]) for manga_data in data] async def get_manga_update( date: Union[str, datetime.datetime] = datetime.datetime.now(), pn: int = 1, ps: int = 8, credential: Credential = None, ) -> List[Manga]: """ 获取更新推荐的漫画 Args: date (Union[str, datetime.datetime]): 日期,默认为今日。 pn (int) : 页码。Defaults to 1. ps (int) : 每页数量。Defaults to 8. credential (Credential) : 凭据类. Defaults to None. Returns: List[Manga]: 漫画列表 """ credential = credential if credential else Credential() api = API["info"]["update"] params = {"device": "pc", "platform": "web"} if isinstance(date, datetime.datetime): date = date.strftime("%Y-%m-%d") data = {"date": date, "page_num": pn, "page_size": ps} manga_data = ( await Api(**api, credential=credential, no_csrf=True) .update_data(**data) .update_params(**params) .result ) return [Manga(manga["comic_id"]) for manga in manga_data["list"]] async def get_manga_home_recommend( pn: int = 1, seed: Optional[str] = "0", credential: Credential = None ) -> List[Manga]: """ 获取首页推荐的漫画 Args: pn (int) : 页码。Defaults to 1. seed (Optional[str]) : Unknown param,无需传入. credential (Credential) : 凭据类. Defaults to None. Returns: List[Manga]: 漫画列表 """ credential = credential if credential else Credential() api = API["info"]["home_recommend"] params = {"device": "pc", "platform": "web"} data = {"page_num": pn, "seed": seed} manga_data = ( await Api(**api, credential=credential, no_csrf=True) .update_data(**data) .update_params(**params) .result ) return [Manga(manga["comic_id"]) for manga in manga_data["list"]]