mirror of
https://github.com/PaiGramTeam/PaiGram.git
synced 2024-11-26 02:11:03 +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()
|