170 lines
5.4 KiB
Python
170 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
测试腾讯云数据库连接和优化历史表
|
||
"""
|
||
|
||
import pymysql
|
||
import sys
|
||
import os
|
||
|
||
# 添加项目路径
|
||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
def test_tencent_db_connection():
|
||
"""测试腾讯云数据库连接"""
|
||
print("🔍 测试腾讯云数据库连接...")
|
||
|
||
try:
|
||
# 腾讯云数据库连接参数
|
||
conn = pymysql.connect(
|
||
host='gz-cynosdbmysql-grp-d26pzce5.sql.tencentcdb.com',
|
||
port=24936,
|
||
user='root',
|
||
password='!Rjb12191',
|
||
database='pro_db',
|
||
charset='utf8mb4'
|
||
)
|
||
|
||
cursor = conn.cursor()
|
||
print("✅ 腾讯云数据库连接成功")
|
||
|
||
# 检查优化历史表是否存在
|
||
cursor.execute("SHOW TABLES LIKE 'optimization_history'")
|
||
result = cursor.fetchone()
|
||
|
||
if result:
|
||
print("✅ optimization_history 表存在")
|
||
|
||
# 检查表结构
|
||
cursor.execute("DESCRIBE optimization_history")
|
||
columns = cursor.fetchall()
|
||
print("📊 表结构:")
|
||
for col in columns:
|
||
print(f" - {col[0]}: {col[1]}")
|
||
|
||
# 检查记录数量
|
||
cursor.execute("SELECT COUNT(*) FROM optimization_history")
|
||
count = cursor.fetchone()[0]
|
||
print(f"📈 当前记录数: {count}")
|
||
|
||
# 检查最近几条记录
|
||
cursor.execute("SELECT id, original_text, optimized_text, created_at FROM optimization_history ORDER BY created_at DESC LIMIT 3")
|
||
records = cursor.fetchall()
|
||
print("📋 最近3条记录:")
|
||
for record in records:
|
||
print(f" ID: {record[0]}, 创建时间: {record[3]}")
|
||
print(f" 原始: {record[1][:50]}...")
|
||
print(f" 优化: {record[2][:50]}...")
|
||
print()
|
||
|
||
else:
|
||
print("❌ optimization_history 表不存在")
|
||
return False
|
||
|
||
cursor.close()
|
||
conn.close()
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 数据库连接失败: {e}")
|
||
return False
|
||
|
||
def test_optimization_history_api():
|
||
"""测试优化历史API"""
|
||
print("\n🔍 测试优化历史API...")
|
||
|
||
try:
|
||
import requests
|
||
|
||
# 测试获取历史记录
|
||
response = requests.get("http://localhost:5002/api/optimization-history", timeout=10)
|
||
print(f"📊 GET API 状态码: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print(f"✅ API响应成功，包含 {len(data.get('data', []))} 条记录")
|
||
return True
|
||
else:
|
||
print(f"❌ API响应失败: {response.text}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ API测试失败: {e}")
|
||
return False
|
||
|
||
def test_add_optimization_record():
|
||
"""测试添加优化历史记录"""
|
||
print("\n🔍 测试添加优化历史记录...")
|
||
|
||
try:
|
||
import requests
|
||
|
||
test_data = {
|
||
"original_text": "测试原始文本",
|
||
"optimized_text": "测试优化后的文本，这是一个用于验证腾讯云数据库连接和优化历史功能的测试记录。",
|
||
"optimization_type": "提示词优化",
|
||
"industry": "测试行业",
|
||
"profession": "测试职业",
|
||
"tags": ["测试", "数据库验证"]
|
||
}
|
||
|
||
response = requests.post(
|
||
"http://localhost:5002/api/optimization-history",
|
||
headers={'Content-Type': 'application/json'},
|
||
json=test_data,
|
||
timeout=10
|
||
)
|
||
|
||
print(f"📊 POST API 状态码: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get('success'):
|
||
print("✅ 添加优化历史记录成功")
|
||
print(f"📋 记录ID: {result.get('data', {}).get('id', 'N/A')}")
|
||
return True
|
||
else:
|
||
print(f"❌ 添加失败: {result.get('message', '未知错误')}")
|
||
return False
|
||
else:
|
||
print(f"❌ 请求失败: {response.text}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 添加测试失败: {e}")
|
||
return False
|
||
|
||
def main():
|
||
"""主测试函数"""
|
||
print("🚀 开始测试腾讯云数据库连接和优化历史功能...")
|
||
print("=" * 60)
|
||
|
||
# 测试数据库连接
|
||
db_ok = test_tencent_db_connection()
|
||
|
||
if not db_ok:
|
||
print("\n❌ 数据库连接失败，请检查腾讯云数据库配置")
|
||
return
|
||
|
||
# 测试API接口
|
||
api_ok = test_optimization_history_api()
|
||
|
||
# 测试添加记录
|
||
add_ok = test_add_optimization_record()
|
||
|
||
print("\n" + "=" * 60)
|
||
print("📊 测试结果汇总:")
|
||
print(f" 数据库连接: {'✅ 正常' if db_ok else '❌ 异常'}")
|
||
print(f" API接口: {'✅ 正常' if api_ok else '❌ 异常'}")
|
||
print(f" 添加记录: {'✅ 正常' if add_ok else '❌ 异常'}")
|
||
|
||
if db_ok and api_ok and add_ok:
|
||
print("\n🎉 腾讯云数据库和优化历史功能测试通过！")
|
||
else:
|
||
print("\n⚠️ 部分功能异常，需要进一步检查")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|