try:
    import ujson as json
except ImportError:
    import json
from pathlib import Path
from typing import Tuple

import requests

app_code = '4ca99fa6b56cc2ba'
curr_path = Path(__file__).parent
header_path = curr_path / 'header.json'
header_login_path = curr_path / 'header_login.json'
# 使用token获得认证代码
grant_code_url = "https://as.hypergryph.com/user/oauth2/v2/grant"
# 使用认证代码获得cred
cred_code_url = "https://zonai.skland.com/api/v1/user/auth/generate_cred_by_code"
# 绑定的角色url
binding_url = "https://zonai.skland.com/api/v1/game/player/binding"
# 签到url
sign_url = "https://zonai.skland.com/api/v1/game/attendance"


class ArknightsException(Exception):
    pass


class ArknightsHelper:
    def __init__(self, cred: str):
        self.cred = cred

    @property
    def header_login(self) -> dict:
        with open(header_login_path, 'r', encoding='utf-8') as f:
            return json.load(f)

    @property
    def header(self) -> dict:
        with open(header_path, 'r', encoding='utf-8') as f:
            header = json.load(f)
            header['cred'] = self.cred
            return header

    def get_binding_list(self) -> list:
        v = []
        resp = requests.get(binding_url, headers=self.header).json()
        if resp['code'] != 0:
            msg = f"请求角色列表出现问题: {resp['message']}"
            if resp.get('message') == '用户未登录':
                msg += '\n用户登录可能失效了, 请重新运行此程序!'
            raise ArknightsException(msg)
        for i in resp['data']['list']:
            if i.get('appCode') != 'arknights':
                continue
            v.extend(i.get('bindingList'))
        return v

    def do_sign(self) -> str:
        characters = self.get_binding_list()
        msgs = []
        for i in characters:
            body = {
                'uid': i.get('uid'),
                'gameId': 1
            }
            resp = requests.post(sign_url, headers=self.header, json=body).json()
            if resp['code'] != 0:
                msgs.append(f'角色{i.get("nickName")}({i.get("channelName")})签到失败了!原因: {resp.get("message")}')
                continue
            awards = resp['data']['awards']
            for j in awards:
                res = j['resource']
                msgs.append(
                    f'角色{i.get("nickName")}({i.get("channelName")})签到成功, 获得了{res["name"]}×{j.get("count") or 1}'
                )
        return '\n\n'.join(msgs)

    async def sign(self) -> Tuple[str, str]:
        """签到
        :return: msg: 消息, info: 状态 (success, warning, error)
        """
        try:
            msgs = self.do_sign()
            return msgs, 'success'
        except ArknightsException as e:
            return f'签到失败了, 请检查cred是否正确 -> 发送 [兔兔设置凭证] 查看使用说明, 失败原因: {e.__str__()}', 'error'
        except Exception as e:
            return f'签到失败了, 原因: {e}', 'error'

    @staticmethod
    def tip() -> bytes:
        pic_path = curr_path / 'tip.png'
        return pic_path.read_bytes()

helper.py
wutongshufqw/amiyabot-skland-sign
import os
from typing import List, Optional

from amiyabot import Message
from amiyabot.builtin.messageChain import Chain
from amiyabot.database import connect_database, ModelClass, table
from amiyabot.factory import BotHandlerFactory
from peewee import AutoField, CharField, BooleanField

from core import AmiyaBotPluginInstance, log, bot as main_bot

from .helper import ArknightsHelper

curr_dir = os.path.dirname(__file__)
message_record = {}
db = connect_database('resource/plugins/tools/skland.db')


class SignBaseModel(ModelClass):
    class Meta:
        database = db


@table
class SklandSign(SignBaseModel):
    id: int = AutoField()
    user_id: str = CharField()
    open: bool = BooleanField(default=False)
    remark: str = CharField(null=True)

    @staticmethod
    def get_skland_sign(user_id: str) -> Optional['SklandSign']:
        return SklandSign.get_or_none(SklandSign.user_id == user_id)

    @staticmethod
    def set_skland_sign(user_id: str, open_: Optional[bool] = None, cred: Optional[str] = None):
        result = SklandSign.get_or_none(SklandSign.user_id == user_id)
        if result:
            if open_:
                result.open = open_
            if cred:
                result.remark = cred
            return result.save()
        else:
            return SklandSign.create(user_id=user_id, open=open_ if open_ else False, remark=cred if cred else None)
            

    @staticmethod
    def get_skland_sign_open() -> List['SklandSign']:
        return SklandSign.select().where(SklandSign.open == True)


class ToolsPluginInstance(AmiyaBotPluginInstance):
    pass


bot = ToolsPluginInstance(
    name='森空岛签到插件',
    version='1.2',
    plugin_id='amiyabot-skland-sign',
    plugin_type='tools',
    description='森空岛每日签到',
    document=f'{curr_dir}/README.md',
    instruction=f'{curr_dir}/README.md',
    global_config_default=f'{curr_dir}/global_config_default.json',
    global_config_schema=f'{curr_dir}/global_config_schema.json',
)


# 每个群聊存储一份data用于发送消息
async def record_msg(data: Message):
    global message_record
    message_record[data.channel_id] = data
    return False


@bot.on_message(verify=record_msg, check_prefix=False)
async def auction_offer(data: Message):
    pass


async def groups_send_message(groups: List[str], msg: str, markdown: bool = True):
    for group_id_ in groups:
        sp = group_id_.split('_')
        if sp[0] in main_bot:
            try:
                data = message_record[sp[1]]
                if markdown:
                    await data.send(Chain(data, at=False).markdown(msg, is_dark=True))
                else:
                    await data.send(Chain(data, at=False).text(msg))
            except KeyError:
                if markdown:
                    await main_bot[sp[0]].send_message(Chain().markdown(msg, is_dark=True), channel_id=sp[1])
                else:
                    await main_bot[sp[0]].send_message(Chain().text(msg), channel_id=sp[1])


async def group_send_message(group: str, msg: str, markdown: bool = True):
    sp = group.split('_')
    if sp[0] in main_bot:
        try:
            data = message_record[sp[1]]
            if markdown:
                await data.send(Chain(data, at=False).markdown(msg, is_dark=True))
            else:
                await data.send(Chain(data, at=False).text(msg))
        except KeyError:
            if markdown:
                await main_bot[sp[0]].send_message(Chain().markdown(msg, is_dark=True), channel_id=sp[1])
            else:
                await main_bot[sp[0]].send_message(Chain().text(msg), channel_id=sp[1])


@bot.on_message(keywords=['设置凭证'], allow_direct=True, level=5)
async def set_skland(data: Message):
    res = await data.wait(Chain(data).image(ArknightsHelper.tip()).text(
        '森空岛网址: \nhttps://www.skland.com/\n方法二代码:\nlocalStorage.getItem("SK_OAUTH_CRED_KEY");\n方法三代码:\njavascript:prompt(undefined, localStorage.getItem("SK_OAUTH_CRED_KEY"));'),
        max_time=120)
    if res:
        await res.recall()
        cred = res.text_original
        SklandSign.set_skland_sign(data.user_id, cred=cred)
        return Chain(data).text('设置成功')
    return Chain(data).text('操作超时')


@bot.on_message(keywords=['方舟签到'], allow_direct=True, level=5)
async def sign_skland(data: Message):
    async def true_or_false(message: Message):
        if message.text_original in ['是', '否']:
            return True

    user_id = data.user_id
    sign = SklandSign.get_skland_sign(user_id)
    if sign and sign.remark:
        if sign.open:
            reply = await data.wait(Chain(data).text('森空岛签到已开启，是否关闭？(是/否)'), data_filter=true_or_false, max_time=20)
            if reply:
                if reply.text_original == '是':
                    SklandSign.set_skland_sign(user_id, open_=False)
                    return Chain(data).text('森空岛签到已关闭')
                reply = await data.wait(Chain(data).text('是否立即签到？(是/否)'), data_filter=true_or_false, max_time=20)
                if reply and reply.text_original == '是':
                    msg, _ = await ArknightsHelper(sign.remark).sign()
                    return Chain(data).text(msg)
            return Chain(data).text('操作已取消')
        reply = await data.wait(Chain(data).text('森空岛签到已关闭，是否开启？(是/否)'), data_filter=true_or_false, max_time=20)
        if reply and reply.text_original == '是':
            SklandSign.set_skland_sign(user_id, open_=True)
            reply = await data.wait(Chain(data).text('森空岛签到已开启，是否立即签到？(是/否)'), data_filter=true_or_false, max_time=20)
            if reply and reply.text_original == '是':
                msg, _ = await ArknightsHelper(sign.remark).sign()
                return Chain(data).text(msg)
        return Chain(data).text('操作已取消')
    return Chain(data).text('博士，您尚未绑定 Cred，请发送 “兔兔设置凭证” 查看绑定说明。')


@bot.timed_task(sub_tag='skland_sign', trigger='cron', hour=6)
async def skland_sign(instance: BotHandlerFactory):
    msgs = []
    users = SklandSign.get_skland_sign_open()
    for user in users:
        user_id = user.user_id
        cred = user.remark
        if cred:
            msg, _ = await ArknightsHelper(cred).sign()
            msgs.append(f'用户 {user_id} 签到结果: {msg}')
        else:
            msgs.append(f'用户 {user_id} 未绑定 Cred')
    if msgs:
        config = bot.get_config('skland_sign')
        if config:
            groups = []
            for group in config:
                groups.append(f'{group["appid"]}_{group["group"]}')
            await groups_send_message(groups, '\n\n'.join(msgs))

main.py
wutongshufqw/amiyabot-skland-sign