221 lines
8.9 KiB
Python
221 lines
8.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
收藏表读写测试脚本
|
|
用于测试收藏功能的增删改查操作
|
|
"""
|
|
from src.flask_prompt_master import create_app, db
|
|
from src.flask_prompt_master.models.favorites import Favorite
|
|
from datetime import datetime
|
|
import json
|
|
|
|
def test_favorites_table():
|
|
"""测试收藏表的读写操作"""
|
|
app = create_app()
|
|
|
|
with app.app_context():
|
|
print("=" * 50)
|
|
print("收藏表读写测试")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
# 1. 测试插入数据
|
|
print("\n1. 测试插入收藏数据...")
|
|
test_favorites = [
|
|
{
|
|
'user_id': '192.168.1.100',
|
|
'template_id': 1,
|
|
'original_text': '如何写一个产品需求文档',
|
|
'generated_prompt': '作为产品经理，请按照以下结构编写产品需求文档：\n1. 产品概述\n2. 用户故事\n3. 功能需求\n4. 非功能需求\n5. 验收标准',
|
|
'system_prompt': '你是一个专业的产品经理',
|
|
'category': '产品管理',
|
|
'industry': '互联网',
|
|
'profession': '产品经理',
|
|
'tags': json.dumps(['PRD', '产品文档', '需求分析']),
|
|
'notes': '这是一个测试收藏，用于验证功能'
|
|
},
|
|
{
|
|
'user_id': '192.168.1.100',
|
|
'template_id': 2,
|
|
'original_text': '设计一个用户注册流程',
|
|
'generated_prompt': '请设计一个简洁高效的用户注册流程，包含：\n1. 注册入口\n2. 信息收集\n3. 验证流程\n4. 完成页面\n5. 异常处理',
|
|
'system_prompt': '你是一个UX设计师',
|
|
'category': '创意设计',
|
|
'industry': '互联网',
|
|
'profession': '设计师',
|
|
'tags': json.dumps(['UX', '注册流程', '用户体验']),
|
|
'notes': '注册流程设计参考'
|
|
},
|
|
{
|
|
'user_id': '192.168.1.101',
|
|
'template_id': 3,
|
|
'original_text': '感冒咳嗽吃什么药',
|
|
'generated_prompt': '作为呼吸内科医生，请为普通成年患者提供感冒咳嗽的药物选择建议...',
|
|
'system_prompt': '你是一个专业的医疗提示词专家',
|
|
'category': '专业服务',
|
|
'industry': '医疗健康',
|
|
'profession': '医生',
|
|
'tags': json.dumps(['医疗', '感冒', '药物建议']),
|
|
'notes': '医疗咨询参考'
|
|
}
|
|
]
|
|
|
|
for i, fav_data in enumerate(test_favorites, 1):
|
|
favorite = Favorite(**fav_data)
|
|
db.session.add(favorite)
|
|
print(f" 插入收藏 {i}: {fav_data['original_text'][:30]}...")
|
|
|
|
db.session.commit()
|
|
print(" ✅ 插入数据成功！")
|
|
|
|
# 2. 测试查询数据
|
|
print("\n2. 测试查询收藏数据...")
|
|
|
|
# 查询所有收藏
|
|
all_favorites = Favorite.query.all()
|
|
print(f" 总收藏数: {len(all_favorites)}")
|
|
|
|
# 按用户ID查询
|
|
user_favorites = Favorite.query.filter_by(user_id='192.168.1.100').all()
|
|
print(f" 用户 192.168.1.100 的收藏数: {len(user_favorites)}")
|
|
|
|
# 按分类查询
|
|
category_favorites = Favorite.query.filter_by(category='产品管理').all()
|
|
print(f" 产品管理分类的收藏数: {len(category_favorites)}")
|
|
|
|
# 按关键词搜索
|
|
search_favorites = Favorite.query.filter(
|
|
Favorite.original_text.contains('产品')
|
|
).all()
|
|
print(f" 包含'产品'关键词的收藏数: {len(search_favorites)}")
|
|
|
|
# 3. 测试更新数据
|
|
print("\n3. 测试更新收藏数据...")
|
|
if all_favorites:
|
|
first_favorite = all_favorites[0]
|
|
old_notes = first_favorite.notes
|
|
first_favorite.notes = "这是更新后的备注 - " + datetime.now().strftime('%H:%M:%S')
|
|
db.session.commit()
|
|
print(f" 更新收藏ID {first_favorite.id} 的备注")
|
|
print(f" 原备注: {old_notes}")
|
|
print(f" 新备注: {first_favorite.notes}")
|
|
|
|
# 4. 测试统计功能
|
|
print("\n4. 测试统计功能...")
|
|
|
|
# 总收藏数
|
|
total_count = Favorite.query.count()
|
|
print(f" 总收藏数: {total_count}")
|
|
|
|
# 按分类统计
|
|
from sqlalchemy import func
|
|
category_stats = db.session.query(
|
|
Favorite.category,
|
|
func.count(Favorite.id)
|
|
).group_by(Favorite.category).all()
|
|
|
|
print(" 分类统计:")
|
|
for category, count in category_stats:
|
|
print(f" {category}: {count} 个")
|
|
|
|
# 按用户统计
|
|
user_stats = db.session.query(
|
|
Favorite.user_id,
|
|
func.count(Favorite.id)
|
|
).group_by(Favorite.user_id).all()
|
|
|
|
print(" 用户统计:")
|
|
for user_id, count in user_stats:
|
|
print(f" {user_id}: {count} 个")
|
|
|
|
# 5. 测试删除数据
|
|
print("\n5. 测试删除收藏数据...")
|
|
if len(all_favorites) > 2:
|
|
# 删除最后一个收藏
|
|
last_favorite = all_favorites[-1]
|
|
favorite_id = last_favorite.id
|
|
original_text = last_favorite.original_text[:30]
|
|
|
|
db.session.delete(last_favorite)
|
|
db.session.commit()
|
|
|
|
print(f" 删除收藏ID {favorite_id}: {original_text}...")
|
|
|
|
# 验证删除
|
|
deleted_favorite = Favorite.query.get(favorite_id)
|
|
if deleted_favorite is None:
|
|
print(" ✅ 删除成功！")
|
|
else:
|
|
print(" ❌ 删除失败！")
|
|
|
|
# 6. 显示最终数据
|
|
print("\n6. 最终数据展示...")
|
|
final_favorites = Favorite.query.order_by(Favorite.created_time.desc()).limit(5).all()
|
|
|
|
print(" 最近5个收藏:")
|
|
for i, fav in enumerate(final_favorites, 1):
|
|
print(f" {i}. ID: {fav.id}")
|
|
print(f" 用户: {fav.user_id}")
|
|
print(f" 内容: {fav.original_text[:50]}...")
|
|
print(f" 分类: {fav.category}")
|
|
print(f" 时间: {fav.created_time}")
|
|
print(f" 备注: {fav.notes}")
|
|
print()
|
|
|
|
print("=" * 50)
|
|
print("✅ 收藏表读写测试完成！")
|
|
print("=" * 50)
|
|
|
|
except Exception as e:
|
|
print(f"❌ 测试过程中出现错误: {str(e)}")
|
|
db.session.rollback()
|
|
raise
|
|
|
|
def show_table_structure():
|
|
"""显示表结构"""
|
|
app = create_app()
|
|
|
|
with app.app_context():
|
|
print("=" * 50)
|
|
print("收藏表结构信息")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
# 获取表信息
|
|
from sqlalchemy import inspect
|
|
inspector = inspect(db.engine)
|
|
|
|
# 检查表是否存在
|
|
tables = inspector.get_table_names()
|
|
if 'favorites' in tables:
|
|
print("✅ favorites表存在")
|
|
|
|
# 获取列信息
|
|
columns = inspector.get_columns('favorites')
|
|
print(f"\n表结构 ({len(columns)} 列):")
|
|
for col in columns:
|
|
print(f" {col['name']}: {col['type']} ({'可空' if col['nullable'] else '非空'})")
|
|
if col.get('comment'):
|
|
print(f" 注释: {col['comment']}")
|
|
|
|
# 获取索引信息
|
|
indexes = inspector.get_indexes('favorites')
|
|
if indexes:
|
|
print(f"\n索引信息 ({len(indexes)} 个):")
|
|
for idx in indexes:
|
|
print(f" {idx['name']}: {idx['column_names']} ({'唯一' if idx['unique'] else '非唯一'})")
|
|
else:
|
|
print("\n索引信息: 无")
|
|
|
|
else:
|
|
print("❌ favorites表不存在")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 获取表结构时出现错误: {str(e)}")
|
|
|
|
if __name__ == '__main__':
|
|
# 显示表结构
|
|
show_table_structure()
|
|
|
|
# 运行测试
|
|
test_favorites_table()
|