朱雀面板---rust写了个超低占用的定时任务面板

弄了试试看，我还在用青龙

https://github.com/linzixuanzz/daidai-panel

还有呆呆面板

不能正常的用xxl-job之类的么?

蹲性能测试，测完麻烦 @ 一下

测速脚本使用AI写的定时加密备份nas上的监控文件到XX云盘，使用openlist的接口及自带加密。

执行脚本的过程中，感觉和青龙没啥区别，通过docker stats大概看了下，CPU也是突然一下就升高了。

而且部分青龙的脚本迁移过来后还不能执行，我还在研究怎么处理。

还有一个缺点：基础包比青龙大很多。

下面是测试脚本，需要配置4个环境变量才能执行，如果需要推送，还要配置2个推送的环境变量。

import requests

import re

import os

import sys



# 强制开启行缓冲，确保日志能实时输出到青龙面板

sys.stdout.reconfigure(line_buffering=True)



# ================= 配置区域 =================

OPENLIST_URL = os.getenv("OPENLIST_URL", "http://192.168.1.x:5244")

OPENLIST_TOKEN = os.getenv("OPENLIST_TOKEN", "")

SOURCE_PATH = os.getenv("CAMERA_PATH", "/本机存储/xiaomi_camera_videos/123456789")

TARGET_PATH = os.getenv("TARGET_PATH", "/XX网盘数据加密/小米摄像头")



WXPUSHER_APP_TOKEN = os.getenv("WXPUSHER_APP_TOKEN", "")

WXPUSHER_TOPIC_ID = os.getenv("WXPUSHER_TOPIC_ID", "")

# ===========================================



headers = {"Authorization": OPENLIST_TOKEN, "Content-Type": "application/json"}



def send_wxpusher(summary):

    """发送WxPusher主题推送并记录详细日志"""

    if not WXPUSHER_APP_TOKEN or not WXPUSHER_TOPIC_ID:

        return

    

    url = "https://wxpusher.zjiecode.com/api/send/message"

    payload = {

        "appToken": WXPUSHER_APP_TOKEN,

        "content": f"<h3>摄像头备份报告</h3><p>{summary.replace(chr(10), '<br/>')}</p>",

        "contentType": 2, 

        "topicIds": [int(WXPUSHER_TOPIC_ID)]

    }

    try:

        res = requests.post(url, json=payload, timeout=10)

        # 获取返回内容进行校验

        res_data = res.json()

        if res_data.get("code") == 1000:

            print("✅ WxPusher 推送成功")

        else:

            print(f"❌ WxPusher 推送失败: {res_data.get('msg')}")

    except Exception as e:

        print(f"❌ WxPusher 推送接口调用异常: {e}")



def get_files_list(path):

    url = f"{OPENLIST_URL}/api/fs/list"

    try:

        res = requests.post(url, headers=headers, json={"path": path}, timeout=30)

        data = res.json()

        if data.get("code") == 200:

            payload = data.get("data")

            if payload and isinstance(payload.get("content"), list):

                return [f['name'] for f in payload['content']]

        return []

    except:

        return []



def make_dir(target_path):

    url = f"{OPENLIST_URL}/api/fs/mkdir"

    requests.post(url, headers=headers, json={"path": target_path}, timeout=30)



def copy_files(src_dir, dst_dir, file_names):

    """批量复制文件，修正payload结构"""

    if not file_names:

        return None

    url = f"{OPENLIST_URL}/api/fs/copy"

    # 已修正：直接传递参数，不嵌套payload层

    payload = {

        "src_dir": src_dir,

        "dst_dir": dst_dir,

        "names": file_names

    }

    try:

        res = requests.post(url, headers=headers, json=payload, timeout=30)

        return res.json()

    except Exception as e:

        print(f"❌ 复制请求出错: {e}")

        return None



def run_organize():

    logs = []

    print(f"🔍 开始扫描路径: {SOURCE_PATH}")

    list_url = f"{OPENLIST_URL}/api/fs/list"

    

    try:

        res = requests.post(list_url, headers=headers, json={"path": SOURCE_PATH}, timeout=30)

        data = res.json()

    except Exception as e:

        return f"连接Alist服务失败: {e}"



    if data.get("code") != 200:

        return f"列表获取失败: {data.get('message', '未知错误')}"



    folder_pattern = re.compile(r'^(\d{4})(\d{2})(\d{2})(\d{2})$')

    items = data.get('data', {}).get('content', [])

    

    for item in items:

        name = item['name']

        match = folder_pattern.match(name)

        if item['is_dir'] and match:

            year, month, day, hour = match.groups()

            target_full = f"{TARGET_PATH}/{year}年/{month}月/{day}日/{hour}时"

            make_dir(target_full)

            

            old_path = f"{SOURCE_PATH}/{name}"

            src_files = get_files_list(old_path)

            dst_files = get_files_list(target_full)

            to_copy = [f for f in src_files if f not in dst_files]



            if to_copy:

                print(f"📄 处理: {name}, 发现新文件: {len(to_copy)}个")

                res = copy_files(old_path, target_full, to_copy)

                if res and res.get("code") == 200:

                    logs.append(f"✅ {name}: 已同步 {len(to_copy)} 个")

                else:

                    msg = res.get("message", "未知错误") if res else "连接超时"

                    logs.append(f"❌ {name}: 失败 ({msg})")

            else:

                print(f"⏩ 跳过: {name} (无更新)")

    

    return "\n".join(logs) if logs else "本次扫描完成，无需备份新文件。"



if __name__ == "__main__":

    print("🚀 任务启动")

    if not OPENLIST_TOKEN:

        print("❌ 错误：未设置 OPENLIST_TOKEN")

    else:

        try:

            msg = run_organize()

            print(msg)

            send_wxpusher(msg)

        except Exception as e:

            err = f"脚本执行崩溃: {str(e)}"

            print(err)

            send_wxpusher(err)

    print("✨ 任务结束")



    

  

@lczws #4

已经测试，如果正在使用青龙，不建议更换，没啥区别。

而且朱雀的功能配置不如白虎，白虎的界面，功能配置舒适度才是最好的。

@feixiang #6 感谢测试 ^-^