yocool.py

# -*- coding: utf-8 -*-
import hoshino, os, json, shutil, zipfile, asyncio, glob
from nonebot import on_command, get_bot, scheduler
from hoshino import aiorequests, util

# 自动更新结果是否通知主人
NOTICE = False

path = os.path.dirname(__file__)

# yobot目录的绝对路径，即__init__.py所在目录
yobot_path = './hoshino/modules/yobot/yobot/'   

yobot_themes_path = os.path.join(yobot_path, 'src', 'client', 'public')
yocool_themes_path = os.path.join(path, 'public')
backup_themes_path = os.path.join(path, 'backup', 'public')
current_info_path = os.path.join(path, 'yocool_info.json')

yocool_info_url = 'https://api.pcrlink.cn/yocool/yocool_info.json'
yocool_themes_url = 'https://api.pcrlink.cn/yocool/yocool_themes.json'
git_yocool_releases = 'https://hub.fastgit.org/A-kirami/YoCool/releases/download/'


async def get_api_url(api_url):
    '''
    获取api链接中的信息
    '''
    try:
        api_result = await aiorequests.get(url=api_url, timeout=10)
    except:
        hoshino.logger.error('api链接无法访问')
    if api_result.status_code != 200:
        hoshino.logger.error(f'获取api信息时发生错误{api_result.status_code}')
    api_result_json = await api_result.json()
    return api_result_json


def get_install_state() -> int:
    '''
    检查安装状态
    '''
    hoshino.logger.info('检查YoCool安装状态')
    if os.path.exists(current_info_path):
        with open(current_info_path, 'r', encoding='utf-8') as lv:
            lvj = json.load(lv)
            current_tms = int(lvj["Install"])
    else:
        current_tms = 0
    return current_tms


def get_current_ver() -> int:
    '''
    获取本地版本号
    '''
    if os.path.exists(current_info_path):
        with open(current_info_path, 'r', encoding='utf-8') as lv:
            lvj = json.load(lv)
            current_ver = int(lvj["ver"])
            current_tag = str(lvj["Version"])
    else:
        current_ver = 0
    hoshino.logger.info(f'本地YoCool版本：YoCool-{current_tag}')
    return current_ver


def get_current_tms() -> int:
    '''
    获取本地安装主题
    '''
    if os.path.exists(current_info_path):
        with open(current_info_path, 'r', encoding='utf-8') as lv:
            lvj = json.load(lv)
            current_tms = int(lvj["Themes"])
    else:
        current_tms = 0
    return current_tms


async def get_yocool_themes(themes_num) -> str:
    '''
    获取主题对应名称
    '''
    themes_list = await get_api_url(yocool_themes_url)
    name_list = themes_list[f"{themes_num}"]
    themes_name = name_list.split(',')
    themes = themes_name[0]
    return themes


async def get_newest_ver() -> str:
    '''
    获取最新版本号与更新日志
    '''
    newest_yocool_ver_json = await get_api_url(yocool_info_url)
    newest_ver = int(newest_yocool_ver_json["ver"])
    newest_tag = str(newest_yocool_ver_json["Version"])
    newest_udn = str(newest_yocool_ver_json["UpdateNote"])
    hoshino.logger.info(f'当前最新YoCool版本：YoCool-{newest_tag}')
    return newest_ver, newest_tag, newest_udn


def update_current_ver(ver,Version,UpdateNote) -> None:
    '''
    更新本地信息
    '''
    with open(current_info_path, 'r', encoding='utf-8') as f:
        current_updata_json = json.load(f)
    current_updata_json['ver'] = ver
    current_updata_json['Version'] = Version
    current_updata_json['UpdateNote'] = UpdateNote
    current_updata_json['Install'] = 1
    with open(current_info_path, 'w+', encoding='utf-8') as f:
        json.dump(current_updata_json, f, indent=4, ensure_ascii=False)
    hoshino.logger.info(f'更新本地YoCool信息文件')


async def get_yocool_file(newest_tag,themes):
    '''
    获取yocool文件
    '''
    hoshino.logger.info(f'开始下载YoCool主题{themes}')
    download_url = git_yocool_releases + newest_tag + '/YoCool-'+ newest_tag + '-' + themes + '-yocool.zip'
    response =  await aiorequests.get(download_url, stream=True, timeout=20)
    status_code = response.status_code
    if status_code != 200:
        hoshino.logger.error(f'下载YoCool最新版本时发生错误{status_code}')
        return "ERROR", status_code
    content = await response.content
    with open(path + '/public.zip', 'wb') as f:
        f.write(content)
    hoshino.logger.info('开始解压缩主题文件')
    zip_files = [file for file in os.listdir(path) if file.endswith('.zip')]
    for zfile in zip_files:
        f = zipfile.ZipFile(os.path.join(path, zfile),'r')
        for file in f.namelist():
            f.extract(file,os.path.join(path, zfile[:-4]))
    hoshino.logger.info('主题文件解压完成')
    hoshino.logger.info('主题文件准备完毕！')
    return "OK", status_code


async def update_yocool(force=False) -> str:
    '''
    从git下载releases安装
    指定force为true, 则不会比较本地版本号是否最新
    '''
    # 获取最新版本编码，版本号,更新日志
    newest_ver, newest_tag, newest_udn = await get_newest_ver()
    if newest_ver < 1000:
        hoshino.error(f'获取YoCool版本时发生错误{newest_ver}')
        return newest_ver, 0, 0

    # 比较本地版本
    current_ver = get_current_ver()
    if force:
        # 指定强制更新
        current_ver = 0
    if newest_ver <= current_ver:
        newest_ver = 0
        return newest_ver, newest_tag, newest_udn

    # 获取本地设置主题
    hoshino.logger.info('检查本地主题配置')
    select = get_current_tms()
    themes = await get_yocool_themes(select)
    hoshino.logger.info(f'本地主题当前配置：{themes}')

    # 下载文件
    getcode, status_code = await get_yocool_file(newest_tag,themes)
    if getcode == "ERROR":
        hoshino.logger.error(f'下载YoCool文件时发生错误')
        return status_code, newest_tag, newest_udn

    # 旧文件备份文件
    ins = get_install_state()
    if ins == 0:
        hoshino.logger.info('正在备份yobot原生主题文件')
        shutil.move(yobot_themes_path,backup_themes_path)

    # 写入新文件
    hoshino.logger.info('正在写入YoCool主题文件')
    shutil.move(yocool_themes_path,yobot_themes_path)
    hoshino.logger.info('YoCool安装完成！')

    # 删除压缩包
    hoshino.logger.info('清理无用文件')
    for infile in glob.glob(os.path.join(path, '*.zip')):
        os.remove(infile)

    # 覆盖本地版本号
    hoshino.logger.info(f'更新版本信息：YoCool-{newest_tag}')
    update_current_ver(newest_ver, newest_tag, newest_udn)
    return newest_ver, newest_tag, newest_udn


async def uninstall_yocool(force=False) -> str:
    '''
    卸载YoCool
    指定force为true, 则不会检查是否安装
    '''
    ins = get_install_state()
    if force:
        ins = 1
    if ins == 0:
        return 0

    hoshino.logger.info('正在验证YoCool配置完整性')
    if not os.path.exists(yobot_themes_path) or not os.path.exists(backup_themes_path):
        return 1
    hoshino.logger.info('正在移除YoCool文件')
    try:
        shutil.rmtree(yobot_themes_path)
    except:
        return 2
    await asyncio.sleep(5)
    hoshino.logger.info('开始从备份恢复')
    try:
        shutil.move(backup_themes_path,yobot_themes_path)
    except:
        return 3
    await asyncio.sleep(5)
    try:
        os.rmdir(path + '/backup')
    except:
        return 4
    hoshino.logger.info('YoCool卸载完成！')
    with open(current_info_path, 'r', encoding='utf-8') as f:
        current_updata_json = json.load(f)
    current_updata_json['Install'] = 0
    current_updata_json['ver'] = 0
    with open(current_info_path, 'w+', encoding='utf-8') as f:
        json.dump(current_updata_json, f, indent=4, ensure_ascii=False)
    return 5


@on_command('一键安装', aliases=('快速安装', '一键YoCool', '一键yocool', '一键YOCOOL'), only_to_me=True)
async def one_key_yocool(session):
    uid = session.event.user_id
    if uid not in hoshino.config.SUPERUSERS:
        return
    if os.path.exists(backup_themes_path):
        await session.finish('您已经安装过了，如需更新请发送【更新YoCool】')
    name = util.normalize_str(session.current_arg_text)
    if not name:
        select = 1
    else:
        if f"{name}" in themes_list:
            select = name
        else:
            await session.finish(f'没有找到主题{name}，请检查输入后再试')
    hoshino.logger.info('正在进行安装前初始化')
    if os.path.exists(current_info_path):
        for infile in glob.glob(os.path.join(path, '*.json')):
            os.remove(infile)
    yocool_info_json = await get_api_url(yocool_info_url)
    yocool_info_json['Themes'] = select
    with open(current_info_path, 'w+', encoding='utf-8') as f:
        json.dump(yocool_info_json, f, indent=4, ensure_ascii=False)
    themes = await get_yocool_themes(select)
    await session.send(f'YoCool初始化完成，准备使用{themes}主题进行安装，安装需要一定时间，请耐心等待……')
    status, version, updatenote = await update_yocool(force=True)
    if status == 0:
        for infile in glob.glob(os.path.join(path, '*.json')):
            os.remove(infile)
        await session.finish('本地版本信息异常！请重新发送指令再试！')
    elif status < 1000:
        await session.finish(f'发生错误{status}，请在控制台日志中查看详细')
    else:
        await session.finish(f'一键安装已完成！\n当前YoCool版本：YoCool-{version}\n使用主题：{themes}\n更新日志：\n{updatenote}')


@on_command('切换主题', aliases=('更换主题', '变更主题', '修改主题'), only_to_me=True)
async def set_yocool_themes(session):
    uid = session.event.user_id
    if uid not in hoshino.config.SUPERUSERS:
        return
    if not os.path.exists(current_info_path):
        await session.send('没有找到YoCool信息配置文件，请安装YoCool后再试')
    ins = get_install_state()
    if ins == 0:
        await session.finish('当前未安装YoCool，无法切换主题')
    name = util.normalize_str(session.current_arg_text)
    themes_list = await get_api_url(yocool_themes_url)
    themes_tip = '请选择需要切换的主题：'
    for num in themes_list:
        name_list = themes_list[f"{num}"]
        themes_name = name_list.split(',')
        themes_tip += f"\n{num}.{themes_name[0]}-{themes_name[1]}"
    themes_tip += '\n*回复“切换主题 序号”即可'
    if not name:
        await session.finish(themes_tip, at_sender=True)
    else:
        if f"{name}" in themes_list:
            select = name
        else:
            await session.finish(f'没有找到主题{name}，请检查输入后再试')
    with open(current_info_path, 'r', encoding='utf-8') as f:
        current_updata_json = json.load(f)
    if current_updata_json['Themes'] == select:
        await session.finish(f'当前已在使用该主题，请选择其他主题')
    else:
        current_updata_json['Themes'] = select
    with open(current_info_path, 'w+', encoding='utf-8') as f:
        json.dump(current_updata_json, f, indent=4, ensure_ascii=False)
    themes = await get_yocool_themes(select)
    hoshino.logger.info(f'设置YoCool主题为{themes}')
    await session.send('开始切换主题，需要一定时间，请耐心等待……')
    yocode = await uninstall_yocool()
    if yocode == 0:
        await session.finish('尚未安装YoCool，无法切换主题')
    elif yocode ==1:
        await session.finish('YoCool完整性验证未通过，没有找到YoCool文件或备份文件夹')
    elif yocode ==2:
        await session.finish('移除YoCool文件出错，请手动卸载')
    elif yocode ==3:
        await session.finish('恢复原生主题出错，请手动恢复')
    elif yocode ==4:
        await session.finish('清理残留出错，请手动删除backup文件夹')
    else:
        hoshino.logger.info(f'原生主题已恢复')
    try:
        status, version, updatenote = await update_yocool(force=True)
    except:
        await session.send('主题切换异常中断，请排查问题后再次尝试')
    if status == 0:
        await session.finish(f'状态{status}')
    elif status < 1000:
        await session.finish(f'发生错误{status}，请在控制台日志中查看详细')
    else:
        await session.finish(f'YoCool主题已切换为{themes}')


@on_command('更新YoCool', aliases=('更新yocool', '更新YOCOOL', '升级YoCool', '升级yocool', '升级YOCOOL'), only_to_me=True)
async def update_yocool_chat(session):
    '''
    手动更新
    '''
    uid = session.event.user_id
    if uid not in hoshino.config.SUPERUSERS:
        return
    ins = get_install_state()
    if ins == 0:
        await session.finish('当前未安装YoCool')
    hoshino.logger.info('开始检查YoCool更新')
    await session.send('开始进行YoCool更新，需要一定时间，请耐心等待……')
    try:
        status, version, updatenote = await update_yocool()
    except:
        await session.send('更新异常中断，请排查问题后再次尝试')
    if status == 0:
        await session.finish('已是最新版本, 仍要更新YoCool请使用【强制更新YoCool】命令')
    elif status < 1000:
        await session.finish(f'发生错误{status}，请在控制台日志中查看详细')
    else:
        await session.finish(f'更新完成！\n当前YoCool版本：YoCool-{version}\n更新日志：\n{updatenote}')


@on_command('强制更新YoCool', aliases=('强制更新yocool', '强制更新YOCOOL', '强制升级YoCool', '强制升级yocool', '强制升级YOCOOL'), only_to_me=True)
async def update_yocool_force_chat(session):
    '''
    强制更新
    '''
    uid = session.event.user_id
    if uid not in hoshino.config.SUPERUSERS:
        return
    hoshino.logger.info('开始检查YoCool更新')
    ins = get_install_state()
    if ins == 0:
        await session.finish('当前未安装YoCool')
    await session.send('开始进行YoCool更新，需要一定时间，请耐心等待……')
    try:
        status, version, updatenote = await update_yocool(force=True)
    except:
        await session.send('更新异常中断，请排查问题后再次尝试')
    if status == 0:
        await session.finish(f'状态{status}')
    elif status < 1000:
        await session.finish(f'发生错误{status}，请在控制台日志中查看详细')
    else:
        await session.finish(f'更新完成！\n当前YoCool版本：YoCool-{version}\n更新日志：\n{updatenote}')


@on_command('卸载YoCool', aliases=('卸载yocool', '卸载YOCOOL'), only_to_me=True)
async def uninstall_yocool_chat(session):
    '''
    卸载
    '''
    uid = session.event.user_id
    if uid not in hoshino.config.SUPERUSERS:
        return
    await session.send('YoCool卸载开始，需要一定时间，请耐心等待……')
    yocode = await uninstall_yocool()
    if yocode == 0:
        await session.finish('尚未安装YoCool，无法操作')
    elif yocode ==1:
        await session.finish('YoCool完整性验证未通过，没有找到YoCool文件或备份文件夹')
    elif yocode ==2:
        await session.finish('移除YoCool文件出错，请手动卸载')
    elif yocode ==3:
        await session.finish('恢复原生主题出错，请手动恢复')
    elif yocode ==4:
        await session.finish('清理残留出错，请手动删除backup文件夹')
    else:
        await session.finish('YoCool卸载完成！')

@on_command('强制卸载YoCool', aliases=('强制卸载yocool', '强制卸载YOCOOL'), only_to_me=True)
async def uninstall_yocool_force_chat(session):
    '''
    强制卸载
    '''
    uid = session.event.user_id
    if uid not in hoshino.config.SUPERUSERS:
        return
    await session.send('YoCool强制卸载开始，需要一定时间，请耐心等待……')
    yocode = await uninstall_yocool(force=True)
    if yocode == 0:
        await session.finish('尚未安装YoCool，无法操作')
    elif yocode ==1:
        await session.finish('YoCool完整性验证未通过，没有找到YoCool文件或备份文件夹')
    elif yocode ==2:
        await session.finish('移除YoCool文件出错，请手动卸载')
    elif yocode ==3:
        await session.finish('恢复原生主题出错，请手动恢复')
    elif yocode ==4:
        await session.finish('清理残留出错，请手动删除backup文件夹')
    else:
        await session.finish('YoCool卸载完成！')


# 每周检查一次，自动进行更新
@scheduler.scheduled_job('cron',day_of_week='1', hour=4, minute=40)
async def update_yocool_sdj():
    bot = get_bot()
    master_id = hoshino.config.SUPERUSERS[0]

    self_ids = bot._wsr_api_clients.keys()
    sid = self_ids[0]

    ins = get_install_state()
    if ins == 0:
        return
    status, version, UpdateNote = await update_yocool()
    if status == 0:
        return
    elif status < 1000 and NOTICE:
        msg = f'自动更新YoCool版本时发生错误{status}，请在控制台日志中查看详细'
        await bot.send_private_msg(seld_id=sid, user_id=master_id, message=msg)
    elif NOTICE:
        msg = f'已自动更新YoCool版本\n当前YoCool版本：YoCool-{version}\n本次更新内容：\n{UpdateNote}'
        await bot.send_private_msg(seld_id=sid, user_id=master_id, message=msg)