""" 12D AI Assessment Engine V2.2 - Gradio SDK entry point for Hugging Face Spaces. Uses demo.launch() to trigger ZeroGPU startup detection. Multi-portal UI integrated: Parent (with Interactive Scale/Assessment), Doctor, and School portals. """ import os import sys import time import logging # ------------------------------------------------------------------- # 环境路径与目录初始化 # ------------------------------------------------------------------- _backend_dir = os.path.join(os.path.dirname(__file__), "backend") sys.path.insert(0, _backend_dir) _data_dir = os.path.join(os.path.dirname(__file__), "data") os.makedirs(_data_dir, exist_ok=True) import gradio as gr import spaces # ZeroGPU placeholder @spaces.GPU(duration=5) def _zerogpu_placeholder(): pass # ------------------------------------------------------------------- # 模拟 12D 评估量表题库数据 (Assessment Scale Item Bank) # ------------------------------------------------------------------- SCALE_ITEMS = [ { "id": "q1", "dimension": "专注力 (Attention)", "question": "1. 孩子在完成日常作业或游戏任务时，容易被无关刺激干扰吗？", "options": ["从不", "偶有", "经常", "总是"], }, { "id": "q2", "dimension": "工作记忆 (Working Memory)", "question": "2. 给孩子连续下达3个简单口头指令（如：关灯、拿书、坐下），能否准确执行？", "options": ["完全不能", "能完成1项", "能完成2项", "全部准确完成"], }, { "id": "q3", "dimension": "抑制控制 (Inhibitory Control)", "question": "3. 在需要轮流等待或遵守规则的场景下，孩子能否控制冲动？", "options": ["极难控制", "需要提醒", "基本可以", "完全没问题"], }, { "id": "q4", "dimension": "空间想象 (Spatial Cognition)", "question": "4. 孩子拼图或进行积木搭建时，对空间位置变换的感知表现如何？", "options": ["较困难", "一般", "良好", "极优秀"], }, ] # ------------------------------------------------------------------- # 构建 Gradio 多端交互界面 # ------------------------------------------------------------------- with gr.Blocks(title="12D AI Assessment Engine V2.2", theme=gr.themes.Soft()) as demo: gr.Markdown( """ # 💡 12D AI Assessment Engine V2.2 **LearningOS V2.0 Enterprise Edition** — 智能化儿童认知能力多端评估与诊断平台 --- """ ) with gr.Tabs(): # ==================== 1. 👶 家长端 Portal (含评估量表) ==================== with gr.Tab("👶 家长端 (Parent Portal)"): gr.Markdown("### 👶 家长工作台 — 儿童档案与 12D 认知能力评估量表") with gr.Row(): # 左侧：档案管理 with gr.Column(scale=1): p_name = gr.Textbox(label="孩子姓名/昵称", value="小敏") p_age = gr.Number(label="年龄", value=7, precision=0) p_gender = gr.Dropdown( label="性别", choices=["男", "女", "其他"], value="男" ) p_create_btn = gr.Button("➕ 建立儿童档案", variant="primary") p_create_output = gr.JSON(label="档案创建结果") # 右侧：量表测评答题区 with gr.Column(scale=2): p_child_id = gr.Textbox(label="儿童 ID", value="child_demo_888") p_start_assess_btn = gr.Button( "🚀 加载/发起 12D 评估量表", variant="secondary" ) # 评估量表交互区域 with gr.Group(visible=False) as scale_group: gr.Markdown("#### 📋 12D AI 认知能力快速筛查量表") q1_radio = gr.Radio( choices=SCALE_ITEMS[0]["options"], label=f"【{SCALE_ITEMS[0]['dimension']}】 {SCALE_ITEMS[0]['question']}", ) q2_radio = gr.Radio( choices=SCALE_ITEMS[1]["options"], label=f"【{SCALE_ITEMS[1]['dimension']}】 {SCALE_ITEMS[1]['question']}", ) q3_radio = gr.Radio( choices=SCALE_ITEMS[2]["options"], label=f"【{SCALE_ITEMS[2]['dimension']}】 {SCALE_ITEMS[2]['question']}", ) q4_radio = gr.Radio( choices=SCALE_ITEMS[3]["options"], label=f"【{SCALE_ITEMS[3]['dimension']}】 {SCALE_ITEMS[3]['question']}", ) p_submit_answers_btn = gr.Button( "📤 提交量表答案并计算分析", variant="primary" ) p_assess_output = gr.JSON(label="评估 Session & 答题结果") p_report_output = gr.JSON(label="📊 智能生成 12D 认知测评报告") # 逻辑响应函数 def _parent_create_child(name, age, gender): return { "status": "success", "message": f"成功为 {name} ({age}岁, {gender}) 建立档案", "child_id": f"child_{name}_888", } def _parent_start_assess(child_id): if not child_id: return gr.update(visible=False), {"error": "请输入有效的儿童 ID"} session_data = { "status": "active", "session_id": "sess_12d_999", "child_id": child_id, "total_items": len(SCALE_ITEMS), "message": "量表加载成功，请在下方完成评估量表勾选。", } return gr.update(visible=True), session_data def _parent_submit_scale(child_id, a1, a2, a3, a4): score_map = { "从不": 100, "偶有": 80, "经常": 60, "总是": 40, "完全不能": 40, "能完成1项": 60, "能完成2项": 80, "全部准确完成": 100, "极难控制": 40, "需要提醒": 60, "基本可以": 80, "完全没问题": 100, "较困难": 40, "一般": 60, "良好": 80, "极优秀": 100, } s1 = score_map.get(a1, 70) s2 = score_map.get(a2, 70) s3 = score_map.get(a3, 70) s4 = score_map.get(a4, 70) avg_score = round((s1 + s2 + s3 + s4) / 4, 1) report_data = { "child_id": child_id or "child_demo_888", "overall_score": avg_score, "dimension_scores": { "专注力": s1, "工作记忆": s2, "抑制控制": s3, "空间想象": s4, }, "strengths": [ k for k, v in { "专注力": s1, "工作记忆": s2, "抑制控制": s3, "空间想象": s4, }.items() if v >= 80 ], "weaknesses": [ k for k, v in { "专注力": s1, "工作记忆": s2, "抑制控制": s3, "空间想象": s4, }.items() if v < 70 ], "ai_suggestions": f"综合得分 {avg_score} 分。" + ( "建议多安排专注力训练游戏及户外有氧运动。" if avg_score < 80 else "各项能力表现优异，保持当前引导策略！" ), } return { "status": "completed", "submitted_answers": {"q1": a1, "q2": a2, "q3": a3, "q4": a4}, }, report_data p_create_btn.click( _parent_create_child, inputs=[p_name, p_age, p_gender], outputs=p_create_output, ) p_start_assess_btn.click( _parent_start_assess, inputs=[p_child_id], outputs=[scale_group, p_assess_output], ) p_submit_answers_btn.click( _parent_submit_scale, inputs=[p_child_id, q1_radio, q2_radio, q3_radio, q4_radio], outputs=[p_assess_output, p_report_output], ) # ==================== 2. 👨‍⚕️ 医生端 Portal ==================== with gr.Tab("👨‍⚕️ 医生端 (Doctor Portal)"): gr.Markdown("### 👨‍⚕️ 医生临床诊断工作台 — 深入认知量表与危险因子诊断") with gr.Row(): with gr.Column(): d_child_id = gr.Textbox(label="患者/儿童 ID", value="child_demo_888") d_adhd_score = gr.Slider(0, 10, value=2, label="ADHD 倾向评估分 (0-10)") d_asd_score = gr.Slider(0, 10, value=1, label="ASD 倾向评估分 (0-10)") d_notes = gr.TextArea( label="临床诊断意见", placeholder="录入医生专业评语或干预方案..." ) d_submit_btn = gr.Button("💾 提交临床诊断报告", variant="primary") d_submit_output = gr.JSON(label="诊断录入结果") with gr.Column(): d_twin_btn = gr.Button( "🧬 提取临床级数字孪生 (Diagnostic Twin)", variant="secondary" ) d_twin_output = gr.JSON(label="深度多维认知矩阵") def _doctor_submit_diag(child_id, adhd, asd, notes): return { "status": "success", "patient_id": child_id, "adhd_risk_index": adhd / 10.0, "asd_risk_index": asd / 10.0, "clinical_notes": notes or "未填写评语", "message": "临床数据已保存至云端电子病历系统", } def _doctor_get_twin(child_id): return { "child_id": child_id, "raw_cognitive_matrix": { "working_memory": 92, "inhibitory_control": 78, "cognitive_flexibility": 85, "processing_speed": 88, }, "neurological_risk": "Low Risk", } d_submit_btn.click( _doctor_submit_diag, inputs=[d_child_id, d_adhd_score, d_asd_score, d_notes], outputs=d_submit_output, ) d_twin_btn.click( _doctor_get_twin, inputs=[d_child_id], outputs=d_twin_output ) # ==================== 3. 🏫 学校端 Portal ==================== with gr.Tab("🏫 学校端 (School Portal)"): gr.Markdown("### 🏫 学校管理工作台 — 班级认知分布与群体预警") with gr.Row(): with gr.Column(): s_school_name = gr.Textbox(label="学校名称", value="第一实验小学") s_class_name = gr.Textbox(label="班级", value="三年级 (2) 班") s_batch_btn = gr.Button("📦 一键导入班级学生名单", variant="primary") s_batch_output = gr.JSON(label="导入结果") with gr.Column(): s_class_id = gr.Textbox(label="班级 ID", value="class_302") s_analytics_btn = gr.Button( "📈 生成班级群体认知分析", variant="secondary" ) s_analytics_output = gr.JSON(label="班级报告与预警名单") def _school_batch_import(school, cls): return { "status": "success", "school": school, "class": cls, "imported_students": 45, } def _school_get_analytics(cls_id): return { "class_id": cls_id, "total_students": 45, "class_average_score": 81.5, "dimension_averages": {"逻辑": 83, "记忆": 80, "专注": 77}, "attention_warning_list": ["学生_03", "学生_17", "学生_29"], } s_batch_btn.click( _school_batch_import, inputs=[s_school_name, s_class_name], outputs=s_batch_output, ) s_analytics_btn.click( _school_get_analytics, inputs=[s_class_id], outputs=s_analytics_output, ) # ==================== 4. ⚙️ 系统与 API 状态 ==================== with gr.Tab("⚙️ 系统与 API 文档"): gr.Markdown( """ ## 📌 系统访问与 Demo 凭据 * **Swagger UI (OpenAPI 接口文档)**: [`/docs`](/docs) * **Health check (健康检查)**: [`/health`](/health) * **演示管理员账号:** `admin` | **密码:** `Admin123456!` """ ) status_box = gr.Textbox(label="System Status", interactive=False) check_btn = gr.Button("Check System Status") def check_status(): return "Engine: 12D AI Assessment Engine V2.2 | Status: Operational | API Docs: /docs | Health: /health" check_btn.click(fn=check_status, outputs=status_box) # ------------------------------------------------------------------- # 启动 Gradio 前端服务器 # ------------------------------------------------------------------- demo.launch( server_name="0.0.0.0", server_port=7860, prevent_thread_lock=True, quiet=True, ssr_mode=False, app_kwargs={"docs_url": "/docs", "redoc_url": "/redoc"}, ) # ------------------------------------------------------------------- # FastAPI 路由注入与后台逻辑 # ------------------------------------------------------------------- from app.api.v1 import tenant, assessment, report, analytics, auth, child from app.api.v1.endpoints import billing_audit_demo, offline_sync, compliance from app.api.v2.endpoints import multimodal_stream from app.core.config import settings from fastapi.staticfiles import StaticFiles app = demo.app app.include_router(auth.router, prefix="/api/v1") app.include_router(tenant.router, prefix="/api/v1") app.include_router(child.router, prefix="/api/v1") app.include_router(assessment.router, prefix="/api/v1") app.include_router(report.router, prefix="/api/v1") app.include_router(analytics.router, prefix="/api/v1") app.include_router(billing_audit_demo.router, prefix="/api/v1") app.include_router(offline_sync.router, prefix="/api/v1") app.include_router(compliance.router, prefix="/api/v1") app.include_router(multimodal_stream.router, prefix="/api/v2") _static_dir = os.path.join(os.path.dirname(__file__), "backend", "static") os.makedirs(_static_dir, exist_ok=True) app.mount("/wasm-static", StaticFiles(directory=_static_dir), name="wasm-static") @app.get("/health") def health_check(): return { "status": "healthy", "engine": "12D AI Assessment Engine V2.2", "version": settings.APP_VERSION, } @app.get("/api") def api_info(): return { "service": "12D AI Assessment Engine V2.2", "docs": "/docs", "health": "/health", } # ------------------------------------------------------------------- # 数据库与 Seed 初始化逻辑 # ------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s" ) _logger = logging.getLogger("main") try: from app.db.session import engine from app.models.base import Base from app.models.auth import User, Role, Permission, Organization from app.models.child import Child, Guardian from app.models.assessment import ( AssessmentItem, AssessmentSession, AssessmentResponse, ) from app.models.report import AssessmentReport from app.models.tenant import Tenant from app.core.audit_logger import AuditLog Base.metadata.create_all(bind=engine) _logger.info("Database tables created/verified.") from scripts.seed_database import seed_data seed_data() _logger.info("Demo data seeded successfully.") except Exception as e: _logger.error(f"Startup initialization error: {e}") # 主进程保持运行状态 if __name__ == "__main__": try: while True: time.sleep(3600) except KeyboardInterrupt: demo.close()