AutoInfo/tools/tools_ql_create_tasks.py

# -*- coding: utf-8 -*-
import os
import json
import urllib.request
import urllib.parse

# 青龙面板的地址
url = "https://ql.erhe.link"


# 登录青龙面板
def login():
    data = json.dumps({"username": "toor", "password": "!QAZ2wsx+0913"}).encode('utf-8')
    req = urllib.request.Request(
        f"{url}/api/user/login",
        data=data,
        headers={'Content-Type': 'application/json'}
    )

    try:
        with urllib.request.urlopen(req) as response:
            result = json.loads(response.read().decode('utf-8'))
            return result['data']['token']
    except urllib.error.HTTPError as e:
        print(f"Login failed with status code: {e.code}")
        print(e.read().decode('utf-8'))
        exit(0)


# 获取任务列表
def get_tasks(token):
    req = urllib.request.Request(
        f"{url}/api/crons",
        headers={"Authorization": f"Bearer {token}"}
    )

    with urllib.request.urlopen(req) as response:
        result = json.loads(response.read().decode('utf-8'))
        return result['data']['data']


# 创建任务
def create_task(task_template, token):
    payload = {
        "name": task_template["name"],
        "command": task_template["command"],
        "schedule": task_template["schedule"],
        "labels": task_template["labels"]
    }

    data = json.dumps(payload).encode('utf-8')
    req = urllib.request.Request(
        f"{url}/api/crons",
        data=data,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
    )

    with urllib.request.urlopen(req) as response:
        return json.loads(response.read().decode('utf-8'))


# 创建视图分类
def create_view_type(token):
    view_type_list = ['base', 'spider_common']
    for view_type in view_type_list:
        payload = {
            "name": view_type,
            "filters": {
                'property': 'labels',
                'operation': 'Reg',
                'value': view_type
            },
            'filterRelation': 'and'
        }

        data = json.dumps(payload).encode('utf-8')
        req = urllib.request.Request(
            f"{url}/api/crons",
            data=data,
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json"
            }
        )

        try:
            with urllib.request.urlopen(req) as response:
                result = json.loads(response.read().decode('utf-8'))
                print(f"View type {view_type} created: {result}")
        except urllib.error.HTTPError as e:
            print(f"Failed to create view type {view_type}: {e}")


# 主逻辑
def main():
    while True:
        try:
            token = login()
            print(f"已连接到 {url}")
            tasks = get_tasks(token)
            tasks_names = [task['name'] for task in tasks]
            if tasks:
                print("Current tasks name: \n{}, \ntotal: {}".format('\n'.join(tasks_names), str(len(tasks_names))))
            else:
                print("Tasks list is empty")

            project_path = '/ql/data/scripts/AutoInfo/'
            base_path = os.path.join(project_path, 'base')
            to_email_tasks_path = os.path.join(project_path, 'to_email')
            to_gotify_tasks_path = os.path.join(project_path, 'to_gotify')
            manual_path = os.path.join(project_path, 'manual')
            tasks_template = [{
                'base': [
                    {
                        "name": "每天开始自动创建日志",
                        "command": "python3 {}/base_daily_logs_generate.py".format(base_path),
                        "schedule": "0 0 * * *",
                        "labels": ["base"]
                    },
                    {
                        "name": "每天结束自动发送日志",
                        "command": "python3 {}/base_daily_logs_send.py".format(base_path),
                        "schedule": "58 23 * * *",
                        "labels": ["base"]
                    },
                    {
                        "name": "每天自动删除旧数据",
                        "command": "python3 {}/base_timing_remove_data.py".format(base_path),
                        "schedule": "1 0 * * *",
                        "labels": ["base"]
                    },
                    {
                        "name": "每天新闻汇总,发送到邮箱",
                        "command": "python3 {}/base_news_data_collation.py".format(base_path),
                        "schedule": "0 10 6,12,18 * * *",
                        "labels": ["base"]
                    }
                ],
                'to-email': [
                    {
                        "name": "对比大乐透最新一期数据,匹配已购买号码,发送消息",
                        "command": "python3 {}/dlt.py".format(to_email_tasks_path),
                        "schedule": "30 22 * * 1,3,6",
                        "labels": ["to-email"]
                    },
                    {
                        "name": "链捕手快讯消息推送",
                        "command": "python3 {}/chaincatcher.py".format(to_email_tasks_path),
                        "schedule": "0 */2 * * *",
                        "labels": ["to-email"]
                    },
                    {
                        "name": "anyknew聚合新闻消息推送",
                        "command": "python3 {}/anyknew.py".format(to_email_tasks_path),
                        "schedule": "0 */3 * * *",
                        "labels": ["to-email"]
                    },
                    {
                        "name": "反斗限免消息推送",
                        "command": "python3 {}/apprcn.py".format(to_email_tasks_path),
                        "schedule": "0 */12 * * *",
                        "labels": ["to-email"]
                    },
                    {
                        "name": "chiphell消息推送",
                        "command": "python3 {}/chiphell.py".format(to_email_tasks_path),
                        "schedule": "0 */12 * * *",
                        "labels": ["to-email"]
                    },
                    {
                        "name": "hello-github消息推送",
                        "command": "python3 {}/hello_github.py".format(to_email_tasks_path),
                        "schedule": "0 */12 * * *",
                        "labels": ["to-email"]
                    }
                ],
                'to-gotify': [
                    {
                        "name": "获取未来 7 天的天气预报",
                        "command": "python3 {}/weather7day.py".format(to_gotify_tasks_path),
                        "schedule": "0 0 6,22 * * *",
                        "labels": ["to-gotify"]
                    },
                    {
                        "name": "获取coin实时数据",
                        "command": "python3 {}/coin_detail.py".format(to_gotify_tasks_path),
                        "schedule": "0 * * * *",
                        "labels": ["to-gotify"]
                    },
                    {
                        "name": "空投任务消息",
                        "command": "python3 {}/airdrop_tasks.py".format(to_gotify_tasks_path),
                        "schedule": "0 8,20 * * *",
                        "labels": ["to-gotify"]
                    },
                    {
                        "name": "币界网消息推送",
                        "command": "python3 {}/coin_world.py".format(to_gotify_tasks_path),
                        "schedule": "0 8,20 * * *",
                        "labels": ["to-gotify"]
                    },
                    {
                        "name": " web3新闻消息推送",
                        "command": "python3 {}/web3_news.py".format(to_gotify_tasks_path),
                        "schedule": "0 9,21 * * *",
                        "labels": ["to-gotify"]
                    }
                ],
                'manual': [

                ]
            }]

            for task_template in tasks_template:
                for task_type, task_list in task_template.items():
                    for task in task_list:
                        task_name = task["name"]
                        if task_name in tasks_names:
                            print("Task {} already exists.".format(task_name))
                        else:
                            result = create_task(task, token)
                            print("Task creation result:", result)

            # 创建所有任务之后, 创建视图分类
            # create_view_type(token)
            break  # 正常执行完成后退出循环

        except Exception as e:
            print("An error occurred: ", e)
            print("Retrying...")


if __name__ == "__main__":
    main()
    print('done!')