🐛 Fix post get official recommended posts

This commit is contained in:
xtaodada 2024-07-18 22:40:42 +08:00
parent c362342b5a
commit f6174f64c9
Signed by: xtaodada
GPG Key ID: 4CBB3F4FA8C85659
3 changed files with 47 additions and 29 deletions

View File

@ -1,4 +1,5 @@
from typing import List
import asyncio
from typing import List, Dict
from .hyperion import HyperionBase
from ..base.hyperionrequest import HyperionRequest
@ -9,7 +10,7 @@ __all__ = ("Hoyolab",)
class Hoyolab(HyperionBase):
POST_FULL_URL = "https://bbs-api-os.hoyolab.com/community/post/wapi/getPostFull"
NEW_LIST_URL = "https://bbs-api-os.hoyolab.com/community/post/wapi/getNewsList"
GET_NEW_LIST_URL = "https://bbs-api-os.hoyolab.com/community/post/wapi/getNewsList"
NEW_BG_URL = "https://bbs-api-os.hoyolab.com/community/painter/wapi/circle/info"
LANG = "zh-cn"
USER_AGENT = (
@ -30,17 +31,12 @@ class Hoyolab(HyperionBase):
async def get_official_recommended_posts(
self, gids: int, page_size: int = 3, type_: int = 1
) -> List[PostRecommend]:
params = {"gids": gids, "page_size": page_size, "type": type_}
response = await self.client.get(url=self.NEW_LIST_URL, params=params)
return [
PostRecommend(
hoyolab=True,
post_id=data["post"]["post_id"],
subject=data["post"]["subject"],
multi_language_info=HoYoPostMultiLang(**data["post"]["multi_language_info"]),
)
for data in response["list"]
]
results = []
tasks = [self.get_new_list_recommended_posts(gids, i, 5) for i in range(1, 4)]
asyncio_results = await asyncio.gather(*tasks)
for result in asyncio_results:
results.extend(result)
return results
async def get_post_info(self, gids: int, post_id: int, read: int = 1, scene: int = 1, lang: str = LANG) -> PostInfo:
params = {"post_id": post_id, "read": read, "scene": scene}
@ -58,6 +54,15 @@ class Hoyolab(HyperionBase):
async def _download_image(self, art_id: int, url: str, page: int = 0) -> List[ArtworkImage]:
return await self.download_image(self.client, art_id, url, page)
async def get_new_list(self, gids: int, type_id: int, page_size: int = 20) -> Dict:
params = {"gids": gids, "page_size": page_size, "type": type_id}
return await self.client.get(url=self.GET_NEW_LIST_URL, params=params)
async def get_new_list_recommended_posts(self, gids: int, type_id: int, page_size: int = 20) -> List[PostRecommend]:
resp = await self.get_new_list(gids, type_id, page_size)
data = resp["list"]
return [PostRecommend.parse(i) for i in data]
async def close(self):
await self.client.shutdown()

View File

@ -3,7 +3,7 @@ import os
import re
from abc import abstractmethod
from time import time
from typing import List, Tuple
from typing import List, Tuple, Dict
from ..base.hyperionrequest import HyperionRequest
from ...models.genshin.hyperion import (
@ -106,6 +106,14 @@ class HyperionBase:
art_id=art_id, page=page, file_name=filename, file_extension=url.split(".")[-1], data=response.content
)
@abstractmethod
async def get_new_list(self, gids: int, type_id: int, page_size: int = 20) -> Dict:
"""获取最新帖子"""
@abstractmethod
async def get_new_list_recommended_posts(self, gids: int, type_id: int, page_size: int = 20) -> List[PostRecommend]:
"""获取最新帖子"""
@abstractmethod
async def get_official_recommended_posts(self, gids: int) -> List[PostRecommend]:
"""获取官方推荐帖子"""
@ -149,9 +157,12 @@ class Hyperion(HyperionBase):
return {"User-Agent": self.USER_AGENT, "Referer": referer}
async def get_official_recommended_posts(self, gids: int) -> List[PostRecommend]:
params = {"gids": gids}
response = await self.client.get(url=self.GET_OFFICIAL_RECOMMENDED_POSTS_URL, params=params)
return [PostRecommend(**data) for data in response["list"]]
results = []
tasks = [self.get_new_list_recommended_posts(gids, i, 5) for i in range(1, 4)]
asyncio_results = await asyncio.gather(*tasks)
for result in asyncio_results:
results.extend(result)
return results
async def get_post_full_in_collection(self, collection_id: int, gids: int = 2, order_type=1) -> JSON_DATA:
params = {"collection_id": collection_id, "gids": gids, "order_type": order_type}
@ -174,14 +185,14 @@ class Hyperion(HyperionBase):
async def _download_image(self, art_id: int, url: str, page: int = 0) -> List[ArtworkImage]:
return await self.download_image(self.client, art_id, url, page)
async def get_new_list(self, gids: int, type_id: int, page_size: int = 20):
"""
?gids=2&page_size=20&type=3
:return:
"""
async def get_new_list(self, gids: int, type_id: int, page_size: int = 20) -> Dict:
params = {"gids": gids, "page_size": page_size, "type": type_id}
response = await self.client.get(url=self.GET_NEW_LIST_URL, params=params)
return response
return await self.client.get(url=self.GET_NEW_LIST_URL, params=params)
async def get_new_list_recommended_posts(self, gids: int, type_id: int, page_size: int = 20) -> List[PostRecommend]:
resp = await self.get_new_list(gids, type_id, page_size)
data = resp["list"]
return [PostRecommend.parse(i) for i in data]
async def get_live_info(self, act_id: str) -> LiveInfo:
headers = {"x-rpc-act_id": act_id}

View File

@ -1,7 +1,7 @@
from datetime import datetime, timedelta
from enum import Enum
from io import BytesIO
from typing import Any, List, Optional
from typing import Any, List, Optional, Dict
from PIL import Image, UnidentifiedImageError
from pydantic import BaseModel, PrivateAttr
@ -167,7 +167,9 @@ class HoYoPostMultiLang(BaseModel):
class PostRecommend(BaseModel):
hoyolab: bool = False
post_id: int
subject: str
banner: Optional[str] = None
official_type: Optional[int] = None
multi_language_info: Optional[HoYoPostMultiLang] = None
@staticmethod
def parse(data: Dict, hoyolab: bool = False):
_post = data.get("post")
post_id = _post.get("post_id")
return PostRecommend(hoyolab=hoyolab, post_id=post_id)