258 lines
10 KiB
Python
258 lines
10 KiB
Python
"""
|
||
将localStorage中的优化历史数据迁移到腾讯云数据库
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
# 添加项目路径
|
||
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from flask import Flask
|
||
from flask_sqlalchemy import SQLAlchemy
|
||
from flask_login import LoginManager
|
||
from src.flask_prompt_master.models.optimization_history import (
|
||
OptimizationHistory, OptimizationTag, UserUsageStats
|
||
)
|
||
|
||
def create_app():
|
||
"""创建Flask应用"""
|
||
app = Flask(__name__)
|
||
|
||
# 配置数据库
|
||
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://username:password@host:port/database'
|
||
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
||
|
||
# 初始化扩展
|
||
db = SQLAlchemy(app)
|
||
login_manager = LoginManager()
|
||
login_manager.init_app(app)
|
||
|
||
return app, db
|
||
|
||
def migrate_localStorage_data():
|
||
"""迁移localStorage数据到数据库"""
|
||
app, db = create_app()
|
||
|
||
with app.app_context():
|
||
print("开始迁移localStorage数据到数据库...")
|
||
|
||
# 模拟localStorage数据（实际使用时需要从浏览器获取）
|
||
sample_localStorage_data = [
|
||
{
|
||
"id": 1704672000000,
|
||
"timestamp": "2024-01-08T12:00:00.000Z",
|
||
"original": "请帮我写一个营销文案",
|
||
"optimized": "以下是针对您需求的专业营销文案：\n\n【产品定位】\n明确产品核心价值主张...",
|
||
"type": "提示词优化"
|
||
},
|
||
{
|
||
"id": 1704675600000,
|
||
"timestamp": "2024-01-08T13:00:00.000Z",
|
||
"original": "如何做金融投资",
|
||
"optimized": "【专业金融投资提示框架】\n一、投资主体定位\n1. 请说明您的风险承受能力等级...",
|
||
"type": "提示词优化"
|
||
}
|
||
]
|
||
|
||
migrated_count = 0
|
||
error_count = 0
|
||
|
||
for data in sample_localStorage_data:
|
||
try:
|
||
# 解析时间戳
|
||
timestamp = datetime.fromisoformat(data['timestamp'].replace('Z', '+00:00'))
|
||
|
||
# 创建历史记录（假设用户ID为1，实际使用时需要根据登录用户确定）
|
||
history = OptimizationHistory(
|
||
user_id=1, # 实际使用时需要从登录用户获取
|
||
original_text=data['original'],
|
||
optimized_text=data['optimized'],
|
||
optimization_type=data.get('type', '提示词优化'),
|
||
created_at=timestamp
|
||
)
|
||
|
||
db.session.add(history)
|
||
db.session.flush() # 获取ID
|
||
|
||
# 添加默认标签
|
||
default_tags = ['逻辑强化', '场景适配', '结构优化']
|
||
for tag_name in default_tags:
|
||
tag = OptimizationTag(
|
||
history_id=history.id,
|
||
tag_name=tag_name,
|
||
tag_type='optimization'
|
||
)
|
||
db.session.add(tag)
|
||
|
||
migrated_count += 1
|
||
print(f"✓ 迁移记录 {migrated_count}: {data['original'][:30]}...")
|
||
|
||
except Exception as e:
|
||
error_count += 1
|
||
print(f"✗ 迁移失败: {str(e)}")
|
||
|
||
# 提交所有更改
|
||
try:
|
||
db.session.commit()
|
||
print(f"\n迁移完成！")
|
||
print(f"成功迁移: {migrated_count} 条记录")
|
||
print(f"失败记录: {error_count} 条")
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
print(f"提交失败: {str(e)}")
|
||
|
||
def create_sample_data():
|
||
"""创建示例数据"""
|
||
app, db = create_app()
|
||
|
||
with app.app_context():
|
||
print("创建示例数据...")
|
||
|
||
# 创建示例历史记录
|
||
sample_histories = [
|
||
{
|
||
'user_id': 1,
|
||
'original_text': '请帮我写一个产品介绍',
|
||
'optimized_text': '以下是专业的产品介绍模板：\n\n【产品概述】\n产品名称：[产品名称]\n核心功能：[主要功能]\n目标用户：[用户群体]\n\n【产品特色】\n1. 独特卖点1\n2. 独特卖点2\n3. 独特卖点3\n\n【使用场景】\n场景1：具体应用情况\n场景2：具体应用情况\n\n【技术优势】\n- 技术特点1\n- 技术特点2\n- 技术特点3',
|
||
'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. 客户沟通邮件',
|
||
'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. 监控告警：及时发现问题',
|
||
'optimization_type': '技术文档',
|
||
'industry': 'IT',
|
||
'profession': '后端开发',
|
||
'satisfaction_rating': 5
|
||
}
|
||
]
|
||
|
||
for i, history_data in enumerate(sample_histories):
|
||
try:
|
||
# 创建历史记录
|
||
history = OptimizationHistory(**history_data)
|
||
db.session.add(history)
|
||
db.session.flush()
|
||
|
||
# 添加标签
|
||
tags = ['逻辑强化', '场景适配', '结构优化']
|
||
for tag_name in tags:
|
||
tag = OptimizationTag(
|
||
history_id=history.id,
|
||
tag_name=tag_name,
|
||
tag_type='optimization'
|
||
)
|
||
db.session.add(tag)
|
||
|
||
print(f"✓ 创建示例记录 {i+1}: {history_data['original_text'][:30]}...")
|
||
|
||
except Exception as e:
|
||
print(f"✗ 创建失败: {str(e)}")
|
||
|
||
# 创建用户统计
|
||
try:
|
||
from datetime import date
|
||
stats = UserUsageStats(
|
||
user_id=1,
|
||
date=date.today(),
|
||
generation_count=3,
|
||
total_time_saved=15,
|
||
avg_rating=4.7,
|
||
total_ratings=3
|
||
)
|
||
db.session.add(stats)
|
||
print("✓ 创建用户统计")
|
||
except Exception as e:
|
||
print(f"✗ 创建统计失败: {str(e)}")
|
||
|
||
# 提交所有更改
|
||
try:
|
||
db.session.commit()
|
||
print("\n示例数据创建完成！")
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
print(f"提交失败: {str(e)}")
|
||
|
||
def backup_localStorage_data():
|
||
"""备份localStorage数据"""
|
||
print("备份localStorage数据...")
|
||
|
||
# 这里应该从浏览器获取localStorage数据
|
||
# 实际实现需要JavaScript代码从浏览器获取数据
|
||
|
||
backup_script = """
|
||
// 在浏览器控制台运行此代码来备份localStorage数据
|
||
function backupOptimizationHistory() {
|
||
const history = JSON.parse(localStorage.getItem('optimization_history') || '[]');
|
||
const stats = JSON.parse(localStorage.getItem('usage_stats') || '{}');
|
||
|
||
const backup = {
|
||
timestamp: new Date().toISOString(),
|
||
history: history,
|
||
stats: stats,
|
||
count: history.length
|
||
};
|
||
|
||
// 下载备份文件
|
||
const blob = new Blob([JSON.stringify(backup, null, 2)], {type: 'application/json'});
|
||
const url = URL.createObjectURL(blob);
|
||
const a = document.createElement('a');
|
||
a.href = url;
|
||
a.download = `optimization_history_backup_${new Date().toISOString().split('T')[0]}.json`;
|
||
document.body.appendChild(a);
|
||
a.click();
|
||
document.body.removeChild(a);
|
||
URL.revokeObjectURL(url);
|
||
|
||
console.log(`备份完成！共 ${history.length} 条记录`);
|
||
return backup;
|
||
}
|
||
|
||
// 运行备份
|
||
backupOptimizationHistory();
|
||
"""
|
||
|
||
print("请在浏览器控制台运行以下代码来备份数据：")
|
||
print(backup_script)
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("=== 优化历史数据迁移工具 ===")
|
||
print("1. 备份localStorage数据")
|
||
print("2. 创建示例数据")
|
||
print("3. 迁移localStorage数据")
|
||
print("4. 退出")
|
||
|
||
while True:
|
||
choice = input("\n请选择操作 (1-4): ").strip()
|
||
|
||
if choice == '1':
|
||
backup_localStorage_data()
|
||
elif choice == '2':
|
||
create_sample_data()
|
||
elif choice == '3':
|
||
migrate_localStorage_data()
|
||
elif choice == '4':
|
||
print("退出程序")
|
||
break
|
||
else:
|
||
print("无效选择，请重新输入")
|
||
|
||
if __name__ == '__main__':
|
||
main()
|