#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 腾讯云数据库修复版部署脚本 解决GTID一致性问题，适配腾讯云数据库 """ import pymysql import sys from datetime import datetime # 腾讯云数据库配置 DB_CONFIG = { 'host': 'gz-cynosdbmysql-grp-d26pzce5.sql.tencentcdb.com', 'port': 24936, 'user': 'root', 'password': '!Rjb12191', 'database': 'pro_db', 'charset': 'utf8mb4', 'autocommit': True # 启用自动提交 } def check_connection(): """检查数据库连接""" try: connection = pymysql.connect(**DB_CONFIG) cursor = connection.cursor() cursor.execute("SELECT 1") result = cursor.fetchone() cursor.close() connection.close() print("✅ 数据库连接成功") return True except Exception as e: print(f"❌ 数据库连接失败: {str(e)}") return False def create_tables(): """创建表结构""" print("📋 开始创建表结构...") try: connection = pymysql.connect(**DB_CONFIG) cursor = connection.cursor() # 1. 创建历史记录表 print("\n1. 创建 prompt_history 表...") cursor.execute(""" CREATE TABLE IF NOT EXISTS prompt_history ( id INT PRIMARY KEY AUTO_INCREMENT, user_id INT NOT NULL, wx_user_id INT, original_input TEXT NOT NULL, generated_prompt TEXT NOT NULL, template_id INT, template_name VARCHAR(100), generation_time INT, satisfaction_rating TINYINT, tags JSON, is_favorite BOOLEAN DEFAULT FALSE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_user_id (user_id), INDEX idx_created_at (created_at), INDEX idx_template_id (template_id), INDEX idx_is_favorite (is_favorite) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) print(" ✅ prompt_history 表创建成功") # 2. 创建标签表 print("\n2. 创建 history_tags 表...") cursor.execute(""" CREATE TABLE IF NOT EXISTS history_tags ( id INT PRIMARY KEY AUTO_INCREMENT, history_id INT NOT NULL, tag_name VARCHAR(50) NOT NULL, tag_type VARCHAR(20) DEFAULT 'custom', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_history_id (history_id), INDEX idx_tag_name (tag_name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) print(" ✅ history_tags 表创建成功") # 3. 创建统计表 print("\n3. 创建 user_statistics 表...") cursor.execute(""" CREATE TABLE IF NOT EXISTS user_statistics ( id INT PRIMARY KEY AUTO_INCREMENT, user_id INT UNIQUE NOT NULL, total_generations INT DEFAULT 0, favorite_count INT DEFAULT 0, avg_rating DECIMAL(3,2) DEFAULT 0.00, last_generation_at TIMESTAMP NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_user_id (user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) print(" ✅ user_statistics 表创建成功") cursor.close() connection.close() print("\n✅ 所有表创建成功") return True except Exception as e: print(f"❌ 创建表失败: {str(e)}") return False def create_sample_data(): """创建示例数据""" print("\n📝 创建示例数据...") try: connection = pymysql.connect(**DB_CONFIG) cursor = connection.cursor() # 检查是否已有数据 cursor.execute("SELECT COUNT(*) FROM prompt_history") count = cursor.fetchone()[0] if count > 0: print(f" ⚠️ 表中已有 {count} 条记录，跳过示例数据创建") return True # 插入示例数据 sample_data = [ (1, '请帮我写一个关于人工智能的提示词', '你是一位专业的人工智能专家，请详细分析人工智能的发展趋势、应用领域和未来前景。', 1, 'AI专家助手', 1500, 4, '["AI", "技术", "分析"]', False), (1, '帮我写一个产品经理的工作提示词', '你是一位资深的产品经理，请从用户需求、市场分析、产品设计等角度提供专业建议。', 2, '产品经理助手', 1200, 5, '["产品", "管理", "分析"]', True) ] for data in sample_data: cursor.execute(""" INSERT INTO prompt_history (user_id, original_input, generated_prompt, template_id, template_name, generation_time, satisfaction_rating, tags, is_favorite) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, data) print(f" ✅ 插入示例数据: {data[1][:30]}...") # 更新用户统计 cursor.execute(""" INSERT INTO user_statistics (user_id, total_generations, favorite_count, avg_rating, last_generation_at) VALUES (1, 2, 1, 4.5, NOW()) ON DUPLICATE KEY UPDATE total_generations = VALUES(total_generations), favorite_count = VALUES(favorite_count), avg_rating = VALUES(avg_rating), last_generation_at = VALUES(last_generation_at), updated_at = NOW() """) cursor.close() connection.close() print(" ✅ 示例数据创建成功") return True except Exception as e: print(f"❌ 创建示例数据失败: {str(e)}") return False def verify_deployment(): """验证部署结果""" print("\n🔍 验证部署结果...") try: connection = pymysql.connect(**DB_CONFIG) cursor = connection.cursor() # 检查表是否存在 tables = ['prompt_history', 'history_tags', 'user_statistics'] 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} 不存在") # 显示示例数据 cursor.execute("SELECT id, original_input, template_name, created_at FROM prompt_history LIMIT 3") records = cursor.fetchall() if records: print("\n📊 示例数据预览:") for record in records: print(f" ID: {record[0]}, 输入: {record[1][:30]}..., 模板: {record[2]}, 时间: {record[3]}") cursor.close() connection.close() return True except Exception as e: print(f"❌ 验证部署失败: {str(e)}") return False def main(): """主函数""" print("🚀 腾讯云数据库优化历史功能部署（修复版）") print("="*60) print(f"部署时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"目标数据库: {DB_CONFIG['host']}:{DB_CONFIG['port']}") print(f"数据库名: {DB_CONFIG['database']}") print("="*60) # 1. 检查连接 print("\n1. 检查数据库连接...") if not check_connection(): print("❌ 数据库连接失败，部署终止") sys.exit(1) # 2. 创建表结构 print("\n2. 创建表结构...") if not create_tables(): print("❌ 创建表失败，部署终止") sys.exit(1) # 3. 创建示例数据 print("\n3. 创建示例数据...") if not create_sample_data(): print("⚠️ 示例数据创建失败，但表结构已创建") # 4. 验证部署 print("\n4. 验证部署结果...") if not verify_deployment(): print("⚠️ 验证失败，但部署可能已成功") print("\n" + "="*60) print("🎉 部署完成！") print("="*60) print("📋 部署摘要:") print(" ✅ 历史记录表 (prompt_history)") print(" ✅ 标签表 (history_tags)") print(" ✅ 统计表 (user_statistics)") print(" ✅ 示例数据") print("\n🔗 下一步操作:") print(" 1. 重启应用: pkill -f gunicorn && gunicorn -c gunicorn.conf.py src.flask_prompt_master:app") print(" 2. 访问历史页面: http://localhost:5002/history") print(" 3. 测试功能: python3 test_history_feature.py") print("\n🌐 访问地址:") print(" 历史页面: http://localhost:5002/history") print(" API接口: http://localhost:5002/api/history") if __name__ == "__main__": main()