刘金宝
241 lines
8.6 KiB
Python
241 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import inspect
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from app.db.session import engine
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TableSpec:
|
|
"""最终表结构规格：定义当前医疗问诊 Agent 完整功能所需的表、字段和关键索引。"""
|
|
|
|
columns: tuple[str, ...]
|
|
indexed_columns: tuple[str, ...] = ()
|
|
|
|
|
|
REQUIRED_SCHEMA: dict[str, TableSpec] = {
|
|
"user": TableSpec(
|
|
columns=(
|
|
"id",
|
|
"username",
|
|
"real_name",
|
|
"phone",
|
|
"role_type",
|
|
"department_id",
|
|
"institution_id",
|
|
"competency_profile",
|
|
"weak_dimensions",
|
|
"strong_dimensions",
|
|
"ai_preference",
|
|
"total_training_count",
|
|
"total_case_count",
|
|
"current_level",
|
|
"status",
|
|
),
|
|
indexed_columns=("id", "username", "phone", "department_id", "institution_id"),
|
|
),
|
|
"department": TableSpec(
|
|
columns=("id", "name", "category", "institution_id", "created_at", "updated_at"),
|
|
indexed_columns=("id", "institution_id"),
|
|
),
|
|
"case_base": TableSpec(
|
|
columns=(
|
|
"id",
|
|
"title",
|
|
"case_type",
|
|
"difficulty",
|
|
"chief_complaint",
|
|
"description",
|
|
"patient_age",
|
|
"patient_gender",
|
|
"publish_status",
|
|
"status",
|
|
"department_id",
|
|
),
|
|
indexed_columns=("id", "case_type", "difficulty", "publish_status", "status", "department_id"),
|
|
),
|
|
"traditional_case": TableSpec(
|
|
columns=("id", "case_id", "standard_diagnosis", "standard_treatment", "guideline_reference"),
|
|
indexed_columns=("id", "case_id"),
|
|
),
|
|
"teaching_case": TableSpec(
|
|
columns=("id", "case_id", "teaching_goal", "discussion_questions", "teacher_guide", "scoring_focus"),
|
|
indexed_columns=("id", "case_id"),
|
|
),
|
|
"scoring_rule": TableSpec(
|
|
columns=("id", "case_id", "dimension", "competency_dimension", "score_weight", "scoring_standard", "rubric_json"),
|
|
indexed_columns=("id", "case_id", "dimension", "competency_dimension"),
|
|
),
|
|
"case_exam_item": TableSpec(
|
|
columns=("id", "case_id", "item_code", "item_name", "item_type", "result_text", "is_key", "is_abnormal"),
|
|
indexed_columns=("id", "case_id", "item_code", "item_type"),
|
|
),
|
|
"training_session": TableSpec(
|
|
columns=(
|
|
"id",
|
|
"session_code",
|
|
"external_user_id",
|
|
"case_id",
|
|
"case_type",
|
|
"training_mode",
|
|
"score_type",
|
|
"status",
|
|
"memory_key",
|
|
),
|
|
indexed_columns=("id", "session_code", "external_user_id", "case_id", "training_mode", "status"),
|
|
),
|
|
"training_order": TableSpec(
|
|
columns=("id", "session_id", "external_user_id", "case_id", "exam_item_id", "item_code", "result_text", "ordered_at"),
|
|
indexed_columns=("id", "session_id", "external_user_id", "case_id"),
|
|
),
|
|
"training_submission": TableSpec(
|
|
columns=("id", "session_id", "external_user_id", "primary_diagnosis", "treatment_measures"),
|
|
indexed_columns=("id", "external_user_id"),
|
|
),
|
|
"training_record": TableSpec(
|
|
columns=(
|
|
"id",
|
|
"user_id",
|
|
"external_user_id",
|
|
"session_id",
|
|
"case_id",
|
|
"total_score",
|
|
"ai_feedback_structured",
|
|
"pdf_file_path",
|
|
),
|
|
indexed_columns=("id", "user_id", "external_user_id", "session_id", "case_id"),
|
|
),
|
|
"training_score_detail": TableSpec(
|
|
columns=(
|
|
"id",
|
|
"record_id",
|
|
"rule_id",
|
|
"dimension",
|
|
"score",
|
|
"deducted_reason",
|
|
"evidence_message_ids",
|
|
"ai_confidence",
|
|
"comment",
|
|
"created_at",
|
|
"updated_at",
|
|
),
|
|
indexed_columns=("id", "record_id", "rule_id", "dimension"),
|
|
),
|
|
"prompt_templates": TableSpec(
|
|
columns=("id", "template_code", "agent_type", "scene", "version_no", "model_type", "output_format", "file_path", "is_active"),
|
|
indexed_columns=("id", "template_code", "agent_type", "scene"),
|
|
),
|
|
"knowledge_sources": TableSpec(
|
|
columns=("id", "source_code", "source_name", "source_type", "authority_level", "is_active"),
|
|
indexed_columns=("id", "source_code", "source_type"),
|
|
),
|
|
"knowledge_documents": TableSpec(
|
|
columns=("id", "source_id", "department_id", "title", "task_type", "file_path", "is_active"),
|
|
indexed_columns=("id", "source_id", "department_id", "task_type"),
|
|
),
|
|
"knowledge_chunks": TableSpec(
|
|
columns=("id", "document_id", "department_id", "task_type", "chunk_text", "keywords", "weight", "is_active"),
|
|
indexed_columns=("id", "document_id", "department_id", "task_type"),
|
|
),
|
|
"audit_logs": TableSpec(
|
|
columns=("id", "user_id", "tenant_id", "session_id", "action", "resource_type", "request_id", "created_at"),
|
|
indexed_columns=("id", "user_id", "session_id", "action", "created_at"),
|
|
),
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
"""最终结构检查：只读校验当前 Agent 完整功能所需数据库结构。"""
|
|
try:
|
|
report = build_schema_report()
|
|
except SQLAlchemyError as exc:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"database_dialect": engine.dialect.name,
|
|
"summary": {
|
|
"can_run_demo": False,
|
|
"database_available": False,
|
|
"error": "database connection 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"]["can_run_demo"]:
|
|
raise SystemExit(1)
|
|
|
|
|
|
def build_schema_report() -> dict[str, Any]:
|
|
"""结构报告：检查必需表、字段和关键索引，不修改数据库。"""
|
|
inspector = inspect(engine)
|
|
existing_tables = set(inspector.get_table_names())
|
|
tables: dict[str, Any] = {}
|
|
missing_tables: list[str] = []
|
|
missing_columns: list[str] = []
|
|
missing_indexes: list[str] = []
|
|
|
|
for table_name, spec in REQUIRED_SCHEMA.items():
|
|
if table_name not in existing_tables:
|
|
missing_tables.append(table_name)
|
|
tables[table_name] = {"exists": False}
|
|
continue
|
|
|
|
actual_columns = {column["name"] for column in inspector.get_columns(table_name)}
|
|
actual_indexes = _indexed_columns(inspector, table_name)
|
|
table_missing_columns = [name for name in spec.columns if name not in actual_columns]
|
|
table_missing_indexes = [name for name in spec.indexed_columns if name not in actual_indexes]
|
|
|
|
missing_columns.extend(f"{table_name}.{column_name}" for column_name in table_missing_columns)
|
|
missing_indexes.extend(f"{table_name}.{column_name}" for column_name in table_missing_indexes)
|
|
|
|
tables[table_name] = {
|
|
"exists": True,
|
|
"missing_columns": table_missing_columns,
|
|
"missing_indexes": table_missing_indexes,
|
|
}
|
|
|
|
can_run_demo = not missing_tables and not missing_columns
|
|
return {
|
|
"database_dialect": engine.dialect.name,
|
|
"identity_rule": "Authorization -> Django /api/user/users/me/ -> data.id -> user isolation fields",
|
|
"tables": tables,
|
|
"summary": {
|
|
"checked_tables": len(REQUIRED_SCHEMA),
|
|
"missing_tables": missing_tables,
|
|
"missing_columns": missing_columns,
|
|
"missing_indexes": missing_indexes,
|
|
"index_warning": bool(missing_indexes),
|
|
"can_run_demo": can_run_demo,
|
|
},
|
|
}
|
|
|
|
|
|
def _indexed_columns(inspector, table_name: str) -> set[str]:
|
|
"""索引读取：汇总普通索引、唯一索引和主键覆盖的字段。"""
|
|
indexed: set[str] = set()
|
|
primary_key = inspector.get_pk_constraint(table_name) or {}
|
|
indexed.update(primary_key.get("constrained_columns") or [])
|
|
for index in inspector.get_indexes(table_name):
|
|
indexed.update(index.get("column_names") or [])
|
|
for unique in inspector.get_unique_constraints(table_name):
|
|
indexed.update(unique.get("column_names") or [])
|
|
return indexed
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|