支持自动签到

ArknightsUID/arknightsuid_config/__init__.py

@@ -0,0 +1,76 @@
+import re
+
+from gsuid_core.bot import Bot
+from gsuid_core.logger import logger
+from gsuid_core.models import Event
+from gsuid_core.sv import SV
+from gsuid_core.utils.error_reply import UID_HINT
+
+from ..utils.database.models import ArknightsBind
+from .set_config import set_config_func, set_push_value
+
+sv_self_config = SV("ark配置")
+
+
+# @sv_self_config.on_fullmatch(("ark配置", "方舟配置"))
+# async def send_config_card(bot: Bot, ev: Event):
+#     logger.info("开始执行[ark配置]")
+#     im = await draw_config_img(ev.bot_id)
+#     await bot.send(im)
+
+
+@sv_self_config.on_prefix(("ark设置"))  # noqa: UP034
+async def send_config_ev(bot: Bot, ev: Event):
+    logger.info("开始执行[设置阈值信息]")
+
+    uid = await ArknightsBind.get_uid_by_game(ev.user_id, bot.bot_id)
+    if uid is None:
+        return await bot.send(UID_HINT)
+
+    func = "".join(re.findall("[\u4e00-\u9fa5]", ev.text.replace("阈值", "")))
+    value = re.findall(r"\d+", ev.text)
+    value = value[0] if value else None
+
+    if value is None:
+        return await bot.send("请输入正确的阈值数字...")
+
+    logger.info(f"[设置阈值信息]func: {func}, value: {value}")
+    im = await set_push_value(ev.bot_id, func, uid, int(value))
+    await bot.send(im)
+
+
+# 开启 自动签到 功能
+@sv_self_config.on_prefix(("ark开启", "ark关闭"))
+async def open_switch_func(bot: Bot, ev: Event):
+    user_id = ev.user_id
+    config_name = ev.text
+
+    logger.info(f"[{user_id}]尝试[{ev.command[2:]}]了[{ev.text}]功能")
+
+    if ev.command == "ark开启":
+        query = True
+        gid = ev.group_id if ev.group_id else "on"
+    else:
+        query = False
+        gid = "off"
+
+    is_admin = ev.user_pm <= 2
+    if ev.at and is_admin:
+        user_id = ev.at
+    elif ev.at:
+        return await bot.send("你没有权限...")
+
+    uid = await ArknightsBind.get_uid_by_game(ev.user_id, bot.bot_id)
+    if uid is None:
+        return await bot.send(UID_HINT)
+
+    im = await set_config_func(
+        ev.bot_id,
+        config_name=config_name,
+        uid=uid,
+        user_id=user_id,
+        option=gid,
+        query=query,
+        is_admin=is_admin,
+    )
+    await bot.send(im)

ArknightsUID/arknightsuid_config/set_config.py

@@ -0,0 +1,86 @@
+from typing import Optional
+
+from gsuid_core.logger import logger
+
+from ..utils.database.models import ArknightsPush, ArknightsUser
+from .ark_config import arkconfig
+from .config_default import CONIFG_DEFAULT
+
+PUSH_MAP = {
+    "理智": "ap",
+    "训练室": "train",
+}
+PRIV_MAP = {
+    "自动签到": "sign",
+    "推送": "push",
+}
+
+
+async def set_push_value(bot_id: str, func: str, uid: str, value: int):
+    if func in PUSH_MAP:
+        status = PUSH_MAP[func]
+    else:
+        return "该配置项不存在!"
+    logger.info(f"[设置推送阈值]func: {status}, value: {value}")
+    if await ArknightsPush.update_push_data(uid, {f"{status}_value": value}):
+        return f"设置成功!\n当前{func}推送阈值:{value}"
+    else:
+        return "设置失败!\n请检查参数是否正确!"
+
+
+async def set_config_func(
+    bot_id: str,
+    config_name: str = "",
+    uid: str = "0",
+    user_id: str = "",
+    option: str = "0",
+    query: Optional[bool] = None,
+    is_admin: bool = False,
+):
+    # 这里将传入的中文config_name转换为英文status
+    for _name in CONIFG_DEFAULT:
+        config = CONIFG_DEFAULT[_name]
+        if config.title == config_name and isinstance(config.data, bool):
+            name = _name
+            break
+    else:
+        logger.info(
+            f"uid: {uid}, option: {option}, config_name: {config_name}"
+        )
+        if config_name in PRIV_MAP:
+            # 执行设置
+            await ArknightsUser.update_user_data(
+                uid,
+                {
+                    f"{PRIV_MAP[config_name]}_switch": option,
+                },
+            )
+        elif config_name.replace("推送", "") in PUSH_MAP:
+            await ArknightsPush.update_push_data(
+                uid,
+                {
+                    f'{PUSH_MAP[config_name.replace("推送", "")]}_push': option,
+                },
+            )
+        else:
+            return "该配置项不存在!"
+
+        if option == "on":
+            succeed_msg = "开启至私聊消息!"
+        elif option == "off":
+            succeed_msg = "关闭!"
+        else:
+            succeed_msg = f"开启至群{option}"
+        return f"{config_name}已{succeed_msg}"
+
+    if is_admin:
+        logger.info(f"config_name:{config_name},query:{query}")
+        # 执行设置
+        if query is not None:
+            arkconfig.set_config(name, query)
+            im = "成功设置{}为{}。".format(config_name, "开" if query else "关")
+        else:
+            im = "未传入参数query!"
+    else:
+        im = "只有管理员才能设置群服务。"
+    return im

ArknightsUID/arknightsuid_signin/sign.py

@@ -2,6 +2,7 @@ import asyncio
 import random
 from copy import deepcopy
 from datetime import datetime
+from typing import List
 
 from gsuid_core.gss import gss
 from gsuid_core.logger import logger
@@ -114,13 +115,15 @@ async def daily_sign():
     global already  # noqa: PLW0603
     tasks = []
     for _ in gss.active_bot:
-        user_list = await ArknightsUser.get_all_user()
+        user_list: List[ArknightsUser] = await ArknightsUser.get_all_user()
+        logger.info(f'[ARK签到] 共有{len(user_list)}个用户需要签到')
+        logger.info(f'[ARK签到] {user_list}')
         for user in user_list:
-            if user.sign_switch != 'off' and user.ark_uid is not None:
+            if user.sign_switch != 'off' and user.uid is not None:
                 tasks.append(
                     single_daily_sign(
                         user.bot_id,
-                        user.ark_uid,
+                        user.uid,
                         user.sign_switch,
                         user.user_id,
                     )

ArknightsUID/arknightsuid_user/deal_skd_cred.py

@@ -1,8 +1,11 @@
 import re
-from ..utils.ark_api import ark_skd_api
-from ..utils.database.models import ArknightsBind, ArknightsUser
 
-ERROR_HINT = '添加失败, 格式为: skd添加cred Cred 例如: skd添加cred VropL583Sb1hClS5buQ4nSASkDlL8tMT'
+from ..utils.ark_api import ark_skd_api
+from ..utils.database.models import ArknightsBind, ArknightsPush, ArknightsUser
+
+ERROR_HINT = (
+    '添加失败, 格式为: skd添加cred Cred 例如: skd添加cred VropL583Sb1hClS5buQ4nSASkDlL8tMT'
+)
 UID_HINT = '添加失败, 请先绑定明日方舟UID'
 
 
@@ -18,7 +21,9 @@ async def deal_skd_cred(bot_id: str, cred: str, user_id: str) -> str:
     # refresh token
     token = await ark_skd_api.refresh_token(match.group())
 
-    check_cred = await ark_skd_api.check_cred_valid(cred=match.group(), token=token)
+    check_cred = await ark_skd_api.check_cred_valid(
+        cred=match.group(), token=token
+    )
 
     if isinstance(check_cred, bool):
         return 'Cred无效!'
@@ -30,10 +35,25 @@ async def deal_skd_cred(bot_id: str, cred: str, user_id: str) -> str:
 
     # 检查是否已经绑定过 Cred, 如果有的话就 update
     skd_data = await ArknightsUser.select_data_by_uid(uid)
+    push_data = await ArknightsPush.select_data_by_uid(uid)
     if not skd_data:
-        await ArknightsUser.insert_data(user_id, bot_id,
+        await ArknightsUser.insert_data(
-                                        cred=match.group(), uid=uid, skd_uid=skd_uid, token=token)
+            user_id,
+            bot_id,
+            cred=match.group(),
+            uid=uid,
+            skd_uid=skd_uid,
+            token=token,
+        )
     else:
-        await ArknightsUser.update_data(user_id, bot_id,
+        await ArknightsUser.update_data(
-                                        cred=match.group(), uid=uid, skd_uid=skd_uid, token=token)
+            user_id,
+            bot_id,
+            cred=match.group(),
+            uid=uid,
+            skd_uid=skd_uid,
+            token=token,
+        )
+    if not push_data:
+        await ArknightsPush.insert_push_data(bot_id, uid=uid, skd_uid=skd_uid)
     return '添加成功!'

ArknightsUID/utils/database/models.py

@@ -1,10 +1,19 @@
-from typing import Dict, Literal, Optional, Type, Union
+from typing import Dict, List, Literal, Optional, Type, TypeVar, Union
 
-from gsuid_core.utils.database.base_models import Bind, Push, T_BaseIDModel, User, with_session, BaseModel
+from gsuid_core.utils.database.base_models import (
+    BaseModel,
+    Bind,
+    Push,
+    T_BaseIDModel,
+    User,
+    with_session,
+)
 from gsuid_core.webconsole.mount_app import GsAdminModel, PageSchema, site
-from sqlmodel import Field
-from sqlalchemy.future import select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+from sqlmodel import Field
+
+T_ARK_User = TypeVar("T_ARK_User", bound="ArknightsUser")
 
 
 class ArknightsBind(Bind, table=True):
@@ -20,23 +29,31 @@ class ArknightsUser(User, table=True):
     @classmethod
     @with_session
     async def select_data_by_cred(
-        cls,
+        cls, session: AsyncSession, cred: str
-        session: AsyncSession,
-        cred: str
     ) -> Union[BaseModel, None]:
-        sql= select(cls).where(cls.cred == cred)
+        sql = select(cls).where(cls.cred == cred)
         result = await session.execute(sql)
         data = result.scalars().all()
         return data[0] if data else None
 
+    @classmethod
+    @with_session
+    async def get_all_user(
+        cls: Type[T_ARK_User], session: AsyncSession
+    ) -> List[T_ARK_User]:
+        sql = select(cls).where(cls.cred is not None, cls.cred != "")
+        result = await session.execute(sql)
+        data: List[T_ARK_User] = result.scalars().all()
+        return data
+
     @classmethod
     async def get_token_by_cred(cls, cred: str) -> Union[str, None]:
-        result =  await cls.select_data_by_cred(cred)
+        result = await cls.select_data_by_cred(cred)
         return getattr(result, 'token') if result else None
 
     @classmethod
     async def get_uid_by_cred(cls, cred: str) -> Union[str, None]:
-        result =  await cls.select_data_by_cred(cred)
+        result = await cls.select_data_by_cred(cred)
         return getattr(result, 'uid') if result else None
 
     @classmethod
@@ -53,6 +70,10 @@ class ArknightsUser(User, table=True):
             )
         return not bool(retcode)
 
+    @classmethod
+    async def update_user_data(cls, uid: str, data: Dict = {}):
+        return await cls.update_data_by_uid(uid, cls.bot_id, None, **data)
+
 
 class ArknightsPush(Push, table=True):
     uid: Union[str, None] = Field(default=None, title='明日方舟UID')
@@ -62,20 +83,22 @@ class ArknightsPush(Push, table=True):
     ap_is_push: Union[bool, None] = Field(default=False, title='理智是否已经推送')
     training_push: Union[bool, None] = Field(default=False, title='训练室推送')
     training_value: Union[int, None] = Field(default=30, title='训练室推送阈值')
-    training_is_push: Union[bool, None] = Field(default=False, title='训练室是否已经推送')
+    training_is_push: Union[bool, None] = Field(
+        default=False, title='训练室是否已经推送'
+    )
 
     @classmethod
-    async def insert_push_data(cls, uid: str, skd_uid: str):
+    async def insert_push_data(cls, bot_id: str, uid: str, skd_uid: str):
         await cls.full_insert_data(
-            bot_id=cls.bot_id,
+            bot_id=bot_id,
             uid=uid,
             skd_uid=skd_uid,
-            ap_push=False,
+            ap_push=True,
-            ap_value=2100,
+            ap_value=110,
             ap_is_push=False,
             training_push=True,
             training_value=140,
-            training_is_push=False
+            training_is_push=False,
         )
 
     @classmethod