PaiGramTeam/PamGram
3
0
Fork 0
mirror of https://github.com/PaiGramTeam/PamGram.git synced 2024-11-21 21:58:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ Use SIMNet and Remove genshin.py

Browse Source 
Co-authored-by: 洛水居室 <luoshuijs@outlook.com>
Co-authored-by: deepsource-autofix[bot] <62050782+deepsource-autofix[bot]@users.noreply.github.com>
This commit is contained in:
omg-xtao 2023-07-19 13:52:30 +08:00 committed by GitHub
parent e0b0156f27
commit fcaf0d398e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
33 changed files with 1572 additions and 2434 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
core/services/cookies/services.py
View File

@ -1,7 +1,7 @@
from typing import List, Optional
import genshin
from genshin import GenshinException, InvalidCookies, TooManyRequests, types, Game
from simnet import StarRailClient, Region, Game
from simnet.errors import InvalidCookies, BadRequest as SimnetBadRequest, TooManyRequests
from core.base_service import BaseService
from core.basemodel import RegionEnum
@ -86,31 +86,29 @@ class PublicCookiesService(BaseService):
await self._cache.delete_public_cookies(public_id, region)
continue
if region == RegionEnum.HYPERION:
client = genshin.Client(cookies=cookies.data, game=types.Game.STARRAIL, region=types.Region.CHINESE)
client = StarRailClient(cookies=cookies.data, region=Region.CHINESE)
elif region == RegionEnum.HOYOLAB:
client = genshin.Client(
cookies=cookies.data, game=types.Game.STARRAIL, region=types.Region.OVERSEAS, lang="zh-cn"
)
client = StarRailClient(cookies=cookies.data, region=Region.OVERSEAS, lang="zh-cn")
else:
raise CookieServiceError
try:
if client.cookie_manager.user_id is None:
if client.account_id is None:
raise RuntimeError("account_id not found")
record_cards = await client.get_record_cards()
for record_card in record_cards:
if record_card.game == Game.STARRAIL:
if record_card.game == Game.GENSHIN:
await client.get_starrail_user(record_card.uid)
break
else:
accounts = await client.get_game_accounts()
for account in accounts:
if account.game == Game.STARRAIL:
if account.game == Game.GENSHIN:
await client.get_starrail_user(account.uid)
break
except InvalidCookies as exc:
if exc.retcode in (10001, -100):
if exc.ret_code in (10001, -100):
logger.warning("用户 [%s] Cookies无效", public_id)
elif exc.retcode == 10103:
elif exc.ret_code == 10103:
logger.warning("用户 [%s] Cookies有效，但没有绑定到游戏帐户", public_id)
else:
logger.warning("Cookies无效 ")
@ -125,10 +123,10 @@ class PublicCookiesService(BaseService):
await self._repository.update(cookies)
await self._cache.delete_public_cookies(cookies.user_id, region)
continue
except GenshinException as exc:
if "invalid content type" in exc.msg:
except SimnetBadRequest as exc:
if "invalid content type" in exc.message:
raise exc
if exc.retcode == 1034:
if exc.ret_code == 1034:
logger.warning("用户 [%s] 触发验证", public_id)
else:
logger.warning("用户 [%s] 获取账号信息发生错误，错误信息为", public_id)
@ -145,6 +143,8 @@ class PublicCookiesService(BaseService):
except Exception as exc:
await self._cache.delete_public_cookies(cookies.user_id, region)
raise exc
finally:
await client.shutdown()
logger.info("用户 user_id[%s] 请求用户 user_id[%s] 的公共Cookies 该Cookies使用次数为%s次 ", user_id, public_id, count)
return cookies

83
modules/apihelper/client/components/authclient.py
View File

@ -34,9 +34,6 @@ class AuthClient:
QRCODE_GET_API = f"https://{HK4E_SDK_HOST}/hk4e_cn/combo/panda/qrcode/query"
GET_COOKIE_ACCOUNT_BY_GAME_TOKEN_API = f"https://{TAKUMI_HOST}/auth/api/getCookieAccountInfoByGameToken"
GET_TOKEN_BY_GAME_LTOKEN_API = f"https://{PASSPORT_HOST}/account/ma-cn-session/app/getTokenByGameToken"
GET_COOKIES_TOKEN_BY_STOKEN_API = f"https://{PASSPORT_HOST}/account/auth/api/getCookieAccountInfoBySToken"
GET_LTOKEN_BY_STOKEN_API = f"https://{PASSPORT_HOST}/account/auth/api/getLTokenBySToken"
get_STOKEN_URL = f"https://{TAKUMI_HOST}/auth/api/getMultiTokenByLoginTicket"
def __init__(
self,
@ -60,24 +57,6 @@ class AuthClient:
else:
self.user_id = self.cookies.user_id
async def get_stoken_by_login_ticket(self) -> bool:
if self.cookies.login_ticket is None and self.user_id is None:
return False
params = {"login_ticket": self.cookies.login_ticket, "uid": self.user_id, "token_types": 3}
data = await self.client.get(self.get_STOKEN_URL, params=params, headers={"User-Agent": self.USER_AGENT})
res_json = data.json()
res_data = res_json.get("data", {}).get("list", [])
for i in res_data:
name = i.get("name")
token = i.get("token")
if name and token and hasattr(self.cookies, name):
setattr(self.cookies, name, token)
if self.cookies.stoken:
if self.cookies.stuid:
self.cookies.stuid = self.user_id
return True
return False
async def get_ltoken_by_game_token(self, game_token: str) -> bool:
if self.user_id is None:
return False
@ -154,68 +133,6 @@ class AuthClient:
if res_data.get("stat", "") == "Confirmed":
return await self._set_cookie_by_game_token(res_json.get("data", {}))
async def get_cookie_token_by_stoken(self) -> bool:
if self.cookies.stoken is None:
return False
user_id = self.cookies.user_id
headers = {
"x-rpc-app_version": "2.11.1",
"User-Agent": (
"Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/2.11.1"
),
"x-rpc-client_type": "5",
"Referer": "https://webstatic.mihoyo.com/",
"Origin": "https://webstatic.mihoyo.com",
}
params = {
"stoken": self.cookies.stoken,
"uid": user_id,
}
res = await self.client.get(
self.GET_COOKIES_TOKEN_BY_STOKEN_API,
headers=headers,
params=params,
)
res_json = res.json()
cookie_token = res_json.get("data", {}).get("cookie_token", "")
if cookie_token:
self.cookies.cookie_token = cookie_token
self.cookies.account_id = user_id
return True
return False
async def get_ltoken_by_stoken(self) -> bool:
if self.cookies.stoken is None:
return False
user_id = self.cookies.user_id
headers = {
"x-rpc-app_version": "2.11.1",
"User-Agent": (
"Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/2.11.1"
),
"x-rpc-client_type": "5",
"Referer": "https://webstatic.mihoyo.com/",
"Origin": "https://webstatic.mihoyo.com",
}
params = {
"stoken": self.cookies.stoken,
"uid": user_id,
}
res = await self.client.get(
self.GET_LTOKEN_BY_STOKEN_API,
headers=headers,
params=params,
)
res_json = res.json()
ltoken = res_json.get("data", {}).get("ltoken", "")
if ltoken:
self.cookies.ltoken = ltoken
self.cookies.ltuid = user_id
return True
return False
@staticmethod
def generate_qrcode(url: str) -> bytes:
qr = qrcode.QRCode(version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=10, border=4)

6
modules/apihelper/client/components/verify.py
View File

@ -1,7 +1,9 @@
import json
import re
import time
from typing import Dict, Optional
from typing import Dict, Optional, Union
from httpx import Cookies
from ..base.hyperionrequest import HyperionRequest
from ...utility.devices import devices_methods
@ -37,7 +39,7 @@ class Verify:
"Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
}
def __init__(self, account_id: int = None, cookies: Dict = None):
def __init__(self, account_id: int = None, cookies: Union[Dict, Cookies] = None):
self.account_id = account_id
self.client = HyperionRequest(headers=self.BBS_HEADERS, cookies=cookies)

2
modules/gacha_log/const.py
View File

@ -1,4 +1,4 @@
from genshin.models.genshin.gacha import StarRailBannerType
from simnet.models.starrail.wish import StarRailBannerType
SRGF_VERSION = "v1.0"

70
modules/gacha_log/log.py
View File

@ -4,13 +4,14 @@ import datetime
import json
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from typing import Dict, List, Optional, Tuple, TYPE_CHECKING
import aiofiles
from genshin import AuthkeyTimeout, Client, InvalidAuthkey
from genshin.models.genshin.gacha import StarRailBannerType
from simnet import StarRailClient, Region
from simnet.errors import AuthkeyTimeout, InvalidAuthkey
from simnet.models.starrail.wish import StarRailBannerType
from simnet.utils.player import recognize_starrail_server
from core.dependence.assets import AssetsService
from metadata.pool.pool import get_pool_by_id
from modules.gacha_log.const import GACHA_TYPE_LIST
from modules.gacha_log.error import (
@ -35,6 +36,10 @@ from modules.gacha_log.models import (
)
from utils.const import PROJECT_ROOT
if TYPE_CHECKING:
from core.dependence.assets import AssetsService
GACHA_LOG_PATH = PROJECT_ROOT.joinpath("data", "apihelper", "warp_log")
GACHA_LOG_PATH.mkdir(parents=True, exist_ok=True)
@ -98,7 +103,7 @@ class GachaLog:
async def save_gacha_log_info(self, user_id: str, uid: str, info: GachaLogInfo):
"""保存跃迁记录数据
:param user_id: 用户id
:param uid: 原神uid
:param uid: 玩家uid
:param info: 跃迁记录数据
"""
save_path = self.gacha_log_path / f"{user_id}-{uid}.json"
@ -166,13 +171,13 @@ class GachaLog:
new_num += 1
return new_num
async def import_gacha_log_data(self, user_id: int, client: Client, data: dict, verify_uid: bool = True) -> int:
async def import_gacha_log_data(self, user_id: int, player_id: int, data: dict, verify_uid: bool = True) -> int:
new_num = 0
try:
uid = data["info"]["uid"]
if not verify_uid:
uid = client.uid
elif int(uid) != client.uid:
uid = player_id
elif int(uid) != player_id:
raise GachaLogAccountNotFound
try:
import_type = ImportType(data["info"]["export_app"])
@ -208,20 +213,29 @@ class GachaLog:
except Exception as exc:
raise GachaLogException from exc
async def get_gacha_log_data(self, user_id: int, client: Client, authkey: str) -> int:
@staticmethod
def get_game_client(player_id: int) -> StarRailClient:
if recognize_starrail_server(player_id) in ["prod_gf_cn", "prod_qd_cn"]:
return StarRailClient(player_id=player_id, region=Region.CHINESE, lang="zh-cn")
else:
return StarRailClient(player_id=player_id, region=Region.OVERSEAS, lang="zh-cn")
async def get_gacha_log_data(self, user_id: int, player_id: int, authkey: str) -> int:
"""使用authkey获取跃迁记录数据，并合并旧数据
:param user_id: 用户id
:param client: genshin client
:param player_id: 玩家id
:param authkey: authkey
:return: 更新结果
"""
new_num = 0
gacha_log, _ = await self.load_history_info(str(user_id), str(client.uid))
gacha_log, _ = await self.load_history_info(str(user_id), str(player_id))
# 将唯一 id 放入临时数据中，加快查找速度
temp_id_data = {pool_name: {i.id: i for i in pool_data} for pool_name, pool_data in gacha_log.item_list.items()}
client = self.get_game_client(player_id)
try:
for pool_id, pool_name in GACHA_TYPE_LIST.items():
async for data in client.warp_history(pool_id, authkey=authkey):
wish_history = await client.wish_history(pool_id.value, authkey=authkey)
for data in wish_history:
item = GachaItem(
id=str(data.id),
name=data.name,
@ -252,11 +266,13 @@ class GachaLog:
raise GachaLogAuthkeyTimeout from exc
except InvalidAuthkey as exc:
raise GachaLogInvalidAuthkey from exc
finally:
await client.shutdown()
for i in gacha_log.item_list.values():
i.sort(key=lambda x: (x.time, x.id))
gacha_log.update_time = datetime.datetime.now()
gacha_log.import_type = ImportType.PaiGram.value
await self.save_gacha_log_info(str(user_id), str(client.uid), gacha_log)
await self.save_gacha_log_info(str(user_id), str(player_id), gacha_log)
return new_num
@staticmethod
@ -265,7 +281,7 @@ class GachaLog:
return False
return True
async def get_all_5_star_items(self, data: List[GachaItem], assets: AssetsService, pool_name: str = "角色跃迁"):
async def get_all_5_star_items(self, data: List[GachaItem], assets: "AssetsService", pool_name: str = "角色跃迁"):
"""
获取所有5星角色
:param data: 跃迁记录
@ -314,7 +330,7 @@ class GachaLog:
return result, count
@staticmethod
async def get_all_4_star_items(data: List[GachaItem], assets: AssetsService):
async def get_all_4_star_items(data: List[GachaItem], assets: "AssetsService"):
"""
获取 no_fout_star
:param data: 跃迁记录
@ -475,16 +491,16 @@ class GachaLog:
return f"{pool_name} · 非"
return pool_name
async def get_analysis(self, user_id: int, client: Client, pool: StarRailBannerType, assets: AssetsService):
async def get_analysis(self, user_id: int, player_id: int, pool: StarRailBannerType, assets: "AssetsService"):
"""
获取跃迁记录分析数据
:param user_id: 用户id
:param client: genshin client
:param player_id: 玩家id
:param pool: 池子类型
:param assets: 资源服务
:return: 分析数据
"""
gacha_log, status = await self.load_history_info(str(user_id), str(client.uid))
gacha_log, status = await self.load_history_info(str(user_id), str(player_id))
if not status:
raise GachaLogNotFound
pool_name = GACHA_TYPE_LIST[pool]
@ -507,7 +523,7 @@ class GachaLog:
last_time = data[0].time.strftime("%Y-%m-%d %H:%M")
first_time = data[-1].time.strftime("%Y-%m-%d %H:%M")
return {
"uid": client.uid,
"uid": player_id,
"allNum": total,
"type": pool.value,
"typeName": pool_name,
@ -519,17 +535,17 @@ class GachaLog:
}
async def get_pool_analysis(
self, user_id: int, client: Client, pool: StarRailBannerType, assets: AssetsService, group: bool
self, user_id: int, player_id: int, pool: StarRailBannerType, assets: "AssetsService", group: bool
) -> dict:
"""获取跃迁记录分析数据
:param user_id: 用户id
:param client: genshin client
:param player_id: 玩家id
:param pool: 池子类型
:param assets: 资源服务
:param group: 是否群组
:return: 分析数据
"""
gacha_log, status = await self.load_history_info(str(user_id), str(client.uid))
gacha_log, status = await self.load_history_info(str(user_id), str(player_id))
if not status:
raise GachaLogNotFound
pool_name = GACHA_TYPE_LIST[pool]
@ -559,20 +575,20 @@ class GachaLog:
)
pool_data = [i for i in pool_data if i["count"] > 0]
return {
"uid": client.uid,
"uid": player_id,
"typeName": pool_name,
"pool": pool_data[:6] if group else pool_data,
"hasMore": len(pool_data) > 6,
}
async def get_all_five_analysis(self, user_id: int, client: Client, assets: AssetsService) -> dict:
async def get_all_five_analysis(self, user_id: int, player_id: int, assets: "AssetsService") -> dict:
"""获取五星跃迁记录分析数据
:param user_id: 用户id
:param client: genshin client
:param player_id: 玩家id
:param assets: 资源服务
:return: 分析数据
"""
gacha_log, status = await self.load_history_info(str(user_id), str(client.uid))
gacha_log, status = await self.load_history_info(str(user_id), str(player_id))
if not status:
raise GachaLogNotFound
pools = []
@ -600,7 +616,7 @@ class GachaLog:
for up_pool in pools
]
return {
"uid": client.uid,
"uid": player_id,
"typeName": "五星列表",
"pool": pool_data,
"hasMore": False,

8
modules/gacha_log/models.py
View File

@ -47,24 +47,24 @@ class GachaItem(BaseModel):
if item_id := (roleToId(v) or lightConeToId(v)):
if item_id not in not_real_roles:
return v
raise ValueError("Invalid name")
raise ValueError(f"Invalid name {v}")
@validator("gacha_type")
def check_gacha_type(cls, v):
if v not in {"1", "2", "11", "12"}:
raise ValueError("gacha_type must be 1, 2, 11 or 12")
raise ValueError(f"gacha_type must be 1, 2, 11 or 12, invalid value: {v}")
return v
@validator("item_type")
def check_item_type(cls, item):
if item not in {"角色", "光锥"}:
raise ValueError("error item type")
raise ValueError(f"error item type {item}")
return item
@validator("rank_type")
def check_rank_type(cls, rank):
if rank not in {"5", "4", "3"}:
raise ValueError("error rank type")
raise ValueError(f"error rank type {rank}")
return rank

26
modules/pay_log/error.py
View File

@ -1,26 +0,0 @@
class PayLogException(Exception):
pass
class PayLogFileError(PayLogException):
pass
class PayLogNotFound(PayLogFileError):
pass
class PayLogAccountNotFound(PayLogException):
pass
class PayLogAuthkeyException(PayLogException):
pass
class PayLogAuthkeyTimeout(PayLogAuthkeyException):
pass
class PayLogInvalidAuthkey(PayLogAuthkeyException):
pass

234
modules/pay_log/log.py
View File

@ -1,234 +0,0 @@
import contextlib
from pathlib import Path
from typing import Tuple, Optional, List, Dict
import aiofiles
from genshin import Client, AuthkeyTimeout, InvalidAuthkey
from genshin.models import TransactionKind, BaseTransaction
from modules.pay_log.error import PayLogAuthkeyTimeout, PayLogInvalidAuthkey, PayLogNotFound
from modules.pay_log.models import PayLog as PayLogModel, BaseInfo
from utils.const import PROJECT_ROOT
try:
import ujson as jsonlib
except ImportError:
import json as jsonlib
PAY_LOG_PATH = PROJECT_ROOT.joinpath("data", "apihelper", "pay_log")
PAY_LOG_PATH.mkdir(parents=True, exist_ok=True)
class PayLog:
def __init__(self, pay_log_path: Path = PAY_LOG_PATH):
self.pay_log_path = pay_log_path
@staticmethod
async def load_json(path):
async with aiofiles.open(path, "r", encoding="utf-8") as f:
return jsonlib.loads(await f.read())
@staticmethod
async def save_json(path, data: PayLogModel):
async with aiofiles.open(path, "w", encoding="utf-8") as f:
return await f.write(data.json(ensure_ascii=False, indent=4))
def get_file_path(
self,
user_id: str,
uid: str,
bak: bool = False,
) -> Path:
"""获取文件路径
:param user_id: 用户 ID
:param uid: UID
:param bak: 是否为备份文件
:return: 文件路径
"""
return self.pay_log_path / f"{user_id}-{uid}.json{'.bak' if bak else ''}"
async def load_history_info(
self,
user_id: str,
uid: str,
only_status: bool = False,
) -> Tuple[Optional[PayLogModel], bool]:
"""读取历史记录数据
:param user_id: 用户id
:param uid: 原神uid
:param only_status: 是否只读取状态
:return: 抽卡记录数据
"""
file_path = self.get_file_path(user_id, uid)
if only_status:
return None, file_path.exists()
if not file_path.exists():
return PayLogModel(info=BaseInfo(uid=uid), list=[]), False
try:
return PayLogModel.parse_obj(await self.load_json(file_path)), True
except jsonlib.JSONDecodeError:
return PayLogModel(info=BaseInfo(uid=uid), list=[]), False
async def remove_history_info(
self,
user_id: str,
uid: str,
) -> bool:
"""删除历史记录数据
:param user_id: 用户id
:param uid: 原神uid
:return: 是否删除成功
"""
file_path = self.get_file_path(user_id, uid)
file_bak_path = self.get_file_path(user_id, uid, bak=True)
with contextlib.suppress(Exception):
file_bak_path.unlink(missing_ok=True)
if file_path.exists():
try:
file_path.unlink()
except PermissionError:
return False
return True
return False
async def save_pay_log_info(self, user_id: str, uid: str, info: PayLogModel) -> None:
"""保存日志记录数据
:param user_id: 用户id
:param uid: 原神uid
:param info: 记录数据
"""
save_path = self.pay_log_path / f"{user_id}-{uid}.json"
save_path_bak = self.pay_log_path / f"{user_id}-{uid}.json.bak"
# 将旧数据备份一次
with contextlib.suppress(PermissionError):
if save_path.exists():
if save_path_bak.exists():
save_path_bak.unlink()
save_path.rename(save_path.parent / f"{save_path.name}.bak")
# 写入数据
await self.save_json(save_path, info)
async def get_log_data(
self,
user_id: int,
client: Client,
authkey: str,
) -> int:
"""使用 authkey 获取历史记录数据，并合并旧数据
:param user_id: 用户id
:param client: genshin client
:param authkey: authkey
:return: 更新结果
"""
new_num = 0
pay_log, have_old = await self.load_history_info(str(user_id), str(client.uid))
history_ids = [i.id for i in pay_log.list]
try:
async for data in client.transaction_log(TransactionKind.CRYSTAL, authkey=authkey):
if data.id not in history_ids:
pay_log.list.append(data)
new_num += 1
except AuthkeyTimeout as exc:
raise PayLogAuthkeyTimeout from exc
except InvalidAuthkey as exc:
raise PayLogInvalidAuthkey from exc
if new_num > 0 or have_old:
pay_log.list.sort(key=lambda x: (x.time, x.id), reverse=True)
pay_log.info.update_now()
await self.save_pay_log_info(str(user_id), str(client.uid), pay_log)
return new_num
@staticmethod
async def get_month_data(pay_log: PayLogModel, price_data: List[Dict]) -> Tuple[int, List[Dict]]:
"""获取月份数据
:param pay_log: 日志数据
:param price_data: 商品数据
:return: 月份数据
"""
all_amount: int = 0
months: List[int] = []
month_datas: List[Dict] = []
last_month: Optional[Dict] = None
month_data: List[Optional[BaseTransaction]] = []
for i in pay_log.list:
if i.amount <= 0:
continue
all_amount += i.amount
if i.time.month not in months:
months.append(i.time.month)
if last_month:
last_month["amount"] = sum(i.amount for i in month_data)
month_data.clear()
if len(months) <= 6:
last_month = {
"month": f"{i.time.month}月",
"amount": 0,
}
month_datas.append(last_month)
else:
last_month = None
for j in price_data:
if i.amount in j["price"]:
j["count"] += 1
break
month_data.append(i)
if last_month:
last_month["amount"] = sum(i.amount for i in month_data)
month_data.clear()
if not month_datas:
raise PayLogNotFound
return all_amount, month_datas
async def get_analysis(self, user_id: int, client: Client):
"""获取分析数据
:param user_id: 用户id
:param client: genshin client
:return: 分析数据
"""
pay_log, status = await self.load_history_info(str(user_id), str(client.uid))
if not status:
raise PayLogNotFound
# 单双倍结晶数
price_data = [
{
"price": price,
"count": 0,
}
for price in [[680], [300], [8080, 12960], [3880, 6560], [2240, 3960], [1090, 1960], [330, 600], [60, 120]]
]
price_data_name = ["大月卡", "小月卡", "648", "328", "198", "98", "30", "6"]
real_price = [68, 30, 648, 328, 198, 98, 30, 6]
all_amount, month_datas = await PayLog.get_month_data(pay_log, price_data)
month_data = sorted(month_datas, key=lambda k: k["amount"], reverse=True)
all_pay = sum((price_data[i]["count"] * real_price[i]) for i in range(len(price_data)))
datas = [
{"value": f"￥{all_pay:.0f}", "name": "总消费"},
{"value": all_amount, "name": "总结晶"},
{"value": f"{month_data[0]['month']}", "name": "消费最多"},
{
"value": f"￥{month_data[0]['amount'] / 10:.0f}",
"name": f"{month_data[0]['month']}消费",
},
*[
{
"value": price_data[i]["count"] if i != 0 else "*",
"name": f"{price_data_name[i]}",
}
for i in range(len(price_data))
],
]
pie_datas = [
{
"value": f"{price_data[i]['count'] * real_price[i]:.0f}",
"name": f"{price_data_name[i]}",
}
for i in range(len(price_data))
if price_data[i]["count"] > 0
]
return {
"uid": client.uid,
"datas": datas,
"bar_data": month_datas,
"pie_data": pie_datas,
}

41
modules/pay_log/models.py
View File

@ -1,41 +0,0 @@
import datetime
from typing import Any, List
from genshin.models import BaseTransaction
from pydantic import BaseModel, BaseConfig
try:
import ujson as jsonlib
except ImportError:
import json as jsonlib
class _ModelConfig(BaseConfig):
json_dumps = jsonlib.dumps
json_loads = jsonlib.loads
class BaseInfo(BaseModel):
Config = _ModelConfig
uid: str = "0"
lang: str = "zh-cn"
export_time: str = ""
export_timestamp: int = 0
export_app: str = "PaimonBot"
def __init__(self, **data: Any):
super().__init__(**data)
if not self.export_time:
self.update_now()
def update_now(self):
self.export_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.export_timestamp = int(datetime.datetime.now().timestamp())
class PayLog(BaseModel):
Config = _ModelConfig
info: BaseInfo
list: List[BaseTransaction]

40
plugins/account/account.py
View File

@ -1,8 +1,13 @@
from datetime import datetime
from typing import Optional, TYPE_CHECKING
import genshin
from genshin import DataNotPublic, GenshinException, types, AccountNotFound, InvalidCookies
from simnet import StarRailClient, Region
from simnet.errors import (
InvalidCookies,
BadRequest as SimnetBadRequest,
DataNotPublic,
AccountNotFound,
)
from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove, TelegramObject
from telegram.ext import ConversationHandler, filters
from telegram.helpers import escape_markdown
@ -132,32 +137,25 @@ class BindAccountPlugin(Plugin.Conversation):
await message.reply_text("用户查询次数过多，请稍后重试", reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
if region == RegionEnum.HYPERION:
client = genshin.Client(cookies=cookies.data, game=types.Game.STARRAIL, region=types.Region.CHINESE)
client = StarRailClient(cookies=cookies.data, region=Region.CHINESE, lang="zh-cn")
elif region == RegionEnum.HOYOLAB:
client = genshin.Client(
cookies=cookies.data, game=types.Game.STARRAIL, region=types.Region.OVERSEAS, lang="zh-cn"
)
client = StarRailClient(cookies=cookies.data, region=Region.OVERSEAS, lang="zh-cn")
else:
return ConversationHandler.END
try:
record_cards = await client.get_record_cards(account_id)
record_card = record_cards[0]
for card in record_cards:
if card.game == types.Game.STARRAIL:
record_card = card
break
record_card = await client.get_record_card(account_id)
if record_card is None:
await message.reply_text("请在设置展示主界面添加崩坏：星穹铁道", reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
except DataNotPublic:
await message.reply_text("角色未公开", reply_markup=ReplyKeyboardRemove())
logger.warning("获取账号信息发生错误 %s 账户信息未公开", account_id)
return ConversationHandler.END
except GenshinException as exc:
except SimnetBadRequest as exc:
await message.reply_text("获取账号信息发生错误", reply_markup=ReplyKeyboardRemove())
logger.error("获取账号信息发生错误")
logger.exception(exc)
return ConversationHandler.END
if record_card.game != types.Game.STARRAIL:
await message.reply_text("角色信息查询返回无星穹铁道游戏信息，请确定你有星穹铁道账号", reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
player_info = await self.players_service.get(
user.id, player_id=record_card.uid, region=bind_account_plugin_data.region
)
@ -201,11 +199,9 @@ class BindAccountPlugin(Plugin.Conversation):
await message.reply_text("用户查询次数过多，请稍后重试", reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
if region == RegionEnum.HYPERION:
client = genshin.Client(cookies=cookies.data, game=types.Game.STARRAIL, region=types.Region.CHINESE)
client = StarRailClient(cookies=cookies.data, region=Region.CHINESE, lang="zh-cn")
elif region == RegionEnum.HOYOLAB:
client = genshin.Client(
cookies=cookies.data, game=types.Game.STARRAIL, region=types.Region.OVERSEAS, lang="zh-cn"
)
client = StarRailClient(cookies=cookies.data, region=Region.OVERSEAS, lang="zh-cn")
else:
return ConversationHandler.END
try:
@ -222,7 +218,7 @@ class BindAccountPlugin(Plugin.Conversation):
await self.public_cookies_service.undo(user.id, cookies, CookiesStatusEnum.INVALID_COOKIES)
await message.reply_text("出错了呜呜呜 ~ 请稍后重试")
return ConversationHandler.END
except GenshinException as exc:
except SimnetBadRequest as exc:
if exc.retcode == 1034:
await self.public_cookies_service.undo(user.id)
await message.reply_text("出错了呜呜呜 ~ 请稍后重试")
@ -234,6 +230,8 @@ class BindAccountPlugin(Plugin.Conversation):
except ValueError:
await message.reply_text("ID 格式有误，请检查", reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
finally:
await client.shutdown()
player_info = await self.players_service.get(
user.id, player_id=player_id, region=bind_account_plugin_data.region
)

98
plugins/account/cookies.py
View File

@ -1,10 +1,10 @@
from datetime import datetime
from typing import Dict, Optional
import genshin
from arkowrapper import ArkoWrapper
from genshin import DataNotPublic, GenshinException, InvalidCookies, types
from genshin.models import GenshinAccount
from simnet import StarRailClient, Region
from simnet.errors import DataNotPublic, InvalidCookies, BadRequest as SimnetBadRequest
from simnet.models.lab.record import Account
from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove, TelegramObject, Update
from telegram.constants import ParseMode
from telegram.ext import CallbackContext, ConversationHandler, filters
@ -34,7 +34,7 @@ class AccountCookiesPluginData(TelegramObject):
cookies: dict = {}
account_id: int = 0
# player_id: int = 0
starrail_account: Optional[GenshinAccount] = None
starrail_account: Optional[Account] = None
def reset(self):
self.player = None
@ -136,40 +136,28 @@ class AccountCookiesPlugin(Plugin.Conversation):
return CHECK_SERVER
account_cookies_plugin_data.region = region
await message.reply_text(f"请输入{bbs_name}的Cookies！或回复退出取消操作", reply_markup=ReplyKeyboardRemove())
if bbs_name == "米游社":
help_message = (
"<b>关于如何获取Cookies</b>\n"
"<b>现在因为网站HttpOnly策略无法通过脚本获取，因此操作只能在PC上运行。</b>\n\n"
"PC：\n"
"1、打开<a href='https://user.mihoyo.com/'>通行证</a>或<a href='https://www.miyoushe.com/ys/'>社区</a>并登录\n"
"2、进入通行证按F12打开开发者工具\n"
"3、将开发者工具切换至网络(Network)并点击过滤栏中的文档(Document)并刷新页面\n"
"4、在请求列表中选择第一个并点击\n"
"5、找到并复制请求标头(Request Headers)中的<b>Cookie</b>\n"
"<u>如发现没有请求标头(Request Headers)大概因为缓存的存在需要你点击禁用缓存(Disable Cache)再次刷新页面</u>"
)
else:
javascript = (
"javascript:(()=>{_=(n)=>{for(i in(r=document.cookie.split(';'))){var a=r[i].split('=');if(a["
"0].trim()==n)return a[1]}};c=_('account_id')||alert('无效的Cookie,请重新登录!');c&&confirm("
"0].trim()==n)return a[1]}};c=_('login_ticket')||alert('无效的Cookie,请重新登录!');c&&confirm("
"'将Cookie复制到剪贴板?')&&copy(document.cookie)})(); "
)
javascript_android = "javascript:(()=>{prompt('',document.cookie)})();"
account_host = "https://user.mihoyo.com" if bbs_name == "米游社" else "https://account.hoyoverse.com"
help_message = (
f"<b>关于如何获取Cookies</b>\n\n"
f"PC：\n"
f"1、<a href='https://www.hoyolab.com/home'>打开社区并登录</a>\n"
"<b>关于如何获取Cookies</b>\n\n"
"PC：\n"
f"1、打开<a href='{account_host}'>通行证</a>并登录\n"
"2、按F12打开开发者工具\n"
"3、将开发者工具切换至控制台(Console)\n"
"4、复制下方的代码，并将其粘贴在控制台中，按下回车\n"
f"<pre><code class='javascript'>{javascript}</code></pre>\n"
"Android：\n"
f"1、<a href='https://www.hoyolab.com/home'>通过 Via 打开 {bbs_name} 并登录</a>\n"
f"1、通过 Via 打开<a href='{account_host}'>通行证</a>并登录\n"
"2、复制下方的代码，并将其粘贴在地址栏中，点击右侧箭头\n"
f"<code>{javascript_android}</code>\n"
"iOS：\n"
"1、在App Store上安装Web Inspector，并在iOS设置- Safari浏览器-扩展-允许这些扩展下找到Web Inspector-打开，允许所有网站\n"
f"2、<a href='https://www.hoyolab.com/home'>通过 Safari 打开 {bbs_name} 并登录</a>\n"
f"2、通过 Safari 打开<a href='{account_host}'>通行证</a>并登录\n"
"3、点击地址栏左侧的大小按钮 - Web Inspector扩展 - Console - 点击下方文本框复制下方代码粘贴：\n"
f"<pre><code class='javascript'>{javascript}</code></pre>\n"
"4、点击Console下的Execute"
@ -214,30 +202,46 @@ class AccountCookiesPlugin(Plugin.Conversation):
account_cookies_plugin_data: AccountCookiesPluginData = context.chat_data.get("account_cookies_plugin_data")
cookies = CookiesModel(**account_cookies_plugin_data.cookies)
if account_cookies_plugin_data.region == RegionEnum.HYPERION:
client = genshin.Client(cookies=cookies.to_dict(), region=types.Region.CHINESE, game=types.Game.STARRAIL)
region = Region.CHINESE
elif account_cookies_plugin_data.region == RegionEnum.HOYOLAB:
client = genshin.Client(cookies=cookies.to_dict(), region=types.Region.OVERSEAS, game=types.Game.STARRAIL)
region = Region.OVERSEAS
else:
logger.error("用户 %s[%s] region 异常", user.full_name, user.id)
await message.reply_text("数据错误", reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
if not cookies.check():
async with StarRailClient(cookies=cookies.to_dict(), region=region, lang="zh-cn") as client:
check_cookie = cookies.check()
if cookies.login_ticket is not None:
try:
cookies.cookie_token = await client.get_cookie_token_by_login_ticket()
cookies.account_id = client.account_id
cookies.ltuid = client.account_id
logger.success("用户 %s[%s] 绑定时获取 cookie_token 成功", user.full_name, user.id)
cookies.stoken = await client.get_stoken_by_login_ticket()
logger.success("用户 %s[%s] 绑定时获取 stoken 成功", user.full_name, user.id)
cookies.ltoken = await client.get_ltoken_by_stoken()
logger.success("用户 %s[%s] 绑定时获取 ltoken 成功", user.full_name, user.id)
check_cookie = True
except SimnetBadRequest as exc:
logger.warning("用户 %s[%s] 获取账号信息发生错误 [%s]%s", user.full_name, user.id, exc.ret_code, exc.original)
except Exception as exc:
logger.error("绑定时获取新Cookie失败 [%s]", (str(exc)))
finally:
cookies.login_ticket = None
cookies.login_uid = None
if not check_cookie:
await message.reply_text("检测到Cookie不完整，可能会出现问题。", reply_markup=ReplyKeyboardRemove())
try:
if client.cookie_manager.user_id is None and cookies.is_v2:
if client.account_id is None and cookies.is_v2:
logger.info("检测到用户 %s[%s] 使用 V2 Cookie 正在尝试获取 account_id", user.full_name, user.id)
if client.region == types.Region.CHINESE:
account_info = await client.get_hoyolab_user()
account_id = account_info.hoyolab_id
account_info = await client.get_user_info()
account_id = account_info.accident_id
account_cookies_plugin_data.account_id = account_id
cookies.set_v2_uid(account_id)
logger.success("获取用户 %s[%s] account_id[%s] 成功", user.full_name, user.id, account_id)
else:
logger.warning("用户 %s[%s] region[%s] 也许是不正确的", user.full_name, user.id, client.region.name)
else:
account_cookies_plugin_data.account_id = client.cookie_manager.user_id
accounts = await client.get_game_accounts()
starrail_accounts = [account for account in accounts if account.game == types.Game.STARRAIL]
account_cookies_plugin_data.account_id = client.account_id
starrail_accounts = await client.get_starrail_accounts()
except DataNotPublic:
logger.info("用户 %s[%s] 账号疑似被注销", user.full_name, user.id)
await message.reply_text("账号疑似被注销，请检查账号状态", reply_markup=ReplyKeyboardRemove())
@ -248,8 +252,8 @@ class AccountCookiesPlugin(Plugin.Conversation):
"获取账号信息失败，返回Cookies已经过期，请尝试在无痕浏览器中登录获取Cookies。", reply_markup=ReplyKeyboardRemove()
)
return ConversationHandler.END
except GenshinException as exc:
logger.info("用户 %s[%s] 获取账号信息发生错误 [%s]%s", user.full_name, user.id, exc.retcode, exc.original)
except SimnetBadRequest as exc:
logger.info("用户 %s[%s] 获取账号信息发生错误 [%s]%s", user.full_name, user.id, exc.ret_code, exc.original)
await message.reply_text(
f"获取账号信息发生错误，错误信息为 {exc.original}，请检查Cookie或者账号是否正常", reply_markup=ReplyKeyboardRemove()
)
@ -263,28 +267,10 @@ class AccountCookiesPlugin(Plugin.Conversation):
logger.debug("用户 %s[%s] Cookies错误", user.full_name, user.id, exc_info=exc)
await message.reply_text("Cookies错误，请检查是否正确", reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
if cookies.login_ticket is not None:
try:
if cookies.login_ticket is not None:
auth_client = AuthClient(cookies=cookies)
if await auth_client.get_stoken_by_login_ticket():
logger.success("用户 %s[%s] 绑定时获取 stoken 成功", user.full_name, user.id)
if await auth_client.get_cookie_token_by_stoken():
logger.success("用户 %s[%s] 绑定时获取 cookie_token 成功", user.full_name, user.id)
if await auth_client.get_ltoken_by_stoken():
logger.success("用户 %s[%s] 绑定时获取 ltoken 成功", user.full_name, user.id)
auth_client.cookies.remove_v2()
except Exception as exc: # pylint: disable=W0703
logger.error("绑定时获取新Cookie失败 [%s]", (str(exc)))
finally:
if cookies.user_id is not None:
account_cookies_plugin_data.account_id = cookies.user_id
cookies.login_ticket = None
cookies.login_uid = None
if account_cookies_plugin_data.account_id is None:
await message.reply_text("无法获取账号ID，请检查Cookie是否正确或请稍后重试")
return ConversationHandler.END
starrail_account: Optional[GenshinAccount] = None
starrail_account: Optional[Account] = None
level: int = 0
# todo : 多账号绑定
for temp in starrail_accounts:

11
plugins/account/players.py
View File

@ -2,6 +2,7 @@ import html
from http.cookies import SimpleCookie
from typing import Tuple, TYPE_CHECKING
from simnet import Region, StarRailClient
from telegram import InlineKeyboardMarkup, InlineKeyboardButton
from telegram.ext import filters
@ -10,7 +11,6 @@ from core.plugin import Plugin, handler
from core.services.cookies import CookiesService
from core.services.players import PlayersService
from core.services.players.services import PlayerInfoService
from modules.apihelper.client.components.authclient import AuthClient
from modules.apihelper.models.genshin.cookies import CookiesModel
from utils.log import logger
@ -211,12 +211,13 @@ class PlayersManagesPlugin(Plugin):
if cookies.stoken is not None:
try:
auth_client = AuthClient(cookies=cookies)
if await auth_client.get_cookie_token_by_stoken():
region = Region.CHINESE if player.region.value == 1 else Region.OVERSEAS
async with StarRailClient(cookies=cookies.to_dict(), region=region) as client:
cookies.cookie_token = await client.get_cookie_token_by_stoken()
logger.success("用户 %s[%s] 刷新 cookie_token 成功", user.full_name, user.id)
if await auth_client.get_ltoken_by_stoken():
cookies.ltoken = await client.get_ltoken_by_stoken()
logger.success("用户 %s[%s] 刷新 ltoken 成功", user.full_name, user.id)
auth_client.cookies.remove_v2()
cookies.remove_v2()
except Exception as exc: # pylint: disable=W0703
logger.error("刷新 cookie_token 失败 [%s]", (str(exc)))
await callback_query.edit_message_text(

24
plugins/app/start.py
View File

@ -7,6 +7,7 @@ from telegram.helpers import escape_markdown
from core.config import config
from core.plugin import handler, Plugin
from core.services.players import PlayersService
from plugins.tools.challenge import ChallengeSystem, ChallengeSystemException
from plugins.tools.genshin import PlayerNotFoundError, CookiesNotFoundError, GenshinHelper
from plugins.tools.sign import SignSystem, NeedChallenge
@ -14,10 +15,17 @@ from utils.log import logger
class StartPlugin(Plugin):
def __init__(self, sign_system: SignSystem, challenge_system: ChallengeSystem, genshin_helper: GenshinHelper):
def __init__(
self,
player: PlayersService,
sign_system: SignSystem,
challenge_system: ChallengeSystem,
genshin_helper: GenshinHelper,
):
self.challenge_system = challenge_system
self.sign_system = sign_system
self.genshin_helper = genshin_helper
self.players_service = player
@handler.command("start", block=False)
async def start(self, update: Update, context: CallbackContext) -> None:
@ -73,9 +81,9 @@ class StartPlugin(Plugin):
async def process_sign_validate(self, message: Message, user: User, validate: str):
try:
client = await self.genshin_helper.get_genshin_client(user.id)
async with self.genshin_helper.genshin(user.id) as client:
await message.reply_chat_action(ChatAction.TYPING)
_, challenge = await self.sign_system.get_challenge(client.uid)
_, challenge = await self.sign_system.get_challenge(client.player_id)
if not challenge:
await message.reply_text("验证请求已过期。", allow_sending_without_reply=True)
return
@ -115,13 +123,13 @@ class StartPlugin(Plugin):
)
async def get_sign_button(self, message: Message, user: User, bot_username: str):
try:
client = await self.genshin_helper.get_genshin_client(user.id)
player = await self.players_service.get_player(user.id)
if player is None:
logger.warning("用户 %s[%s] 账号信息未找到", user.full_name, user.id)
return
await message.reply_chat_action(ChatAction.TYPING)
button = await self.sign_system.get_challenge_button(bot_username, client.uid, user.id, callback=False)
button = await self.sign_system.get_challenge_button(bot_username, player.player_id, user.id, callback=False)
if not button:
await message.reply_text("验证请求已过期。", allow_sending_without_reply=True)
return
await message.reply_text("请尽快点击下方按钮进行验证。", allow_sending_without_reply=True, reply_markup=button)
except (PlayerNotFoundError, CookiesNotFoundError):
logger.warning("用户 %s[%s] 账号信息未找到", user.full_name, user.id)

25
plugins/starrail/activity.py
View File

@ -1,7 +1,7 @@
from typing import Optional, List, Dict
from typing import Optional, List, Dict, TYPE_CHECKING
from genshin import Client, GenshinException
from genshin.models import StarRailStarFight
from simnet.errors import BadRequest as SimnetBadRequest
from simnet.models.starrail.chronicle.activity import StarRailStarFight
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, Message
from telegram.constants import ChatAction
from telegram.ext import CallbackContext, filters
@ -15,6 +15,10 @@ from core.services.template.services import TemplateService
from plugins.tools.genshin import GenshinHelper, PlayerNotFoundError, CookiesNotFoundError
from utils.log import logger
if TYPE_CHECKING:
from simnet import StarRailClient
__all__ = ("PlayerActivityPlugins",)
@ -71,11 +75,12 @@ class PlayerActivityPlugins(Plugin):
try:
uid = await self.get_uid(user.id, context.args, message.reply_to_message)
try:
client = await self.helper.get_genshin_client(user.id)
if client.uid != uid:
async with self.helper.genshin(user.id) as client:
if client.player_id != uid:
raise CookiesNotFoundError(uid)
render_result = await self.star_fight_render(client, uid)
except CookiesNotFoundError:
client, _ = await self.helper.get_public_genshin_client(user.id)
async with self.helper.public_genshin(user.id) as client:
render_result = await self.star_fight_render(client, uid)
except PlayerNotFoundError:
buttons = [[InlineKeyboardButton("点我绑定账号", url=create_deep_linked_url(context.bot.username, "set_cookie"))]]
@ -88,7 +93,7 @@ class PlayerActivityPlugins(Plugin):
else:
await message.reply_text("未查询到您所绑定的账号信息，请先绑定账号", reply_markup=InlineKeyboardMarkup(buttons))
return
except GenshinException as exc:
except SimnetBadRequest as exc:
if exc.retcode == 1034:
await message.reply_text("出错了呜呜呜 ~ 请稍后重试")
return
@ -114,7 +119,7 @@ class PlayerActivityPlugins(Plugin):
self.add_delete_message_job(reply_message)
return
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await render_result.reply_photo(message, filename=f"{client.uid}.png", allow_sending_without_reply=True)
await render_result.reply_photo(message, filename=f"{user.id}.png", allow_sending_without_reply=True)
async def get_star_fight_rander_data(self, uid: int, data: StarRailStarFight) -> Dict:
if not data.exists_data:
@ -129,9 +134,9 @@ class PlayerActivityPlugins(Plugin):
"avatar_icons": avatar_icons,
}
async def star_fight_render(self, client: Client, uid: Optional[int] = None) -> RenderResult:
async def star_fight_render(self, client: "StarRailClient", uid: Optional[int] = None) -> RenderResult:
if uid is None:
uid = client.uid
uid = client.player_id
act_data = await client.get_starrail_activity(uid)
try:

24
plugins/starrail/avatar_list.py
View File

@ -1,8 +1,8 @@
from typing import List, Optional, TYPE_CHECKING
from genshin import Client, InvalidCookies, Game
from genshin.models.starrail.chronicle import StarRailDetailCharacter
from pydantic import BaseModel
from simnet.errors import InvalidCookies
from simnet.models.starrail.chronicle.characters import StarRailDetailCharacter
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.constants import ChatAction, ParseMode
from telegram.ext import filters
@ -18,6 +18,7 @@ from plugins.tools.genshin import CookiesNotFoundError, GenshinHelper, PlayerNot
from utils.log import logger
if TYPE_CHECKING:
from simnet import StarRailClient
from telegram.ext import ContextTypes
from telegram import Update
@ -58,7 +59,9 @@ class AvatarListPlugin(Plugin):
self.wiki_service = wiki_service
self.helper = helper
async def get_user_client(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE") -> Optional[Client]:
async def get_user_client(
self, update: "Update", context: "ContextTypes.DEFAULT_TYPE"
) -> Optional["StarRailClient"]:
message = update.effective_message
user = update.effective_user
try:
@ -91,7 +94,7 @@ class AvatarListPlugin(Plugin):
)
@staticmethod
async def get_avatars_data(client: Client) -> List[StarRailDetailCharacter]:
async def get_avatars_data(client: "StarRailClient") -> List[StarRailDetailCharacter]:
task_results = (await client.get_starrail_characters()).avatar_list
return sorted(
list(filter(lambda x: x, task_results)),
@ -150,21 +153,18 @@ class AvatarListPlugin(Plugin):
await message.reply_chat_action(ChatAction.TYPING)
try:
characters: List[StarRailDetailCharacter] = await self.get_avatars_data(client)
record_cards = await client.get_record_cards()
record_card = record_cards[0]
for card in record_cards:
if card.game == Game.STARRAIL:
record_card = card
break
record_card = await client.get_record_card()
nickname = record_card.nickname
except InvalidCookies as exc:
await client.get_starrail_user(client.uid)
await client.get_starrail_user(client.player_id)
logger.warning("用户 %s[%s] 无法请求角色数数据 API返回信息为 [%s]%s", user.full_name, user.id, exc.retcode, exc.original)
reply_message = await message.reply_text("出错了呜呜呜 ~ 当前访问令牌无法请求角色数数据，请尝试重新获取Cookie。")
if filters.ChatType.GROUPS.filter(message):
self.add_delete_message_job(reply_message, delay=30)
self.add_delete_message_job(message, delay=30)
return
finally:
await client.shutdown()
has_more = (not all_avatars) and len(characters) > 20
if has_more:
@ -172,7 +172,7 @@ class AvatarListPlugin(Plugin):
avatar_datas = await self.get_final_data(characters)
render_data = {
"uid": client.uid, # 玩家uid
"uid": client.player_id, # 玩家uid
"nickname": nickname, # 玩家昵称
"avatar_datas": avatar_datas, # 角色数据
"has_more": has_more, # 是否显示了全部角色

45
plugins/starrail/challenge.py
View File

@ -2,11 +2,11 @@
import asyncio
import re
from functools import lru_cache
from typing import Any, List, Optional, Tuple, Union
from typing import Any, List, Optional, Tuple, Union, TYPE_CHECKING
from arkowrapper import ArkoWrapper
from genshin import Client, GenshinException
from pytz import timezone
from simnet.errors import BadRequest as SimnetBadRequest
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Message, Update
from telegram.constants import ChatAction, ParseMode
from telegram.ext import CallbackContext, filters
@ -26,6 +26,9 @@ try:
except ImportError:
import json as jsonlib
if TYPE_CHECKING:
from simnet import StarRailClient
TZ = timezone("Asia/Shanghai")
cmd_pattern = r"(?i)^/challenge\s*((?:\d+)|(?:all))?\s*(pre)?"
@ -126,13 +129,26 @@ class ChallengePlugin(Plugin):
extra={"markup": True},
)
async def reply_message_func(content: str) -> None:
_reply_msg = await message.reply_text(f"开拓者 (<code>{uid}</code>) {content}", parse_mode=ParseMode.HTML)
reply_text: Optional[Message] = None
try:
try:
client = await self.helper.get_genshin_client(user.id)
if client.uid != uid:
async with self.helper.genshin(user.id) as client:
if client.player_id != uid:
raise CookiesNotFoundError(uid)
if total:
reply_text = await message.reply_text("彦卿需要时间整理混沌回忆数据，还请耐心等待哦~")
await message.reply_chat_action(ChatAction.TYPING)
images = await self.get_rendered_pic(client, uid, floor, total, previous)
except CookiesNotFoundError:
client, _ = await self.helper.get_public_genshin_client(user.id)
async with self.helper.public_genshin(user.id) as client:
if total:
reply_text = await message.reply_text("彦卿需要时间整理混沌回忆数据，还请耐心等待哦~")
await message.reply_chat_action(ChatAction.TYPING)
images = await self.get_rendered_pic(client, uid, floor, total, previous)
except PlayerNotFoundError: # 若未找到账号
buttons = [[InlineKeyboardButton("点我绑定账号", url=create_deep_linked_url(context.bot.username, "set_uid"))]]
if filters.ChatType.GROUPS.filter(message):
@ -150,27 +166,14 @@ class ChallengePlugin(Plugin):
self.add_delete_message_job(reply_message)
self.add_delete_message_job(message)
return
async def reply_message_func(content: str) -> None:
_reply_msg = await message.reply_text(f"开拓者 (<code>{uid}</code>) {content}", parse_mode=ParseMode.HTML)
reply_text: Optional[Message] = None
if total:
reply_text = await message.reply_text("彦卿需要时间整理混沌回忆数据，还请耐心等待哦~")
await message.reply_chat_action(ChatAction.TYPING)
try:
images = await self.get_rendered_pic(client, uid, floor, total, previous)
except AbyssUnlocked: # 若混沌回忆未解锁
await reply_message_func("还未解锁混沌回忆哦~")
return
except IndexError: # 若混沌回忆为挑战此层
await reply_message_func("还没有挑战本层呢，咕咕咕~")
return
except GenshinException as exc:
if exc.retcode == 1034 and client.uid != uid:
except SimnetBadRequest as exc:
if exc.retcode == 1034 and client.player_id != uid:
await message.reply_text("出错了呜呜呜 ~ 请稍后重试 ~ 米游社风控太严力")
return
raise exc
@ -216,7 +219,7 @@ class ChallengePlugin(Plugin):
return avatar_data
async def get_rendered_pic(
self, client: Client, uid: int, floor: int, total: bool, previous: bool
self, client: "StarRailClient", uid: int, floor: int, total: bool, previous: bool
) -> Union[
Tuple[
Union[BaseException, Any],

21
plugins/starrail/daily_note.py
View File

@ -1,9 +1,8 @@
import datetime
from datetime import datetime
from typing import Optional
from typing import Optional, TYPE_CHECKING
import genshin
from genshin import DataNotPublic
from simnet.errors import DataNotPublic
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import ChatAction
from telegram.ext import ConversationHandler, filters, CallbackContext
@ -15,6 +14,10 @@ from core.services.template.services import TemplateService
from plugins.tools.genshin import GenshinHelper, CookiesNotFoundError, PlayerNotFoundError
from utils.log import logger
if TYPE_CHECKING:
from simnet import StarRailClient
__all__ = ("DailyNotePlugin",)
@ -29,8 +32,8 @@ class DailyNotePlugin(Plugin):
self.template_service = template
self.helper = helper
async def _get_daily_note(self, client: genshin.Client) -> RenderResult:
daily_info = await client.get_starrail_notes(client.uid)
async def _get_daily_note(self, client: "StarRailClient") -> RenderResult:
daily_info = await client.get_starrail_notes(client.player_id)
day = datetime.now().strftime("%m-%d %H:%M") + " 星期" + "一二三四五六日"[datetime.now().weekday()]
resin_recovery_time = (
@ -50,7 +53,7 @@ class DailyNotePlugin(Plugin):
remained_time = (datetime.now().astimezone() + remained_time).strftime("%m-%d %H:%M")
render_data = {
"uid": client.uid,
"uid": client.player_id,
"day": day,
"resin_recovery_time": resin_recovery_time,
"current_resin": daily_info.current_stamina,
@ -77,9 +80,7 @@ class DailyNotePlugin(Plugin):
logger.info("用户 %s[%s] 每日便签命令请求", user.full_name, user.id)
try:
# 获取当前用户的 genshin.Client
client = await self.helper.get_genshin_client(user.id)
# 渲染
async with self.helper.genshin(user.id) as client:
render_result = await self._get_daily_note(client)
except (CookiesNotFoundError, PlayerNotFoundError):
buttons = [
@ -106,4 +107,4 @@ class DailyNotePlugin(Plugin):
return ConversationHandler.END
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await render_result.reply_photo(message, filename=f"{client.uid}.png", allow_sending_without_reply=True)
await render_result.reply_photo(message, filename=f"{client.player_id}.png", allow_sending_without_reply=True)

23
plugins/starrail/ledger.py
View File

@ -1,9 +1,10 @@
import os
import re
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
from genshin import DataNotPublic, InvalidCookies, GenshinException
from genshin.models.genshin.diary import StarRailDiary
from simnet.errors import BadRequest as SimnetBadRequest, DataNotPublic, InvalidCookies
from simnet.models.starrail.diary import StarRailDiary
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import ChatAction
from telegram.ext import filters, CallbackContext
@ -16,6 +17,10 @@ from core.services.template.services import TemplateService
from plugins.tools.genshin import CookiesNotFoundError, GenshinHelper, PlayerNotFoundError
from utils.log import logger
if TYPE_CHECKING:
from simnet import StarRailClient
__all__ = ("LedgerPlugin",)
@ -33,9 +38,9 @@ class LedgerPlugin(Plugin):
self.current_dir = os.getcwd()
self.helper = helper
async def _start_get_ledger(self, client, year, month) -> RenderResult:
async def _start_get_ledger(self, client: "StarRailClient", year, month) -> RenderResult:
req_month = f"{year}0{month}" if month < 10 else f"{year}{month}"
diary_info: StarRailDiary = await client.get_starrail_diary(client.uid, month=req_month)
diary_info: StarRailDiary = await client.get_starrail_diary(client.player_id, month=req_month)
color = ["#73a9c6", "#d56565", "#70b2b4", "#bd9a5a", "#739970", "#7a6da7", "#597ea0"]
categories = [
{
@ -53,7 +58,7 @@ class LedgerPlugin(Plugin):
return f"{round(amount / 10000, 2)}w" if amount >= 10000 else amount
ledger_data = {
"uid": client.uid,
"uid": client.player_id,
"day": month,
"current_hcoin": format_amount(diary_info.month_data.current_hcoin),
"gacha": int(diary_info.month_data.current_hcoin / 160),
@ -108,11 +113,11 @@ class LedgerPlugin(Plugin):
logger.info("用户 %s[%s] 查询开拓月历", user.full_name, user.id)
await message.reply_chat_action(ChatAction.TYPING)
try:
client = await self.helper.get_genshin_client(user.id)
async with self.helper.genshin(user.id) as client:
try:
render_result = await self._start_get_ledger(client, year, month)
except InvalidCookies as exc: # 如果抛出InvalidCookies 判断是否真的玄学过期（或权限不足？）
await client.get_starrail_user(client.uid)
await client.get_starrail_user(client.player_id)
logger.warning(
"用户 %s[%s] 无法请求开拓月历数据 API返回信息为 [%s]%s", user.full_name, user.id, exc.retcode, exc.original
)
@ -144,10 +149,10 @@ class LedgerPlugin(Plugin):
self.add_delete_message_job(reply_message, delay=30)
self.add_delete_message_job(message, delay=30)
return
except GenshinException as exc:
except SimnetBadRequest as exc:
if exc.retcode == -120:
await message.reply_text("当前角色开拓等级不足，暂时无法获取信息")
return
raise exc
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await render_result.reply_photo(message, filename=f"{client.uid}.png", allow_sending_without_reply=True)
await render_result.reply_photo(message, filename=f"{client.player_id}.png", allow_sending_without_reply=True)

45
plugins/starrail/museum.py
View File

@ -1,7 +1,7 @@
from typing import Optional, List, Dict
from typing import Optional, List, Dict, TYPE_CHECKING
from genshin import Client, GenshinException
from genshin.models import StarRailMuseumBasic, StarRailMuseumDetail
from simnet.errors import BadRequest as SimnetBadRequest
from simnet.models.starrail.chronicle.museum import StarRailMuseumBasic, StarRailMuseumDetail
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, Message
from telegram.constants import ChatAction
from telegram.ext import CallbackContext, filters
@ -15,13 +15,13 @@ from core.services.template.services import TemplateService
from plugins.tools.genshin import GenshinHelper, PlayerNotFoundError, CookiesNotFoundError
from utils.log import logger
if TYPE_CHECKING:
from simnet import StarRailClient
__all__ = ("PlayerMuseumPlugins",)
class NotSupport(Exception):
"""不支持的服务器"""
class NotHaveData(Exception):
"""没有数据"""
@ -70,15 +70,13 @@ class PlayerMuseumPlugins(Plugin):
logger.info("用户 %s[%s] 查询博物馆信息命令请求", user.full_name, user.id)
try:
uid = await self.get_uid(user.id, context.args, message.reply_to_message)
if uid and str(uid)[0] not in ["1", "2", "5"]:
# todo: 支持国际服
raise NotSupport
try:
client = await self.helper.get_genshin_client(user.id)
if client.uid != uid:
async with self.helper.genshin(user.id) as client:
if client.player_id != uid:
raise CookiesNotFoundError(uid)
render_result = await self.render(client, uid)
except CookiesNotFoundError:
client, _ = await self.helper.get_public_genshin_client(user.id)
async with self.helper.public_genshin(user.id) as client:
render_result = await self.render(client, uid)
except PlayerNotFoundError:
buttons = [[InlineKeyboardButton("点我绑定账号", url=create_deep_linked_url(context.bot.username, "set_cookie"))]]
@ -91,7 +89,7 @@ class PlayerMuseumPlugins(Plugin):
else:
await message.reply_text("未查询到您所绑定的账号信息，请先绑定账号", reply_markup=InlineKeyboardMarkup(buttons))
return
except GenshinException as exc:
except SimnetBadRequest as exc:
if exc.retcode == 1034:
await message.reply_text("出错了呜呜呜 ~ 请稍后重试")
return
@ -104,12 +102,6 @@ class PlayerMuseumPlugins(Plugin):
logger.exception(exc)
await message.reply_text("冬城博物珍奇簿数据有误 估计是彦卿晕了")
return
except NotSupport:
reply_message = await message.reply_text("暂不支持该服务器查询冬城博物珍奇簿数据")
if filters.ChatType.GROUPS.filter(reply_message):
self.add_delete_message_job(message)
self.add_delete_message_job(reply_message)
return
except NotHaveData:
reply_message = await message.reply_text("没有查找到冬城博物珍奇簿数据")
if filters.ChatType.GROUPS.filter(reply_message):
@ -117,9 +109,10 @@ class PlayerMuseumPlugins(Plugin):
self.add_delete_message_job(reply_message)
return
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await render_result.reply_photo(message, filename=f"{client.uid}.png", allow_sending_without_reply=True)
await render_result.reply_photo(message, filename=f"{user.id}.png", allow_sending_without_reply=True)
async def get_rander_data(self, uid: int, basic: StarRailMuseumBasic, detail: StarRailMuseumDetail) -> Dict:
@staticmethod
async def get_rander_data(uid: int, basic: StarRailMuseumBasic, detail: StarRailMuseumDetail) -> Dict:
exhibitions = []
for region in detail.regions:
for exhibition in region.exhibitions:
@ -132,14 +125,14 @@ class PlayerMuseumPlugins(Plugin):
"directors": detail.director,
}
async def render(self, client: Client, uid: Optional[int] = None) -> RenderResult:
async def render(self, client: "StarRailClient", uid: Optional[int] = None) -> RenderResult:
if uid is None:
uid = client.uid
uid = client.player_id
basic = await client.get_starrail_museum_info(uid)
try:
basic = await client.get_starrail_museum_info(uid)
detail = await client.get_starrail_museum_detail(uid)
except GenshinException as e:
except SimnetBadRequest as e:
if e.retcode == 10301:
raise NotHaveData from e
raise e

26
plugins/starrail/rogue.py
View File

@ -1,7 +1,8 @@
from typing import Optional, List, Dict, Tuple
from typing import Optional, List, Dict, Tuple, TYPE_CHECKING
from genshin import Client, GenshinException
from genshin.models import StarRailRogue, RogueCharacter
from simnet.errors import BadRequest as SimnetBadRequest
from simnet.models.starrail.character import RogueCharacter
from simnet.models.starrail.chronicle.rogue import StarRailRogue
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, Message
from telegram.constants import ChatAction
from telegram.ext import CallbackContext, filters
@ -15,6 +16,10 @@ from core.services.template.services import TemplateService
from plugins.tools.genshin import GenshinHelper, PlayerNotFoundError, CookiesNotFoundError
from utils.log import logger
if TYPE_CHECKING:
from simnet import StarRailClient
__all__ = ("PlayerRoguePlugins",)
@ -73,11 +78,12 @@ class PlayerRoguePlugins(Plugin):
try:
uid, pre = await self.get_uid(user.id, context.args, message.reply_to_message)
try:
client = await self.helper.get_genshin_client(user.id)
if client.uid != uid:
async with self.helper.genshin(user.id) as client:
if client.player_id != uid:
raise CookiesNotFoundError(uid)
render_result = await self.render(client, pre, uid)
except CookiesNotFoundError:
client, _ = await self.helper.get_public_genshin_client(user.id)
async with self.helper.public_genshin(user.id) as client:
render_result = await self.render(client, pre, uid)
except PlayerNotFoundError:
buttons = [[InlineKeyboardButton("点我绑定账号", url=create_deep_linked_url(context.bot.username, "set_cookie"))]]
@ -90,7 +96,7 @@ class PlayerRoguePlugins(Plugin):
else:
await message.reply_text("未查询到您所绑定的账号信息，请先绑定账号", reply_markup=InlineKeyboardMarkup(buttons))
return
except GenshinException as exc:
except SimnetBadRequest as exc:
if exc.retcode == 1034:
await message.reply_text("出错了呜呜呜 ~ 请稍后重试")
return
@ -116,7 +122,7 @@ class PlayerRoguePlugins(Plugin):
self.add_delete_message_job(reply_message)
return
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await render_result.reply_photo(message, filename=f"{client.uid}.png", allow_sending_without_reply=True)
await render_result.reply_photo(message, filename=f"{user.id}.png", allow_sending_without_reply=True)
async def get_rander_data(self, uid: int, data: StarRailRogue, pre: bool) -> Dict:
luo_ma_bum = ["", "Ⅰ", "Ⅱ", "Ⅲ", "Ⅳ", "Ⅴ", "Ⅵ"]
@ -155,9 +161,9 @@ class PlayerRoguePlugins(Plugin):
"miracles": record.miracles,
}
async def render(self, client: Client, pre: bool, uid: Optional[int] = None) -> RenderResult:
async def render(self, client: "StarRailClient", pre: bool, uid: Optional[int] = None) -> RenderResult:
if uid is None:
uid = client.uid
uid = client.player_id
rogue = await client.get_starrail_rogue(uid)
data = await self.get_rander_data(uid, rogue, pre)

4
plugins/starrail/sign.py
View File

@ -95,9 +95,9 @@ class Sign(Plugin):
try:
client = await self.genshin_helper.get_genshin_client(user.id)
await message.reply_chat_action(ChatAction.TYPING)
_, challenge = await self.sign_system.get_challenge(client.uid)
_, challenge = await self.sign_system.get_challenge(client.player_id)
if validate:
_, challenge = await self.sign_system.get_challenge(client.uid)
_, challenge = await self.sign_system.get_challenge(client.player_id)
if challenge:
sign_text = await self.sign_system.start_sign(client, challenge=challenge, validate=validate)
else:

25
plugins/starrail/stats.py
View File

@ -1,6 +1,6 @@
from typing import Optional, List
from typing import Optional, List, TYPE_CHECKING
from genshin import Client, GenshinException
from simnet.errors import BadRequest as SimnetBadRequest
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, Message
from telegram.constants import ChatAction
from telegram.ext import CallbackContext, filters
@ -13,6 +13,10 @@ from core.services.template.services import TemplateService
from plugins.tools.genshin import GenshinHelper, PlayerNotFoundError, CookiesNotFoundError
from utils.log import logger
if TYPE_CHECKING:
from simnet import StarRailClient
__all__ = ("PlayerStatsPlugins",)
@ -59,11 +63,12 @@ class PlayerStatsPlugins(Plugin):
try:
uid: int = await self.get_uid(user.id, context.args, message.reply_to_message)
try:
client = await self.helper.get_genshin_client(user.id)
if client.uid != uid:
async with self.helper.genshin(user.id) as client:
if client.player_id != uid:
raise CookiesNotFoundError(uid)
render_result = await self.render(client, uid)
except CookiesNotFoundError:
client, _ = await self.helper.get_public_genshin_client(user.id)
async with self.helper.public_genshin(user.id) as client:
render_result = await self.render(client, uid)
except PlayerNotFoundError:
buttons = [[InlineKeyboardButton("点我绑定账号", url=create_deep_linked_url(context.bot.username, "set_cookie"))]]
@ -76,7 +81,7 @@ class PlayerStatsPlugins(Plugin):
else:
await message.reply_text("未查询到您所绑定的账号信息，请先绑定账号", reply_markup=InlineKeyboardMarkup(buttons))
return
except GenshinException as exc:
except SimnetBadRequest as exc:
if exc.retcode == 1034:
await message.reply_text("出错了呜呜呜 ~ 请稍后重试")
return
@ -90,16 +95,16 @@ class PlayerStatsPlugins(Plugin):
await message.reply_text("角色数据有误 估计是彦卿晕了")
return
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await render_result.reply_photo(message, filename=f"{client.uid}.png", allow_sending_without_reply=True)
await render_result.reply_photo(message, filename=f"{user.id}.png", allow_sending_without_reply=True)
async def render(self, client: Client, uid: Optional[int] = None) -> RenderResult:
async def render(self, client: "StarRailClient", uid: Optional[int] = None) -> RenderResult:
if uid is None:
uid = client.uid
uid = client.player_id
user_info = await client.get_starrail_user(uid)
try:
rogue = await client.get_starrail_rogue(uid)
except GenshinException:
except SimnetBadRequest:
rogue = None
logger.debug(user_info)

64
plugins/starrail/wish_log.py
View File

@ -1,6 +1,7 @@
from io import BytesIO
from typing import Optional
from genshin.models.genshin.gacha import StarRailBannerType
from simnet.models.starrail.wish import StarRailBannerType
from telegram import Document, InlineKeyboardButton, InlineKeyboardMarkup, Message, Update, User
from telegram.constants import ChatAction
from telegram.ext import CallbackContext, CommandHandler, ConversationHandler, MessageHandler, filters
@ -22,7 +23,7 @@ from modules.gacha_log.error import (
)
from modules.gacha_log.helpers import from_url_get_authkey
from modules.gacha_log.log import GachaLog
from plugins.tools.genshin import PlayerNotFoundError, GenshinHelper
from plugins.tools.genshin import PlayerNotFoundError
from utils.log import logger
try:
@ -43,14 +44,20 @@ class WishLogPlugin(Plugin.Conversation):
players_service: PlayersService,
assets: AssetsService,
cookie_service: CookiesService,
helper: GenshinHelper,
):
self.template_service = template_service
self.players_service = players_service
self.assets_service = assets
self.cookie_service = cookie_service
self.gacha_log = GachaLog()
self.helper = helper
async def get_player_id(self, uid: int) -> Optional[int]:
"""获取绑定的游戏ID"""
logger.debug("尝试获取已绑定的星穹铁道账号")
player = await self.players_service.get_player(uid)
if player is None:
raise PlayerNotFoundError(uid)
return player.player_id
async def _refresh_user_data(
self, user: User, data: dict = None, authkey: str = None, verify_uid: bool = True
@ -63,12 +70,12 @@ class WishLogPlugin(Plugin.Conversation):
"""
try:
logger.debug("尝试获取已绑定的星穹铁道账号")
client = await self.helper.get_genshin_client(user.id, need_cookie=False)
player_id = await self.get_player_id(user.id)
if authkey:
new_num = await self.gacha_log.get_gacha_log_data(user.id, client, authkey)
new_num = await self.gacha_log.get_gacha_log_data(user.id, player_id, authkey)
return "更新完成，本次没有新增数据" if new_num == 0 else f"更新完成，本次共新增{new_num}条跃迁记录"
if data:
new_num = await self.gacha_log.import_gacha_log_data(user.id, client, data, verify_uid)
new_num = await self.gacha_log.import_gacha_log_data(user.id, player_id, data, verify_uid)
return "更新完成，本次没有新增数据" if new_num == 0 else f"更新完成，本次共新增{new_num}条跃迁记录"
except GachaLogNotFound:
return "彦卿没有找到你的跃迁记录，快来私聊彦卿导入吧~"
@ -136,23 +143,6 @@ class WishLogPlugin(Plugin.Conversation):
args = self.get_args(context)
logger.info("用户 %s[%s] 导入跃迁记录命令请求", user.full_name, user.id)
authkey = from_url_get_authkey(args[0] if args else "")
# if not args:
# player_info = await self.players_service.get_player(user.id, region=RegionEnum.HYPERION)
# if player_info is not None:
# cookies = await self.cookie_service.get(user.id, account_id=player_info.account_id)
# if cookies is not None and cookies.data and "stoken" in cookies.data:
# if stuid := next(
# (value for key, value in cookies.data.items() if key in ["ltuid", "login_uid"]), None
# ):
# cookies.data["stuid"] = stuid
# client = genshin.Client(
# cookies=cookies.data,
# game=genshin.types.Game.STARRAIL,
# region=genshin.Region.CHINESE,
# lang="zh-cn",
# uid=player_info.player_id,
# )
# authkey = await get_authkey_by_stoken(client)
if not authkey:
await message.reply_text(
"<b>开始导入跃迁历史记录：请通过 https://starrailstation.com/cn/warp#import 获取跃迁记录链接后发送给我"
@ -198,13 +188,13 @@ class WishLogPlugin(Plugin.Conversation):
user = update.effective_user
logger.info("用户 %s[%s] 删除跃迁记录命令请求", user.full_name, user.id)
try:
client = await self.helper.get_genshin_client(user.id, need_cookie=False)
context.chat_data["uid"] = client.uid
player_id = await self.get_player_id(user.id)
context.chat_data["uid"] = player_id
except PlayerNotFoundError:
logger.info("未查询到用户 %s[%s] 所绑定的账号信息", user.full_name, user.id)
await message.reply_text("未查询到您所绑定的账号信息，请先绑定账号")
return ConversationHandler.END
_, status = await self.gacha_log.load_history_info(str(user.id), str(client.uid), only_status=True)
_, status = await self.gacha_log.load_history_info(str(user.id), str(player_id), only_status=True)
if not status:
await message.reply_text("你还没有导入跃迁记录哦~")
return ConversationHandler.END
@ -234,12 +224,12 @@ class WishLogPlugin(Plugin.Conversation):
cid = int(args[0])
if cid < 0:
raise ValueError("Invalid cid")
client = await self.helper.get_genshin_client(cid, need_cookie=False)
_, status = await self.gacha_log.load_history_info(str(cid), str(client.uid), only_status=True)
player_id = await self.get_player_id(cid)
_, status = await self.gacha_log.load_history_info(str(cid), str(player_id), only_status=True)
if not status:
await message.reply_text("该用户还没有导入跃迁记录")
return
status = await self.gacha_log.remove_history_info(str(cid), str(client.uid))
status = await self.gacha_log.remove_history_info(str(cid), str(player_id))
await message.reply_text("跃迁记录已强制删除" if status else "跃迁记录删除失败")
except GachaLogNotFound:
await message.reply_text("该用户还没有导入跃迁记录")
@ -255,9 +245,9 @@ class WishLogPlugin(Plugin.Conversation):
user = update.effective_user
logger.info("用户 %s[%s] 导出跃迁记录命令请求", user.full_name, user.id)
try:
client = await self.helper.get_genshin_client(user.id, need_cookie=False)
await message.reply_chat_action(ChatAction.TYPING)
path = await self.gacha_log.gacha_log_to_srgf(str(user.id), str(client.uid))
player_id = await self.get_player_id(user.id)
path = await self.gacha_log.gacha_log_to_srgf(str(user.id), str(player_id))
await message.reply_chat_action(ChatAction.UPLOAD_DOCUMENT)
await message.reply_document(document=open(path, "rb+"), caption="跃迁记录导出文件 - SRGF V1.0")
except GachaLogNotFound:
@ -289,9 +279,9 @@ class WishLogPlugin(Plugin.Conversation):
pool_type = StarRailBannerType.NOVICE
logger.info("用户 %s[%s] 跃迁记录命令请求 || 参数 %s", user.full_name, user.id, pool_type.name)
try:
client = await self.helper.get_genshin_client(user.id, need_cookie=False)
await message.reply_chat_action(ChatAction.TYPING)
data = await self.gacha_log.get_analysis(user.id, client, pool_type, self.assets_service)
player_id = await self.get_player_id(user.id)
data = await self.gacha_log.get_analysis(user.id, player_id, pool_type, self.assets_service)
if isinstance(data, str):
reply_message = await message.reply_text(data)
if filters.ChatType.GROUPS.filter(message):
@ -346,13 +336,13 @@ class WishLogPlugin(Plugin.Conversation):
all_five = True
logger.info("用户 %s[%s] 跃迁统计命令请求 || 参数 %s || 仅五星 %s", user.full_name, user.id, pool_type.name, all_five)
try:
client = await self.helper.get_genshin_client(user.id, need_cookie=False)
group = filters.ChatType.GROUPS.filter(message)
await message.reply_chat_action(ChatAction.TYPING)
player_id = await self.get_player_id(user.id)
if all_five:
data = await self.gacha_log.get_all_five_analysis(user.id, client, self.assets_service)
data = await self.gacha_log.get_all_five_analysis(user.id, player_id, self.assets_service)
else:
data = await self.gacha_log.get_pool_analysis(user.id, client, pool_type, self.assets_service, group)
data = await self.gacha_log.get_pool_analysis(user.id, player_id, pool_type, self.assets_service, group)
if isinstance(data, str):
reply_message = await message.reply_text(data)
if filters.ChatType.GROUPS.filter(message):

27
plugins/system/errorhandler.py
View File

@ -5,8 +5,14 @@ from typing import Optional
import aiofiles
from aiohttp import ClientError, ClientConnectorError
from genshin import DataNotPublic, GenshinException, InvalidCookies, TooManyRequests
from httpx import HTTPError, TimeoutException
from simnet.errors import (
DataNotPublic,
BadRequest as SIMNetBadRequest,
InvalidCookies,
TooManyRequests,
CookieException,
)
from telegram import ReplyKeyboardRemove, Update, InlineKeyboardMarkup, InlineKeyboardButton
from telegram.constants import ParseMode
from telegram.error import BadRequest, Forbidden, TelegramError, TimedOut, NetworkError
@ -100,7 +106,7 @@ class ErrorHandler(Plugin):
@error_handler()
async def process_genshin_exception(self, update: object, context: CallbackContext):
if not isinstance(context.error, GenshinException) or not isinstance(update, Update):
if not isinstance(context.error, SIMNetBadRequest) or not isinstance(update, Update):
return
exc = context.error
notice: Optional[str] = None
@ -114,6 +120,12 @@ class ErrorHandler(Plugin):
else:
logger.error("未知Cookie错误", exc_info=exc)
notice = self.ERROR_MSG_PREFIX + f"Cookie 无效 错误信息为 {exc.original} 请尝试重新绑定"
elif isinstance(exc, CookieException):
if exc.retcode == 0:
notice = self.ERROR_MSG_PREFIX + "Cookie 已经被刷新，请尝试重试操作~"
else:
logger.error("未知Cookie错误", exc_info=exc)
notice = self.ERROR_MSG_PREFIX + f"Cookie 无效 错误信息为 {exc.original} 请尝试重新绑定"
elif isinstance(exc, DataNotPublic):
notice = self.ERROR_MSG_PREFIX + "查询的用户数据未公开"
else:
@ -124,14 +136,17 @@ class ErrorHandler(Plugin):
elif exc.retcode == -500001:
notice = self.ERROR_MSG_PREFIX + "网络出小差了，请稍后重试~"
elif exc.retcode == -1:
notice = self.ERROR_MSG_PREFIX + "系统发生错误，请稍后重试~"
logger.warning("内部数据库错误 [%s]%s", exc.ret_code, exc.original)
notice = self.ERROR_MSG_PREFIX + "系统内部数据库错误，请稍后重试~"
elif exc.retcode == -10001: # 参数异常 不应该抛出异常 进入下一步处理
pass
else:
logger.error("GenshinException", exc_info=exc)
notice = (
self.ERROR_MSG_PREFIX + f"获取账号信息发生错误 错误信息为 {exc.original if exc.original else exc.retcode} ~ 请稍后再试"
)
message = exc.original if exc.original else exc.message
if message:
notice = self.ERROR_MSG_PREFIX + f"获取信息发生错误 错误信息为 {message} ~ 请稍后再试"
else:
notice = self.ERROR_MSG_PREFIX + "获取信息发生错误 请稍后再试"
if notice:
self.create_notice_task(update, context, notice)
raise ApplicationHandlerStop

36
plugins/tools/challenge.py
View File

@ -1,13 +1,15 @@
from typing import Tuple, Optional
from genshin import Region, GenshinException
from simnet import Region
from simnet.errors import BadRequest as SIMNetBadRequest
from core.dependence.redisdb import RedisDB
from core.plugin import Plugin
from core.services.cookies import CookiesService
from core.services.players import PlayersService
from modules.apihelper.client.components.verify import Verify
from modules.apihelper.error import ResponseException, APIHelperException
from plugins.tools.genshin import GenshinHelper, PlayerNotFoundError, CookiesNotFoundError
from plugins.tools.genshin import PlayerNotFoundError, CookiesNotFoundError, GenshinHelper
from utils.log import logger
__all__ = ("ChallengeSystemException", "ChallengeSystem")
@ -25,11 +27,13 @@ class ChallengeSystem(Plugin):
cookies_service: CookiesService,
redis: RedisDB,
genshin_helper: GenshinHelper,
player: PlayersService,
) -> None:
self.cookies_service = cookies_service
self.genshin_helper = genshin_helper
self.cache = redis.client
self.qname = "plugin:challenge:"
self.players_service = player
async def get_challenge(self, uid: int) -> Tuple[Optional[str], Optional[str]]:
data = await self.cache.get(f"{self.qname}{uid}")
@ -56,12 +60,16 @@ class ChallengeSystem(Plugin):
if need_verify:
try:
await client.get_starrail_notes()
except GenshinException as exc:
except SIMNetBadRequest as exc:
if exc.retcode != 1034:
raise exc
else:
raise ChallengeSystemException("账户正常，无需验证")
verify = Verify(account_id=client.hoyolab_id, cookies=client.cookie_manager.cookies)
finally:
await client.shutdown()
else:
await client.shutdown()
verify = Verify(cookies=client.cookies)
try:
data = await verify.create()
challenge = data["challenge"]
@ -74,26 +82,28 @@ class ChallengeSystem(Plugin):
validate = await verify.ajax(referer="https://webstatic.mihoyo.com/", gt=gt, challenge=challenge)
if validate:
await verify.verify(challenge, validate)
return client.uid, "ajax", "ajax"
return client.player_id, "ajax", "ajax"
except APIHelperException as exc:
logger.warning("用户 %s ajax 验证失效 错误信息为 %s", user_id, str(exc))
logger.warning("用户 %s ajax 验证失败 重新申请验证", user_id)
return await self.create_challenge(user_id, need_verify, False)
await self.set_challenge(client.uid, gt, challenge)
return client.uid, gt, challenge
await self.set_challenge(client.player_id, gt, challenge)
return client.player_id, gt, challenge
async def pass_challenge(self, user_id: int, validate: str, challenge: Optional[str] = None) -> bool:
try:
client = await self.genshin_helper.get_genshin_client(user_id)
except PlayerNotFoundError:
player = await self.players_service.get_player(user_id)
if player is None:
raise ChallengeSystemException("用户未找到")
if client.region != Region.CHINESE:
if player.region != Region.CHINESE:
raise ChallengeSystemException("非法用户")
cookie_model = await self.cookies_service.get(player.user_id, player.account_id, player.region)
if cookie_model is None:
raise ChallengeSystemException("无需验证")
if challenge is None:
_, challenge = await self.get_challenge(client.uid)
_, challenge = await self.get_challenge(player.player_id)
if challenge is None:
raise ChallengeSystemException("验证失效 请求已经过期")
verify = Verify(account_id=client.hoyolab_id, cookies=client.cookie_manager.cookies)
verify = Verify(cookies=cookie_model.data)
try:
await verify.verify(challenge=challenge, validate=validate)
except ResponseException as exc:

224
plugins/tools/genshin.py
View File

@ -1,46 +1,37 @@
import asyncio
import random
from contextlib import asynccontextmanager
from datetime import datetime, time, timedelta
from typing import Optional, TYPE_CHECKING, Tuple, Union
from typing import Optional
from typing import TYPE_CHECKING, Union
import genshin
from genshin.errors import GenshinException
from genshin.models import BaseCharacter, CalculatorCharacterDetails
from pydantic import ValidationError
from simnet import StarRailClient, Region
from simnet.errors import BadRequest as SimnetBadRequest, InvalidCookies, NetworkError, CookieException
from simnet.models.genshin.calculator import CalculatorCharacterDetails
from simnet.models.genshin.chronicle.characters import Character
from sqlalchemy.exc import SQLAlchemyError
from sqlmodel import BigInteger, Column, DateTime, Field, Index, Integer, SQLModel, String, delete, func, select
from telegram.ext import ContextTypes
from core.basemodel import RegionEnum
from core.config import config
from core.dependence.database import Database
from core.dependence.redisdb import RedisDB
from core.error import ServiceNotFoundError
from core.plugin import Plugin
from core.services.cookies.services import CookiesService, PublicCookiesService
from core.services.devices import DevicesService
from core.services.players.services import PlayersService
from core.services.users.services import UserService
from core.sqlmodel.session import AsyncSession
from utils.const import REGION_MAP
from utils.log import logger
if TYPE_CHECKING:
from sqlalchemy import Table
from genshin import Client as GenshinClient
__all__ = ("GenshinHelper", "PlayerNotFoundError", "CookiesNotFoundError", "CharacterDetails")
class PlayerNotFoundError(Exception):
def __init__(self, user_id):
super().__init__(f"User not found, user_id: {user_id}")
class CookiesNotFoundError(Exception):
def __init__(self, user_id):
super().__init__(f"{user_id} cookies not found")
class CharacterDetailsSQLModel(SQLModel, table=True):
__tablename__ = "character_details"
__table_args__ = (
@ -164,82 +155,69 @@ class CharacterDetails(Plugin):
return None
async def get_character_details(
self, client: "GenshinClient", character: "Union[int,BaseCharacter]"
self, client: "StarRailClient", character: "Union[int,Character]"
) -> Optional["CalculatorCharacterDetails"]:
"""缓存 character_details 并定时对其进行数据存储 当遇到 Too Many Requests 可以获取以前的数据
:param client: genshin.py
:param character:
:return:
"""
uid = client.uid
if uid is not None:
if isinstance(character, BaseCharacter):
"""缓存 character_details 并定时对其进行数据存储 当遇到 Too Many Requests 可以获取以前的数据"""
uid = client.player_id
if isinstance(character, Character):
character_id = character.id
else:
character_id = character
if uid is not None:
detail = await self.get_character_details_for_redis(uid, character_id)
if detail is not None:
return detail
try:
detail = await client.get_character_details(character)
except GenshinException as exc:
if "Too Many Requests" in exc.msg:
detail = await client.get_character_details(character_id)
except SimnetBadRequest as exc:
if "Too Many Requests" in exc.message:
return await self.get_character_details_for_mysql(uid, character_id)
raise exc
asyncio.create_task(self.set_character_details(uid, character_id, detail.json()))
asyncio.create_task(self.set_character_details(uid, character_id, detail.json(by_alias=True)))
return detail
try:
return await client.get_character_details(character)
except GenshinException as exc:
if "Too Many Requests" in exc.msg:
return await client.get_character_details(character_id)
except SimnetBadRequest as exc:
if "Too Many Requests" in exc.message:
logger.warning("Too Many Requests")
else:
raise exc
return None
class PlayerNotFoundError(Exception):
def __init__(self, user_id):
super().__init__(f"User not found, user_id: {user_id}")
class CookiesNotFoundError(Exception):
def __init__(self, user_id):
super().__init__(f"{user_id} cookies not found")
class GenshinHelper(Plugin):
def __init__(
self,
cookies: CookiesService,
public_cookies: PublicCookiesService,
user: UserService,
redis: RedisDB,
player: PlayersService,
devices: DevicesService,
) -> None:
self.cookies_service = cookies
self.public_cookies_service = public_cookies
self.user_service = user
self.redis_db = redis
self.players_service = player
if self.redis_db and config.genshin_ttl:
self.genshin_cache = genshin.RedisCache(self.redis_db.client, ttl=config.genshin_ttl)
else:
self.genshin_cache = None
self.devices_service = devices
if None in (temp := [self.user_service, self.cookies_service, self.players_service]):
raise ServiceNotFoundError(*filter(lambda x: x is None, temp))
@staticmethod
def region_server(uid: Union[int, str]) -> RegionEnum:
if isinstance(uid, (int, str)):
region = REGION_MAP.get(str(uid)[0])
else:
raise TypeError("UID variable type error")
if region:
return region
raise ValueError(f"UID {uid} isn't associated with any region.")
async def get_genshin_client(
self, user_id: int, region: Optional[RegionEnum] = None, need_cookie: bool = True
) -> Optional[genshin.Client]:
"""通过 user_id 和 region 获取私有的 `genshin.Client`"""
@asynccontextmanager
async def genshin(self, user_id: int, region: Optional[RegionEnum] = None) -> StarRailClient:
player = await self.players_service.get_player(user_id, region)
if player is None:
raise PlayerNotFoundError(user_id)
cookies = None
if need_cookie:
if player.account_id is None:
raise CookiesNotFoundError(user_id)
cookie_model = await self.cookies_service.get(player.user_id, player.account_id, player.region)
@ -247,49 +225,121 @@ class GenshinHelper(Plugin):
raise CookiesNotFoundError(user_id)
cookies = cookie_model.data
uid = player.player_id
region = player.region
if region == RegionEnum.HYPERION: # 国服
game_region = genshin.types.Region.CHINESE
elif region == RegionEnum.HOYOLAB: # 国际服
game_region = genshin.types.Region.OVERSEAS
if player.region == RegionEnum.HYPERION: # 国服
region = Region.CHINESE
elif player.region == RegionEnum.HOYOLAB: # 国际服
region = Region.OVERSEAS
else:
raise TypeError("Region is not None")
client = genshin.Client(
device_id: Optional[str] = None
device_fp: Optional[str] = None
devices = await self.devices_service.get(player.account_id)
if devices:
device_id = devices.device_id
device_fp = devices.device_fp
async with StarRailClient(
cookies,
region=region,
account_id=player.account_id,
player_id=player.player_id,
lang="zh-cn",
game=genshin.types.Game.STARRAIL,
region=game_region,
uid=uid,
hoyolab_id=player.account_id,
device_id=device_id,
device_fp=device_fp,
) as client:
try:
yield client
except InvalidCookies as exc:
stoken = client.cookies.get("stoken")
if stoken is not None:
try:
await client.get_cookie_token_by_stoken()
logger.success("用户 %s 刷新 cookie_token 成功", user_id)
await client.get_ltoken_by_stoken()
logger.success("用户 %s 刷新 ltoken 成功", user_id)
except SimnetBadRequest as _exc:
logger.warning(
"用户 %s 刷新 token 失败 [%s]%s", user_id, _exc.ret_code, _exc.original or _exc.message
)
except NetworkError:
logger.warning("用户 %s 刷新 Cookies 失败 网络错误", user_id)
except Exception as _exc:
logger.error("用户 %s 刷新 Cookies 失败", user_id, exc_info=_exc)
else:
raise CookieException(message="The cookie has been refreshed.") from exc
raise exc
async def get_genshin_client(self, user_id: int, region: Optional[RegionEnum] = None) -> StarRailClient:
player = await self.players_service.get_player(user_id, region)
if player is None:
raise PlayerNotFoundError(user_id)
if player.account_id is None:
raise CookiesNotFoundError(user_id)
cookie_model = await self.cookies_service.get(player.user_id, player.account_id, player.region)
if cookie_model is None:
raise CookiesNotFoundError(user_id)
cookies = cookie_model.data
if player.region == RegionEnum.HYPERION:
region = Region.CHINESE
elif player.region == RegionEnum.HOYOLAB:
region = Region.OVERSEAS
else:
raise TypeError("Region is not None")
device_id: Optional[str] = None
device_fp: Optional[str] = None
devices = await self.devices_service.get(player.account_id)
if devices:
device_id = devices.device_id
device_fp = devices.device_fp
return StarRailClient(
cookies,
region=region,
account_id=player.account_id,
player_id=player.player_id,
lang="zh-cn",
device_id=device_id,
device_fp=device_fp,
)
if self.genshin_cache is not None:
client.cache = self.genshin_cache
return client
async def get_public_genshin_client(self, user_id: int) -> Tuple[genshin.Client, int]:
"""通过 user_id 获取公共的 `genshin.Client`"""
player = await self.players_service.get_player(user_id)
@asynccontextmanager
async def public_genshin(self, user_id: int, region: Optional[RegionEnum] = None) -> StarRailClient:
player = await self.players_service.get_player(user_id, region)
region = player.region
cookies = await self.public_cookies_service.get_cookies(user_id, region)
uid = player.player_id
if region is RegionEnum.HYPERION:
game_region = genshin.types.Region.CHINESE
elif region is RegionEnum.HOYOLAB:
game_region = genshin.types.Region.OVERSEAS
if player.region == RegionEnum.HYPERION:
region = Region.CHINESE
elif player.region == RegionEnum.HOYOLAB:
region = Region.OVERSEAS
else:
raise TypeError("Region is not `RegionEnum.NULL`")
client = genshin.Client(
cookies.data, region=game_region, uid=uid, game=genshin.types.Game.STARRAIL, lang="zh-cn"
)
device_id: Optional[str] = None
device_fp: Optional[str] = None
devices = await self.devices_service.get(cookies.account_id)
if devices:
device_id = devices.device_id
device_fp = devices.device_fp
if self.genshin_cache is not None:
client.cache = self.genshin_cache
return client, uid
async with StarRailClient(
cookies.data,
region=region,
account_id=player.account_id,
player_id=uid,
lang="zh-cn",
device_id=device_id,
device_fp=device_fp,
) as client:
try:
yield client
except SimnetBadRequest as exc:
if exc.ret_code == 1034:
await self.public_cookies_service.undo(user_id)
raise exc

74
plugins/tools/sign.py
View File

@ -6,9 +6,10 @@ from enum import Enum
from typing import Optional, Tuple, List, TYPE_CHECKING
from aiohttp import ClientConnectorError
from genshin import Game, GenshinException, AlreadyClaimed, Client, InvalidCookies
from genshin.utility import recognize_genshin_server
from httpx import TimeoutException
from simnet import Game
from simnet.errors import BadRequest as SimnetBadRequest, AlreadyClaimed, InvalidCookies
from simnet.utils.player import recognize_starrail_server
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.constants import ParseMode
from telegram.error import Forbidden, BadRequest
@ -21,11 +22,12 @@ from core.services.sign.models import SignStatusEnum
from core.services.sign.services import SignServices
from core.services.users.services import UserService
from modules.apihelper.client.components.verify import Verify
from plugins.tools.genshin import GenshinHelper, CookiesNotFoundError, PlayerNotFoundError
from plugins.tools.genshin import PlayerNotFoundError, CookiesNotFoundError, GenshinHelper
from plugins.tools.recognize import RecognizeSystem
from utils.log import logger
if TYPE_CHECKING:
from simnet import StarRailClient
from telegram.ext import ContextTypes
@ -104,7 +106,7 @@ class SignSystem(Plugin):
async def start_sign(
self,
client: Client,
client: "StarRailClient",
challenge: Optional[str] = None,
validate: Optional[str] = None,
is_sleep: bool = False,
@ -112,28 +114,28 @@ class SignSystem(Plugin):
title: Optional[str] = "签到结果",
) -> str:
if is_sleep:
if recognize_genshin_server(client.uid) in ("cn_gf01", "cn_qd01"):
if recognize_starrail_server(client.player_id) in ("prod_gf_cn", "prod_qd_cn"):
await asyncio.sleep(random.randint(10, 300)) # nosec
else:
await asyncio.sleep(random.randint(0, 3)) # nosec
try:
rewards = await client.get_monthly_rewards(game=Game.STARRAIL, lang="zh-cn")
except GenshinException as error:
logger.warning("UID[%s] 获取签到信息失败，API返回信息为 %s", client.uid, str(error))
except SimnetBadRequest as error:
logger.warning("UID[%s] 获取签到信息失败，API返回信息为 %s", client.player_id, str(error))
if is_raise:
raise error
return f"获取签到信息失败，API返回信息为 {str(error)}"
try:
daily_reward_info = await client.get_reward_info(game=Game.STARRAIL, lang="zh-cn") # 获取签到信息失败
except GenshinException as error:
logger.warning("UID[%s] 获取签到状态失败，API返回信息为 %s", client.uid, str(error))
except SimnetBadRequest as error:
logger.warning("UID[%s] 获取签到状态失败，API返回信息为 %s", client.player_id, str(error))
if is_raise:
raise error
return f"获取签到状态失败，API返回信息为 {str(error)}"
if not daily_reward_info.signed_in:
try:
if validate:
logger.info("UID[%s] 正在尝试通过验证码\nchallenge[%s]\nvalidate[%s]", client.uid, challenge, validate)
logger.info("UID[%s] 正在尝试通过验证码\nchallenge[%s]\nvalidate[%s]", client.player_id, challenge, validate)
request_daily_reward = await client.request_daily_reward(
"sign",
method="POST",
@ -147,8 +149,8 @@ class SignSystem(Plugin):
# 尝试通过 ajax 请求绕过签到
gt = request_daily_reward.get("gt", "")
challenge = request_daily_reward.get("challenge", "")
logger.warning("UID[%s] 触发验证码\ngt[%s]\nchallenge[%s]", client.uid, gt, challenge)
self.verify.account_id = client.hoyolab_id
logger.warning("UID[%s] 触发验证码\ngt[%s]\nchallenge[%s]", client.player_id, gt, challenge)
self.verify.account_id = client.account_id
validate = await self.verify.ajax(
referer=RecognizeSystem.REFERER,
gt=gt,
@ -166,16 +168,16 @@ class SignSystem(Plugin):
)
logger.debug("request_daily_reward 返回 %s", request_daily_reward)
if request_daily_reward and request_daily_reward.get("success", 0) == 1:
logger.warning("UID[%s] 触发验证码\nchallenge[%s]", client.uid, challenge)
logger.warning("UID[%s] 触发验证码\nchallenge[%s]", client.player_id, challenge)
raise NeedChallenge(
uid=client.uid,
uid=client.player_id,
gt=request_daily_reward.get("gt", ""),
challenge=request_daily_reward.get("challenge", ""),
)
elif config.pass_challenge_app_key:
# 如果无法绕过 检查配置文件是否配置识别 API 尝试请求绕过
# 注意 需要重新获取没有进行任何请求的 Challenge
logger.info("UID[%s] 正在使用 recognize 重新请求签到", client.uid)
logger.info("UID[%s] 正在使用 recognize 重新请求签到", client.player_id)
_request_daily_reward = await client.request_daily_reward(
"sign",
method="POST",
@ -186,8 +188,8 @@ class SignSystem(Plugin):
if _request_daily_reward and _request_daily_reward.get("success", 0) == 1:
_gt = _request_daily_reward.get("gt", "")
_challenge = _request_daily_reward.get("challenge", "")
logger.info("UID[%s] 创建验证码\ngt[%s]\nchallenge[%s]", client.uid, _gt, _challenge)
_validate = await RecognizeSystem.recognize(_gt, _challenge, uid=client.uid)
logger.info("UID[%s] 创建验证码\ngt[%s]\nchallenge[%s]", client.player_id, _gt, _challenge)
_validate = await RecognizeSystem.recognize(_gt, _challenge, uid=client.player_id)
if _validate:
logger.success("recognize 通过验证成功\nchallenge[%s]\nvalidate[%s]", _challenge, _validate)
request_daily_reward = await client.request_daily_reward(
@ -199,55 +201,57 @@ class SignSystem(Plugin):
validate=_validate,
)
if request_daily_reward and request_daily_reward.get("success", 0) == 1:
logger.warning("UID[%s] 触发验证码\nchallenge[%s]", client.uid, _challenge)
logger.warning("UID[%s] 触发验证码\nchallenge[%s]", client.player_id, _challenge)
gt = request_daily_reward.get("gt", "")
challenge = request_daily_reward.get("challenge", "")
logger.success("UID[%s] 创建验证成功\ngt[%s]\nchallenge[%s]", client.uid, gt, challenge)
logger.success(
"UID[%s] 创建验证成功\ngt[%s]\nchallenge[%s]", client.player_id, gt, challenge
)
raise NeedChallenge(
uid=client.uid,
uid=client.player_id,
gt=gt,
challenge=challenge,
)
logger.success("UID[%s] 通过 recognize 签到成功", client.uid)
logger.success("UID[%s] 通过 recognize 签到成功", client.player_id)
else:
request_daily_reward = await client.request_daily_reward(
"sign", method="POST", game=Game.STARRAIL, lang="zh-cn"
)
gt = request_daily_reward.get("gt", "")
challenge = request_daily_reward.get("challenge", "")
logger.success("UID[%s] 创建验证成功\ngt[%s]\nchallenge[%s]", client.uid, gt, challenge)
raise NeedChallenge(uid=client.uid, gt=gt, challenge=challenge)
logger.success("UID[%s] 创建验证成功\ngt[%s]\nchallenge[%s]", client.player_id, gt, challenge)
raise NeedChallenge(uid=client.player_id, gt=gt, challenge=challenge)
else:
request_daily_reward = await client.request_daily_reward(
"sign", method="POST", game=Game.STARRAIL, lang="zh-cn"
)
gt = request_daily_reward.get("gt", "")
challenge = request_daily_reward.get("challenge", "")
logger.success("UID[%s] 创建验证成功\ngt[%s]\nchallenge[%s]", client.uid, gt, challenge)
raise NeedChallenge(uid=client.uid, gt=gt, challenge=challenge)
logger.success("UID[%s] 创建验证成功\ngt[%s]\nchallenge[%s]", client.player_id, gt, challenge)
raise NeedChallenge(uid=client.player_id, gt=gt, challenge=challenge)
else:
logger.success("UID[%s] 签到成功", client.uid)
logger.success("UID[%s] 签到成功", client.player_id)
except TimeoutException as error:
logger.warning("UID[%s] 签到请求超时", client.uid)
logger.warning("UID[%s] 签到请求超时", client.player_id)
if is_raise:
raise error
return "签到失败了呜呜呜 ~ 服务器连接超时 服务器熟啦 ~ "
except AlreadyClaimed as error:
logger.warning("UID[%s] 已经签到", client.uid)
logger.warning("UID[%s] 已经签到", client.player_id)
if is_raise:
raise error
result = "今天开拓者已经签到过了~"
except GenshinException as error:
logger.warning("UID %s 签到失败，API返回信息为 %s", client.uid, str(error))
except SimnetBadRequest as error:
logger.warning("UID %s 签到失败，API返回信息为 %s", client.player_id, str(error))
if is_raise:
raise error
return f"获取签到状态失败，API返回信息为 {str(error)}"
else:
result = "OK"
else:
logger.info("UID[%s] 已经签到", client.uid)
logger.info("UID[%s] 已经签到", client.player_id)
result = "今天开拓者已经签到过了~"
logger.info("UID[%s] 签到结果 %s", client.uid, result)
logger.info("UID[%s] 签到结果 %s", client.player_id, result)
reward = rewards[daily_reward_info.claimed_rewards - (1 if daily_reward_info.signed_in else 0)]
today = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
cn_timezone = datetime.timezone(datetime.timedelta(hours=8))
@ -258,7 +262,7 @@ class SignSystem(Plugin):
message = (
f"#### {title} ####\n"
f"时间：{today} (UTC+8)\n"
f"UID: {client.uid}\n"
f"UID: {client.player_id}\n"
f"今日奖励: {reward.name} × {reward.amount}\n"
f"本月漏签次数：{missed_days}\n"
f"签到结果: {result}"
@ -284,7 +288,7 @@ class SignSystem(Plugin):
continue
user_id = sign_db.user_id
try:
client = await self.genshin_helper.get_genshin_client(user_id)
async with self.genshin_helper.genshin(user_id) as client:
text = await self.start_sign(client, is_sleep=True, is_raise=True, title=title)
except InvalidCookies:
text = "自动签到执行失败，Cookie无效"
@ -292,7 +296,7 @@ class SignSystem(Plugin):
except AlreadyClaimed:
text = "今天开拓者已经签到过了~"
sign_db.status = SignStatusEnum.ALREADY_CLAIMED
except GenshinException as exc:
except SimnetBadRequest as exc:
text = f"自动签到执行失败，API返回信息为 {str(exc)}"
sign_db.status = SignStatusEnum.GENSHIN_EXCEPTION
except ClientConnectorError:

1942
poetry.lock generated
View File

File diff suppressed because it is too large Load Diff

28
pyproject.toml
View File

@ -1,30 +1,29 @@
[tool.poetry]
name = "PaiGram"
name = "PamGram"
version = "0.4.0"
description = "Telegarm robot, query the official genshin information."
description = "Telegarm robot, query the official starrail information."
authors = ["洛水居室", "zhxy-CN", "Chuangbo Li", "kotoriのねこ", "omg-xtao", "艾迪", "Karako", "SiHuaN"]
license = "AGPL-3.0"
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.8"
httpx = "^0.23.3"
ujson = "^5.7.0"
genshin = { git = "https://github.com/thesadru/genshin.py" }
httpx = "^0.24.0"
ujson = "^5.8.0"
Jinja2 = "^3.1.2"
python-telegram-bot = { version = "^20.1", extras = ["ext", "rate-limiter"] }
sqlmodel = "^0.0.8"
colorlog = "^6.7.0"
fakeredis = "^2.10.3"
redis = "^4.5.4"
fakeredis = "^2.16.0"
redis = "^4.6.0"
beautifulsoup4 = "^4.12.1"
asyncmy = "^0.2.7"
pyppeteer = "^1.0.2"
aiofiles = "^23.1.0"
python-dotenv = "^1.0.0"
alembic = "^1.10.3"
alembic = "^1.11.1"
black = "^23.3.0"
rich = "^13.3.1"
rich = "^13.4.1"
enkanetwork-py = { git = "https://github.com/mrwan200/EnkaNetwork.py" }
TgCrypto = { version = "^1.2.5", optional = true }
Pyrogram = { version = "^2.0.102", optional = true }
@ -33,18 +32,19 @@ pytest-asyncio = { version = "^0.21.0", optional = true }
flaky = { version = "^3.7.0", optional = true }
lxml = "^4.9.2"
arko-wrapper = "^0.2.8"
fastapi = "^0.95.0"
uvicorn = { extras = ["standard"], version = "^0.21.1" }
sentry-sdk = "^1.19.1"
fastapi = "<0.100.0"
uvicorn = { extras = ["standard"], version = "^0.23.1" }
sentry-sdk = "^1.28.1"
GitPython = "^3.1.30"
openpyxl = "^3.1.1"
async-lru = "^2.0.2"
thefuzz = "^0.19.0"
qrcode = "^7.4.2"
cryptography = "^40.0.1"
pillow = "^9.4.0"
cryptography = "^41.0.2"
pillow = "^10.0.0"
playwright = "^1.27.1"
aiosqlite = { extras = ["sqlite"], version = "^0.19.0" }
simnet = { git = "https://github.com/PaiGramTeam/SIMNet" }
[tool.poetry.extras]
pyro = ["Pyrogram", "TgCrypto"]

104
requirements.txt
View File

@ -1,106 +1,104 @@
aiofiles==23.1.0 ; python_version >= "3.8" and python_version < "4.0"
aiohttp==3.8.4 ; python_version >= "3.8" and python_version < "4.0"
aiolimiter==1.0.0 ; python_version >= "3.8" and python_version < "4.0"
aiolimiter==1.1.0 ; python_version >= "3.8" and python_version < "4.0"
aiosignal==1.3.1 ; python_version >= "3.8" and python_version < "4.0"
aiosqlite[sqlite]==0.19.0 ; python_version >= "3.8" and python_version < "4.0"
alembic==1.10.3 ; python_version >= "3.8" and python_version < "4.0"
anyio==3.6.2 ; python_version >= "3.8" and python_version < "4.0"
alembic==1.11.1 ; python_version >= "3.8" and python_version < "4.0"
anyio==3.7.1 ; python_version >= "3.8" and python_version < "4.0"
appdirs==1.4.4 ; python_version >= "3.8" and python_version < "4.0"
apscheduler==3.10.1 ; python_version >= "3.8" and python_version < "4.0"
arko-wrapper==0.2.8 ; python_version >= "3.8" and python_version < "4.0"
async-lru==2.0.2 ; python_version >= "3.8" and python_version < "4.0"
async-lru==2.0.3 ; python_version >= "3.8" and python_version < "4.0"
async-timeout==4.0.2 ; python_version >= "3.8" and python_version < "4.0"
asyncmy==0.2.7 ; python_version >= "3.8" and python_version < "4.0"
asyncmy==0.2.8 ; python_version >= "3.8" and python_version < "4.0"
attrs==23.1.0 ; python_version >= "3.8" and python_version < "4.0"
backports-zoneinfo==0.2.1 ; python_version >= "3.8" and python_version < "3.9"
beautifulsoup4==4.12.2 ; python_version >= "3.8" and python_version < "4.0"
black==23.3.0 ; python_version >= "3.8" and python_version < "4.0"
cachetools==5.3.0 ; python_version >= "3.8" and python_version < "4.0"
certifi==2022.12.7 ; python_version >= "3.8" and python_version < "4.0"
black==23.7.0 ; python_version >= "3.8" and python_version < "4.0"
cachetools==5.3.1 ; python_version >= "3.8" and python_version < "4.0"
certifi==2023.5.7 ; python_version >= "3.8" and python_version < "4.0"
cffi==1.15.1 ; python_version >= "3.8" and python_version < "4.0"
charset-normalizer==3.1.0 ; python_version >= "3.8" and python_version < "4.0"
click==8.1.3 ; python_version >= "3.8" and python_version < "4.0"
colorama==0.4.6 ; python_version >= "3.8" and python_version < "4.0" and sys_platform == "win32" or python_version >= "3.8" and python_version < "4.0" and platform_system == "Windows"
charset-normalizer==3.2.0 ; python_version >= "3.8" and python_version < "4.0"
click==8.1.5 ; python_version >= "3.8" and python_version < "4.0"
colorama==0.4.6 ; python_version >= "3.8" and python_version < "4.0" and (sys_platform == "win32" or platform_system == "Windows")
colorlog==6.7.0 ; python_version >= "3.8" and python_version < "4.0"
cryptography==40.0.2 ; python_version >= "3.8" and python_version < "4.0"
cryptography==41.0.2 ; python_version >= "3.8" and python_version < "4.0"
enkanetwork-py @ git+https://github.com/mrwan200/EnkaNetwork.py@master ; python_version >= "3.8" and python_version < "4.0"
et-xmlfile==1.1.0 ; python_version >= "3.8" and python_version < "4.0"
exceptiongroup==1.1.1 ; python_version >= "3.8" and python_version < "3.11"
fakeredis==2.10.3 ; python_version >= "3.8" and python_version < "4.0"
fastapi==0.95.1 ; python_version >= "3.8" and python_version < "4.0"
exceptiongroup==1.1.2 ; python_version >= "3.8" and python_version < "3.11"
fakeredis==2.16.0 ; python_version >= "3.8" and python_version < "4.0"
fastapi==0.99.1 ; python_version >= "3.8" and python_version < "4.0"
flaky==3.7.0 ; python_version >= "3.8" and python_version < "4.0"
frozenlist==1.3.3 ; python_version >= "3.8" and python_version < "4.0"
genshin @ git+https://github.com/thesadru/genshin.py@master ; python_version >= "3.8" and python_version < "4.0"
frozenlist==1.4.0 ; python_version >= "3.8" and python_version < "4.0"
gitdb==4.0.10 ; python_version >= "3.8" and python_version < "4.0"
gitpython==3.1.31 ; python_version >= "3.8" and python_version < "4.0"
gitpython==3.1.32 ; python_version >= "3.8" and python_version < "4.0"
greenlet==1.1.3 ; python_version >= "3.8" and python_version < "4.0"
h11==0.14.0 ; python_version >= "3.8" and python_version < "4.0"
httpcore==0.16.3 ; python_version >= "3.8" and python_version < "4.0"
httptools==0.5.0 ; python_version >= "3.8" and python_version < "4.0"
httpx==0.23.3 ; python_version >= "3.8" and python_version < "4.0"
httpcore==0.17.3 ; python_version >= "3.8" and python_version < "4.0"
httptools==0.6.0 ; python_version >= "3.8" and python_version < "4.0"
httpx==0.24.1 ; python_version >= "3.8" and python_version < "4.0"
idna==3.4 ; python_version >= "3.8" and python_version < "4.0"
importlib-metadata==6.5.0 ; python_version >= "3.8" and python_version < "4.0"
importlib-resources==5.12.0 ; python_version >= "3.8" and python_version < "3.9"
importlib-metadata==6.8.0 ; python_version >= "3.8" and python_version < "4.0"
importlib-resources==6.0.0 ; python_version >= "3.8" and python_version < "3.9"
iniconfig==2.0.0 ; python_version >= "3.8" and python_version < "4.0"
jinja2==3.1.2 ; python_version >= "3.8" and python_version < "4.0"
lxml==4.9.2 ; python_version >= "3.8" and python_version < "4.0"
lxml==4.9.3 ; python_version >= "3.8" and python_version < "4.0"
mako==1.2.4 ; python_version >= "3.8" and python_version < "4.0"
markdown-it-py==2.2.0 ; python_version >= "3.8" and python_version < "4.0"
markupsafe==2.1.2 ; python_version >= "3.8" and python_version < "4.0"
markdown-it-py==3.0.0 ; python_version >= "3.8" and python_version < "4.0"
markupsafe==2.1.3 ; python_version >= "3.8" and python_version < "4.0"
mdurl==0.1.2 ; python_version >= "3.8" and python_version < "4.0"
multidict==6.0.4 ; python_version >= "3.8" and python_version < "4.0"
mypy-extensions==1.0.0 ; python_version >= "3.8" and python_version < "4.0"
openpyxl==3.1.2 ; python_version >= "3.8" and python_version < "4.0"
packaging==23.1 ; python_version >= "3.8" and python_version < "4.0"
pathspec==0.11.1 ; python_version >= "3.8" and python_version < "4.0"
pillow==9.5.0 ; python_version >= "3.8" and python_version < "4.0"
platformdirs==3.2.0 ; python_version >= "3.8" and python_version < "4.0"
pillow==10.0.0 ; python_version >= "3.8" and python_version < "4.0"
platformdirs==3.9.1 ; python_version >= "3.8" and python_version < "4.0"
playwright==1.27.1 ; python_version >= "3.8" and python_version < "4.0"
pluggy==1.0.0 ; python_version >= "3.8" and python_version < "4.0"
pluggy==1.2.0 ; python_version >= "3.8" and python_version < "4.0"
pyaes==1.6.1 ; python_version >= "3.8" and python_version < "4.0"
pycparser==2.21 ; python_version >= "3.8" and python_version < "4.0"
pydantic==1.10.7 ; python_version >= "3.8" and python_version < "4.0"
pydantic==1.10.11 ; python_version >= "3.8" and python_version < "4.0"
pyee==8.1.0 ; python_version >= "3.8" and python_version < "4.0"
pygments==2.15.1 ; python_version >= "3.8" and python_version < "4.0"
pypng==0.20220715.0 ; python_version >= "3.8" and python_version < "4.0"
pyppeteer==1.0.2 ; python_version >= "3.8" and python_version < "4.0"
pyrogram==2.0.104 ; python_version >= "3.8" and python_version < "4.0"
pyrogram==2.0.106 ; python_version >= "3.8" and python_version < "4.0"
pysocks==1.7.1 ; python_version >= "3.8" and python_version < "4.0"
pytest-asyncio==0.21.0 ; python_version >= "3.8" and python_version < "4.0"
pytest==7.3.1 ; python_version >= "3.8" and python_version < "4.0"
pytest-asyncio==0.21.1 ; python_version >= "3.8" and python_version < "4.0"
pytest==7.4.0 ; python_version >= "3.8" and python_version < "4.0"
python-dotenv==1.0.0 ; python_version >= "3.8" and python_version < "4.0"
python-telegram-bot[ext,rate-limiter]==20.2 ; python_version >= "3.8" and python_version < "4.0"
pytz-deprecation-shim==0.1.0.post0 ; python_version >= "3.8" and python_version < "4.0"
python-telegram-bot[ext,rate-limiter]==20.4 ; python_version >= "3.8" and python_version < "4.0"
pytz==2023.3 ; python_version >= "3.8" and python_version < "4.0"
pyyaml==6.0 ; python_version >= "3.8" and python_version < "4.0"
pyyaml==6.0.1 ; python_version >= "3.8" and python_version < "4.0"
qrcode==7.4.2 ; python_version >= "3.8" and python_version < "4.0"
redis==4.5.4 ; python_version >= "3.8" and python_version < "4.0"
rfc3986[idna2008]==1.5.0 ; python_version >= "3.8" and python_version < "4.0"
rich==13.3.4 ; python_version >= "3.8" and python_version < "4.0"
sentry-sdk==1.20.0 ; python_version >= "3.8" and python_version < "4.0"
setuptools==67.7.1 ; python_version >= "3.8" and python_version < "4.0"
redis==4.6.0 ; python_version >= "3.8" and python_version < "4.0"
rich==13.4.2 ; python_version >= "3.8" and python_version < "4.0"
sentry-sdk==1.28.1 ; python_version >= "3.8" and python_version < "4.0"
setuptools==68.0.0 ; python_version >= "3.8" and python_version < "4.0"
simnet @ git+https://github.com/PaiGramTeam/SIMNet@main ; python_version >= "3.8" and python_version < "4.0"
six==1.16.0 ; python_version >= "3.8" and python_version < "4.0"
smmap==5.0.0 ; python_version >= "3.8" and python_version < "4.0"
sniffio==1.3.0 ; python_version >= "3.8" and python_version < "4.0"
sortedcontainers==2.4.0 ; python_version >= "3.8" and python_version < "4.0"
soupsieve==2.4.1 ; python_version >= "3.8" and python_version < "4.0"
sqlalchemy2-stubs==0.0.2a34 ; python_version >= "3.8" and python_version < "4.0"
sqlalchemy2-stubs==0.0.2a35 ; python_version >= "3.8" and python_version < "4.0"
sqlalchemy==1.4.41 ; python_version >= "3.8" and python_version < "4.0"
sqlmodel==0.0.8 ; python_version >= "3.8" and python_version < "4.0"
starlette==0.26.1 ; python_version >= "3.8" and python_version < "4.0"
starlette==0.27.0 ; python_version >= "3.8" and python_version < "4.0"
tgcrypto==1.2.5 ; python_version >= "3.8" and python_version < "4.0"
thefuzz==0.19.0 ; python_version >= "3.8" and python_version < "4.0"
tomli==2.0.1 ; python_version >= "3.8" and python_version < "3.11"
tornado==6.3 ; python_version >= "3.8" and python_version < "4.0"
tornado==6.3.2 ; python_version >= "3.8" and python_version < "4.0"
tqdm==4.65.0 ; python_version >= "3.8" and python_version < "4.0"
typing-extensions==4.5.0 ; python_version >= "3.8" and python_version < "4.0"
tzdata==2023.3 ; python_version >= "3.8" and python_version < "4.0"
tzlocal==4.3 ; python_version >= "3.8" and python_version < "4.0"
ujson==5.7.0 ; python_version >= "3.8" and python_version < "4.0"
urllib3==1.26.15 ; python_version >= "3.8" and python_version < "4.0"
uvicorn[standard]==0.21.1 ; python_version >= "3.8" and python_version < "4.0"
uvloop==0.17.0 ; sys_platform != "win32" and sys_platform != "cygwin" and platform_python_implementation != "PyPy" and python_version >= "3.8" and python_version < "4.0"
typing-extensions==4.7.1 ; python_version >= "3.8" and python_version < "4.0"
tzdata==2023.3 ; python_version >= "3.8" and python_version < "4.0" and platform_system == "Windows"
tzlocal==5.0.1 ; python_version >= "3.8" and python_version < "4.0"
ujson==5.8.0 ; python_version >= "3.8" and python_version < "4.0"
urllib3==1.26.16 ; python_version >= "3.8" and python_version < "4.0"
uvicorn[standard]==0.22.0 ; python_version >= "3.8" and python_version < "4.0"
uvloop==0.17.0 ; (sys_platform != "win32" and sys_platform != "cygwin") and platform_python_implementation != "PyPy" and python_version >= "3.8" and python_version < "4.0"
watchfiles==0.19.0 ; python_version >= "3.8" and python_version < "4.0"
websockets==10.4 ; python_version >= "3.8" and python_version < "4.0"
yarl==1.8.2 ; python_version >= "3.8" and python_version < "4.0"
zipp==3.15.0 ; python_version >= "3.8" and python_version < "4.0"
yarl==1.9.2 ; python_version >= "3.8" and python_version < "4.0"
zipp==3.16.2 ; python_version >= "3.8" and python_version < "4.0"

64
utils/genshin.py
View File

@ -1,64 +0,0 @@
from typing import Optional
from genshin import Client
from genshin.client.routes import InternationalRoute # noqa F401
from genshin.utility import recognize_starrail_server
from modules.apihelper.utility.devices import devices_methods
from modules.apihelper.utility.helpers import hex_digest, get_ds
AUTHKEY_API = "https://api-takumi.mihoyo.com/binding/api/genAuthKey"
HK4E_LOGIN_URL = InternationalRoute(
overseas="https://sg-public-api.hoyoverse.com/common/badge/v1/login/account",
chinese="https://api-takumi.mihoyo.com/common/badge/v1/login/account",
)
GACHA_HEADERS = {
"User-Agent": "okhttp/4.8.0",
"x-rpc-sys_version": "12",
"x-rpc-channel": "mihoyo",
"x-rpc-device_name": "",
"x-rpc-device_model": "",
"Referer": "https://app.mihoyo.com",
"Host": "api-takumi.mihoyo.com",
}
def recognize_starrail_game_biz(game_uid: int) -> str:
return "hkrpg_cn" if game_uid < 600000000 else "hkrpg_global"
async def get_authkey_by_stoken(client: Client) -> Optional[str]:
"""通过 stoken 获取 authkey"""
headers = GACHA_HEADERS.copy()
json = {
"auth_appid": "webview_gacha",
"game_biz": recognize_starrail_game_biz(client.uid),
"game_uid": client.uid,
"region": recognize_starrail_server(client.uid),
}
device_id = hex_digest(str(client.uid))
device = f"Paimon Build {device_id[:5]}"
await devices_methods.update_device_headers(client.hoyolab_id, headers)
headers["x-rpc-device_name"] = device
headers["x-rpc-device_model"] = device
app_version, client_type, ds_sign = get_ds()
headers["x-rpc-app_version"] = app_version
headers["x-rpc-client_type"] = client_type
headers["ds"] = ds_sign
data = await client.cookie_manager.request(AUTHKEY_API, method="POST", json=json, headers=headers)
return data.get("authkey")
async def fetch_hk4e_token_by_cookie(client: Client) -> None:
"""通过 cookie_token 获取 hk4e_token 保存到 client"""
url = HK4E_LOGIN_URL.get_url(client.region)
headers = {
"Content-Type": "application/json;charset=UTF-8",
}
json = {
"game_biz": recognize_starrail_game_biz(client.uid),
"lang": "zh-cn",
"uid": str(client.uid),
"region": recognize_starrail_server(client.uid),
}
await client.cookie_manager.request(url, method="POST", json=json, headers=headers)

4
utils/patch/__init__.py
View File

@ -1,3 +1,3 @@
from utils.patch import aiohttp, genshin
from utils.patch import aiohttp
__all__ = ["aiohttp", "genshin"]
__all__ = ["aiohttp"]

358
utils/patch/genshin.py
View File

@ -1,358 +0,0 @@
import asyncio
import typing
import warnings
import aiohttp.typedefs
import genshin # pylint: disable=W0406
import yarl
from genshin import constants, types, utility
from genshin.client import routes
from genshin.utility import generate_dynamic_secret, ds
from modules.apihelper.utility.devices import devices_methods
from modules.apihelper.utility.helpers import get_ds, get_ua, get_device_id, hex_digest
from utils.patch.methods import patch, patchable
DEVICE_ID = get_device_id()
UPDATE_CHARACTERS = False
def get_account_mid_v2(cookies: typing.Dict[str, str]) -> typing.Optional[str]:
return next(
(value for name, value in cookies.items() if name == "account_mid_v2"),
None,
)
@patch(genshin.client.components.calculator.CalculatorClient) # noqa
class CalculatorClient:
@patchable
async def request_calculator(
self,
endpoint: str,
*,
method: str = "POST",
lang: typing.Optional[str] = None,
params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
data: typing.Optional[typing.Mapping[str, typing.Any]] = None,
headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
**kwargs: typing.Any,
) -> typing.Mapping[str, typing.Any]:
global UPDATE_CHARACTERS # pylint: disable=W0603
params = dict(params or {})
headers = dict(headers or {})
base_url = routes.CALCULATOR_URL.get_url(self.region)
url = base_url / endpoint
if method == "GET":
params["lang"] = lang or self.lang
data = None
else:
data = dict(data or {})
data["lang"] = lang or self.lang
if self.region == types.Region.CHINESE:
headers["referer"] = str(routes.CALCULATOR_REFERER_URL.get_url())
update_task = (
None
if UPDATE_CHARACTERS
else asyncio.create_task(utility.update_characters_any(lang or self.lang, lenient=True))
)
data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
if update_task:
try:
await update_task
UPDATE_CHARACTERS = True
except Exception as e: # pylint: disable=W0703
warnings.warn(f"Failed to update characters: {e!r}")
return data
@patchable
async def get_character_details(
self,
character: genshin.types.IDOr[genshin.models.genshin.Character],
*,
uid: typing.Optional[int] = None,
lang: typing.Optional[str] = None,
):
uid = uid or await self._get_uid(genshin.types.Game.GENSHIN)
data = await self.request_calculator(
"sync/avatar/detail",
method="GET",
lang=lang,
params=dict(
avatar_id=int(character),
uid=uid,
region=genshin.utility.recognize_genshin_server(uid),
),
)
if data.get("weapon") is None:
weapon = {
"id": character.weapon.id,
"name": character.weapon.name,
"icon": character.weapon.icon,
"weapon_cat_id": character.weapon.type,
"weapon_level": character.weapon.rarity,
"max_level": 90,
"level_current": character.weapon.level,
}
data["weapon"] = weapon
return genshin.models.genshin.CalculatorCharacterDetails(**data)
@patch(genshin.client.components.base.BaseClient) # noqa
class BaseClient:
@patchable
async def request_hoyolab(
self: "genshin.Client",
url: aiohttp.typedefs.StrOrURL,
*,
lang: typing.Optional[str] = None,
region: typing.Optional[types.Region] = None,
method: typing.Optional[str] = None,
params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
data: typing.Any = None,
headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
**kwargs: typing.Any,
) -> typing.Mapping[str, typing.Any]:
"""Make a request any hoyolab endpoint."""
if lang is not None and lang not in constants.LANGS:
raise ValueError(f"{lang} is not a valid language, must be one of: " + ", ".join(constants.LANGS))
lang = lang or self.lang
region = region or self.region
url = routes.TAKUMI_URL.get_url(region).join(yarl.URL(url))
if region == types.Region.OVERSEAS:
headers = {
"x-rpc-app_version": "1.5.0",
"x-rpc-client_type": "5",
"x-rpc-language": lang,
"ds": generate_dynamic_secret(),
}
elif region == types.Region.CHINESE:
account_id = self.cookie_manager.user_id
if account_id:
device_id = hex_digest(str(account_id))
else:
account_mid_v2 = get_account_mid_v2(self.cookie_manager.cookies)
if account_mid_v2:
device_id = hex_digest(account_mid_v2)
else:
device_id = DEVICE_ID
app_version, client_type, ds_sign = get_ds(new_ds=True, data=data, params=params)
ua = get_ua(device="Paimon Build " + device_id[0:5], version=app_version)
headers = {
"User-Agent": ua,
"X_Requested_With": "com.mihoyo.hoyolab",
"Referer": "https://webstatic.mihoyo.com",
"x-rpc-app_version": app_version,
"x-rpc-client_type": client_type,
"ds": ds_sign,
}
await devices_methods.update_device_headers(self.hoyolab_id, headers)
else:
raise TypeError(f"{region!r} is not a valid region.")
data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
return data
@patchable
async def request(
self: "genshin.Client",
url: aiohttp.typedefs.StrOrURL,
*,
method: typing.Optional[str] = None,
params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
data: typing.Any = None,
headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
cache: typing.Any = None,
static_cache: typing.Any = None,
**kwargs: typing.Any,
) -> typing.Mapping[str, typing.Any]:
"""Make a request and return a parsed json response."""
if cache is not None:
value = await self.cache.get(cache)
if value is not None:
return value
elif static_cache is not None:
value = await self.cache.get_static(static_cache)
if value is not None:
return value
# actual request
headers = dict(headers or {})
headers.setdefault("User-Agent", self.USER_AGENT)
await devices_methods.update_device_headers(self.hoyolab_id, headers)
if method is None:
method = "POST" if data else "GET"
if "json" in kwargs:
raise TypeError("Use data instead of json in request.")
await self._request_hook(method, url, params=params, data=data, headers=headers, **kwargs)
response = await self.cookie_manager.request(
url,
method=method,
params=params,
json=data,
headers=headers,
**kwargs,
)
# cache
if cache is not None:
await self.cache.set(cache, response)
elif static_cache is not None:
await self.cache.set_static(static_cache, response)
return response
@patchable
async def request_bbs(
self: "genshin.Client",
url: aiohttp.typedefs.StrOrURL,
*,
lang: typing.Optional[str] = None,
region: typing.Optional[types.Region] = None,
method: typing.Optional[str] = None,
params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
data: typing.Any = None,
headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
**kwargs: typing.Any,
) -> typing.Mapping[str, typing.Any]:
"""Make a request any bbs endpoint."""
if lang is not None and lang not in constants.LANGS:
raise ValueError(f"{lang} is not a valid language, must be one of: " + ", ".join(constants.LANGS))
lang = lang or self.lang
region = region or self.region
url = routes.BBS_URL.get_url(region).join(yarl.URL(url))
headers = dict(headers or {})
if self.region == types.Region.CHINESE:
if self.region == types.Region.CHINESE:
account_id = self.cookie_manager.user_id
if account_id:
device_id = hex_digest(str(account_id))
else:
account_mid_v2 = get_account_mid_v2(self.cookie_manager.cookies)
if account_mid_v2:
device_id = hex_digest(account_mid_v2)
else:
device_id = DEVICE_ID
app_version, _, ds_sign = get_ds()
ua = get_ua(device="Paimon Build " + device_id[0:5], version=app_version)
add_headers = {
"User-Agent": ua,
"Referer": "https://www.miyoushe.com/ys/",
"x-rpc-app_version": app_version,
"x-rpc-client_type": "4",
"ds": ds_sign,
}
headers.update(add_headers)
await devices_methods.update_device_headers(self.hoyolab_id, headers)
elif self.region == types.Region.OVERSEAS:
headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))
headers["Referer"] = str(routes.BBS_REFERER_URL.get_url(self.region))
else:
raise TypeError(f"{region!r} is not a valid region.")
data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
return data
@patch(genshin.client.components.daily.DailyRewardClient) # noqa
class DailyRewardClient:
@patchable
async def request_daily_reward(
self: "genshin.Client",
endpoint: str,
*,
game: typing.Optional[types.Game] = None,
method: str = "GET",
lang: typing.Optional[str] = None,
params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
**kwargs: typing.Any,
) -> typing.Mapping[str, typing.Any]:
"""Make a request towards the daily reward endpoint."""
params = dict(params or {})
headers = dict(headers or {})
if game is None:
if self.default_game is None:
raise RuntimeError("No default game set.")
game = self.default_game
base_url = routes.REWARD_URL.get_url(self.region, game)
url = (base_url / endpoint).update_query(**base_url.query)
if self.region == types.Region.OVERSEAS:
params["lang"] = lang or self.lang
elif self.region == types.Region.CHINESE:
# TODO: Support cn honkai
player_id = await self._get_uid(types.Game.STARRAIL)
params["uid"] = player_id
params["region"] = utility.recognize_server(player_id, types.Game.STARRAIL)
account_id = self.cookie_manager.user_id
if account_id:
device_id = hex_digest(str(account_id))
else:
account_mid_v2 = get_account_mid_v2(self.cookie_manager.cookies)
if account_mid_v2:
device_id = hex_digest(account_mid_v2)
else:
device_id = DEVICE_ID
if endpoint == "sign":
app_version, client_type, ds_sign = get_ds()
else:
app_version, client_type, ds_sign = get_ds(new_ds=True, params=params)
device = "Paimon Build " + device_id[0:5]
ua = get_ua(device=device)
headers["User-Agent"] = ua
headers["X_Requested_With"] = "com.mihoyo.hoyolab"
headers["Referer"] = (
"https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html?"
"bbs_auth_required=true&act_id=e202009291139501&utm_source=bbs&utm_medium=mys&utm_campaign=icon"
)
headers["x-rpc-device_name"] = device
headers["x-rpc-app_version"] = app_version
headers["x-rpc-client_type"] = client_type
headers["x-rpc-sys_version"] = "12"
headers["x-rpc-platform"] = "android"
headers["x-rpc-channel"] = "miyousheluodi"
headers["x-rpc-device_model"] = device
headers["ds"] = ds_sign
validate = kwargs.get("validate")
challenge = kwargs.get("challenge")
if validate and challenge:
headers["x-rpc-challenge"] = challenge
headers["x-rpc-validate"] = validate
headers["x-rpc-seccode"] = f"{validate}|jordan"
await devices_methods.update_device_headers(self.hoyolab_id, headers)
else:
raise TypeError(f"{self.region!r} is not a valid region.")
kwargs.pop("challenge", None)
kwargs.pop("validate", None)
return await self.request(url, method=method, params=params, headers=headers, **kwargs)