Dify备份恢复应用python版

脚本所用Dify平台为v1.11.1版本

需要注意使用此脚本前提为dify平台的console接口可用！！！

测试接口，浏览器访问：

1	http://192.168.11.15:80/console/api/apps

有返回值就表示可用。

1	{"code":"unauthorized","message":"Invalid Authorization token.","status":401}

🛠️ 运行前，需要配置信息：

dify平台地址。
平台的账号密码。
可以仅备份包含这些标签的应用 (留空则不按标签过滤)。
还可以排除不备份某些标签的应用。
yml文件备份到当前目录的【dify_backups/年月日】文件夹中。

1.备份

📝备份backup-dify.py完整代码：

import requests
import json
import os,time

# ------------------------------- ！！！配置区域  -------------------------------
# 平台配置
BASE_URL = "http://172.16.11.53/console/api"        # 请替换为你的 Dify 地址
EMAIL = "[email protected]"                             # 请替换为你的Dify用户
PASSWORD = "*****"                                  # 请替换为你的Dify密码
SAVE_DIR = f"./dify_backups/{time.strftime('%Y%m%d')}"

# 过滤配置
INCLUDE_TAGS = ["备份"]  # 仅备份包含这些标签的应用 (留空则不按标签过滤)
EXCLUDE_APP_NAMES = ["测试", "弃用"]  # 排除这些名称的应用
# ----------------------------------------------------------------------------

class DifyBackup:
    def __init__(self):
        self.session = requests.Session()
        self.access_token = None
        self.csrf_token = None

    def login(self):
        """登录并获取 Token"""
        login_url = f"{BASE_URL}/login"
        payload = {"email": EMAIL, "password": PASSWORD}

        response = self.session.post(login_url, json=payload)
        if response.status_code == 200:
            data = response.json()
            self.access_token = data.get("data", {}).get("access_token")
            # 从 cookies 中获取 csrf_token
            self.csrf_token = self.session.cookies.get("csrf_token")

            # 更新 Session Header
            self.session.headers.update({
                "Authorization": f"Bearer {self.access_token}",
                "X-Csrf-Token": self.csrf_token
            })
            print("✅ 登录成功，Token 已获取")
        else:
            print(f"❌ 登录失败: {response.text}")
            exit()

    def get_apps(self):
        """获取应用列表"""
        apps_url = f"{BASE_URL}/apps?page=1&limit=100" # 假设应用不超过100个
        response = self.session.get(apps_url)
        if response.status_code == 200:
            return response.json().get("data", [])
        return []

    def backup_apps(self):
        """执行备份逻辑"""
        if not os.path.exists(SAVE_DIR):
            os.makedirs(SAVE_DIR)

        apps = self.get_apps()
        for app in apps:
            app_name = app.get("name")
            app_id = app.get("id")
            app_tags = [t.get("name") for t in app.get("tags", [])]

            # 1. 排除特定名称
            if app_name in EXCLUDE_APP_NAMES:
                print(f"⏩ 跳过 (排除名单): {app_name}")
                continue

            # 2. 标签过滤 (如果设置了 INCLUDE_TAGS)
            if INCLUDE_TAGS and not any(tag in INCLUDE_TAGS for tag in app_tags):
                print(f"⏩ 跳过 (标签不匹配): {app_name}")
                continue

            # 3. 导出 DSL
            print(f"📦 正在备份应用: {app_name} ({app_id})...")
            export_url = f"{BASE_URL}/apps/{app_id}/export"
            export_res = self.session.get(export_url)

            if export_res.status_code == 200:
                # 提取 JSON 中的 data 字段（即真正的 YAML 内容）
                real_yaml_content = export_res.json().get("data", "")
                # 保存为 .yml 文件
                file_path = os.path.join(SAVE_DIR, f"{app_name}_{time.strftime('%Y%m%d')}.yml")
                with open(file_path, "w", encoding="utf-8") as f:
                    f.write(real_yaml_content)
                print(f"💾 已保存至: {file_path}")
            else:
                print(f"⚠️ 备份失败: {app_name}")

if __name__ == "__main__":
    backup_tool = DifyBackup()
    backup_tool.login()
    backup_tool.backup_apps()

备份成功输出

user@user-P10DRG:~/Dify-Backup$ python3 dify-backup.py
✅ 登录成功，Token 已获取
⏩ 跳过 (标签不匹配): 服务器查询助手
📦 正在备份应用: 查询助手(sovia) (e7e3cae8-82ac-4590-922c-628c261af632)...
💾 已保存至: ./dify_backups/20260204/查询助手(sovia)_20260204.yml
💾 已保存至: ./dify_backups/20260204/筹划辅助智能体-1216-kj迁移（备份）_20260204.yml
⏩ 跳过 (标签不匹配): 问答助手
⏩ 跳过 (排除名单): 测试
⏩ 跳过 (标签不匹配): AI语料库-分段添加标签
⏩ 跳过 (标签不匹配): 多模态模型测试

2.恢复

📝恢复recovery-dify.py完整代码：

import requests
import os,time

# --- 配置区域 ---
BASE_URL = "http://172.16.11.53/console/api"  # 请替换为你的 Dify 地址
EMAIL = "[email protected]"                       # 请替换为你的Dify用户
PASSWORD = "****"                             # 请替换为你的Dify密码
IMPORT_DIR = f"./dify_backups/{time.strftime('%Y%m%d')}"
# ----------------

class DifyImporter:
    def __init__(self):
        self.session = requests.Session()

    def login(self):
        url = f"{BASE_URL}/login"
        res = self.session.post(url, json={"email": EMAIL, "password": PASSWORD})
        if res.status_code == 200:
            data = res.json().get("data", {})
            # 登录成功后，设置后续所有请求的通用 Header
            self.session.headers.update({
                "Authorization": f"Bearer {data.get('access_token')}",
                "X-Csrf-Token": self.session.cookies.get("csrf_token"),
                "Content-Type": "application/json"
            })
            print("✅ 登录成功")
            return True
        print(f"❌ 登录失败: {res.text}")
        return False

    def do_import(self, file_path):
        url = f"{BASE_URL}/apps/imports"
        file_name = os.path.basename(file_path)
        
        try:
            # 使用 utf-8-sig 以处理可能存在的 BOM 头
            with open(file_path, 'r', encoding='utf-8-sig') as f:
                dsl_content = f.read()
                
                # 打印前 100 个字符进行调试（确认没有奇怪的嵌套）
                # print(f"DEBUG: {file_name} content start: {dsl_content[:100]}")

                payload = {
                    "mode": "yaml-content",
                    "yaml_content": dsl_content
                }
                
                # 明确不使用额外的 headers 修改，只保留 login 时的 session headers
                response = self.session.post(url, json=payload)
                
                res_json = response.json()
                if response.status_code in [200, 201] and res_json.get("status") != "failed":
                    print(f"✨ 成功导入: {file_name}")
                else:
                    print(f"❌ 导入失败: {file_name}")
                    print(f"   状态码: {response.status_code}")
                    # 报错：Missing app data in YAML content
                    print(f"   详细错误: {res_json.get('error', response.text)}")
                    
        except Exception as e:
            print(f"🚨 脚本执行异常: {e}")

    def run(self):
        if self.login():
            # 获取目录下所有 yaml 文件
            files = [f for f in os.listdir(IMPORT_DIR) if f.endswith(('.yml', '.yaml'))]
            if not files:
                print("📭 备份目录中没有发现 .yml 文件")
                return
                
            for f in files:
                print(f"🚀 正在尝试导入: {f}")
                self.do_import(os.path.join(IMPORT_DIR, f))

if __name__ == "__main__":
    DifyImporter().run()

恢复成功输出

user@user-P10DRG:~/Dify-Backup$ python3 recovery-dify.py
✅ 登录成功
🚀 正在尝试导入: 查询助手(sovia)_20260204.yml
✨ 成功导入: 查询助手(sovia)_20260204.yml

The End