# -*- coding: utf-8 -*- import os import requests # 青龙面板的地址 url = "https://auto.erhe.top" # 登录青龙面板 def login(): response = requests.post(f"{url}/api/user/login", json={"username": "toor", "password": "!QAZ2wsx+0913"}) print(response) if response.status_code != 200: print(response.status_code) print(response.text) exit(0) return response.json()['data']['token'] # 获取任务列表 def get_tasks(token): response = requests.get(f"{url}/api/crons", headers={"Authorization": f"Bearer {token}"}) return response.json()['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"] } headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } response = requests.post(f"{url}/api/crons", headers=headers, json=payload) return response.json() # 创建视图分类 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' } headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } response = requests.post(f"{url}/api/crons", headers=headers, json=payload) # 主逻辑 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/auto/' base_path = os.path.join(project_path, 'base') spider_path = os.path.join(project_path, 'spider') message_path = os.path.join(project_path, 'message') manual_path = os.path.join(project_path, 'manual') daily_path = os.path.join(project_path, 'daily') tasks_template = [{ 'base_tasks': [ { "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"] }, { "name": "定时刷新 freshrss 订阅源", "command": "python3 {}/update_feed.py".format(base_path), "schedule": "0 6,22 * * *", "labels": ["base"] } ], 'spider': [ { "name": "自动执行反斗限免爬虫", "command": "python3 {}/news_get_apprcn.py".format(spider_path), "schedule": "0 0 3,6,9,12,15,18,21 * * *", "labels": ["spider"] }, { "name": "自动执行chiphell爬虫", "command": "python3 {}/news_get_chiphell.py".format(spider_path), "schedule": "0 0 3,6,9,12,15,18,21 * * *", "labels": ["spider"] }, { "name": "自动执行hello_github爬虫", "command": "python3 {}/news_get_hello_github.py".format(spider_path), "schedule": "0 0 3,6,9,12,15,18,21 * * *", "labels": ["spider"] }, { "name": "自动执行Anyknew爬虫", "command": "python3 {}/news_get_news.py".format(spider_path), "schedule": "0 0 3,6,9,12,15,18,21 * * *", "labels": ["spider"] }, { "name": "自动执行币界网文章爬虫", "command": "python3 {}/spider_web3_coin_world.py".format(spider_path), "schedule": "0 0 3,6,9,12,15,18,21 * * *", "labels": ["spider"] }, { "name": "获取 web3 新闻", "command": "python3 {}/spider_web3_news.py".format(spider_path), "schedule": "0 0 3,6,9,12,15,18,21 * * *", "labels": ["spider"] }, { "name": "自动执行dlt爬虫", "command": "python3 {}/spider_get_and_check_dlt.py".format(spider_path), "schedule": "30 22 * * 1,3,6", "labels": ["spider"] } ], 'message_tasks': [ { "name": "获取coin实时数据", "command": "python3 {}/message_coin_detail.py".format(message_path), "schedule": "0 * * * *", "labels": ["message"] }, { "name": "对比大乐透最新一期数据,匹配已购买号码,发送消息", "command": "python3 {}/message_dlt.py".format(message_path), "schedule": "30 22 * * 1,3,6", "labels": ["message"] }, { "name": "获取未来 7 天的天气预报", "command": "python3 {}/message_get_one_week_weather.py".format(message_path), "schedule": "0 0 6,22 * * *", "labels": ["message"] }, { "name": "从 freshrss-psql 数据库中读取数据并发送", "command": "python3 {}/message_rss_data_handel.py".format(message_path), "schedule": "30 6,9,12,18,22 * * *", "labels": ["message"] }, { "name": "空投任务消息", "command": "python3 {}/message_airdrop_tasks.py".format(message_path), "schedule": "0 8,20 * * *", "labels": ["message"] }, { "name": "链捕手快讯消息推送", "command": "python3 {}/message_chaincatcher.py".format(message_path), "schedule": "0 */2 * * *", "labels": ["message"] } ], 'manual': [ { "name": "手动读取rss订阅新闻", "command": "python3 {}/read_news.py".format(manual_path), "schedule": "0 0 1 1 *", "labels": ["manual"] } ], 'daily': [ { "name": "3dos自动签到", "command": "python3 {}/daily_3dos.py".format(daily_path), "schedule": "*/5 * * * *", "labels": ["daily"] } ], }] 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!')