wecom_it_smart_desk/scripts/dify_export.py

"""
Dify 应用 DSL 导出工具 v2
通过 Console API 登录 → 获取应用列表 → 匹配应用名 → 导出 DSL YAML
"""
import sys
import json
import httpx

# === 配置 ===
DIFY_CONSOLE_URL = "http://yw-dify.dc.servyou-it.com/console/api"
EMAIL = "sxn@servyou.com.cn"
PASSWORD = "t6C@VTYv5V"
TARGET_APP_NAME = "智能IT支持-员工咨询"  # 要导出的应用名
TIMEOUT = 30.0


def login(client: httpx.Client) -> dict:
    """登录 Dify Console，获取认证信息"""
    login_url = f"{DIFY_CONSOLE_URL}/login"
    payload = {"email": EMAIL, "password": PASSWORD}

    print(f"[1] 登录 Dify Console: {login_url}")
    resp = client.post(login_url, json=payload, timeout=TIMEOUT)
    print(f"    状态码: {resp.status_code}")

    if resp.status_code != 200:
        print(f"    响应: {resp.text[:500]}")
        raise RuntimeError(f"登录失败: HTTP {resp.status_code}")

    data = resp.json()
    print(f"    登录响应: {json.dumps(data, ensure_ascii=False, indent=2)[:2000]}")
    return data


def list_apps(client: httpx.Client, login_data: dict) -> list:
    """获取应用列表，找到目标应用"""
    # 尝试多个可能的应用列表 API
    urls = [
        f"{DIFY_CONSOLE_URL}/apps",
        f"{DIFY_CONSOLE_URL}/apps?page=1&limit=50",
    ]

    headers = {}
    token = login_data.get("access_token") or login_data.get("data", {}).get("access_token")
    if token:
        headers["Authorization"] = f"Bearer {token}"

    for url in urls:
        print(f"[2] 获取应用列表: {url}")
        resp = client.get(url, headers=headers, timeout=TIMEOUT)
        print(f"    状态码: {resp.status_code}")

        if resp.status_code == 200:
            data = resp.json()
            apps = data.get("data", data) if isinstance(data, dict) else data
            if isinstance(apps, list):
                print(f"    应用总数: {len(apps)}")
                # 列出所有应用
                for app in apps:
                    name = app.get("name", "N/A")
                    app_id = app.get("id", "N/A")
                    mode = app.get("mode", "N/A")
                    print(f"    - {name} (id={app_id}, mode={mode})")
                return apps
            else:
                print(f"    响应格式异常: {json.dumps(data, ensure_ascii=False)[:500]}")
        else:
            print(f"    响应: {resp.text[:300]}")

    raise RuntimeError("无法获取应用列表")


def export_app(client: httpx.Client, login_data: dict, app_id: str) -> str:
    """导出指定应用 DSL"""
    export_url = f"{DIFY_CONSOLE_URL}/apps/{app_id}/export"
    print(f"[3] 导出应用 DSL: {export_url}")

    headers = {"Accept": "text/yaml, application/yaml, application/json, */*"}
    token = login_data.get("access_token") or login_data.get("data", {}).get("access_token")
    if token:
        headers["Authorization"] = f"Bearer {token}"

    resp = client.get(export_url, headers=headers, timeout=TIMEOUT)
    print(f"    状态码: {resp.status_code}")
    print(f"    Content-Type: {resp.headers.get('content-type', 'unknown')}")

    if resp.status_code == 200:
        return resp.text

    print(f"    响应: {resp.text[:1000]}")
    raise RuntimeError(f"导出失败: HTTP {resp.status_code}")


def main():
    output_file = "D:\\资料\\03-项目开发\\wecom_it_smart_desk\\scripts\\dify_export_result.yaml"

    with httpx.Client(timeout=TIMEOUT, follow_redirects=True) as client:
        # Step 1: 登录
        login_data = login(client)

        # Step 2: 获取应用列表
        apps = list_apps(client, login_data)

        # Step 3: 找到目标应用
        target_app = None
        for app in apps:
            if app.get("name") == TARGET_APP_NAME:
                target_app = app
                break

        if not target_app:
            print(f"\n❌ 未找到应用: {TARGET_APP_NAME}")
            print(f"   可用的应用: {[a['name'] for a in apps]}")
            return 1

        app_id = target_app["id"]
        print(f"\n✅ 找到目标应用: {TARGET_APP_NAME} (id={app_id})")

        # Step 4: 导出 DSL
        dsl_content = export_app(client, login_data, app_id)

        # Step 5: 保存
        with open(output_file, "w", encoding="utf-8") as f:
            f.write(dsl_content)
        print(f"\n[4] DSL 已保存到: {output_file}")
        print(f"    文件大小: {len(dsl_content)} 字符")

    return 0


if __name__ == "__main__":
    sys.exit(main())