Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2026-10-09 13:50:00 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

update v0.1

Browse Source
This commit is contained in:
HibiKier
2022-02-09 20:05:49 +08:00
parent 4cb91b47e6
commit 0efb2bbc60
33 changed files with 875 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/genshin/query_user/genshin_sign/__init__.py
+101
View File
@@ -0,0 +1,101 @@
from .data_source import get_sign_reward_list, genshin_sign
from nonebot.adapters.cqhttp import Bot, MessageEvent, GroupMessageEvent
from nonebot import on_command
from nonebot.typing import T_State
from services.log import logger
from .init_task import add_job, scheduler, _sign
from apscheduler.jobstores.base import JobLookupError
from ..models import Genshin
__zx_plugin_name__ = "原神自动签到"
__plugin_usage__ = """
usage：
米游社原神签到，需要uid以及cookie
且在第二天自动排序签到时间
# 不听，就要手动签到！（使用命令 “原神我硬签 or 米游社我硬签”
指令：
开/关原神自动签到
原神我硬签
""".strip()
__plugin_des__ = "原神懒人签到"
__plugin_cmd__ = ["开启/关闭原神自动签到", "原神我硬签"]
__plugin_type__ = ("原神相关",)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["原神签到"],
}
genshin_matcher = on_command(
"开原神自动签到", aliases={"关原神自动签到", "原神我硬签"}, priority=5, block=True
)
@genshin_matcher.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
uid = await Genshin.get_user_uid(event.user_id)
if not uid or not await Genshin.get_user_cookie(uid, True):
await genshin_matcher.finish("请先绑定uid和cookie！")
if "account_id" not in await Genshin.get_user_cookie(uid, True):
await genshin_matcher.finish("请更新cookie！")
if state["_prefix"]["raw_command"] == "原神我硬签":
try:
msg = await genshin_sign(uid)
logger.info(
f"(USER {event.user_id}, "
f"GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'}) UID：{uid} 原神签到"
)
# 硬签，移除定时任务
try:
for i in range(3):
scheduler.remove_job(f"genshin_auto_sign_{uid}_{event.user_id}_{i}",)
except JobLookupError:
pass
u = await Genshin.get_user_by_uid(uid)
if u and u.auto_sign:
await u.clear_sign_time(uid)
next_date = await Genshin.random_sign_time(uid)
add_job(event.user_id, uid, next_date)
msg += f"因开启自动签到\n下一次签到时间为：{next_date.replace(microsecond=0)}"
except Exception as e:
msg = "原神签到失败..请尝试检查cookie或报告至管理员！"
logger.info(
f"(USER {event.user_id}, "
f"GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'}) UID：{uid} 原神签到发生错误 "
f"{type(e)}：{e}"
)
msg = msg or "请检查cookie是否更新！"
await genshin_matcher.send(msg, at_sender=True)
else:
for i in range(3):
try:
scheduler.remove_job(f"genshin_auto_sign_{uid}_{event.user_id}_{i}")
except JobLookupError:
pass
if state["_prefix"]["raw_command"][0] == "开":
await Genshin.set_auto_sign(uid, True)
next_date = await Genshin.random_sign_time(uid)
add_job(event.user_id, uid, next_date)
await genshin_matcher.send(
f"已开启原神自动签到！\n下一次签到时间为：{next_date.replace(microsecond=0)}",
at_sender=True,
)
logger.info(
f"(USER {event.user_id}, GROUP "
f"{event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 开启原神自动签到"
)
else:
await Genshin.set_auto_sign(uid, False)
await Genshin.clear_sign_time(uid)
await genshin_matcher.send(f"已关闭原神自动签到！", at_sender=True)
logger.info(
f"(USER {event.user_id}, GROUP "
f"{event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 关闭原神自动签到"
)
plugins/genshin/query_user/genshin_sign/data_source.py
+104
View File
@@ -0,0 +1,104 @@
from utils.http_utils import AsyncHttpx
from configs.config import Config
from services.log import logger
from ..utils import random_hex, get_old_ds
from ..models import Genshin
from typing import Optional, Dict
async def genshin_sign(uid: int) -> Optional[str]:
"""
原神签到信息
:param uid: uid
"""
data = await _sign(uid)
if not data:
return "签到失败..."
status = data["message"]
if status == "OK":
sign_info = await _get_sign_info(uid)
if sign_info:
sign_info = sign_info["data"]
sign_list = await get_sign_reward_list()
get_reward = sign_list["data"]["awards"][
int(sign_info["total_sign_day"]) - 1
]["name"]
reward_num = sign_list["data"]["awards"][
int(sign_info["total_sign_day"]) - 1
]["cnt"]
get_im = f"本次签到获得：{get_reward}x{reward_num}"
if status == "OK" and sign_info["is_sign"]:
return f"\n原神签到成功！\n{get_im}\n本月漏签次数：{sign_info['sign_cnt_missed']}"
else:
return status
return None
async def _sign(uid: int, server_id: str = "cn_gf01") -> Optional[Dict[str, str]]:
"""
米游社签到
:param uid: uid
:param server_id: 服务器id
"""
if str(uid)[0] == "5":
server_id = "cn_qd01"
try:
req = await AsyncHttpx.post(
url="https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign",
headers={
"User_Agent": "Mozilla/5.0 (Linux; Android 10; MIX 2 Build/QKQ1.190825.002; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/83.0.4103.101 Mobile Safari/537.36 miHoYoBBS/2.3.0",
"Cookie": await Genshin.get_user_cookie(int(uid), True),
"x-rpc-device_id": random_hex(32),
"Origin": "https://webstatic.mihoyo.com",
"X_Requested_With": "com.mihoyo.hyperion",
"DS": get_old_ds(),
"x-rpc-client_type": "5",
"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",
"x-rpc-app_version": "2.3.0",
},
json={"act_id": "e202009291139501", "uid": uid, "region": server_id},
)
return req.json()
except Exception as e:
logger.error(f"米游社签到发生错误 UID：{uid} {type(e)}：{e}")
return None
async def get_sign_reward_list():
"""
获取签到奖励列表
"""
try:
req = await AsyncHttpx.get(
url="https://api-takumi.mihoyo.com/event/bbs_sign_reward/home?act_id=e202009291139501",
headers={
"x-rpc-app_version": str(Config.get_config("genshin", "mhyVersion")),
"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": str(Config.get_config("genshin", "client_type")),
"Referer": "https://webstatic.mihoyo.com/",
},
)
return req.json()
except Exception as e:
logger.error(f"获取签到奖励列表发生错误 {type(e)}：{e}")
return None
async def _get_sign_info(uid: int, server_id: str = "cn_gf01"):
if str(uid)[0] == "5":
server_id = "cn_qd01"
try:
req = await AsyncHttpx.get(
url=f"https://api-takumi.mihoyo.com/event/bbs_sign_reward/info?act_id=e202009291139501&region={server_id}&uid={uid}",
headers={
"x-rpc-app_version": str(Config.get_config("genshin", "mhyVersion")),
"Cookie": await Genshin.get_user_cookie(int(uid), True),
"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": str(Config.get_config("genshin", "client_type")),
"Referer": "https://webstatic.mihoyo.com/",
},
)
return req.json()
except Exception as e:
logger.error(f"获取签到信息发生错误 UID：{uid} {type(e)}：{e}")
return None
plugins/genshin/query_user/genshin_sign/init_task.py
+110
View File
@@ -0,0 +1,110 @@
from .data_source import genshin_sign
from models.group_member_info import GroupInfoUser
from utils.message_builder import at
from services.log import logger
from utils.utils import scheduler, get_bot
from apscheduler.jobstores.base import ConflictingIdError
from ..models import Genshin
from datetime import datetime, timedelta
from nonebot import Driver
import nonebot
import random
import pytz
driver: Driver = nonebot.get_driver()
@driver.on_startup
async def _():
"""
启动时分配定时任务
"""
g_list = await Genshin.get_all_auto_sign_user()
for u in g_list:
if u.auto_sign_time:
date = await Genshin.random_sign_time(u.uid)
scheduler.add_job(
_sign,
"date",
run_date=date.replace(microsecond=0),
id=f"genshin_auto_sign_{u.uid}_{u.user_qq}_0",
args=[u.user_qq, u.uid, 0],
)
logger.info(
f"genshin_sign add_job：USER：{u.user_qq} UID：{u.uid} " f"{date} 原神自动签到"
)
def add_job(user_id: int, uid: int, date: datetime):
try:
scheduler.add_job(
_sign,
"date",
run_date=date.replace(microsecond=0),
id=f"genshin_auto_sign_{uid}_{user_id}_0",
args=[user_id, uid, 0],
)
logger.debug(f"genshin_sign add_job：{date.replace(microsecond=0)} 原神自动签到")
except ConflictingIdError:
pass
async def _sign(user_id: int, uid: int, count: int):
"""
执行签到任务
:param user_id: 用户id
:param uid: uid
:param count: 执行次数
"""
if count < 3:
try:
msg = await genshin_sign(uid)
next_time = await Genshin.random_sign_time(uid)
msg += f"\n下一次签到时间为：{next_time.replace(microsecond=0)}"
logger.info(f"USER：{user_id} UID：{uid} 原神自动签到任务发生成功...")
try:
scheduler.add_job(
_sign,
"date",
run_date=next_time.replace(microsecond=0),
id=f"genshin_auto_sign_{uid}_{user_id}_0",
args=[user_id, uid, 0],
)
except ConflictingIdError:
msg += "\n定时任务设定失败..."
except Exception as e:
logger.error(f"USER：{user_id} UID：{uid} 原神自动签到任务发生错误 {type(e)}：{e}")
msg = None
if not msg:
now = datetime.now(pytz.timezone("Asia/Shanghai"))
if now.hour < 23:
random_hours = random.randint(1, 23 - now.hour)
next_time = now + timedelta(hours=random_hours)
scheduler.add_job(
_sign,
"date",
run_date=next_time.replace(microsecond=0),
id=f"genshin_auto_sign_{uid}_{user_id}_{count}",
args=[user_id, uid, count + 1],
)
msg = (
f"{now.replace(microsecond=0)} 原神"
f"签到失败，将在 {next_time.replace(microsecond=0)} 时重试！"
)
else:
msg = "今日原神签到失败，请手动签到..."
logger.debug(f"USER：{user_id} UID：{uid} 原神今日签到失败...")
else:
msg = "今日原神自动签到重试次数已达到3次，请手动签到。"
logger.debug(f"USER：{user_id} UID：{uid} 原神今日签到失败次数打到 3 次...")
bot = get_bot()
if bot:
if user_id in [x["user_id"] for x in await bot.get_friend_list()]:
await bot.send_private_msg(user_id=user_id, message=msg)
else:
group_list = await GroupInfoUser.get_user_all_group(user_id)
if group_list:
await bot.send_group_msg(
group_id=group_list[0], message=at(user_id) + msg
)