#!/usr/bin/env python3 """ 腾讯云数据库优化历史表创建脚本 创建优化历史功能所需的所有数据库表 """ import os import sys import pymysql from datetime import datetime # 添加项目路径 sys.path.append(os.path.dirname(os.path.abspath(__file__))) # 导入配置 import importlib.util spec = importlib.util.spec_from_file_location("config", "config.py") config_module = importlib.util.module_from_spec(spec) spec.loader.exec_module(config_module) Config = config_module.Config def get_tencent_db_connection(): """获取腾讯云数据库连接""" try: # 从配置中解析腾讯云数据库连接信息 db_uri = Config.TENCENT_SQLALCHEMY_DATABASE_URI # 解析连接字符串: mysql+pymysql://root:!Rjb12191@gz-cynosdbmysql-grp-d26pzce5.sql.tencentcdb.com:24936/pro_db?charset=utf8mb4 if db_uri.startswith('mysql+pymysql://'): db_uri = db_uri.replace('mysql+pymysql://', '') # 分离参数部分 if '?' in db_uri: db_uri, params = db_uri.split('?', 1) # 解析连接信息 auth_host_db = db_uri.split('@') if len(auth_host_db) != 2: raise ValueError("数据库连接字符串格式错误") auth_part = auth_host_db[0] host_db_part = auth_host_db[1] # 解析用户名和密码 if ':' in auth_part: username, password = auth_part.split(':', 1) else: username = auth_part password = '' # 解析主机、端口和数据库名 if ':' in host_db_part: host_port, database = host_db_part.split('/', 1) if ':' in host_port: host, port = host_port.split(':') port = int(port) else: host = host_port port = 3306 else: host = host_db_part port = 3306 database = 'pro_db' print(f"连接信息: {username}@{host}:{port}/{database}") # 创建数据库连接 connection = pymysql.connect( host=host, port=port, user=username, password=password, database=database, charset='utf8mb4', autocommit=True ) return connection except Exception as e: print(f"❌ 数据库连接失败: {str(e)}") return None def create_optimization_history_tables(connection): """创建优化历史相关表""" # 表创建SQL语句 tables_sql = { 'optimization_history': """ CREATE TABLE IF NOT EXISTS `optimization_history` ( `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键ID', `user_id` BIGINT NOT NULL COMMENT '用户ID', `original_text` TEXT NOT NULL COMMENT '原始输入文本', `optimized_text` TEXT NOT NULL COMMENT '优化后文本', `optimization_type` VARCHAR(50) DEFAULT '提示词优化' COMMENT '优化类型', `industry` VARCHAR(100) DEFAULT NULL COMMENT '行业分类', `profession` VARCHAR(100) DEFAULT NULL COMMENT '职业分类', `sub_category` VARCHAR(100) DEFAULT NULL COMMENT '子分类', `template_id` BIGINT DEFAULT NULL COMMENT '使用的模板ID', `satisfaction_rating` TINYINT DEFAULT NULL COMMENT '满意度评分(1-5)', `generation_time` INT DEFAULT NULL COMMENT '生成耗时(毫秒)', `ip_address` VARCHAR(45) DEFAULT NULL COMMENT '用户IP地址', `user_agent` TEXT DEFAULT NULL COMMENT '用户代理信息', `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', PRIMARY KEY (`id`), KEY `idx_user_id` (`user_id`), KEY `idx_created_at` (`created_at`), KEY `idx_optimization_type` (`optimization_type`), KEY `idx_industry` (`industry`), KEY `idx_template_id` (`template_id`), KEY `idx_user_created` (`user_id`, `created_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='优化历史记录表'; """, 'optimization_tags': """ CREATE TABLE IF NOT EXISTS `optimization_tags` ( `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键ID', `history_id` BIGINT NOT NULL COMMENT '历史记录ID', `tag_name` VARCHAR(50) NOT NULL COMMENT '标签名称', `tag_type` VARCHAR(20) DEFAULT 'optimization' COMMENT '标签类型', `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', PRIMARY KEY (`id`), KEY `idx_history_id` (`history_id`), KEY `idx_tag_name` (`tag_name`), CONSTRAINT `fk_optimization_tags_history` FOREIGN KEY (`history_id`) REFERENCES `optimization_history` (`id`) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='优化历史标签表'; """, 'optimization_favorites': """ CREATE TABLE IF NOT EXISTS `optimization_favorites` ( `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键ID', `user_id` BIGINT NOT NULL COMMENT '用户ID', `history_id` BIGINT NOT NULL COMMENT '历史记录ID', `favorite_name` VARCHAR(100) DEFAULT NULL COMMENT '收藏名称', `notes` TEXT DEFAULT NULL COMMENT '备注', `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', PRIMARY KEY (`id`), UNIQUE KEY `uk_user_history` (`user_id`, `history_id`), KEY `idx_user_id` (`user_id`), KEY `idx_history_id` (`history_id`), CONSTRAINT `fk_favorites_history` FOREIGN KEY (`history_id`) REFERENCES `optimization_history` (`id`) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='优化历史收藏表'; """, 'user_usage_stats': """ CREATE TABLE IF NOT EXISTS `user_usage_stats` ( `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键ID', `user_id` BIGINT NOT NULL COMMENT '用户ID', `date` DATE NOT NULL COMMENT '统计日期', `generation_count` INT DEFAULT 0 COMMENT '生成次数', `total_time_saved` INT DEFAULT 0 COMMENT '节省时间(分钟)', `avg_rating` DECIMAL(3,2) DEFAULT NULL COMMENT '平均评分', `total_ratings` INT DEFAULT 0 COMMENT '总评分次数', `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', PRIMARY KEY (`id`), UNIQUE KEY `uk_user_date` (`user_id`, `date`), KEY `idx_date` (`date`), KEY `idx_user_id` (`user_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户使用统计表'; """, 'system_config': """ CREATE TABLE IF NOT EXISTS `system_config` ( `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键ID', `config_key` VARCHAR(100) NOT NULL COMMENT '配置键', `config_value` TEXT NOT NULL COMMENT '配置值', `config_type` VARCHAR(20) DEFAULT 'string' COMMENT '配置类型', `description` VARCHAR(255) DEFAULT NULL COMMENT '配置描述', `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', PRIMARY KEY (`id`), UNIQUE KEY `uk_config_key` (`config_key`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='系统配置表'; """ } # 索引创建SQL (MySQL不支持IF NOT EXISTS，需要先检查) indexes_sql = [ ("idx_user_type_created", "CREATE INDEX `idx_user_type_created` ON `optimization_history` (`user_id`, `optimization_type`, `created_at`);"), ("idx_user_rating", "CREATE INDEX `idx_user_rating` ON `optimization_history` (`user_id`, `satisfaction_rating`);") ] # 默认配置数据 default_configs = [ ("max_history_per_user", "1000", "integer", "每个用户最大历史记录数"), ("history_retention_days", "365", "integer", "历史记录保留天数"), ("enable_analytics", "true", "boolean", "是否启用使用分析"), ("default_optimization_type", "提示词优化", "string", "默认优化类型") ] try: cursor = connection.cursor() print("🚀 开始创建优化历史相关表...") # 创建表 for table_name, sql in tables_sql.items(): print(f"📋 创建表: {table_name}") cursor.execute(sql) print(f"✅ 表 {table_name} 创建成功") # 创建索引 print("🔍 创建索引...") for index_name, index_sql in indexes_sql: try: # 检查索引是否已存在 cursor.execute(f"SHOW INDEX FROM optimization_history WHERE Key_name = '{index_name}'") if not cursor.fetchone(): cursor.execute(index_sql) print(f"✅ 索引 {index_name} 创建成功") else: print(f"⚠️ 索引 {index_name} 已存在，跳过") except Exception as e: print(f"⚠️ 索引 {index_name} 创建失败: {str(e)}") print("✅ 索引创建完成") # 插入默认配置 print("⚙️ 插入默认配置...") for config_key, config_value, config_type, description in default_configs: # 检查配置是否已存在 cursor.execute("SELECT id FROM system_config WHERE config_key = %s", (config_key,)) if not cursor.fetchone(): cursor.execute(""" INSERT INTO system_config (config_key, config_value, config_type, description) VALUES (%s, %s, %s, %s) """, (config_key, config_value, config_type, description)) print(f"✅ 配置 {config_key} 插入成功") else: print(f"⚠️ 配置 {config_key} 已存在，跳过") print("\n🎉 所有表创建完成！") return True except Exception as e: print(f"❌ 创建表失败: {str(e)}") return False finally: if 'cursor' in locals(): cursor.close() def insert_sample_data(connection): """插入示例数据""" try: cursor = connection.cursor() print("\n📝 插入示例数据...") # 示例历史记录 sample_histories = [ { 'user_id': 1, 'original_text': '请帮我写一个产品介绍', 'optimized_text': '以下是专业的产品介绍模板：\n\n【产品概述】\n产品名称：AI智能助手\n核心功能：智能对话、任务管理、数据分析\n目标用户：企业用户、个人用户\n\n【产品特色】\n1. 智能对话：自然语言处理，理解用户意图\n2. 任务管理：自动规划，提高工作效率\n3. 数据分析：深度洞察，支持决策制定\n\n【使用场景】\n场景1：企业客服自动化\n场景2：个人日程管理\n场景3：数据分析报告生成\n\n【技术优势】\n- 先进的AI算法\n- 云端部署，稳定可靠\n- 支持多平台接入', 'optimization_type': '产品文案', 'industry': '科技', 'profession': '产品经理', 'satisfaction_rating': 5 }, { 'user_id': 1, 'original_text': '如何写邮件给客户', 'optimized_text': '【专业商务邮件写作指南】\n\n一、邮件结构\n1. 主题行：简洁明确，突出要点\n2. 称呼：根据关系选择正式或非正式\n3. 正文：逻辑清晰，分段明确\n4. 结尾：礼貌用语+签名\n\n二、写作要点\n- 语言简洁：避免冗长句子\n- 重点突出：使用加粗或列表\n- 语气恰当：根据收件人调整\n- 检查无误：发送前仔细校对\n\n三、常见类型\n1. 工作汇报邮件\n2. 会议邀请邮件\n3. 项目协调邮件\n4. 客户沟通邮件\n\n四、注意事项\n- 避免使用过于正式或随意的语言\n- 注意时区和时间格式\n- 附件大小控制在合理范围\n- 及时回复，保持沟通效率', 'optimization_type': '商务邮件', 'industry': '商务', 'profession': '行政助理', 'satisfaction_rating': 4 }, { 'user_id': 1, 'original_text': '数据库设计', 'optimized_text': '【数据库设计最佳实践】\n\n一、需求分析\n1. 业务需求梳理\n2. 数据关系分析\n3. 性能要求评估\n\n二、表结构设计\n- 主键设计：自增ID或UUID\n- 字段类型：选择合适的数据类型\n- 索引设计：提高查询性能\n- 约束设置：保证数据完整性\n\n三、优化策略\n1. 分库分表：应对大数据量\n2. 读写分离：提高并发性能\n3. 缓存策略：减少数据库压力\n4. 监控告警：及时发现问题\n\n四、安全考虑\n- 数据加密：敏感信息保护\n- 访问控制：权限管理\n- 备份策略：数据安全\n- 审计日志：操作追踪', 'optimization_type': '技术文档', 'industry': 'IT', 'profession': '后端开发', 'satisfaction_rating': 5 } ] # 插入历史记录 for i, history in enumerate(sample_histories): cursor.execute(""" INSERT INTO optimization_history (user_id, original_text, optimized_text, optimization_type, industry, profession, satisfaction_rating) VALUES (%s, %s, %s, %s, %s, %s, %s) """, ( history['user_id'], history['original_text'], history['optimized_text'], history['optimization_type'], history['industry'], history['profession'], history['satisfaction_rating'] )) history_id = cursor.lastrowid print(f"✅ 历史记录 {i+1} 插入成功 (ID: {history_id})") # 插入标签 tags = ['逻辑强化', '场景适配', '结构优化'] for tag_name in tags: cursor.execute(""" INSERT INTO optimization_tags (history_id, tag_name, tag_type) VALUES (%s, %s, %s) """, (history_id, tag_name, 'optimization')) # 插入用户统计 from datetime import date today = date.today() cursor.execute(""" INSERT INTO user_usage_stats (user_id, date, generation_count, total_time_saved, avg_rating, total_ratings) VALUES (%s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE generation_count = VALUES(generation_count), total_time_saved = VALUES(total_time_saved), avg_rating = VALUES(avg_rating), total_ratings = VALUES(total_ratings) """, (1, today, 3, 15, 4.7, 3)) print("✅ 用户统计插入成功") print("✅ 示例数据插入完成") return True except Exception as e: print(f"❌ 插入示例数据失败: {str(e)}") return False finally: if 'cursor' in locals(): cursor.close() def verify_tables(connection): """验证表创建结果""" try: cursor = connection.cursor() print("\n🔍 验证表创建结果...") # 检查表是否存在 tables = ['optimization_history', 'optimization_tags', 'optimization_favorites', 'user_usage_stats', 'system_config'] for table in tables: cursor.execute(f"SHOW TABLES LIKE '{table}'") if cursor.fetchone(): print(f"✅ 表 {table} 存在") # 检查记录数 cursor.execute(f"SELECT COUNT(*) FROM {table}") count = cursor.fetchone()[0] print(f" 📊 记录数: {count}") else: print(f"❌ 表 {table} 不存在") # 检查索引 print("\n🔍 检查索引...") cursor.execute("SHOW INDEX FROM optimization_history") indexes = cursor.fetchall() print(f"✅ optimization_history 表有 {len(indexes)} 个索引") print("\n🎉 表验证完成！") return True except Exception as e: print(f"❌ 验证表失败: {str(e)}") return False finally: if 'cursor' in locals(): cursor.close() def main(): """主函数""" print("=" * 60) print("🚀 腾讯云数据库优化历史表创建脚本") print("=" * 60) # 获取数据库连接 connection = get_tencent_db_connection() if not connection: print("❌ 无法连接到腾讯云数据库，请检查配置") return False try: # 创建表 if not create_optimization_history_tables(connection): return False # 插入示例数据 if not insert_sample_data(connection): return False # 验证表 if not verify_tables(connection): return False print("\n" + "=" * 60) print("🎉 腾讯云数据库优化历史表创建完成！") print("=" * 60) print("📋 已创建的表:") print(" - optimization_history (优化历史记录表)") print(" - optimization_tags (优化历史标签表)") print(" - optimization_favorites (优化历史收藏表)") print(" - user_usage_stats (用户使用统计表)") print(" - system_config (系统配置表)") print("\n📊 已插入示例数据:") print(" - 3条历史记录") print(" - 9个标签") print(" - 1条用户统计") print(" - 4个系统配置") print("\n✅ 现在可以启用完整的优化历史功能了！") return True except Exception as e: print(f"❌ 执行失败: {str(e)}") return False finally: if connection: connection.close() print("\n🔌 数据库连接已关闭") if __name__ == '__main__': success = main() sys.exit(0 if success else 1)