#!/usr/bin/env python3 """ 租户配置迁移脚本 功能：将各租户的 .env 文件配置迁移到 kaopeilian_admin 数据库 使用方法： python scripts/migrate_env_to_db.py 说明： 1. 读取各租户的 .env 文件 2. 创建租户记录 3. 将配置写入 tenant_configs 表 4. 保留原 .env 文件作为备份 """ import os import sys import re import pymysql from datetime import datetime from typing import Dict, List, Tuple, Optional # 添加项目根目录到路径 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # ============================================ # 配置 # ============================================ # 管理库连接配置 ADMIN_DB_CONFIG = { "host": os.getenv("ADMIN_DB_HOST", "120.79.247.16"), "port": int(os.getenv("ADMIN_DB_PORT", "3309")), "user": os.getenv("ADMIN_DB_USER", "root"), "password": os.getenv("ADMIN_DB_PASSWORD", "ProdMySQL2025!@#"), "db": os.getenv("ADMIN_DB_NAME", "kaopeilian_admin"), "charset": "utf8mb4", } # 租户配置 TENANTS = [ { "code": "demo", "name": "演示版", "display_name": "考培练系统-演示版", "domain": "aiedu.ireborn.com.cn", "env_file": "/root/aiedu/kaopeilian-backend/.env.production", "industry": "medical_beauty", }, { "code": "hua", "name": "华尔倍丽", "display_name": "华尔倍丽-考培练系统", "domain": "hua.ireborn.com.cn", "env_file": "/root/aiedu/kaopeilian-backend/.env.hua", "industry": "medical_beauty", }, { "code": "yy", "name": "杨扬宠物", "display_name": "杨扬宠物-考培练系统", "domain": "yy.ireborn.com.cn", "env_file": "/root/aiedu/kaopeilian-backend/.env.yy", "industry": "pet", }, { "code": "hl", "name": "武汉禾丽", "display_name": "武汉禾丽-考培练系统", "domain": "hl.ireborn.com.cn", "env_file": "/root/aiedu/kaopeilian-backend/.env.hl", "industry": "medical_beauty", }, { "code": "xy", "name": "芯颜定制", "display_name": "芯颜定制-考培练系统", "domain": "xy.ireborn.com.cn", "env_file": "/root/aiedu/kaopeilian-backend/.env.xy", "industry": "medical_beauty", }, { "code": "fw", "name": "飞沃", "display_name": "飞沃-考培练系统", "domain": "fw.ireborn.com.cn", "env_file": "/root/aiedu/kaopeilian-backend/.env.fw", "industry": "medical_beauty", }, { "code": "ex", "name": "恩喜成都总院", "display_name": "恩喜成都总院-考培练系统", "domain": "ex.ireborn.com.cn", "env_file": "/root/aiedu/kaopeilian-backend/.env.ex", "industry": "medical_beauty", }, { "code": "kpl", "name": "瑞小美", "display_name": "瑞小美-考培练系统", "domain": "kpl.ireborn.com.cn", "env_file": "/root/aiedu/kaopeilian-backend/.env.kpl", "industry": "medical_beauty", }, ] # 配置键到分组的映射 CONFIG_MAPPING = { # 数据库配置 "MYSQL_HOST": ("database", "string", False), "MYSQL_PORT": ("database", "int", False), "MYSQL_USER": ("database", "string", False), "MYSQL_PASSWORD": ("database", "string", True), "MYSQL_DATABASE": ("database", "string", False), # Redis配置 "REDIS_HOST": ("redis", "string", False), "REDIS_PORT": ("redis", "int", False), "REDIS_DB": ("redis", "int", False), "REDIS_URL": ("redis", "string", False), # 安全配置 "SECRET_KEY": ("security", "string", True), "CORS_ORIGINS": ("security", "json", False), # Dify配置 "DIFY_API_KEY": ("dify", "string", True), "DIFY_EXAM_GENERATOR_API_KEY": ("dify", "string", True), "DIFY_PRACTICE_API_KEY": ("dify", "string", True), "DIFY_COURSE_CHAT_API_KEY": ("dify", "string", True), "DIFY_YANJI_ANALYSIS_API_KEY": ("dify", "string", True), # Coze配置 "COZE_PRACTICE_BOT_ID": ("coze", "string", False), "COZE_BROADCAST_WORKFLOW_ID": ("coze", "string", False), "COZE_BROADCAST_SPACE_ID": ("coze", "string", False), "COZE_BROADCAST_BOT_ID": ("coze", "string", False), "COZE_OAUTH_CLIENT_ID": ("coze", "string", False), "COZE_OAUTH_PUBLIC_KEY_ID": ("coze", "string", False), "COZE_OAUTH_PRIVATE_KEY_PATH": ("coze", "string", False), # AI配置 "AI_PRIMARY_API_KEY": ("ai", "string", True), "AI_PRIMARY_BASE_URL": ("ai", "string", False), "AI_FALLBACK_API_KEY": ("ai", "string", True), "AI_FALLBACK_BASE_URL": ("ai", "string", False), "AI_DEFAULT_MODEL": ("ai", "string", False), "AI_TIMEOUT": ("ai", "int", False), # 言迹配置 "YANJI_CLIENT_ID": ("yanji", "string", False), "YANJI_CLIENT_SECRET": ("yanji", "string", True), "YANJI_TENANT_ID": ("yanji", "string", False), "YANJI_ESTATE_ID": ("yanji", "string", False), # 其他配置 "APP_NAME": ("basic", "string", False), "PUBLIC_DOMAIN": ("basic", "string", False), } def parse_env_file(filepath: str) -> Dict[str, str]: """解析 .env 文件""" config = {} if not os.path.exists(filepath): print(f" 警告: 文件不存在 {filepath}") return config with open(filepath, 'r', encoding='utf-8') as f: for line in f: line = line.strip() # 跳过注释和空行 if not line or line.startswith('#'): continue # 解析 KEY=VALUE if '=' in line: key, value = line.split('=', 1) key = key.strip() value = value.strip() # 去除引号 if (value.startswith('"') and value.endswith('"')) or \ (value.startswith("'") and value.endswith("'")): value = value[1:-1] config[key] = value return config def create_tenant(cursor, tenant: Dict) -> int: """创建租户记录，返回租户ID""" # 检查是否已存在 cursor.execute( "SELECT id FROM tenants WHERE code = %s", (tenant["code"],) ) row = cursor.fetchone() if row: print(f" 租户已存在，ID: {row['id']}") return row["id"] # 创建新租户 cursor.execute( """ INSERT INTO tenants (code, name, display_name, domain, industry, status, created_by) VALUES (%s, %s, %s, %s, %s, 'active', 1) """, (tenant["code"], tenant["name"], tenant["display_name"], tenant["domain"], tenant["industry"]) ) tenant_id = cursor.lastrowid print(f" 创建租户成功，ID: {tenant_id}") return tenant_id def migrate_config(cursor, tenant_id: int, config: Dict[str, str]) -> Tuple[int, int]: """迁移配置到数据库""" inserted = 0 updated = 0 for key, value in config.items(): if key not in CONFIG_MAPPING: continue config_group, value_type, is_secret = CONFIG_MAPPING[key] # 检查是否已存在 cursor.execute( """ SELECT id FROM tenant_configs WHERE tenant_id = %s AND config_group = %s AND config_key = %s """, (tenant_id, config_group, key) ) row = cursor.fetchone() if row: # 更新 cursor.execute( """ UPDATE tenant_configs SET config_value = %s, value_type = %s, is_encrypted = %s, updated_at = NOW() WHERE id = %s """, (value, value_type, is_secret, row["id"]) ) updated += 1 else: # 插入 cursor.execute( """ INSERT INTO tenant_configs (tenant_id, config_group, config_key, config_value, value_type, is_encrypted) VALUES (%s, %s, %s, %s, %s, %s) """, (tenant_id, config_group, key, value, value_type, is_secret) ) inserted += 1 return inserted, updated def main(): """主函数""" print("=" * 60) print("租户配置迁移脚本") print("=" * 60) print(f"\n目标数据库: {ADMIN_DB_CONFIG['host']}:{ADMIN_DB_CONFIG['port']}/{ADMIN_DB_CONFIG['db']}") print(f"待迁移租户: {len(TENANTS)} 个\n") # 连接数据库 conn = pymysql.connect(**ADMIN_DB_CONFIG, cursorclass=pymysql.cursors.DictCursor) try: with conn.cursor() as cursor: total_inserted = 0 total_updated = 0 for tenant in TENANTS: print(f"\n处理租户: {tenant['name']} ({tenant['code']})") print(f" 环境文件: {tenant['env_file']}") # 解析 .env 文件 config = parse_env_file(tenant['env_file']) print(f" 读取配置: {len(config)} 项") # 创建租户 tenant_id = create_tenant(cursor, tenant) # 迁移配置 if config: inserted, updated = migrate_config(cursor, tenant_id, config) print(f" 迁移结果: 新增 {inserted} 项, 更新 {updated} 项") total_inserted += inserted total_updated += updated else: print(" 跳过迁移（无配置）") # 提交事务 conn.commit() print("\n" + "=" * 60) print("迁移完成!") print(f"总计: 新增 {total_inserted} 项, 更新 {total_updated} 项") print("=" * 60) except Exception as e: conn.rollback() print(f"\n错误: {e}") raise finally: conn.close() if __name__ == "__main__": main()