Merge pull request #274 from Ljzd-PRO/dev

更新至 v2.3.0

README.md

@@ -18,16 +18,16 @@
 > v2.0.0 突破性更新
 > https://github.com/Ljzd-PRO/nonebot-plugin-mystool/releases/tag/v2.0.0
 
-### 2024.3.8 - v2.2.0
+### 2024.3.8 - v2.3.0
 
 #### 💡 新特性
-- 在未开启 崩坏：星穹铁道 便笺通知的情况下输出便笺检查结果日志
+- 增加微博超话兑换码推送功能 - by @Joseandluue (#272)
+- 更改每日任务执行顺序。先执行游戏签到，再执行米游币任务，以降低执行米游币任务时出现人机验证的概率 - by @Sakamakiiizayoi @Joseandluue
 
 #### 🐛 Bug 修复
-- 修复 OneBotV11 适配器私信发送失败的问题 (#260, #264)
-- 修复 崩坏：星穹铁道 便笺提醒失效的问题 - by @Joseandluue
-- 更改 崩坏：星穹铁道 便笺检查的每日实训/模拟宇宙通知逻辑 - by @Joseandluue
-- 更正 崩坏：星穹铁道 便笺检查的开拓力阈值范围 - by @Joseandluue
+- 修复每日任务自动进行游戏签到后，QQ聊天的主动私信推送失败的问题 (#270)
+- 更改 `UserAccount.mission_games`（用户米游币任务目标分区） 默认值为 `["BBSMission"]`，并在执行时检查该配置是否未空 (#261)
+- 修复可能出现的启动失败的问题（`AttributeError: 'NoneType' object has no attribute 'metadata'`） (#271)
 
 ## ⚡ 功能和特性
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nonebot-plugin-mystool"
-version = "v2.2.0"
+version = "v2.3.0"
 description = "QQ聊天、频道机器人插件 | 米游社工具-每日米游币任务、游戏签到、商品兑换、免抓包登录、原神崩铁便笺提醒"
 license = "MIT"
 authors = [

src/nonebot_plugin_mystool/_version.py

@@ -1 +1 @@
-__version__ = "v2.2.0"
+__version__ = "v2.3.0"

src/nonebot_plugin_mystool/api/weibo.py

@@ -0,0 +1,109 @@
+import re
+from urllib.parse import unquote
+
+import httpx
+
+from ..model import UserAccount
+from ..utils import logger
+
+
+def cookie_to_dict(cookie):
+    if cookie and '=' in cookie:
+        cookie = dict([line.strip().split('=', 1) for line in cookie.split(';')])
+    return cookie
+
+
+def nested_lookup(obj, key, with_keys=False, fetch_first=False):
+    result = list(_nested_lookup(obj, key, with_keys=with_keys))
+    if with_keys:
+        values = [v for k, v in _nested_lookup(obj, key, with_keys=with_keys)]
+        result = {key: values}
+    if fetch_first:
+        result = result[0] if result else result
+    return result
+
+
+def _nested_lookup(obj, key, with_keys=False):
+    if isinstance(obj, list):
+        for i in obj:
+            yield from _nested_lookup(i, key, with_keys=with_keys)
+    if isinstance(obj, dict):
+        for k, v in obj.items():
+            if key == k:
+                if with_keys:
+                    yield k, v
+                else:
+                    yield v
+            if isinstance(v, (list, dict)):
+                yield from _nested_lookup(v, key, with_keys=with_keys)
+
+
+class WeiboCode:
+    def __init__(self, account: UserAccount):
+        self.params = cookie_to_dict(account.weibo_params.replace('&', ';')) if account.weibo_params else None
+        """params: s=xxxxxx; gsid=xxxxxx; aid=xxxxxx; from=xxxxxx"""
+        self.cookie = cookie_to_dict(account.weibo_cookie)
+        self.container_id = {'原神': '100808fc439dedbb06ca5fd858848e521b8716',
+                             '星铁': '100808e1f868bf9980f09ab6908787d7eaf0f0'}
+        self.ua = 'WeiboOverseas/4.4.6 (iPhone; iOS 14.0.1; Scale/2.00)'
+        self.headers = {'User-Agent': self.ua}
+        self.follow_data_url = 'https://api.weibo.cn/2/cardlist'
+        self.sign_url = 'https://api.weibo.cn/2/page/button'
+        self.event_url = 'https://m.weibo.cn/api/container/getIndex?containerid={container_id}_-_activity_list'
+        self.draw_url = 'https://games.weibo.cn/prize/aj/lottery'
+
+    @property
+    async def get_ticket_id(self):
+        logger.info('开始获取微博兑换码ticket_id')
+        ticket_id = {}
+        for key, value in self.container_id.items():
+            url = self.event_url.replace('{container_id}', value)
+            async with httpx.AsyncClient() as client:
+                response = await client.get(url)
+            responses = response.json()
+            group = nested_lookup(responses, 'group', fetch_first=True)
+            if group:
+                ticket_id[key] = [i
+                                  for id in group
+                                  for i in re.findall(r'ticket_id=(\d*)', unquote(unquote(id['scheme'])))]
+            else:
+                logger.info(f'{key}超话未存在活动签到')
+        if not ticket_id:
+            return None
+        return ticket_id
+
+    async def get_code(self, id: str):
+        url = self.draw_url
+        self.headers.update({
+            'Referer': f'https://games.weibo.cn/prize/lottery?ua={self.ua}&from=10E2295010&ticket_id={id}&ext='
+        })
+        data = {
+            'ext': '', 'ticket_id': id, 'aid': self.params['aid'], 'from': self.params['from']
+        }
+        async with httpx.AsyncClient() as client:
+            response = await client.get(url, params=data, headers=self.headers, cookies=self.cookie)
+        responses = response.json()
+        code = responses['data']['prize_data']['card_no'] if responses['msg'] == 'success' or responses[
+            'msg'] == 'recently' else False
+        if responses['msg'] == 'fail':
+            responses['msg'] = responses['data']['fail_desc1']
+        result = {'success': True, 'id': id, 'code': code} if code else {'success': False, 'id': id,
+                                                                         'response': responses['msg']}
+        return result['code'] if result['success'] else responses['msg']
+
+    async def get_code_list(self):
+        ticket_id = await self.get_ticket_id
+        msg = ""
+        code = {key: [] for key in ticket_id.keys()}
+        for key, value in ticket_id.items():
+            for item in value:
+                code[key].append(await self.get_code(item))
+        for key, values in code.items():
+            msg += f"{key}微博兑换码：" \
+                   "\n1️⃣" \
+                   f"  \n{values[0]}" \
+                   "\n2️⃣" \
+                   f"  \n{values[1]}" \
+                   "\n3️⃣" \
+                   f"  \n{values[2]}"
+        return msg

src/nonebot_plugin_mystool/command/plan.py

@@ -12,17 +12,19 @@
 from nonebot.internal.matcher import Matcher
 from nonebot.params import CommandArg
 from nonebot_plugin_apscheduler import scheduler
+from nonebot_plugin_saa import Image
 from pydantic import BaseModel
 
 from ..api import BaseGameSign
 from ..api import BaseMission, get_missions_state
 from ..api.common import genshin_note, get_game_record, starrail_note
+from ..api.weibo import WeiboCode
 from ..command.common import CommandRegistry
 from ..command.exchange import generate_image
 from ..model import (MissionStatus, PluginDataManager, plugin_config, UserData, CommandUsage, GenshinNoteNotice,
                      StarRailNoteNotice)
-from ..utils import get_file, logger, COMMAND_BEGIN, GeneralMessageEvent, send_private_msg, \
-    get_all_bind, \
+from ..utils import get_file, logger, COMMAND_BEGIN, GeneralMessageEvent, GeneralGroupMessageEvent, \
+    send_private_msg, get_all_bind, \
     get_unique_users, get_validate, read_admin_list
 
 __all__ = [
@@ -190,6 +192,8 @@ async def _(event: Union[GeneralMessageEvent], matcher: Matcher):
     )
 )
 
+weibo_check = on_command(plugin_config.preference.command_start + '微博兑换码', priority=5, block=True)
+
 
 @manually_starrail_note_check.handle()
 async def _(event: Union[GeneralMessageEvent], matcher: Matcher):
@@ -285,7 +289,7 @@ async def perform_game_sign(
 
             # 用户打开通知或手动签到时，进行通知
             if user.enable_notice or matcher:
-                onebot_img_msg, qq_guild_img_msg = "", ""
+                onebot_img_msg, saa_img, qq_guild_img_msg = "", "", ""
                 get_info_status, info = await signer.get_info(account.platform)
                 get_award_status, awards = await signer.get_rewards()
                 if not get_info_status or not get_award_status:
@@ -303,6 +307,7 @@ async def perform_game_sign(
                               f"\n\n📅本月签到次数：{info.total_sign_day}"
                         img_file = await get_file(award.icon)
                         onebot_img_msg = OneBotV11MessageSegment.image(img_file)
+                        saa_img = Image(img_file)
                         qq_guild_img_msg = QQGuildMessageSegment.file_image(img_file)
                     else:
                         msg = (f"⚠️账户 {account.display_name} 🎮『{signer.name}』签到失败！请尝试重新签到，"
@@ -320,7 +325,7 @@ async def perform_game_sign(
                     for adapter in get_adapters().values():
                         if isinstance(adapter, OneBotV11Adapter):
                             for user_id in user_ids:
-                                await send_private_msg(use=adapter, user_id=user_id, message=msg + onebot_img_msg)
+                                await send_private_msg(use=adapter, user_id=user_id, message=msg + saa_img)
                         elif isinstance(adapter, QQGuildAdapter):
                             for user_id in user_ids:
                                 await send_private_msg(use=adapter, user_id=user_id, message=msg)
@@ -382,6 +387,9 @@ async def perform_bbs_sign(user: UserData, user_ids: Iterable[str], matcher: Mat
         # 在此处进行判断。因为如果在多个分区执行任务，会在完成之前就已经达成米游币任务目标，导致其他分区任务不会执行。
         finished = all(current == mission.threshold for mission, current in missions_state.state_dict.values())
         if not finished:
+            if not account.mission_games:
+                await matcher.send(
+                    f'⚠️🆔账户 {account.display_name} 未设置米游币任务目标分区，将跳过执行')
             for class_name in account.mission_games:
                 class_type = BaseMission.available_games.get(class_name)
                 if not class_type:
@@ -618,12 +626,13 @@ async def starrail_note_check(user: UserData, user_ids: Iterable[str], matcher:
 
                 # 每周模拟宇宙积分提醒
                 if note.current_rogue_score != note.max_rogue_score:
-                    if plugin_config.preference.notice_time:  
+                    if plugin_config.preference.notice_time:
                         msg += '❕您的模拟宇宙积分还没打满\n\n'
                         do_notice = True
 
                 if not do_notice:
-                    logger.info(f"崩铁实时便笺：账户 {account.display_name} 开拓力:{note.current_stamina},未满足推送条件")
+                    logger.info(
+                        f"崩铁实时便笺：账户 {account.display_name} 开拓力:{note.current_stamina},未满足推送条件")
                     return
 
             msg += "❖星穹铁道·实时便笺❖" \
@@ -641,6 +650,22 @@ async def starrail_note_check(user: UserData, user_ids: Iterable[str], matcher:
                     await send_private_msg(user_id=user_id, message=msg)
 
 
+async def weibo_code_check(user: UserData, user_ids: Iterable[str]):
+    """
+    是否开启微博兑换码功能的函数，并发送给用户任务执行消息。
+
+    :param user: 用户对象
+    :param user_ids: 发送通知的所有用户ID
+    """
+    for account in user.accounts.values():
+        if account.enable_weibo:
+            # account = UserAccount(account) 
+            weibo = WeiboCode(account)
+            msg = await weibo.get_code_list()
+            for user_id in user_ids:
+                await send_private_msg(user_id=user_id, message=msg)
+
+
 @scheduler.scheduled_job("cron", hour='0', minute='0', id="daily_goodImg_update")
 def daily_update():
     """
@@ -661,8 +686,8 @@ async def daily_schedule():
     logger.info(f"{plugin_config.preference.log_head}开始执行每日自动任务")
     for user_id, user in get_unique_users():
         user_ids = [user_id] + list(get_all_bind(user_id))
-        await perform_bbs_sign(user=user, user_ids=user_ids)
         await perform_game_sign(user=user, user_ids=user_ids)
+        await perform_bbs_sign(user=user, user_ids=user_ids)
     logger.info(f"{plugin_config.preference.log_head}每日自动任务执行完成")
 
 
@@ -679,3 +704,28 @@ async def auto_note_check():
         await genshin_note_check(user=user, user_ids=user_ids)
         await starrail_note_check(user=user, user_ids=user_ids)
     logger.info(f"{plugin_config.preference.log_head}自动便笺检查执行完成")
+
+
+@scheduler.scheduled_job("cron",
+                         hour=str(int(plugin_config.preference.plan_time.split(':')[0]) + 1),
+                         minute=plugin_config.preference.plan_time.split(':')[1],
+                         id="weibo_schedule")
+async def auto_weibo_check():
+    """
+    每日检查微博活动签到兑换码函数
+    """
+    logger.info(f"{plugin_config.preference.log_head}开始执行微博自动任务")
+    for user_id, user in get_unique_users():
+        user_ids = [user_id] + list(get_all_bind(user_id))
+        await weibo_code_check(user=user, user_ids=user_ids)
+    logger.info(f"{plugin_config.preference.log_head}微博自动任务执行完成")
+
+
+@weibo_check.handle()
+async def weibo_schedule(event: Union[GeneralMessageEvent], matcher: Matcher):
+    if isinstance(event, GeneralGroupMessageEvent):
+        await matcher.send("⚠️为了保护您的隐私，请私聊进行查询。")
+    else:
+        user_id = event.get_user_id()
+        user = PluginDataManager.plugin_data.users.get(user_id)
+        await weibo_code_check(user=user, user_ids=[user_id])

src/nonebot_plugin_mystool/model/config.py

@@ -256,11 +256,12 @@ class Config(BaseSettings.Config):
         logger.info(f"插件配置文件 {plugin_config_path} 不存在，已创建默认插件配置文件。")
 
 
-@_driver.on_startup
-def _():
-    """将 ``PluginMetadata.config`` 设为定义的插件配置对象 ``plugin_config``"""
-    plugin = nonebot.plugin.get_plugin(plugin_config.preference.plugin_name)
-    plugin.metadata.config = plugin_config
+# TODO: 可能产生 #271 的问题 https://github.com/Ljzd-PRO/nonebot-plugin-mystool/issues/271
+# @_driver.on_startup
+# def _():
+#     """将 ``PluginMetadata.config`` 设为定义的插件配置对象 ``plugin_config``"""
+#     plugin = nonebot.plugin.get_plugin(plugin_config.preference.plugin_name)
+#     plugin.metadata.config = plugin_config
 
 
 plugin_env = PluginEnv()

src/nonebot_plugin_mystool/model/data.py

@@ -241,14 +241,20 @@ class UserAccount(BaseModelWithSetter):
     '''是否开启米游社游戏签到计划'''
     enable_resin: bool = True
     '''是否开启便笺提醒'''
+    enable_weibo: bool = False
+    '''是否开启微博兑换码功能'''
     platform: Literal["ios", "android"] = "ios"
     '''设备平台'''
-    mission_games: List[str] = []
+    mission_games: List[str] = ["BBSMission"]
     '''在哪些板块执行米游币任务计划 为 BaseMission 子类名称'''
     user_stamina_threshold: int = 240
     '''崩铁便笺体力提醒阈值，0为一直提醒'''
     user_resin_threshold: int = 160
     '''原神便笺树脂提醒阈值，0为一直提醒'''
+    weibo_cookie: str = ""
+    '''微博查询活动签到用的 cookie'''
+    weibo_params: str = ""
+    '''微博查询活动签到用的 params'''
 
     def __init__(self, **data: Any):
         if not data.get("device_id_ios") or not data.get("device_id_android"):