179 lines
5.0 KiB
Python
179 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
数据库连接和迁移测试脚本
|
|
用于验证环境配置是否正确，以及迁移是否能够成功
|
|
"""
|
|
|
|
import sys
|
|
import psycopg2
|
|
from psycopg2 import sql, Error
|
|
from datetime import datetime
|
|
|
|
def print_header(text):
|
|
"""打印标题"""
|
|
print("\n" + "=" * 80)
|
|
print(f"【{text}】")
|
|
print("=" * 80)
|
|
|
|
def test_db_connection(host, port, database, user, password):
|
|
"""测试数据库连接"""
|
|
print_header("数据库连接测试")
|
|
|
|
try:
|
|
print(f"连接参数：")
|
|
print(f" 主机: {host}")
|
|
print(f" 端口: {port}")
|
|
print(f" 数据库: {database}")
|
|
print(f" 用户: {user}")
|
|
|
|
conn = psycopg2.connect(
|
|
host=host,
|
|
port=port,
|
|
database=database,
|
|
user=user,
|
|
password=password,
|
|
connect_timeout=10
|
|
)
|
|
|
|
print("\n✓ 数据库连接成功！")
|
|
cur = conn.cursor()
|
|
|
|
# 获取服务器版本
|
|
cur.execute("SELECT version();")
|
|
version = cur.fetchone()[0]
|
|
print(f"✓ PostgreSQL 版本: {version.split(',')[0]}")
|
|
|
|
conn.close()
|
|
return True
|
|
except Error as e:
|
|
print(f"\n❌ 连接失败: {e}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"\n❌ 未知错误: {e}")
|
|
return False
|
|
|
|
|
|
def test_table_structure(host, port, database, user, password):
|
|
"""测试表结构和字段"""
|
|
print_header("表结构检查")
|
|
|
|
try:
|
|
conn = psycopg2.connect(
|
|
host=host,
|
|
port=port,
|
|
database=database,
|
|
user=user,
|
|
password=password
|
|
)
|
|
|
|
cur = conn.cursor()
|
|
|
|
# 检查表是否存在
|
|
cur.execute("""
|
|
SELECT COUNT(*) FROM information_schema.tables
|
|
WHERE table_name = 'journal_entries'
|
|
""")
|
|
|
|
if cur.fetchone()[0] == 0:
|
|
print("❌ 表 journal_entries 不存在！")
|
|
print("请检查数据库架构是否已初始化。")
|
|
conn.close()
|
|
return False
|
|
|
|
print("✓ 表 journal_entries 存在")
|
|
|
|
# 检查新字段
|
|
required_fields = [
|
|
('is_latest', 'BOOLEAN'),
|
|
('revision_count', 'INTEGER'),
|
|
('original_entry_id', 'INTEGER')
|
|
]
|
|
|
|
for field_name, expected_type in required_fields:
|
|
cur.execute("""
|
|
SELECT data_type FROM information_schema.columns
|
|
WHERE table_name = 'journal_entries' AND column_name = %s
|
|
""", (field_name,))
|
|
|
|
row = cur.fetchone()
|
|
if row:
|
|
actual_type = row[0]
|
|
print(f"✓ 字段 {field_name}: {actual_type}")
|
|
else:
|
|
print(f"⚠️ 字段 {field_name} 不存在（将由迁移脚本创建）")
|
|
|
|
conn.close()
|
|
return True
|
|
except Error as e:
|
|
print(f"❌ 检查失败: {e}")
|
|
return False
|
|
|
|
|
|
def test_migration(host, port, database, user, password):
|
|
"""运行迁移测试"""
|
|
print_header("迁移测试（模拟运行）")
|
|
|
|
try:
|
|
from app.db_auto_migration import run_auto_migration
|
|
|
|
print("执行自动迁移...")
|
|
run_auto_migration()
|
|
print("\n✓ 迁移测试完成！")
|
|
return True
|
|
except ImportError:
|
|
print("⚠️ 无法导入迁移模块（正常，应用启动时会运行）")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 迁移失败: {e}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("\n" + "=" * 80)
|
|
print("【日记条目版本追踪系统 - 环境验证】")
|
|
print("=" * 80)
|
|
print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
|
|
|
|
# 数据库连接参数
|
|
host = '192.168.0.61'
|
|
port = 55432
|
|
database = 'njts_acct'
|
|
user = 'njts_app'
|
|
password = 'njts_app2025'
|
|
|
|
# 运行测试
|
|
results = []
|
|
|
|
# 1. 数据库连接
|
|
results.append(("数据库连接", test_db_connection(host, port, database, user, password)))
|
|
|
|
if results[-1][1]: # 如果连接成功
|
|
# 2. 表结构
|
|
results.append(("表结构检查", test_table_structure(host, port, database, user, password)))
|
|
|
|
# 3. 迁移测试
|
|
results.append(("迁移测试", test_migration(host, port, database, user, password)))
|
|
|
|
# 打印摘要
|
|
print_header("检查摘要")
|
|
|
|
for test_name, result in results:
|
|
status = "✓ 通过" if result else "❌ 失败"
|
|
print(f"{status} - {test_name}")
|
|
|
|
# 最终结果
|
|
print("\n" + "=" * 80)
|
|
if all(r for _, r in results):
|
|
print("✓ 所有检查通过！可以启动应用。")
|
|
print("=" * 80 + "\n")
|
|
return 0
|
|
else:
|
|
print("❌ 某些检查失败。请参考上面的错误信息。")
|
|
print("=" * 80 + "\n")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|