255 lines
8.9 KiB
Python
Executable File
255 lines
8.9 KiB
Python
Executable File
#!/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()
|