diff --git a/.vscode/extensions.json b/.vscode/extensions.json index 54fc7aa..5b5f9de 100644 --- a/.vscode/extensions.json +++ b/.vscode/extensions.json @@ -1,7 +1,7 @@ { "recommendations": [ "ms-python.python", - "ms-python.vscode-pylance", + "detachhead.basedpyright", "ms-python.isort", "ms-python.black-formatter" ] diff --git a/.vscode/settings.json b/.vscode/settings.json index badf942..c549d5f 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,11 +1,5 @@ { - "python.languageServer": "Pylance", - "python.analysis.typeCheckingMode": "basic", - "cSpell.words": [ - "enka", - "genshin", - "genshinuid" - ], + "python.languageServer": "None", "editor.formatOnSave": true, "[python]": { "editor.defaultFormatter": "ms-python.black-formatter", @@ -14,22 +8,14 @@ "source.organizeImports": "explicit" } }, + "python.autoComplete.extraPaths": [ + "${workspaceFolder}/../../../../" + ], "isort.args": [ "--profile", "black" ], - "python.formatting.provider": "black", - "python.linting.flake8Enabled": true, - "python.linting.flake8CategorySeverity.W": "Warning", - "python.linting.flake8CategorySeverity.F": "Warning", - "python.linting.flake8CategorySeverity.E": "Warning", - "python.analysis.extraPaths": [ + "basedpyright.analysis.extraPaths": [ "${workspaceFolder}/../../../" ], - "python.autoComplete.extraPaths": [ - "${workspaceFolder}/../../../" - ], - "python.analysis.include": [ - "${workspaceFolder}/../../../" - ] } \ No newline at end of file diff --git a/StarRailUID/starrailuid_config/__init__.py b/StarRailUID/starrailuid_config/__init__.py new file mode 100644 index 0000000..4cc8443 --- /dev/null +++ b/StarRailUID/starrailuid_config/__init__.py @@ -0,0 +1,21 @@ +from gsuid_core.sv import SV +from gsuid_core.bot import Bot +from gsuid_core.models import Event +from gsuid_core.logger import logger +from gsuid_core.utils.database.models import GsBind + +from .set_config import set_config_func +from ..starrailuid_config.sr_config import srconfig + +PREFIX = srconfig.get_config('StarRailPrefix').data + +sv_self_config = SV('星穹铁道配置') + + +@sv_self_config.on_prefix((f'{PREFIX}开启', f'{PREFIX}关闭')) +async def open_switch_func(bot: Bot, ev: Event): + uid = await GsBind.get_uid_by_game(ev.user_id, ev.bot_id, 'sr') + if uid is None: + return await bot.send('[星穹铁道] 你还没有绑定UID哦！') + logger.info(f'[{ev.user_id}] [UID{uid}]尝试[{ev.command[2:]}]了[{ev.text}]功能') + await bot.send(await set_config_func(uid, ev)) diff --git a/StarRailUID/starrailuid_config/config_default.py b/StarRailUID/starrailuid_config/config_default.py index a025653..22ef0d0 100644 --- a/StarRailUID/starrailuid_config/config_default.py +++ b/StarRailUID/starrailuid_config/config_default.py @@ -11,10 +11,10 @@ CONIFG_DEFAULT: Dict[str, GSC] = { 'SignTime': GsListStrConfig( '每晚签到时间设置', '每晚米游社签到时间设置(时,分)', ['0', '38'] ), - 'SignReportSimple': GsBoolConfig( - '简洁签到报告', - '开启后可以大大减少每日签到报告字数', - True, + 'PrivateSignReport': GsBoolConfig( + '签到私聊报告', + '关闭后将不再给任何人推送当天签到任务完成情况', + False, ), 'SchedSignin': GsBoolConfig( '定时签到', diff --git a/StarRailUID/starrailuid_config/set_config.py b/StarRailUID/starrailuid_config/set_config.py new file mode 100644 index 0000000..16e3513 --- /dev/null +++ b/StarRailUID/starrailuid_config/set_config.py @@ -0,0 +1,35 @@ +from gsuid_core.models import Event +from gsuid_core.logger import logger +from gsuid_core.utils.database.models import GsUser +from gsuid_core.utils.database.config_switch import set_database_value + + +async def set_config_func( + uid: str, + ev: Event, +): + if '开启' in ev.command: + if ev.user_type == 'direct': + value = 'on' + else: + if ev.group_id: + value = ev.group_id + else: + value = 'on' + else: + value = 'off' + + text = await set_database_value( + GsUser, + 'sr', + 'sr开启', + ev.text.strip(), + uid, + ev.bot_id, + value, + ) + if text is None: + return '[星穹铁道] 未找到配置项' + else: + logger.success(f'[UID{uid}]成功将配置[SR自动签到]设置为[{value}]!') + return text diff --git a/StarRailUID/starrailuid_config/sr_config.py b/StarRailUID/starrailuid_config/sr_config.py index 28c60ac..0e2fa23 100644 --- a/StarRailUID/starrailuid_config/sr_config.py +++ b/StarRailUID/starrailuid_config/sr_config.py @@ -3,4 +3,8 @@ from gsuid_core.utils.plugins_config.gs_config import StringConfig from .config_default import CONIFG_DEFAULT from ..utils.resource.RESOURCE_PATH import CONFIG_PATH -srconfig = StringConfig('StarRailUID', CONFIG_PATH, CONIFG_DEFAULT) +srconfig = StringConfig( + 'StarRailUID', + CONFIG_PATH, + CONIFG_DEFAULT, +) diff --git a/StarRailUID/starrailuid_signin/__init__.py b/StarRailUID/starrailuid_signin/__init__.py index 7cee914..38be2b7 100644 --- a/StarRailUID/starrailuid_signin/__init__.py +++ b/StarRailUID/starrailuid_signin/__init__.py @@ -1,105 +1,49 @@ -import random -import asyncio - from gsuid_core.sv import SV from gsuid_core.bot import Bot -from gsuid_core.gss import gss from gsuid_core.models import Event from gsuid_core.aps import scheduler from gsuid_core.logger import logger from gsuid_core.utils.database.models import GsBind +from gsuid_core.utils.sign.sign import sign_in, daily_sign +from gsuid_core.utils.boardcast.send_msg import send_board_cast_msg from ..utils.sr_prefix import PREFIX -from .sign import sign_in, daily_sign from ..utils.error_reply import UID_HINT from ..starrailuid_config.sr_config import srconfig SIGN_TIME = srconfig.get_config('SignTime').data +IS_REPORT = srconfig.get_config('PrivateSignReport').data sv_sign = SV('星穹铁道签到') sv_sign_config = SV('星穹铁道管理', pm=2) +@sv_sign.on_fullmatch(f'{PREFIX}签到') +async def get_sign_func(bot: Bot, ev: Event): + logger.info(f'[星穹铁道] [签到] 用户: {ev.user_id}') + uid = await GsBind.get_uid_by_game(ev.user_id, ev.bot_id, 'sr') + if uid is None: + return await bot.send(UID_HINT) + logger.info(f'[星穹铁道] [签到] UID: {uid}') + await bot.send(await sign_in(uid, 'sr')) + + +@sv_sign_config.on_fullmatch(f'{PREFIX}全部重签') +async def recheck(bot: Bot, ev: Event): + await bot.logger.info('开始执行[全部重签]') + await bot.send('[星穹铁道] [全部重签] 已开始执行!') + result = await daily_sign('sr') + if not IS_REPORT: + result['private_msg_dict'] = {} + await send_board_cast_msg(result) + await bot.send('[星穹铁道] [全部重签] 执行完成!') + + # 每日零点半执行米游社星穹铁道签到 @scheduler.scheduled_job('cron', hour=SIGN_TIME[0], minute=SIGN_TIME[1]) async def sr_sign_at_night(): if srconfig.get_config('SchedSignin').data: - await send_daily_sign() - - -# 群聊内 签到 功能 -@sv_sign.on_fullmatch(f'{PREFIX}签到') -async def get_sign_func(bot: Bot, ev: Event): - await bot.logger.info(f'[SR签到]QQ号: {ev.user_id}') - sr_uid = await GsBind.get_uid_by_game(ev.user_id, ev.bot_id, 'sr') - if sr_uid is None: - return await bot.send(UID_HINT) - await bot.logger.info(f'[SR签到]UID: {sr_uid}') - await bot.send(await sign_in(sr_uid)) - return None - - -@sv_sign_config.on_fullmatch(f'{PREFIX}全部重签') -async def recheck(bot: Bot, ev: Event): - await bot.logger.info('开始执行[SR全部重签]') - await bot.send('已开始执行') - await send_daily_sign() - await bot.send('执行完成') - - -async def send_daily_sign(): - logger.info('开始执行[SR每日全部签到]') - # 执行签到 并获得推送消息 - result = await daily_sign() - private_msg_list = result['private_msg_list'] - group_msg_list = result['group_msg_list'] - logger.info('[SR每日全部签到]完成') - - # 执行私聊推送 - for qid in private_msg_list: - try: - for bot_id in gss.active_bot: - for single in private_msg_list[qid]: - await gss.active_bot[bot_id].target_send( - single['msg'], 'direct', qid, single['bot_id'], '', '' - ) - except Exception as e: - logger.warning( - f'[SR每日全部签到] QQ {qid} 私聊推送失败!错误信息:{e}' - ) - await asyncio.sleep(0.5) - logger.info('[SR每日全部签到]私聊推送完成') - - # 执行群聊推送 - for gid in group_msg_list: - # 根据succee数判断是否为简洁推送 - if group_msg_list[gid]['success'] >= 0: - report = ( - '以下为签到失败报告:{}'.format( - group_msg_list[gid]['push_message'] - ) - if group_msg_list[gid]['push_message'] != '' - else '' - ) - msg_title = ( - f"星穹铁道今日自动签到已完成!\n" - f"本群共签到成功{group_msg_list[gid]['success']}人," - f"共签到失败{group_msg_list[gid]['failed']}人。{report}" - ) - else: - msg_title = group_msg_list[gid]['push_message'] - # 发送群消息 - try: - for bot_id in gss.active_bot: - await gss.active_bot[bot_id].target_send( - msg_title, - 'group', - gid, - group_msg_list[gid]['bot_id'], - '', - '', - ) - except Exception as e: - logger.warning(f'[SR每日全部签到]群 {gid} 推送失败!错误信息:{e}') - await asyncio.sleep(0.5 + random.randint(1, 3)) - logger.info('[SR每日全部签到]群聊推送完成') + result = await daily_sign('sr') + if not IS_REPORT: + result['private_msg_dict'] = {} + await send_board_cast_msg(result) diff --git a/StarRailUID/starrailuid_signin/sign.py b/StarRailUID/starrailuid_signin/sign.py deleted file mode 100644 index 24b72f8..0000000 --- a/StarRailUID/starrailuid_signin/sign.py +++ /dev/null @@ -1,191 +0,0 @@ -import asyncio -import random -from copy import deepcopy - -from gsuid_core.gss import gss -from gsuid_core.logger import logger -from gsuid_core.utils.database.models import GsUser -from gsuid_core.utils.plugins_config.gs_config import core_plugins_config - -from ..starrailuid_config.sr_config import srconfig -from ..utils.mys_api import mys_api - -private_msg_list = {} -group_msg_list = {} -already = 0 - - -# 签到函数 -async def sign_in(sr_uid: str) -> str: - logger.info(f'[SR签到] {sr_uid} 开始执行签到') - # 获得签到信息 - sign_info = await mys_api.get_sign_info(sr_uid) - # 初步校验数据 - if isinstance(sign_info, int): - logger.warning(f'[SR签到] {sr_uid} 出错, 请检查Cookies是否过期!') - return '签到失败...请检查Cookies是否过期!' - # 检测是否已签到 - if sign_info.is_sign: - logger.info(f'[SR签到] {sr_uid} 该用户今日已签到,跳过...') - global already - already += 1 - day_of_month = int(sign_info.today.split('-')[-1]) - signed_count = int(sign_info.total_sign_day) - sign_missed = day_of_month - signed_count - return f'今日已签到!本月漏签次数:{sign_missed}' - - # 实际进行签到 - Header = {} - for index in range(4): - # 进行一次签到 - sign_data = await mys_api.mys_sign(uid=sr_uid, header=Header) - # 检测数据 - if isinstance(sign_data, int): - logger.warning(f'[SR签到] {sr_uid} 出错, 请检查Cookies是否过期!') - return 'sr签到失败...请检查Cookies是否过期!' - if sign_data.risk_code == 5001 or sign_data.risk_code == 0: - # 出现校验码 - if sign_data.risk_code == 5001 and sign_data.is_risk: - if core_plugins_config.get_config('CaptchaPass').data: - gt = sign_data.gt - ch = sign_data.challenge - vl, ch = await mys_api._pass(gt, ch, Header) # noqa: SLF001 - if vl: - delay = 1 - Header['x-rpc-challenge'] = ch - Header['x-rpc-validate'] = vl - Header['x-rpc-seccode'] = f'{vl}|jordan' - logger.info( - f'[SR签到] {sr_uid} 已获取验证码, 等待时间{delay}秒' - ) - await asyncio.sleep(delay) - else: - delay = 605 + random.randint(1, 120) - logger.info( - f'[SR签到] {sr_uid} 未获取验证码,等待{delay}秒后重试...' - ) - await asyncio.sleep(delay) - continue - logger.info('配置文件暂未开启[跳过无感验证],结束本次任务...') - return '签到失败...出现验证码!' - # 成功签到! - if index == 0: - logger.info(f'[SR签到] {sr_uid} 该用户无校验码!') - else: - logger.info( - f'[SR签到] [无感验证] {sr_uid} 该用户重试 {index} 次验证成功!' - ) - break - elif (int(str(sr_uid)[0]) > 5) and (sign_data.code == 'ok'): - # 国际服签到无risk_code字段 - logger.info(f'[SR国际服签到] {sr_uid} 签到成功!') - break - # 重试超过阈值 - else: - logger.warning('[SR签到] 超过请求阈值...') - return 'sr签到失败...出现验证码!\n请过段时间使用[签到]或由管理员[全部重签]或手动至米游社进行签到!' - # 签到失败 - else: - im = 'sr签到失败!' - logger.warning(f'[SR签到] {sr_uid} 签到失败, 结果: {im}') - return im - # 获取签到列表 - sign_list = await mys_api.get_sign_list(sr_uid) - new_sign_info = await mys_api.get_sign_info(sr_uid) - if isinstance(sign_list, int) or isinstance(new_sign_info, int): - logger.warning(f'[SR签到] {sr_uid} 出错, 请检查Cookies是否过期!') - return 'sr签到失败...请检查Cookies是否过期!' - # 获取签到奖励物品,拿旧的总签到天数 + 1 为新的签到天数,再 -1 即为今日奖励物品的下标 - getitem = sign_list.awards[int(sign_info.total_sign_day) + 1 - 1] - get_im = f'本次sr签到获得{getitem.name}x{getitem.cnt}' - day_of_month = int(new_sign_info.today.split('-')[-1]) - signed_count = int(new_sign_info.total_sign_day) - sign_missed = day_of_month - signed_count - if new_sign_info.is_sign: - mes_im = 'sr签到成功' - else: - mes_im = 'sr签到失败...' - sign_missed -= 1 - sign_missed = sign_info.sign_cnt_missed or sign_missed - im = f'{mes_im}!\n{get_im}\n本月漏签次数:{sign_missed}' - logger.info( - f'[SR签到] {sr_uid} 签到完成, 结果: {mes_im}, 漏签次数: {sign_missed}' - ) - return im - - -async def single_daily_sign(bot_id: str, sr_uid: str, gid: str, qid: str): - im = await sign_in(sr_uid) - if gid == 'on': - if qid not in private_msg_list: - private_msg_list[qid] = [] - private_msg_list[qid].append( - { - 'bot_id': bot_id, - 'uid': sr_uid, - 'msg': im, - } - ) - else: - # 向群消息推送列表添加这个群 - if gid not in group_msg_list: - group_msg_list[gid] = { - 'bot_id': bot_id, - 'success': 0, - 'failed': 0, - 'push_message': '', - } - # 检查是否开启简洁签到 - if srconfig.get_config('SignReportSimple').data: - # 如果失败, 则添加到推送列表 - if im.startswith(('sr签到失败', '网络有点忙', 'OK', 'ok')): - message = f'[CQ:at,qq={qid}] {im}' - group_msg_list[gid]['failed'] += 1 - group_msg_list[gid]['push_message'] += '\n' + message - else: - group_msg_list[gid]['success'] += 1 - # 没有开启简洁签到, 则每条消息都要携带@信息 - else: - # 不用MessageSegment.at(row[2]),因为不方便移植 - message = f'[CQ:at,qq={qid}] {im}' - group_msg_list[gid]['push_message'] += '\n' + message - group_msg_list[gid]['success'] -= 1 - - -async def daily_sign(): - global already - tasks = [] - for _ in gss.active_bot: - user_list = await GsUser.get_all_user() - for user in user_list: - if user.sign_switch != 'off' and user.sr_uid is not None: - tasks.append( - single_daily_sign( - user.bot_id, - user.sr_uid, - user.sign_switch, - user.user_id, - ) - ) - if len(tasks) >= 1: - await asyncio.gather(*tasks) - if already >= 1: - delay = 1 - else: - delay = 50 + random.randint(3, 45) - logger.info( - f'[SR签到] 已签到{len(tasks)}个用户, 等待{delay}秒进行下一次签到' - ) - tasks.clear() - already = 0 - await asyncio.sleep(delay) - await asyncio.gather(*tasks) - tasks.clear() - result = { - 'private_msg_list': deepcopy(private_msg_list), - 'group_msg_list': deepcopy(group_msg_list), - } - private_msg_list.clear() - group_msg_list.clear() - logger.info(result) - return result