🧩 Dify 社区版多租户批量导入用户实战指南（修正版）

基于社区实践，针对 Dify 初次启动后数据库结构变化 和 常见报错问题 进行修正与增强。

---

🔍 问题背景

参考博客《Dify 社区版手动开启多租户并批量导入用户实战指南》中提到的 accounts.id 为字符串类型，但实际 Dify（v0.6+）中 accounts.id 是 UUID 类型，且 Excel 中的 WORKCODE 并非合法 UUID，直接插入会失败。

此外，未设置时区或语言字段 会导致前端报错：

• 知识库访问：Application error: a client-side exception...
• 工作流创建：Internal Server Error

---

✅ 正确方案概览

要点	说明
✅ accounts.id	必须为 合法 UUID（由代码生成）
✅ 密码	统一设为 CnAI@2025
✅ Excel 列	仅需 LOGINID（姓名）、EMAIL
✅ 时区	必须设置为 'Asia/Shanghai'
✅ 语言	可选，但建议在数据库中允许为空（见后文）

---

📊 所需 Excel 格式（仅 2 列）

LOGINID	EMAIL
张三	zhangsan@example.com
李四	lisi@example.com

⚠️ 不再使用 WORKCODE 作为 ID！

---

🐘 验证数据库结构

进入 PostgreSQL 容器：

psql -h localhost -p 5432 -U postgres -d dify

查看 accounts 表结构：

\d accounts

输出示例：

       Column       |            Type             | ...
--------------------+-----------------------------+-----
 id                 | uuid                        | ← 必须是 UUID
 name               | character varying(255)      |
 email              | character varying(255)      |
 password           | text                        |
 password_salt      | text                        |
 timezone           | character varying(255)      | ← 关键！
 ...

✅ 确认无 custom_id 字段，不要尝试用字符串替代 id。

---

🐍 批量导入脚本（修正版）

import pandas as pd
import psycopg2
from psycopg2 import sql
import uuid
from datetime import datetime
import secrets
import hashlib
import binascii
import base64
import os
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization

# --- 配置 ---
DB_CONFIG = {
    'dbname': 'dify',
    'user': 'postgres',
    'password': 'difyai123456',  # ⚠️ 来自 .env 中的 POSTGRES_PASSWORD
    'host': 'localhost',
    'port': '5432'
}

XLS_FILE_PATH = r'E:\dify\dify相关\users.xls'
OUTPUT_KEYS_DIR = 'private_keys_to_upload'
FIXED_PASSWORD = "CnAI@2025"  # 所有用户统一密码


# --- Dify 密码加密函数（PBKDF2 + Base64）---
def generate_dify_password_hash(plain_password: str):
    salt_bytes = secrets.token_bytes(16)
    dk_bytes = hashlib.pbkdf2_hmac("sha256", plain_password.encode("utf-8"), salt_bytes, 10000)
    hex_hash = binascii.hexlify(dk_bytes)
    final_hash = base64.b64encode(hex_hash).decode('utf-8')
    final_salt = base64.b64encode(salt_bytes).decode('utf-8')
    return final_hash, final_salt


# --- 生成 RSA 密钥对（用于 tenant 加密）---
def generate_key_pair():
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    public_pem = private_key.public_key().public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    private_pem = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    )
    return public_pem.decode('utf-8'), private_pem


# --- 主函数 ---
def main():
    conn = None
    successful = 0
    failed = 0
    passwords = {}

    os.makedirs(OUTPUT_KEYS_DIR, exist_ok=True)

    try:
        print(f"[*] 读取 Excel: {XLS_FILE_PATH}")
        df = pd.read_excel(XLS_FILE_PATH)
        print(f"[+] 成功读取 {len(df)} 条记录。")

        conn = psycopg2.connect(**DB_CONFIG)
        cur = conn.cursor()
        print("[+] 数据库连接成功！\n--- 开始导入 ---")

        for idx, row in df.iterrows():
            name = str(row['LOGINID']).strip()
            email = str(row['EMAIL']).strip()

            if not name or not email:
                print(f"[!] 跳过第 {idx + 2} 行：LOGINID 或 EMAIL 为空")
                failed += 1
                continue

            # 检查邮箱是否已存在
            cur.execute("SELECT 1 FROM accounts WHERE email = %s;", (email,))
            if cur.fetchone():
                print(f"[!] 邮箱已存在，跳过: {email}")
                failed += 1
                continue

            try:
                # 生成密码哈希
                hashed_password, salt = generate_dify_password_hash(FIXED_PASSWORD)
                # 生成密钥对
                public_key, private_key = generate_key_pair()
                # 生成 UUID
                account_id = str(uuid.uuid4())
                tenant_id = str(uuid.uuid4())
                workspace_name = f"{name}'s Workspace"
                ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

                # 1. 插入 accounts（含 timezone！）
                cur.execute(
                    sql.SQL("""
                        INSERT INTO accounts (
                            id, name, email, password, password_salt,
                            status, created_at, updated_at, timezone
                        ) VALUES (%s, %s, %s, %s, %s, 'active', %s, %s, 'Asia/Shanghai')
                    """),
                    (account_id, name, email, hashed_password, salt, ts, ts)
                )

                # 2. 插入 tenants
                cur.execute(
                    sql.SQL("""
                        INSERT INTO tenants (
                            id, name, plan, status, created_at, updated_at, encrypt_public_key
                        ) VALUES (%s, %s, 'basic', 'normal', %s, %s, %s)
                    """),
                    (tenant_id, workspace_name, ts, ts, public_key)
                )

                # 3. 插入关联关系
                cur.execute(
                    sql.SQL("INSERT INTO tenant_account_joins (tenant_id, account_id, role, current) VALUES (%s, %s, 'owner', TRUE)"),
                    (tenant_id, account_id)
                )

                # 4. 保存私钥（用于后续 API 调用）
                key_dir = os.path.join(OUTPUT_KEYS_DIR, tenant_id)
                os.makedirs(key_dir, exist_ok=True)
                with open(os.path.join(key_dir, 'private.pem'), 'wb') as f:
                    f.write(private_key)

                conn.commit()
                successful += 1
                passwords[email] = FIXED_PASSWORD
                print(f"  -> 成功: {name} ({email})")

            except Exception as e:
                conn.rollback()
                print(f"[!] 导入失败 {name} ({email}): {e}")
                failed += 1

        print("\n--- 导入完成 ---")

    except Exception as e:
        print(f"\n[!] 致命错误: {e}")
        if conn:
            conn.rollback()
    finally:
        if conn:
            cur.close()
            conn.close()
            print("\n[*] 数据库连接已关闭。")

    # 输出结果
    print(f"\n✅ 成功: {successful} | ❌ 失败/跳过: {failed}")
    if passwords:
        print("\n" + "="*50)
        print("🔐 用户登录信息")
        print("="*50)
        print("所有用户密码均为: CnAI@2025")
        print("-" * 50)
        for email in passwords:
            print(email)
        print("-" * 50)


if __name__ == "__main__":
    print("🚀 批量导入用户到 Dify（固定密码 CnAI@2025）")
    input("按 Enter 开始...")
    main()

---

⚠️ 常见报错及修复

1. 前端报错：Application error / Internal Server Error

原因：

• accounts.timezone 为 NULL
• sites.default_language 为 NULL 且字段 不允许为空

解决方案：

✅ 方案 A：建账号时显式设置时区（推荐）

已在脚本中处理：

timezone = 'Asia/Shanghai'

✅ 方案 B：允许 default_language 为空（一次性修复）

-- 进入 psql 后执行
ALTER TABLE sites ALTER COLUMN default_language DROP NOT NULL;

💡 社区版注册流程中会强制填写时区和语言，批量导入必须模拟该行为。

---

🔐 数据库连接配置说明

Dify 默认的 PostgreSQL 凭据来自 .env 文件：

POSTGRES_USER=postgres
POSTGRES_PASSWORD=difyai123456  # ← 这是 DB 密码！
POSTGRES_DB=dify

因此 Python 脚本中的 DB_CONFIG 应为：

DB_CONFIG = {
    'dbname': 'dify',
    'user': 'postgres',
    'password': 'difyai123456',  # ⚠️ 不是 'postgres'
    'host': 'localhost',
    'port': '5432'
}

---

📌 总结

步骤	操作
✅ 1	Excel 仅保留 LOGINID 和 EMAIL
✅ 2	使用 uuid.uuid4() 生成 accounts.id
✅ 3	密码统一为 CnAI@2025，使用 Dify 的 PBKDF2 加密逻辑
✅ 4	必须设置 timezone = 'Asia/Shanghai'
✅ 5	（可选）执行 ALTER TABLE sites ALTER COLUMN default_language DROP NOT NULL;
✅ 6	私钥保存至本地，用于后续 API 调用或调试

---

✅ 此方案已在 Dify v0.6.10 社区版验证通过。

如需支持自定义密码、部门、角色等扩展字段，可在此基础上继续增强。欢迎交流！
​
参考：Dify 社区版手动开启多租户并批量导入用户实战指南_dify 多租户-CSDN博客

​