✨ 使用GsCore.utils.sign替换默认sr签到, 增加sr开启自动签到 (警告: 请重新开启) (#13)

.vscode/extensions.json

@@ -1,7 +1,7 @@
 {
   "recommendations": [
     "ms-python.python",
-    "ms-python.vscode-pylance",
+    "detachhead.basedpyright",
     "ms-python.isort",
     "ms-python.black-formatter"
   ]

.vscode/settings.json

@@ -1,11 +1,5 @@
 {
-  "python.languageServer": "Pylance",
+  "python.languageServer": "None",
-  "python.analysis.typeCheckingMode": "basic",
-  "cSpell.words": [
-    "enka",
-    "genshin",
-    "genshinuid"
-  ],
   "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",
+  "basedpyright.analysis.extraPaths": [
-  "python.linting.flake8Enabled": true,
-  "python.linting.flake8CategorySeverity.W": "Warning",
-  "python.linting.flake8CategorySeverity.F": "Warning",
-  "python.linting.flake8CategorySeverity.E": "Warning",
-  "python.analysis.extraPaths": [
     "${workspaceFolder}/../../../"
   ],
-  "python.autoComplete.extraPaths": [
-    "${workspaceFolder}/../../../"
-  ],
-  "python.analysis.include": [
-    "${workspaceFolder}/../../../"
-  ]
 }

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))

StarRailUID/starrailuid_config/config_default.py

@@ -11,10 +11,10 @@ CONIFG_DEFAULT: Dict[str, GSC] = {
     'SignTime': GsListStrConfig(
         '每晚签到时间设置', '每晚米游社签到时间设置(时,分)', ['0', '38']
     ),
-    'SignReportSimple': GsBoolConfig(
+    'PrivateSignReport': GsBoolConfig(
-        '简洁签到报告',
+        '签到私聊报告',
-        '开启后可以大大减少每日签到报告字数',
+        '关闭后将不再给任何人推送当天签到任务完成情况',
-        True,
+        False,
     ),
     'SchedSignin': GsBoolConfig(
         '定时签到',

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

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,
+)

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()
+        result = await daily_sign('sr')
-
+        if not IS_REPORT:
-
+            result['private_msg_dict'] = {}
-# 群聊内 签到 功能
+        await send_board_cast_msg(result)
-@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每日全部签到]群聊推送完成')

StarRailUID/starrailuid_signin/sign.py

@@ -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