cosmo_backend/scripts/check_config.py

#!/usr/bin/env python3
"""
配置验证脚本 - 检查 PostgreSQL 和 Redis 配置是否正确

Usage:
    python scripts/check_config.py
"""
import asyncio
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

from app.config import settings
import asyncpg
import redis.asyncio as redis
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


async def check_postgresql():
    """检查 PostgreSQL 连接"""
    print("\n" + "=" * 60)
    print("检查 PostgreSQL 配置")
    print("=" * 60)

    try:
        # 连接参数
        print(f"主机: {settings.database_host}")
        print(f"端口: {settings.database_port}")
        print(f"数据库: {settings.database_name}")
        print(f"用户: {settings.database_user}")
        print(f"连接池大小: {settings.database_pool_size}")

        # 尝试连接
        conn = await asyncpg.connect(
            host=settings.database_host,
            port=settings.database_port,
            user=settings.database_user,
            password=settings.database_password,
            database=settings.database_name,
        )

        # 查询版本
        version = await conn.fetchval("SELECT version()")
        print(f"\n✓ PostgreSQL 连接成功")
        print(f"版本: {version.split(',')[0]}")

        # 查询数据库大小
        db_size = await conn.fetchval(
            "SELECT pg_size_pretty(pg_database_size($1))",
            settings.database_name
        )
        print(f"数据库大小: {db_size}")

        # 查询表数量
        table_count = await conn.fetchval("""
            SELECT COUNT(*)
            FROM information_schema.tables
            WHERE table_schema = 'public'
        """)
        print(f"数据表数量: {table_count}")

        await conn.close()
        return True

    except Exception as e:
        print(f"\n✗ PostgreSQL 连接失败: {e}")
        print("\n请检查:")
        print("  1. PostgreSQL 是否正在运行")
        print("  2. 数据库是否已创建 (运行: python scripts/create_db.py)")
        print("  3. .env 文件中的账号密码是否正确")
        return False


async def check_redis():
    """检查 Redis 连接"""
    print("\n" + "=" * 60)
    print("检查 Redis 配置")
    print("=" * 60)

    try:
        # 连接参数
        print(f"主机: {settings.redis_host}")
        print(f"端口: {settings.redis_port}")
        print(f"数据库: {settings.redis_db}")
        print(f"密码: {'(无)' if not settings.redis_password else '******'}")
        print(f"最大连接数: {settings.redis_max_connections}")

        # 尝试连接
        client = redis.from_url(
            settings.redis_url,
            encoding="utf-8",
            decode_responses=True,
        )

        # 测试连接
        await client.ping()
        print(f"\n✓ Redis 连接成功")

        # 获取 Redis 信息
        info = await client.info()
        print(f"版本: {info.get('redis_version')}")
        print(f"使用内存: {info.get('used_memory_human')}")
        print(f"已连接客户端: {info.get('connected_clients')}")
        print(f"运行天数: {info.get('uptime_in_days')} 天")

        await client.close()
        return True

    except Exception as e:
        print(f"\n⚠ Redis 连接失败: {e}")
        print("\n说明:")
        print("  Redis 是可选的缓存服务")
        print("  如果 Redis 不可用，应用会自动降级为内存缓存")
        print("  不影响核心功能，但会失去跨进程缓存能力")
        print("\n如需启用 Redis:")
        print("  - macOS: brew install redis && brew services start redis")
        print("  - Ubuntu: sudo apt install redis && sudo systemctl start redis")
        return False


def check_env_file():
    """检查 .env 文件"""
    print("\n" + "=" * 60)
    print("检查配置文件")
    print("=" * 60)

    env_path = Path(__file__).parent.parent / ".env"

    if env_path.exists():
        print(f"✓ .env 文件存在: {env_path}")
        print(f"文件大小: {env_path.stat().st_size} bytes")
        return True
    else:
        print(f"✗ .env 文件不存在")
        print(f"请从 .env.example 创建: cp .env.example .env")
        return False


def check_upload_dir():
    """检查上传目录"""
    print("\n" + "=" * 60)
    print("检查上传目录")
    print("=" * 60)

    upload_dir = Path(__file__).parent.parent / settings.upload_dir

    if upload_dir.exists():
        print(f"✓ 上传目录存在: {upload_dir}")
        return True
    else:
        print(f"⚠ 上传目录不存在: {upload_dir}")
        print(f"自动创建...")
        upload_dir.mkdir(parents=True, exist_ok=True)
        print(f"✓ 上传目录创建成功")
        return True


async def main():
    """主函数"""
    print("\n" + "=" * 60)
    print("  Cosmo 配置验证工具")
    print("=" * 60)

    results = []

    # 1. 检查配置文件
    results.append(("配置文件", check_env_file()))

    # 2. 检查上传目录
    results.append(("上传目录", check_upload_dir()))

    # 3. 检查 PostgreSQL
    results.append(("PostgreSQL", await check_postgresql()))

    # 4. 检查 Redis
    results.append(("Redis", await check_redis()))

    # 总结
    print("\n" + "=" * 60)
    print("  配置检查总结")
    print("=" * 60)

    for name, status in results:
        status_str = "✓" if status else "✗"
        print(f"{status_str} {name}")

    # 判断是否所有必需服务都正常
    required_services = [results[0], results[1], results[2]]  # 配置文件、上传目录、PostgreSQL
    all_required_ok = all(status for _, status in required_services)

    if all_required_ok:
        print("\n" + "=" * 60)
        print("  ✓ 所有必需服务配置正确！")
        print("=" * 60)
        print("\n可以启动服务：")
        print("  python -m uvicorn app.main:app --reload")
        print("\n或者：")
        print("  python app/main.py")
        return 0
    else:
        print("\n" + "=" * 60)
        print("  ✗ 部分必需服务配置有问题")
        print("=" * 60)
        print("\n请先解决上述问题，然后重新运行此脚本")
        return 1


if __name__ == "__main__":
    exit_code = asyncio.run(main())
    sys.exit(exit_code)