diff --git a/hibiapi/api/pixiv/api.py b/hibiapi/api/pixiv/api.py index 5b15e221..c351f101 100644 --- a/hibiapi/api/pixiv/api.py +++ b/hibiapi/api/pixiv/api.py @@ -1,6 +1,8 @@ +import json +import re from datetime import date, timedelta from enum import Enum -from typing import Any, Dict, Optional, cast +from typing import Any, Dict, Literal, Optional, Union, cast, overload from hibiapi.api.pixiv.constants import PixivConstants from hibiapi.api.pixiv.net import NetRequest as PixivNetClient @@ -129,11 +131,33 @@ def _parse_accept_language(accept_language: str) -> str: language_code, *_ = first_language.partition(";") return language_code.lower().strip() + @overload + async def request( + self, + endpoint: str, + *, + params: Optional[Dict[str, Any]] = None, + return_text: Literal[False] = False, + ) -> Dict[str, Any]: ... + + @overload + async def request( + self, + endpoint: str, + *, + params: Optional[Dict[str, Any]] = None, + return_text: Literal[True], + ) -> str: ... + @dont_route @catch_network_error async def request( - self, endpoint: str, *, params: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: + self, + endpoint: str, + *, + params: Optional[Dict[str, Any]] = None, + return_text: bool = False, + ) -> Union[Dict[str, Any], str]: headers = self.client.headers.copy() net_client = cast(PixivNetClient, self.client.net_client) @@ -155,6 +179,8 @@ async def request( ), headers=headers, ) + if return_text: + return response.text return response.json() @cache_config(ttl=timedelta(days=3)) @@ -261,6 +287,7 @@ async def search( page: int = 1, size: int = 30, include_translated_tag_results: bool = True, + search_ai_type: bool = True, # 搜索结果是否包含AI作品 ): return await self.request( "v1/search/illust", @@ -271,6 +298,7 @@ async def search( "duration": duration, "offset": (page - 1) * size, "include_translated_tag_results": include_translated_tag_results, + "search_ai_type": 1 if search_ai_type else 0, }, ) @@ -466,8 +494,25 @@ async def novel_series(self, *, id: int): async def novel_detail(self, *, id: int): return await self.request("/v2/novel/detail", params={"novel_id": id}) + # 已被官方移除,调用 webview/v2/novel 作兼容处理 async def novel_text(self, *, id: int): - return await self.request("/v1/novel/text", params={"novel_id": id}) + # return await self.request("/v1/novel/text", params={"novel_id": id}) + response = await self.webview_novel(id=id) + return {"novel_text": response["text"] or ""} + + # 获取小说 HTML 后解析 JSON + async def webview_novel(self, *, id: int): + response = await self.request( + "webview/v2/novel", + params={ + "id": id, + "viewer_version": "20221031_ai", + }, + return_text=True, + ) + + novel_match = re.search(r"novel:\s+(?P{.+?}),\s+isOwnWork", response) + return json.loads(novel_match["data"] if novel_match else response) @cache_config(ttl=timedelta(hours=12)) async def tags_novel(self): @@ -484,6 +529,7 @@ async def search_novel( duration: Optional[SearchDurationType] = None, page: int = 1, size: int = 30, + search_ai_type: bool = True, # 搜索结果是否包含AI作品 ): return await self.request( "/v1/search/novel", @@ -495,6 +541,7 @@ async def search_novel( "include_translated_tag_results": include_translated_tag_results, "duration": duration, "offset": (page - 1) * size, + "search_ai_type": 1 if search_ai_type else 0, }, ) @@ -523,3 +570,44 @@ async def novel_new(self, *, max_novel_id: Optional[int] = None): return await self.request( "/v1/novel/new", params={"max_novel_id": max_novel_id} ) + + # 人气直播列表 + async def live_list(self, *, page: int = 1, size: int = 30): + params = {"list_type": "popular", "offset": (page - 1) * size} + if not params["offset"]: + del params["offset"] + return await self.request("v1/live/list", params=params) + + # 相关小说作品 + async def related_novel(self, *, id: int, page: int = 1, size: int = 30): + return await self.request( + "v1/novel/related", + params={ + "novel_id": id, + "offset": (page - 1) * size, + }, + ) + + # 相关用户 + async def related_member(self, *, id: int): + return await self.request("v1/user/related", params={"seed_user_id": id}) + + # 漫画系列 + async def illust_series(self, *, id: int, page: int = 1, size: int = 30): + return await self.request( + "v1/illust/series", + params={"illust_series_id": id, "offset": (page - 1) * size}, + ) + + # 用户的漫画系列 + async def member_illust_series(self, *, id: int, page: int = 1, size: int = 30): + return await self.request( + "v1/user/illust-series", + params={"user_id": id, "offset": (page - 1) * size}, + ) + + # 用户的小说系列 + async def member_novel_series(self, *, id: int, page: int = 1, size: int = 30): + return await self.request( + "v1/user/novel-series", params={"user_id": id, "offset": (page - 1) * size} + ) diff --git a/test/test_pixiv.py b/test/test_pixiv.py index 3716ab4d..e981e507 100644 --- a/test/test_pixiv.py +++ b/test/test_pixiv.py @@ -179,6 +179,48 @@ def test_novel_text(client: TestClient): assert response.json().get("novel_text") +def test_webview_novel(client: TestClient): + response = client.get("webview_novel", params={"id": 19791013}) + assert response.status_code == 200 + assert response.json().get("text") + + +def test_live_list(client: TestClient): + response = client.get("live_list") + assert response.status_code == 200 + assert response.json().get("lives") + + +def test_related_novel(client: TestClient): + response = client.get("related_novel", params={"id": 19791013}) + assert response.status_code == 200 + assert response.json().get("novels") + + +def test_related_member(client: TestClient): + response = client.get("related_member", params={"id": 10109777}) + assert response.status_code == 200 + assert response.json().get("user_previews") + + +def test_illust_series(client: TestClient): + response = client.get("illust_series", params={"id": 218893}) + assert response.status_code == 200 + assert response.json().get("illust_series_detail") + + +def test_member_illust_series(client: TestClient): + response = client.get("member_illust_series", params={"id": 4087934}) + assert response.status_code == 200 + assert response.json().get("illust_series_details") + + +def test_member_novel_series(client: TestClient): + response = client.get("member_novel_series", params={"id": 86832559}) + assert response.status_code == 200 + assert response.json().get("novel_series_details") + + def test_tags_novel(client: TestClient): response = client.get("tags_novel") assert response.status_code == 200