427 lines
19 KiB
Python
427 lines
19 KiB
Python
#!/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)
|