ai-contract-platform/backend/app/api/system.py

from datetime import datetime, timezone
from typing import Any, Dict, List

from fastapi import APIRouter, Depends
from sqlalchemy import select, text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.response import success_response
from app.db import get_db_session
from app.core.security import verify_password
from app.models.user import User
from app.services.auth_service import DEFAULT_ADMIN_PASSWORD, DEFAULT_ADMIN_USERNAME

router = APIRouter(prefix="/system", tags=["system"])
ROUTE_PERMISSION_SCOPE = "system"
ROUTE_PERMISSION_ACTIONS = {
    "readiness": "system:readiness:get",
}


def _build_check(key: str, label: str, status: str, detail: str, required: bool = True) -> Dict[str, Any]:
    return {
        "key": key,
        "label": label,
        "status": status,
        "detail": detail,
        "required": required,
    }


def _check_database(db: Session) -> Dict[str, Any]:
    try:
        db.execute(text("SELECT 1")).scalar()
        return _build_check("database", "数据库连接", "ok", "数据库连接可用")
    except Exception as exc:
        return _build_check("database", "数据库连接", "failed", f"数据库连接失败：{exc.__class__.__name__}")


def _check_runtime_config() -> Dict[str, Any]:
    if settings.secret_key.strip() == "change-me":
        return _build_check(
            "runtime_config",
            "运行配置",
            "warning",
            "当前 SECRET_KEY 仍为默认值，试点前建议替换为环境专用密钥",
            required=False,
        )

    return _build_check("runtime_config", "运行配置", "ok", "关键运行配置已设置", required=False)


def _check_object_storage_config() -> Dict[str, Any]:
    if not settings.minio_endpoint.strip() or not settings.minio_bucket.strip():
        return _build_check("object_storage_config", "对象存储配置", "failed", "对象存储 endpoint 或 bucket 未配置")

    return _build_check(
        "object_storage_config",
        "对象存储配置",
        "ok",
        f"对象存储配置已就绪：{settings.minio_endpoint} / {settings.minio_bucket}",
    )


def _check_default_admin_password(db: Session) -> Dict[str, Any]:
    try:
        admin = db.execute(select(User).where(User.username == DEFAULT_ADMIN_USERNAME)).scalar_one_or_none()
    except Exception as exc:
        return _build_check(
            "default_admin_password",
            "默认管理员密码",
            "warning",
            f"无法确认默认管理员密码状态：{exc.__class__.__name__}",
            required=False,
        )

    if admin is None:
        return _build_check(
            "default_admin_password",
            "默认管理员密码",
            "warning",
            "默认管理员账号尚未初始化",
            required=False,
        )

    if verify_password(DEFAULT_ADMIN_PASSWORD, admin.password_hash):
        return _build_check(
            "default_admin_password",
            "默认管理员密码",
            "warning",
            "默认管理员仍在使用初始验收密码，试点前应完成轮换",
            required=False,
        )

    return _build_check("default_admin_password", "默认管理员密码", "ok", "默认管理员密码已轮换", required=False)


def _build_pilot_assets(password_rotated: bool) -> Dict[str, Any]:
    return {
        "default_admin": {
            "username": DEFAULT_ADMIN_USERNAME,
            "role": "admin",
            "seed_endpoint": f"{settings.api_v1_prefix}/auth/seed-admin",
            "password_rotated": password_rotated,
            "rotation_required": not password_rotated,
        },
        "demo_flow": [
            "登录默认管理员账号",
            "发起一份高风险合同审查",
            "从审查详情进入 Chat 追问修改建议",
            "把审查上下文带入合同生成",
            "上传知识库文档并验证重试语义",
            "查看审计日志确认关键动作留痕",
        ],
        "acceptance_gates": [
            "前端 lint/build 通过",
            "后端 pytest 通过",
            "阶段 1/2/3 E2E 均可复跑",
            "生产入口 health/readiness 通过",
        ],
    }


@router.get("/readiness")
def get_system_readiness(db: Session = Depends(get_db_session)):
    default_admin_password_check = _check_default_admin_password(db)
    checks: List[Dict[str, Any]] = [
        _build_check("api", "API 服务", "ok", "FastAPI 服务已响应"),
        _check_database(db),
        _check_runtime_config(),
        _check_object_storage_config(),
        _build_check("auth_seed", "默认管理员初始化", "ok", "默认管理员初始化入口可用", required=False),
        default_admin_password_check,
    ]
    failed_required = [item for item in checks if item["required"] and item["status"] == "failed"]
    warning_count = len([item for item in checks if item["status"] == "warning"])
    status = "ready" if not failed_required else "degraded"

    return success_response(
        {
            "status": status,
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "app": {
                "name": settings.app_name,
                "environment": settings.app_env,
                "api_prefix": settings.api_v1_prefix,
            },
            "checks": checks,
            "summary": {
                "total": len(checks),
                "failed_required": len(failed_required),
                "warnings": warning_count,
            },
            "pilot_assets": _build_pilot_assets(default_admin_password_check["status"] == "ok"),
        }
    )