mirror of
https://github.com/PaiGramTeam/PamGram.git
synced 2024-11-22 14:26:45 +00:00
246 lines
9.5 KiB
Python
246 lines
9.5 KiB
Python
import asyncio
|
||
import os
|
||
import re
|
||
from abc import abstractmethod
|
||
from time import time
|
||
from typing import List, Tuple, Dict
|
||
|
||
from ..base.hyperionrequest import HyperionRequest
|
||
from ...models.genshin.hyperion import (
|
||
PostInfo,
|
||
ArtworkImage,
|
||
LiveInfo,
|
||
LiveCode,
|
||
LiveCodeHoYo,
|
||
PostRecommend,
|
||
PostTypeEnum,
|
||
)
|
||
from ...typedefs import JSON_DATA
|
||
|
||
__all__ = (
|
||
"HyperionBase",
|
||
"Hyperion",
|
||
)
|
||
|
||
|
||
class HyperionBase:
|
||
@staticmethod
|
||
def extract_post_id(text: str) -> Tuple[int, PostTypeEnum]:
|
||
"""
|
||
:param text:
|
||
# https://bbs.mihoyo.com/ys/article/8808224
|
||
# https://m.bbs.mihoyo.com/ys/article/8808224
|
||
# https://www.miyoushe.com/ys/article/32497914
|
||
# https://m.miyoushe.com/ys/#/article/32497914
|
||
:return: post_id
|
||
"""
|
||
rgx = re.compile(r"(?:bbs|www\.)?(?:miyoushe|mihoyo)\.(.*)/[^.]+/article/(?P<article_id>\d+)")
|
||
rgx2 = re.compile(r"(?:bbs|www\.)?(?:hoyolab|hoyoverse)\.(.*)/article/(?P<article_id>\d+)")
|
||
matches = rgx.search(text) or rgx2.search(text)
|
||
if matches is None:
|
||
return -1, PostTypeEnum.NULL
|
||
entries = matches.groupdict()
|
||
if entries is None:
|
||
return -1, PostTypeEnum.NULL
|
||
try:
|
||
art_id = int(entries.get("article_id"))
|
||
post_type = PostTypeEnum.CN if "miyoushe" in text or "mihoyo" in text else PostTypeEnum.OS
|
||
except (IndexError, ValueError, TypeError):
|
||
return -1, PostTypeEnum.NULL
|
||
return art_id, post_type
|
||
|
||
@staticmethod
|
||
def get_list_url_params(forum_id: int, is_good: bool = False, is_hot: bool = False, page_size: int = 20) -> dict:
|
||
return {
|
||
"forum_id": forum_id,
|
||
"gids": 2,
|
||
"is_good": is_good,
|
||
"is_hot": is_hot,
|
||
"page_size": page_size,
|
||
"sort_type": 1,
|
||
}
|
||
|
||
@staticmethod
|
||
def get_images_params(
|
||
resize: int = 600, quality: int = 80, auto_orient: int = 0, interlace: int = 1, images_format: str = "jpg"
|
||
):
|
||
"""
|
||
image/resize,s_600/quality,q_80/auto-orient,0/interlace,1/format,jpg
|
||
:param resize: 图片大小
|
||
:param quality: 图片质量
|
||
:param auto_orient: 自适应
|
||
:param interlace: 未知
|
||
:param images_format: 图片格式
|
||
:return:
|
||
"""
|
||
params = (
|
||
f"image/resize,s_{resize}/quality,q_{quality}/auto-orient,"
|
||
f"{auto_orient}/interlace,{interlace}/format,{images_format}"
|
||
)
|
||
return {"x-oss-process": params}
|
||
|
||
@staticmethod
|
||
async def get_images_by_post_id_tasks(task_list: List) -> List[ArtworkImage]:
|
||
art_list = []
|
||
result_lists = await asyncio.gather(*task_list)
|
||
for result_list in result_lists:
|
||
for result in result_list:
|
||
if isinstance(result, ArtworkImage):
|
||
art_list.append(result)
|
||
|
||
def take_page(elem: ArtworkImage):
|
||
return elem.page
|
||
|
||
art_list.sort(key=take_page)
|
||
return art_list
|
||
|
||
@staticmethod
|
||
async def download_image(client: "HyperionRequest", art_id: int, url: str, page: int = 0) -> List[ArtworkImage]:
|
||
filename = os.path.basename(url)
|
||
_, _file_extension = os.path.splitext(filename)
|
||
file_extension = _file_extension.lower()
|
||
is_image = file_extension in ".jpg" or file_extension in ".jpeg" or file_extension in ".png"
|
||
response = await client.get(
|
||
url, params=Hyperion.get_images_params(resize=2000) if is_image else None, de_json=False
|
||
)
|
||
return ArtworkImage.gen(
|
||
art_id=art_id,
|
||
page=page,
|
||
file_name=filename,
|
||
file_extension=url.split(".")[-1],
|
||
data=response.content,
|
||
url=url,
|
||
)
|
||
|
||
@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]:
|
||
"""获取官方推荐帖子"""
|
||
|
||
@abstractmethod
|
||
async def get_post_info(self, gids: int, post_id: int, read: int = 1) -> PostInfo:
|
||
"""获取帖子信息"""
|
||
|
||
@abstractmethod
|
||
async def get_images_by_post_id(self, gids: int, post_id: int) -> List[ArtworkImage]:
|
||
"""获取帖子图片"""
|
||
|
||
@abstractmethod
|
||
async def close(self):
|
||
"""关闭请求会话"""
|
||
|
||
|
||
class Hyperion(HyperionBase):
|
||
"""米忽悠bbs相关API请求
|
||
|
||
该名称来源于米忽悠的安卓BBS包名结尾，考虑到大部分重要的功能确实是在移动端实现了
|
||
"""
|
||
|
||
POST_FULL_URL = "https://bbs-api.miyoushe.com/post/wapi/getPostFull"
|
||
POST_FULL_IN_COLLECTION_URL = "https://bbs-api.miyoushe.com/post/wapi/getPostFullInCollection"
|
||
GET_NEW_LIST_URL = "https://bbs-api-static.miyoushe.com/painter/wapi/getNewsList"
|
||
GET_OFFICIAL_RECOMMENDED_POSTS_URL = "https://bbs-api.miyoushe.com/post/wapi/getOfficialRecommendedPosts"
|
||
LIVE_INFO_URL = "https://api-takumi.mihoyo.com/event/miyolive/index"
|
||
LIVE_CODE_URL = "https://api-takumi-static.mihoyo.com/event/miyolive/refreshCode"
|
||
LIVE_CODE_HOYO_URL = "https://bbs-api-os.hoyolab.com/community/painter/wapi/circle/channel/guide/material"
|
||
|
||
USER_AGENT = (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/90.0.4430.72 Safari/537.36"
|
||
)
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
self.client = HyperionRequest(headers=self.get_headers(), *args, **kwargs)
|
||
|
||
def get_headers(self, referer: str = "https://www.miyoushe.com/ys/"):
|
||
return {"User-Agent": self.USER_AGENT, "Referer": referer}
|
||
|
||
async def get_official_recommended_posts(self, gids: int) -> List[PostRecommend]:
|
||
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}
|
||
response = await self.client.get(url=self.POST_FULL_IN_COLLECTION_URL, params=params)
|
||
return response
|
||
|
||
async def get_post_info(self, gids: int, post_id: int, read: int = 1) -> PostInfo:
|
||
params = {"gids": gids, "post_id": post_id, "read": read}
|
||
response = await self.client.get(self.POST_FULL_URL, params=params)
|
||
return PostInfo.paste_data(response)
|
||
|
||
async def get_images_by_post_id(self, gids: int, post_id: int) -> List[ArtworkImage]:
|
||
post_info = await self.get_post_info(gids, post_id)
|
||
task_list = [
|
||
self._download_image(post_info.post_id, post_info.image_urls[page], page)
|
||
for page in range(len(post_info.image_urls))
|
||
]
|
||
return await self.get_images_by_post_id_tasks(task_list)
|
||
|
||
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 get_live_info(self, act_id: str) -> LiveInfo:
|
||
headers = {"x-rpc-act_id": act_id}
|
||
response = await self.client.get(url=self.LIVE_INFO_URL, headers=headers)
|
||
return LiveInfo(**response["live"])
|
||
|
||
async def get_live_code(self, act_id: str, ver_code: str) -> List[LiveCode]:
|
||
headers = {"x-rpc-act_id": act_id}
|
||
params = {
|
||
"version": ver_code,
|
||
"time": str(int(time())),
|
||
}
|
||
response = await self.client.get(url=self.LIVE_CODE_URL, headers=headers, params=params)
|
||
codes = []
|
||
for code_data in response.get("code_list", []):
|
||
codes.append(LiveCode(**code_data))
|
||
return codes
|
||
|
||
async def get_live_code_hoyo(self, gid: int) -> List[LiveCodeHoYo]:
|
||
headers = self.get_headers("https://www.hoyolab.com/")
|
||
headers.update(
|
||
{
|
||
"x-rpc-app_version": "2.50.0",
|
||
"x-rpc-client_type": "4",
|
||
"x-rpc-language": "zh-cn",
|
||
}
|
||
)
|
||
params = {
|
||
"game_id": str(gid),
|
||
}
|
||
codes = []
|
||
response = await self.client.get(url=self.LIVE_CODE_HOYO_URL, headers=headers, params=params)
|
||
guess_offline_at = LiveCodeHoYo.guess_offline_at()
|
||
for module in response.get("modules", []):
|
||
if exchange_group := module.get("exchange_group"):
|
||
for code_data in exchange_group.get("bonuses", []):
|
||
codes.append(LiveCodeHoYo(**code_data))
|
||
break
|
||
for _ in range(len(codes), 3):
|
||
codes.append(LiveCodeHoYo(exchange_code="", offline_at=guess_offline_at))
|
||
return codes
|
||
|
||
async def close(self):
|
||
await self.client.shutdown()
|