Files
fastapi/scripts/check_final_demo_readiness.py
2026-06-04 10:55:23 +08:00

131 lines
4.9 KiB
Python

from __future__ import annotations
import json
import sys
from pathlib import Path
from typing import Any
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from app.core.config import settings
from app.db.session import SessionLocal
from scripts.check_final_schema import build_schema_report
COUNT_SQL = {
"departments": "SELECT COUNT(*) FROM department",
"active_cases": "SELECT COUNT(*) FROM case_base WHERE status = 1 AND publish_status = 1",
"traditional_cases": "SELECT COUNT(*) FROM traditional_case",
"teaching_cases": "SELECT COUNT(*) FROM teaching_case",
"exam_items": "SELECT COUNT(*) FROM case_exam_item",
"scoring_rules": "SELECT COUNT(*) FROM scoring_rule",
"active_prompt_templates": "SELECT COUNT(*) FROM prompt_templates WHERE is_active = 1",
"knowledge_sources": "SELECT COUNT(*) FROM knowledge_sources",
"knowledge_documents": "SELECT COUNT(*) FROM knowledge_documents",
"knowledge_chunks": "SELECT COUNT(*) FROM knowledge_chunks",
}
def main() -> None:
"""最终就绪检查:校验当前 Demo 功能运行所需的结构、基础数据、提示词和配置。"""
try:
report = build_readiness_report()
except SQLAlchemyError as exc:
print(
json.dumps(
{
"ready": False,
"error": "database operation failed",
"detail": str(exc).splitlines()[0],
},
ensure_ascii=False,
indent=2,
)
)
raise SystemExit(2) from exc
print(json.dumps(report, ensure_ascii=False, indent=2))
if not report["summary"]["ready"]:
raise SystemExit(1)
def build_readiness_report() -> dict[str, Any]:
"""就绪报告:聚合数据库结构、基础业务数据、提示词文件和关键环境配置。"""
schema_report = build_schema_report()
counts = _collect_counts()
prompt_files = _prompt_files()
config = _public_config()
critical_checks = {
"schema_complete": schema_report["summary"]["can_run_demo"],
"has_active_cases": counts["active_cases"] > 0,
"has_department": counts["departments"] > 0,
"has_case_detail": counts["traditional_cases"] + counts["teaching_cases"] > 0,
"has_exam_items": counts["exam_items"] > 0,
"has_scoring_rules": counts["scoring_rules"] > 0,
"has_prompt_templates": counts["active_prompt_templates"] > 0,
"has_prompt_files": len(prompt_files) > 0,
"auth_user_center_configured": bool(settings.auth_user_me_url),
}
warnings = []
if not settings.llm_api_key:
warnings.append("LLM_API_KEY is not configured; real LLM calls will not work unless mock is enabled.")
if counts["knowledge_chunks"] == 0:
warnings.append("knowledge_chunks is empty; scoring can run but guideline reference retrieval has no data.")
missing_indexes = schema_report["summary"].get("missing_indexes") or []
if missing_indexes:
warnings.append("Some recommended indexes are missing; functions can run, but high-concurrency query performance may be affected.")
return {
"summary": {
"ready": all(critical_checks.values()),
"critical_checks": critical_checks,
"warnings": warnings,
},
"database": {
"dialect": schema_report["database_dialect"],
"counts": counts,
"schema_summary": schema_report["summary"],
},
"prompts": {
"markdown_count": len(prompt_files),
"files": prompt_files,
},
"config": config,
}
def _collect_counts() -> dict[str, int]:
"""数据计数:统计 Demo 闭环运行依赖的基础数据。"""
with SessionLocal() as db:
return {name: int(db.execute(text(sql)).scalar() or 0) for name, sql in COUNT_SQL.items()}
def _prompt_files() -> list[str]:
"""提示词检查:读取 prompts 目录下所有 Markdown 模板。"""
prompt_root = Path(__file__).resolve().parents[1] / "app" / "prompts"
return sorted(str(path.relative_to(prompt_root)).replace("\\", "/") for path in prompt_root.rglob("*.md"))
def _public_config() -> dict[str, Any]:
"""配置摘要:只输出可公开的配置状态,不暴露密钥。"""
return {
"auth_validate_enabled": settings.auth_validate_enabled,
"auth_user_me_url_configured": bool(settings.auth_user_me_url),
"llm_base_url_configured": bool(settings.llm_base_url),
"llm_model": settings.llm_model,
"llm_fast_model": settings.llm_fast_model,
"llm_reason_model": settings.llm_reason_model,
"llm_api_key_configured": bool(settings.llm_api_key),
"llm_mock_enabled": settings.llm_mock_enabled,
"runtime_memory_backend": settings.runtime_memory_backend,
"redis_url_configured": bool(settings.redis_url),
}
if __name__ == "__main__":
main()