check/utils.py

import json
import os
import random
import re
import sqlite3
import time
import traceback
# import tomli_w
from functools import wraps


def pip_install():
    print("正在安装依赖")
    os.system("pip3 install requests rsa tomli tomli_w beautifulsoup4")


try:
    import tomli
except ModuleNotFoundError:
    pip_install()
    import tomli

from checksendNotify import send


# def toml_to_json(toml_path, to_json_path):
#     """
#     :param toml_path: 需要转换的toml文件的路径
#     :param to_json_path: 需要输出的json文件路径
#     :return: None
#     """
#     with open(toml_path, "rb") as f:
#         toml_dict = tomli.load(f)
#         json_date = json.dumps(toml_dict, indent=4, ensure_ascii=False)
#         with open(to_json_path, 'w', encoding="utf8") as s:
#             s.write(json_date)


# def json_to_toml(json_path, to_toml_path):
#     with open(json_path, "r", encoding="utf8") as f:
#         json_dict = json.load(f)
#         with open(to_toml_path, "wb") as f:
#             tomli_w.dump(json_dict, f)

class config_get(object):
    def __init__(self, custom_path=None):
        """
        config_path: 自定义配置文件路径
        config_file: 实际使用的配置文件路径
        config_format: 实际使用的配置文件格式
        """
        if custom_path is None:
            self.config_path = self.get_config_path()
            self.config_file = self.get_config_file()
            self.config_format = self.get_config_format()
        else:
            self.config_file = custom_path
            self.config_format = self.get_config_format()

    def get_config_format(self):
        if self.config_file.endswith('.toml'):
            return "toml"
        else:
            return "json"

    @staticmethod
    def get_config_path():
        ql_old = "/ql/config/"
        ql_new = "/ql/data/config/"
        if os.path.isdir(ql_old):
            print('成功 当前环境为青龙面板v2.12- 继续执行\n')
            return ql_old
        elif os.path.isdir(ql_new):
            print('成功 当前环境为青龙面板v2.12+ 继续执行\n')
            return ql_new
        else:
            print('失败 请检查环境')
            exit(0)

    def get_config_file(self):
        toml_file = f"{self.config_path}check.toml"
        json_file = f"{self.config_path}check.json"
        if os.path.exists(toml_file):
            print(f"启用了toml配置文件\n路径为{toml_file}\n")
            return toml_file
        elif os.path.exists(json_file):
            print(f"启用了json配置文件\n路径为{json_file}\n")
            return json_file
        else:
            print("未找到配置文件")
            self.move_config_file()
            return toml_file

    def move_config_file(self):
        print("尝试移动配置文件到目录")
        if self.config_path == "/ql/config/":
            self.move_configuration_file_old()
        else:
            self.move_configuration_file_new()

    def get_real_key(self, expression):
        """
        从配置文件中获取，re表达式想要的KEY
        :return:
        """
        pattern = re.compile(expression, re.I)
        real_key = ''
        if self.config_format == "toml":
            for key in self.get_key_for_toml(self.config_file):
                if pattern.match(key) is not None:
                    real_key = key
        else:
            for key in self.get_key_for_json(self.config_file):
                if pattern.match(key) is not None:
                    real_key = key
        if real_key != '':
            return real_key
        else:
            print("啊哦没有找到")
            exit(1)

    def get_value(self, expression):
        real_key = self.get_real_key(expression)
        if self.config_format == "toml":
            return self.get_value_for_toml(self.config_file, real_key)
        else:
            return self.get_value_for_json(self.config_file, real_key)

    @staticmethod
    def move_configuration_file_old():
        print("移动配置文件")
        os.system("cp /ql/repo/yuxian158_check/check.sample.toml /ql/config/check.toml")

    @staticmethod
    def move_configuration_file_new():
        print("移动配置文件")
        os.system("cp /ql/data/repo/yuxian158_check/check.sample.toml /ql/data/config/check.toml")

    @staticmethod
    def get_value_for_toml(toml_path, key):
        with open(toml_path, "rb") as f:
            try:
                toml_dict = tomli.load(f)
                return toml_dict.get(key)
            except tomli.TOMLDecodeError:
                print(
                    f"错误：配置文件 {toml_path} 格式不对，请学习 https://toml.io/cn/v1.0.0\n错误信息：\n{traceback.format_exc()}"
                )
                exit(1)

    @staticmethod
    def get_value_for_json(json_path, key):
        with open(json_path, "r", encoding="utf8") as f:
            try:
                json_dict = json.load(f)
                return json_dict.get(key)
            except json.decoder.JSONDecodeError:
                print(f"错误：配置文件 {json_path} 格式不对，错误信息{traceback.format_exc()}")

    @staticmethod
    def get_key_for_toml(toml_path):
        with open(toml_path, "rb") as f:
            try:
                toml_dict = tomli.load(f)
                return toml_dict.keys()
            except tomli.TOMLDecodeError:
                print(
                    f"错误：配置文件 {toml_path} 格式不对，请学习 https://toml.io/cn/v1.0.0\n错误信息：\n{traceback.format_exc()}"
                )
                exit(1)

    @staticmethod
    def get_key_for_json(json_path):
        with open(json_path, "r", encoding="utf8") as f:
            try:
                json_dict = json.load(f)
                return json_dict.keys()
            except json.decoder.JSONDecodeError:
                print(f"错误：配置文件 {json_path} 格式不对，错误信息{traceback.format_exc()}")


class check(object):
    def __init__(self, run_script_name, run_script_expression, Configuration_flag=False):
        """
        :param run_script_name: 执行脚本的说明
        :param run_script_expression: 需要获取的配置键的re表达式
        :param Configuration_flag: 是否只检测True或False(默认为False)
        """
        self.run_script_name = run_script_name
        self.run_script_expression = run_script_expression
        self.Configuration_flag = Configuration_flag

    @staticmethod
    def other_task():
        # change_db()
        pass

    def __call__(self, func):
        @wraps(func)
        def wrapper():
            if not self.Configuration_flag:
                config = config_get()
                value_list = config.get_value(self.run_script_expression)
                push_message = ""
                num = 1
                for value in value_list:
                    print(f"<----------------账号【{num}】---------------->")
                    print(f"获取到的账号信息为:{value}\n")
                    num += 1
                    try:
                        result = func(value=value) + '\n\n'
                        print(f"执行结果:\n{result}")
                        push_message += result
                    except IndexError:
                        print("可能是示例格式被运行\n错误信息:")
                        print(f"{traceback.format_exc()}")
                        push_message += ''
                    except AttributeError:
                        print("可能是配置文件的键名出现问题\n"
                              "例如:在此次更新中什么值得买的键名从smzdm_cookie变成了cookie\n")
                        print(f"{traceback.format_exc()}")
                        push_message += ''
                    except TypeError:
                        print(f"{traceback.format_exc()}")
                        push_message += ''
                send(self.run_script_name, push_message)
            else:
                config = config_get()
                flag = config.get_value(self.run_script_expression)
                if flag is not None and flag:
                    print(f"开始执行{self.run_script_name}")
                    func()
                else:
                    print(f"设置为不执行{self.run_script_name}")

        return wrapper


def change_cron_new(cron_file_path="/ql/data/db/database.sqlite", repositories="yuxian158_check"):
    print("尝试修改定时时间")
    os.system(f"cp {cron_file_path} {cron_file_path}.back")
    con = sqlite3.connect(cron_file_path)
    cur = con.cursor()

    def change_time(time_str: str):
        words = re.sub("\\s+", " ", time_str).split()
        words[0] = str(random.randrange(60))
        words[1] = str(random.randrange(22))
        return " ".join(words)

    cur.execute("select id,name,command,schedule from Crontabs")
    res = cur.fetchall()
    for line in res:
        if line[2].find(repositories) != -1:
            sql = f" UPDATE Crontabs SET schedule = \"{change_time(line[3])}\" WHERE id = {line[0]}"
            print(f"任务名称 {line[1]} 修改为{sql}")
            cur.execute(sql)

    con.commit()
    con.close()


def change_cron_old(cron_file_path="/ql/db/crontab.db", repositories="yuxian158_check"):
    print("尝试修改定时时间")

    def change_time(time_str: str):
        words = re.sub("\\s+", " ", time_str).split()
        words[0] = str(random.randrange(60))
        words[1] = str(random.randrange(22))
        return " ".join(words)

    time_str = time.strftime("%Y-%m-%d", time.localtime())
    os.system(f"cp /ql/db/crontab.db /ql/db/crontab.db.{time_str}.back")
    lines = []
    with open(cron_file_path, "r", encoding="UTF-8") as f:
        for i in f.readlines():
            # print(record.get("command"))
            if i.find(repositories) != -1:
                record = json.loads(i)
                record["schedule"] = change_time(record["schedule"])
                lines.append(json.dumps(record, ensure_ascii=False) + "\n")
            else:
                lines.append(i)

    with open(cron_file_path, "w", encoding="UTF-8") as f:
        f.writelines(lines)


if __name__ == "__main__":
    pip_install()
    config = config_get()
    if config.config_path == "/ql/config/":
        if os.path.isfile("/ql/db/database.sqlite"):
            change_cron_new(cron_file_path="/ql/db/database.sqlite")
        else:
            change_cron_old()
    else:
        change_cron_new()
        print("修改完成请重启容器")