🐛 修复获取小说文字接口并新增几条 API @asadahimeka (#399)

* 修复获取小说文字接口并新增几条 API * fix lint errors * fix some glitches * fix: remove webview_novel `raw` param * fix: remove `raw` param * Add overload for request method in PixivEndpoints --------- Co-authored-by: Mix <[email protected]>
mixmoe · Mar 9, 2024 · 0c91dd9 · 0c91dd9
1 parent 468c98c
commit 0c91dd9
Show file tree

Hide file tree

Showing 2 changed files with 134 additions and 4 deletions.
diff --git 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<data>{.+?}),\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
@@ -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