mirror of
https://github.com/PaiGramTeam/PamGram.git
synced 2024-11-17 20:33:41 +00:00
252 lines
8.7 KiB
Python
252 lines
8.7 KiB
Python
import asyncio
|
||
import re
|
||
from typing import List
|
||
|
||
import httpx
|
||
from httpx import AsyncClient
|
||
|
||
from .base import HyperionResponse, ArtworkImage, BaseResponseData
|
||
from .helpers import get_ds, get_device_id
|
||
|
||
|
||
class Hyperion:
|
||
"""
|
||
米忽悠bbs相关API请求
|
||
该名称来源于米忽悠的安卓BBS包名结尾，考虑到大部分重要的功能确实是在移动端实现了
|
||
"""
|
||
|
||
POST_FULL_URL = "https://bbs-api.mihoyo.com/post/wapi/getPostFull"
|
||
POST_FULL_IN_COLLECTION_URL = "https://bbs-api.mihoyo.com/post/wapi/getPostFullInCollection"
|
||
GET_NEW_LIST_URL = "https://bbs-api.mihoyo.com/post/wapi/getNewsList"
|
||
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):
|
||
self.client = httpx.AsyncClient(headers=self.get_headers())
|
||
|
||
@staticmethod
|
||
def extract_post_id(text: str) -> int:
|
||
"""
|
||
:param text:
|
||
# https://bbs.mihoyo.com/ys/article/8808224
|
||
# https://m.bbs.mihoyo.com/ys/article/8808224
|
||
:return: post_id
|
||
"""
|
||
rgx = re.compile(r"(?:bbs\.)?mihoyo\.com/[^.]+/article/(?P<article_id>\d+)")
|
||
matches = rgx.search(text)
|
||
if matches is None:
|
||
return -1
|
||
entries = matches.groupdict()
|
||
if entries is None:
|
||
return -1
|
||
try:
|
||
art_id = int(entries.get('article_id'))
|
||
except (IndexError, ValueError, TypeError):
|
||
return -1
|
||
return art_id
|
||
|
||
def get_headers(self, referer: str = "https://bbs.mihoyo.com/"):
|
||
return {
|
||
"User-Agent": self.USER_AGENT,
|
||
"Referer": referer
|
||
}
|
||
|
||
@staticmethod
|
||
def get_list_url_params(forum_id: int, is_good: bool = False, is_hot: bool = False,
|
||
page_size: int = 20) -> dict:
|
||
params = {
|
||
"forum_id": forum_id,
|
||
"gids": 2,
|
||
"is_good": is_good,
|
||
"is_hot": is_hot,
|
||
"page_size": page_size,
|
||
"sort_type": 1
|
||
}
|
||
|
||
return params
|
||
|
||
@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}
|
||
|
||
async def get_post_full_in_collection(self, collection_id: int, gids: int = 2, order_type=1) -> BaseResponseData:
|
||
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)
|
||
if response.is_error:
|
||
return BaseResponseData(error_message="请求错误")
|
||
return BaseResponseData(response.json())
|
||
|
||
async def get_artwork_info(self, gids: int, post_id: int, read: int = 1) -> HyperionResponse:
|
||
params = {
|
||
"gids": gids,
|
||
"post_id": post_id,
|
||
"read": read
|
||
}
|
||
response = await self.client.get(self.POST_FULL_URL, params=params)
|
||
if response.is_error:
|
||
return HyperionResponse(error_message="请求错误")
|
||
return HyperionResponse(response.json())
|
||
|
||
async def get_post_full_info(self, gids: int, post_id: int, read: int = 1) -> BaseResponseData:
|
||
params = {
|
||
"gids": gids,
|
||
"post_id": post_id,
|
||
"read": read
|
||
}
|
||
response = await self.client.get(self.POST_FULL_URL, params=params)
|
||
if response.is_error:
|
||
return BaseResponseData(error_message="请求错误")
|
||
return BaseResponseData(response.json())
|
||
|
||
async def get_images_by_post_id(self, gids: int, post_id: int) -> List[ArtworkImage]:
|
||
artwork_info = await self.get_artwork_info(gids, post_id)
|
||
if artwork_info.error:
|
||
return []
|
||
urls = artwork_info.results.image_url_list
|
||
art_list = []
|
||
task_list = [
|
||
self.download_image(artwork_info.post_id, urls[page], page) for page in range(len(urls))
|
||
]
|
||
result_list = await asyncio.gather(*task_list)
|
||
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
|
||
|
||
async def download_image(self, art_id: int, url: str, page: int = 0) -> ArtworkImage:
|
||
response = await self.client.get(url, params=self.get_images_params(resize=2000), timeout=5)
|
||
if response.is_error:
|
||
return ArtworkImage(art_id, page, True)
|
||
return ArtworkImage(art_id, page, data=response.content)
|
||
|
||
async def get_new_list(self, gids: int, type_id: int, page_size: int = 20):
|
||
"""
|
||
?gids=2&page_size=20&type=3
|
||
:return:
|
||
"""
|
||
params = {
|
||
"gids": gids,
|
||
"page_size": page_size,
|
||
"type": type_id
|
||
}
|
||
response = await self.client.get(url=self.GET_NEW_LIST_URL, params=params)
|
||
if response.is_error:
|
||
return BaseResponseData(error_message="请求错误")
|
||
return BaseResponseData(response.json())
|
||
|
||
async def close(self):
|
||
await self.client.aclose()
|
||
|
||
|
||
class YuanShen:
|
||
SIGN_INFO_URL = "https://api-takumi.mihoyo.com/event/bbs_sign_reward/info"
|
||
SIGN_URL = "https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign"
|
||
SIGN_HOME_URL = "https://api-takumi.mihoyo.com/event/bbs_sign_reward/home"
|
||
|
||
APP_VERSION = "2.3.0"
|
||
USER_AGENT = "Mozilla/5.0 (Linux; Android 9; Unspecified Device) AppleWebKit/537.36 (KHTML, like Gecko) " \
|
||
"Version/4.0 Chrome/39.0.0.0 Mobile Safari/537.36 miHoYoBBS/2.3.0"
|
||
REFERER = "https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html?" \
|
||
"bbs_auth_required=true&act_id=e202009291139501&utm_source=hyperion&utm_medium=mys&utm_campaign=icon"
|
||
ORIGIN = "https://webstatic.mihoyo.com"
|
||
|
||
ACT_ID = "e202009291139501"
|
||
DS_SALT = "h8w582wxwgqvahcdkpvdhbh2w9casgfl"
|
||
|
||
def __init__(self):
|
||
self.headers = {
|
||
"Origin": self.ORIGIN,
|
||
'DS': get_ds(self.DS_SALT),
|
||
'x-rpc-app_version': self.APP_VERSION,
|
||
'User-Agent': self.USER_AGENT,
|
||
'x-rpc-client_type': '5', # 1为ios 2为安卓 4为pc_web 5为mobile_web
|
||
'Referer': self.REFERER,
|
||
'x-rpc-device_id': get_device_id(self.USER_AGENT)}
|
||
self.client = AsyncClient(headers=self.headers)
|
||
|
||
async def is_sign(self, uid: int, region: str = "cn_gf01", cookies: dict = None):
|
||
"""
|
||
检查是否签到
|
||
:param uid: 游戏UID
|
||
:param region: 服务器
|
||
:param cookies: cookie
|
||
:return:
|
||
"""
|
||
params = {
|
||
"act_id": self.ACT_ID,
|
||
"region": region,
|
||
"uid": uid
|
||
}
|
||
req = await self.client.get(self.SIGN_INFO_URL, params=params, cookies=cookies)
|
||
if req.is_error:
|
||
return BaseResponseData(error_message="请求错误")
|
||
return BaseResponseData(req.json())
|
||
|
||
async def sign(self, uid: int, region: str = "cn_gf01", cookies: dict = None):
|
||
"""
|
||
执行签到
|
||
:param uid: 游戏UID
|
||
:param region: 服务器
|
||
:param cookies: cookie
|
||
:return:
|
||
"""
|
||
data = {
|
||
"act_id": self.ACT_ID,
|
||
"region": region,
|
||
"uid": uid
|
||
}
|
||
req = await self.client.post(self.SIGN_URL, json=data, cookies=cookies)
|
||
if req.is_error:
|
||
return BaseResponseData(error_message="签到失败")
|
||
return BaseResponseData(req.json())
|
||
|
||
async def get_sign_give(self, cookies: dict = None):
|
||
"""
|
||
返回今日签到信息
|
||
:param cookies:
|
||
:return:
|
||
"""
|
||
params = {
|
||
"act_id": self.ACT_ID
|
||
}
|
||
req = await self.client.get(self.SIGN_HOME_URL, params=params, cookies=cookies)
|
||
if req.is_error:
|
||
return
|
||
return BaseResponseData(req.json())
|
||
|
||
async def __aenter__(self):
|
||
"""
|
||
:return:
|
||
"""
|
||
pass
|
||
|
||
async def __aexit__(self, exc_type, exc, tb):
|
||
"""
|
||
:param exc_type:
|
||
:param exc:
|
||
:param tb:
|
||
:return:
|
||
"""
|
||
await self.client.aclose()
|