✨ 支持notice推送

ArknightsUID/arknightsuid_ap/__init__.py

@@ -7,63 +7,45 @@ from gsuid_core.logger import logger
 from gsuid_core.models import Event
 from gsuid_core.segment import MessageSegment
 from gsuid_core.sv import SV
-from gsuid_core.utils.database.api import get_uid
-from gsuid_core.utils.error_reply import UID_HINT
 
 from ..utils.ark_prefix import PREFIX
-from ..utils.database.models import ArknightsBind
-
-# from .ap_text import get_ap_text
 from .draw_ap_card import get_ap_img
-
-# from .notice import get_notice_list
+from .notice import get_notice_list
 
 sv_get_ap = SV('ark查询体力')
 sv_get_ap_admin = SV('ark强制推送', pm=1)
 
 
-# @sv_get_ap.on_fullmatch((f'{PREFIX}当前状态'))
-# async def send_daily_info(bot: Bot, ev: Event):
-#     await bot.logger.info('开始执行[ark每日信息文字版]')
-#     uid = await get_uid(bot, ev, ArknightsBind)
-#     if uid is None:
-#         return await bot.send(UID_HINT)
-#     await bot.logger.info(f'[ark每日信息文字版]UID: {uid}')
-
-#     im = await get_ap_text(uid)
-#     await bot.send(im)
+@sv_get_ap_admin.on_fullmatch((f'{PREFIX}强制推送体力提醒'))  # noqa: UP034
+async def force_notice_job(bot: Bot, ev: Event):
+    await bot.logger.info('开始执行[ark强制推送体力信息]')
+    await ark_notice_job()
 
 
-# @sv_get_ap_admin.on_fullmatch((f'{PREFIX}强制推送体力提醒'))
-# async def force_notice_job(bot: Bot, ev: Event):
-#     await bot.logger.info('开始执行[ark强制推送体力信息]')
-#     await ark_notice_job()
+@scheduler.scheduled_job('cron', minute='*/30')
+async def ark_notice_job():
+    result = await get_notice_list()
+    logger.info('[ark推送检查]完成!等待消息推送中...')
+    logger.debug(result)
 
-
-# @scheduler.scheduled_job('cron', minute='*/30')
-# async def ark_notice_job():
-#     result = await get_notice_list()
-#     logger.info('[ark推送检查]完成!等待消息推送中...')
-#     logger.debug(result)
-
-#     # 执行私聊推送
-#     for bot_id in result:
-#         for BOT_ID in gss.active_bot:
-#             bot = gss.active_bot[BOT_ID]
-#             for user_id in result[bot_id]['direct']:
-#                 msg = result[bot_id]['direct'][user_id]
-#                 await bot.target_send(msg, 'direct', user_id, bot_id, '', '')
-#                 await asyncio.sleep(0.5)
-#             logger.info('[ark推送检查] 私聊推送完成')
-#             for gid in result[bot_id]['group']:
-#                 msg_list = []
-#                 for user_id in result[bot_id]['group'][gid]:
-#                     msg_list.append(MessageSegment.at(user_id))
-#                     msg = result[bot_id]['group'][gid][user_id]
-#                     msg_list.append(MessageSegment.text(msg))
-#                 await bot.target_send(msg_list, 'group', gid, bot_id, '', '')
-#                 await asyncio.sleep(0.5)
-#             logger.info('[ark推送检查] 群聊推送完成')
+    # 执行私聊推送
+    for bot_id in result:
+        for BOT_ID in gss.active_bot:
+            bot = gss.active_bot[BOT_ID]
+            for user_id in result[bot_id]['direct']:
+                msg = result[bot_id]['direct'][user_id]
+                await bot.target_send(msg, 'direct', user_id, bot_id, '', '')
+                await asyncio.sleep(0.5)
+            logger.info('[ark推送检查] 私聊推送完成')
+            for gid in result[bot_id]['group']:
+                msg_list = []
+                for user_id in result[bot_id]['group'][gid]:
+                    msg_list.append(MessageSegment.at(user_id))
+                    msg = result[bot_id]['group'][gid][user_id]
+                    msg_list.append(MessageSegment.text(msg))
+                await bot.target_send(msg_list, 'group', gid, bot_id, '', '')
+                await asyncio.sleep(0.5)
+            logger.info('[ark推送检查] 群聊推送完成')
 
 
 @sv_get_ap.on_fullmatch(

ArknightsUID/arknightsuid_ap/ap_text.py

@@ -1,99 +0,0 @@
-# import math
-# from datetime import datetime
-
-# from gsuid_core.data_store import get_res_path
-# from gsuid_core.logger import logger
-# from gsuid_core.utils.error_reply import get_error
-# from msgspec import json as msgjson
-
-# # from ..arknightsuid_resource.constants import Excel
-# from ..utils.ark_api import ark_skd_api
-# from ..utils.models.skland.models import PlayerStatusAp
-
-# daily_im = """*数据刷新可能存在一定延迟，请以当前游戏实际数据为准
-# ==============
-# 理智：{}/{}
-# 公开招募：{}/{}
-# 公招刷新：{}
-# 训练室：{}
-# 每周报酬合成玉：{}/{}
-# 每日任务：{}/{}
-# 每周任务：{}/{}
-# 数据增补仪：{}/{}
-# 数据增补条：{}/{}
-# =============="""
-
-
-# def seconds2hours(seconds: int) -> str:
-#     m, s = divmod(int(seconds), 60)
-#     h, m = divmod(m, 60)
-#     return '%02d:%02d:%02d' % (h, m, s)
-
-# def now_ap(ap: PlayerStatusAp) -> int:
-#     _ap =  ap.current + math.floor((datetime.now().timestamp() - ap.lastApAddTime) / 360)
-#     return _ap if _ap <= ap.max else ap.max
-
-
-# async def get_ap_text(uid: str) -> str:
-#     try:
-#         player_info = await ark_skd_api.get_game_player_info(uid)
-#         if isinstance(player_info, int):
-#             return get_error(player_info)
-
-#         player_save_path = get_res_path(['ArknightsUID', 'players'])
-
-#         with open(player_save_path / f'{player_info.status.uid}.json', 'wb') as file:
-#             file.write(msgjson.format(msgjson.encode(player_info), indent=4))
-
-#         ap = player_info.status.ap
-#         current_ap = now_ap(ap)
-#         max_ap = ap.max
-#         rec_time = ''
-#         if current_ap < max_ap:
-#             ap_recover_time = seconds2hours(
-#                 ap.completeRecoveryTime
-#             )
-#             next_ap_rec_time = seconds2hours(
-#                 8 * 60
-#                 - (
-#                     (max_ap - current_ap)
-#                     * 8
-#                     * 60
-#                     - int(ap.completeRecoveryTime)
-#                 )
-#             )
-#             rec_time = f' ({next_ap_rec_time}/{ap_recover_time})'
-
-#         accepted_epedition_num = dailydata['accepted_epedition_num']
-#         total_expedition_num = dailydata['total_expedition_num']
-#         finished_expedition_num = 0
-#         expedition_info: list[str] = []
-#         for expedition in dailydata['expeditions']:
-#             expedition_name = expedition['name']
-
-#             if expedition['status'] == 'Finished':
-#                 expedition_info.append(f'{expedition_name} 探索完成')
-#                 finished_expedition_num += 1
-#             else:
-#                 remaining_time: str = seconds2hours(
-#                     expedition['remaining_time']
-#                 )
-#                 expedition_info.append(
-#                     f'{expedition_name} 剩余时间{remaining_time}'
-#                 )
-
-#         expedition_data = '\n'.join(expedition_info)
-#         print(expedition_data)
-#         send_mes = daily_im.format(
-#             current_ap,
-#             max_ap,
-#             rec_time,
-#             accepted_epedition_num,
-#             finished_expedition_num,
-#             total_expedition_num,
-#             expedition_data,
-#         )
-#         return send_mes
-#     except TypeError:
-#         logger.exception('[查询当前状态]查询失败!')
-#         return '你绑定过的UID中可能存在过期CK~请重新绑定一下噢~'

ArknightsUID/arknightsuid_ap/draw_ap_card.py

@@ -1,10 +1,7 @@
 import asyncio
-import math
 from datetime import datetime, timedelta
-from io import BytesIO
 from pathlib import Path
 
-import aiohttp
 from gsuid_core.logger import logger
 from gsuid_core.utils.image.convert import convert_img
 from PIL import Image, ImageDraw
@@ -17,7 +14,7 @@ from ..utils.fonts.source_han_sans import (
     sans_font_26,
     sans_font_34,
 )
-from ..utils.models.skland.models import PlayerStatusAp
+from .utils import now_ap, seconds2hours_zhcn
 
 TEXT_PATH = Path(__file__).parent / 'texture2D'
 
@@ -39,24 +36,6 @@ white_color = (255, 255, 255)
 red_color = (235, 61, 75)
 
 
-def seconds2hours(seconds: int) -> str:
-    m, s = divmod(int(seconds), 60)
-    h, m = divmod(m, 60)
-    return '%02d:%02d:%02d' % (h, m, s)
-
-def now_ap(ap: PlayerStatusAp) -> int:
-    _ap =  ap.current + math.floor((datetime.now().timestamp() - ap.lastApAddTime) / 360)
-    return _ap if _ap <= ap.max else ap.max
-
-
-async def download_image(url: str) -> Image.Image:
-    async with aiohttp.ClientSession() as session:
-        async with session.get(url) as response:
-            img_data = await response.read()
-            img = Image.open(BytesIO(img_data))
-            return img
-
-
 async def get_ap_img(bot_id: str, user_id: str):
     try:
         uid_list = await ArknightsBind.get_uid_list_by_game(user_id, bot_id)
@@ -115,12 +94,6 @@ def get_error(img: Image.Image, uid: str, daily_data: int):
     return img
 
 
-async def seconds2hours_zhcn(seconds: int) -> str:
-    m, s = divmod(int(seconds), 60)
-    h, m = divmod(m, 60)
-    return '%02d小时%02d分钟' % (h, m)
-
-
 async def draw_ap_img(uid: str) -> Image.Image:
     # char
     char_pic = Image.open(TEXT_PATH / 'char_1028_texas2_1b.png').resize((1700, 1700)).convert('RGBA')
@@ -297,7 +270,7 @@ async def draw_ap_img(uid: str) -> Image.Image:
         remain_time = 0
         if remain_secs != -1:
             # 将remainSecs(剩余秒数) ，转换为几小时几分钟
-            remain_time = await seconds2hours_zhcn(remain_secs)
+            remain_time = seconds2hours_zhcn(remain_secs)
 
         char_cn_name = Excel.CHARATER_TABLE[training_char].name
         blue_bar_bg1_img = blue_bar_bg1.copy()

ArknightsUID/arknightsuid_ap/notice.py

@@ -1,105 +1,105 @@
-# from gsuid_core.gss import gss
-# from gsuid_core.logger import logger
+from gsuid_core.gss import gss
+from gsuid_core.logger import logger
 
-# from ..sruid_utils.api.mys.models import DailyNoteData
-# from ..starrailuid_config.sr_config import srconfig
-# from ..utils.api import get_sqla
-# from ..utils.mys_api import mys_api
+from ..arknightsuid_config.ark_config import arkconfig
+from ..utils.ark_api import ark_skd_api
+from ..utils.database.models import ArknightsPush, ArknightsUser
+from ..utils.models.skland.models import ArknightsPlayerInfoModel
+from .utils import now_ap
 
-# MR_NOTICE = '\n可发送[srmr]或者[sr每日]来查看更多信息！\n'
+MR_NOTICE = '\n可发送[arkmr]或者[ark每日]来查看更多信息！\n'
 
-# NOTICE = {
-#     'stamina': f'你的开拓力快满啦！{MR_NOTICE}',
-#     'go': f'你有派遣信息即将可收取！{MR_NOTICE}',
-# }
+NOTICE = {
+    'ap': f'你的理智快满啦！{MR_NOTICE}',
+    'training': f'你的专精即将可收取！{MR_NOTICE}',
+}
 
 
-# async def get_notice_list() -> dict[str, dict[str, dict]]:
-#     msg_dict: dict[str, dict[str, dict]] = {}
-#     for bot_id in gss.active_bot:
-#         sqla = get_sqla(bot_id)
-#         user_list = await sqla.get_all_push_user_list()
-#         for user in user_list:
-#             if user.sr_uid is not None:
-#                 raw_data = await mys_api.get_daily_data(user.sr_uid)
-#                 if isinstance(raw_data, int):
-#                     logger.error(f'[sr推送提醒]获取{user.sr_uid}的数据失败!')
-#                     continue
-#                 push_data = await sqla.select_push_data(user.sr_uid)
-#                 msg_dict = await all_check(
-#                     user.bot_id,
-#                     raw_data,
-#                     push_data.__dict__,
-#                     msg_dict,
-#                     user.user_id,
-#                     user.sr_uid,
-#                 )
-#     return msg_dict
+async def get_notice_list() -> dict[str, dict[str, dict]]:
+    msg_dict: dict[str, dict[str, dict]] = {}
+    for bot_id in gss.active_bot:
+        user_list = await ArknightsUser.get_all_push_user_list()
+        for user in user_list:
+            if user.uid is not None:
+                raw_data = await ark_skd_api.get_game_player_info(user.uid)
+                if isinstance(raw_data, int):
+                    logger.error(f'[ark推送提醒]获取{user.uid}的数据失败!')
+                    continue
+                push_data = await ArknightsPush.select_push_data(user.uid)
+                msg_dict = await all_check(
+                    user.bot_id,
+                    raw_data,
+                    push_data.__dict__,
+                    msg_dict,
+                    user.user_id,
+                    user.uid,
+                )
+    return msg_dict
 
 
-# async def all_check(
-#     bot_id: str,
-#     raw_data: DailyNoteData,
-#     push_data: dict,
-#     msg_dict: dict[str, dict[str, dict]],
-#     user_id: str,
-#     uid: str,
-# ) -> dict[str, dict[str, dict]]:
-#     sqla = get_sqla(bot_id)
-#     for mode in NOTICE.keys():
-#         # 检查条件
-#         if push_data[f'{mode}_is_push'] == 'on':
-#             if srconfig.get_config('CrazyNotice').data:
-#                 if not await check(mode, raw_data, push_data[f'{mode}_value']):
-#                     await sqla.update_push_data(
-#                         uid, {f'{mode}_is_push': 'off'}
-#                     )
-#                 continue
-#         # 准备推送
-#         if await check(mode, raw_data, push_data[f'{mode}_value']):
-#             if push_data[f'{mode}_push'] == 'off':
-#                 pass
-#             # on 推送到私聊
-#             else:
-#                 # 初始化
-#                 if bot_id not in msg_dict:
-#                     msg_dict[bot_id] = {'direct': {}, 'group': {}}
+async def all_check(
+    bot_id: str,
+    raw_data: ArknightsPlayerInfoModel,
+    push_data: dict,
+    msg_dict: dict[str, dict[str, dict]],
+    user_id: str,
+    uid: str,
+) -> dict[str, dict[str, dict]]:
+    for mode in NOTICE.keys():
+        # 检查条件
+        if push_data[f'{mode}_is_push'] is True:
+            if arkconfig.get_config('CrazyNotice').data:
+                if not await check(mode, raw_data, push_data[f'{mode}_value']):
+                    await ArknightsPush.update_push_data(
+                        uid, {f'{mode}_is_push': False}
+                    )
+                continue
+        # 准备推送
+        if await check(mode, raw_data, push_data[f'{mode}_value']):
+            if push_data[f'{mode}_push'] is False:
+                pass
+            # on 推送到私聊
+            else:
+                # 初始化
+                if bot_id not in msg_dict:
+                    msg_dict[bot_id] = {'direct': {}, 'group': {}}
 
-#                 if push_data[f'{mode}_push'] == 'on':
-#                     # 添加私聊信息
-#                     if user_id not in msg_dict[bot_id]['direct']:
-#                         msg_dict[bot_id]['direct'][user_id] = NOTICE[mode]
-#                     else:
-#                         msg_dict[bot_id]['direct'][user_id] += NOTICE[mode]
-#                     await sqla.update_push_data(uid, {f'{mode}_is_push': 'on'})
-#                 # 群号推送到群聊
-#                 else:
-#                     # 初始化
-#                     gid = push_data[f'{mode}_push']
-#                     if gid not in msg_dict[bot_id]['group']:
-#                         msg_dict[bot_id]['group'][gid] = {}
+                if push_data[f'{mode}_push'] is True:
+                    # 添加私聊信息
+                    if user_id not in msg_dict[bot_id]['direct']:
+                        msg_dict[bot_id]['direct'][user_id] = NOTICE[mode]
+                    else:
+                        msg_dict[bot_id]['direct'][user_id] += NOTICE[mode]
+                    await ArknightsPush.update_push_data(uid, {f'{mode}_is_push': True})
+                # 群号推送到群聊
+                else:
+                    # 初始化
+                    gid = push_data[f'{mode}_push']
+                    if gid not in msg_dict[bot_id]['group']:
+                        msg_dict[bot_id]['group'][gid] = {}
 
-#                     if user_id not in msg_dict[bot_id]['group'][gid]:
-#                         msg_dict[bot_id]['group'][gid][user_id] = NOTICE[mode]
-#                     else:
-#                         msg_dict[bot_id]['group'][gid][user_id] += NOTICE[mode]
-#                     await sqla.update_push_data(uid, {f'{mode}_is_push': 'on'})
-#     return msg_dict
+                    if user_id not in msg_dict[bot_id]['group'][gid]:
+                        msg_dict[bot_id]['group'][gid][user_id] = NOTICE[mode]
+                    else:
+                        msg_dict[bot_id]['group'][gid][user_id] += NOTICE[mode]
+                    await ArknightsPush.update_push_data(uid, {f'{mode}_is_push': True})
+    return msg_dict
 
 
-# async def check(mode: str, data: DailyNoteData, limit: int) -> bool:
-#     if mode == 'resin':
-#         if data['current_stamina'] >= limit:
-#             return True
-#         elif data['current_stamina'] >= data['max_stamina']:
-#             return True
-#         else:
-#             return False
-#     if mode == 'go':
-#         for i in data['expeditions']:
-#             if i['status'] == 'Ongoing':
-#                 if int(i['remaining_time']) <= limit * 60:
-#                     return True
-#             else:
-#                 return True
-#         return False
+async def check(mode: str, data: ArknightsPlayerInfoModel, limit: int) -> bool:
+    if mode == 'ap':
+        current_ap = now_ap(data.status.ap)
+        if current_ap >= limit:
+            return True
+        elif current_ap >= data.status.ap.max:
+            return True
+        else:
+            return False
+    if mode == 'training':
+        if data.building.training:
+            remain_secs = data.building.training.remainSecs
+            if remain_secs <= limit * 60:
+                return True
+            else:
+                return True
+    return False

ArknightsUID/arknightsuid_ap/utils.py

@@ -0,0 +1,15 @@
+import math
+from datetime import datetime
+
+from ..utils.models.skland.models import PlayerStatusAp
+
+
+def seconds2hours_zhcn(seconds: int) -> str:
+    m, s = divmod(int(seconds), 60)
+    h, m = divmod(m, 60)
+    return '%02d小时%02d分钟' % (h, m)
+
+
+def now_ap(ap: PlayerStatusAp) -> int:
+    _ap =  ap.current + math.floor((datetime.now().timestamp() - ap.lastApAddTime) / 360)
+    return _ap if _ap <= ap.max else ap.max

ArknightsUID/arknightsuid_config/config_default.py

@@ -1,5 +1,6 @@
 from gsuid_core.utils.plugins_config.models import (
     GSC,
+    GsBoolConfig,
     GsStrConfig,
 )
 
@@ -9,4 +10,9 @@ CONIFG_DEFAULT: dict[str, GSC] = {
         '用于本插件的前缀设定',
         'ark',
     ),
+    'CrazyNotice': GsBoolConfig(
+        '催命模式',
+        '开启后当达到推送阈值将会一直推送',
+        False,
+    ),
 }

ArknightsUID/arknightsuid_roleinfo/__init__.py

@@ -17,6 +17,4 @@ async def send_role_info(bot: Bot, ev: Event):
         return '你还没有绑定UID噢,请使用[ark绑定uid123]完成绑定!'
 
     await bot.logger.info('开始执行[ark查询信息]')
-    # await get_role_img(uid)
-    # await bot.send('WIP')
     await bot.send(await get_role_img(uid))

ArknightsUID/utils/database/models.py

@@ -1,7 +1,6 @@
-from gsuid_core.utils.database.base_models import (
-    Bind,
-    User,
-)
+from typing import Literal
+
+from gsuid_core.utils.database.base_models import Bind, Push, T_BaseIDModel, User
 from gsuid_core.webconsole.mount_app import GsAdminModel, PageSchema, site
 from sqlmodel import Field
 
@@ -16,6 +15,59 @@ class ArknightsUser(User, table=True):
     cred: str | None = Field(default=None, title='SKD凭证')
 
 
+class ArknightsPush(Push, table=True):
+    uid: str | None = Field(default=None, title='明日方舟UID')
+    skd_uid: str | None = Field(default=None, title='森空岛用户ID')
+    ap_push: bool | None = Field(default=False, title='理智推送')
+    ap_value: int | None = Field(default=130, title='理智推送阈值')
+    ap_is_push: bool | None = Field(default=False, title='理智是否已经推送')
+    training_push: bool | None = Field(default=False, title='训练室推送')
+    training_value: int | None = Field(default=30, title='训练室推送阈值')
+    training_is_push: bool | None = Field(default=False, title='训练室是否已经推送')
+
+    @classmethod
+    async def insert_push_data(cls, uid: str, skd_uid: str):
+        await cls.full_insert_data(
+            bot_id=cls.bot_id,
+            uid=uid,
+            skd_uid=skd_uid,
+            ap_push=False,
+            ap_value=2100,
+            ap_is_push=False,
+            training_push=True,
+            training_value=140,
+            training_is_push=False
+        )
+
+    @classmethod
+    async def update_push_data(cls, uid: str, data: dict) -> bool:
+        retcode = -1
+        if await cls.data_exist(uid=uid):
+            retcode = await cls.update_data_by_uid(
+                uid, cls.bot_id, None, **data
+            )
+        return not bool(retcode)
+
+    @classmethod
+    async def change_push_status(
+        cls,
+        mode: Literal['ap', 'training'],
+        uid: str,
+        status: str,
+    ):
+        await cls.update_push_data(uid, {f'{mode}_is_push': status})
+
+    @classmethod
+    async def select_push_data(
+        cls: type[T_BaseIDModel], uid: str
+    ) -> T_BaseIDModel | None:
+        return await cls.base_select_data(uid=uid)
+
+    @classmethod
+    async def push_exists(cls, uid: str) -> bool:
+        return await cls.data_exist(uid=uid)
+
+
 @site.register_admin
 class ArknightsBindadmin(GsAdminModel):
     pk_name = 'id'
@@ -32,3 +84,12 @@ class ArknightsUseradmin(GsAdminModel):
 
     # 配置管理模型
     model = ArknightsUser
+
+
+@site.register_admin
+class ArknightsPushadmin(GsAdminModel):
+    pk_name = 'id'
+    page_schema = PageSchema(label='明日方舟推送管理', icon='fa fa-database')  # type: ignore
+
+    # 配置管理模型
+    model = ArknightsPush