""" FastAPI REST 后端 - 多模态情绪识别系统 运行: python api_server.py (端口 8088) """ import sys import os # 修复 Windows 控制台编码问题 if sys.platform == "win32": sys.stdout.reconfigure(encoding="utf-8", errors="replace") sys.stderr.reconfigure(encoding="utf-8", errors="replace") os.environ["PYTHONIOENCODING"] = "utf-8" import json import tempfile import io as std_io from pathlib import Path from datetime import datetime from typing import Any from fastapi import FastAPI, UploadFile, File, Form, Query, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, StreamingResponse import uvicorn # 确保项目根目录在 sys.path PROJECT_ROOT = Path(__file__).resolve().parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) # ====== 导入现有模块 ====== from text_module.text_emotion import TextEmotionAnalyzer from fusion.multimodal_va import fuse_multimodal_va from persistence.database import ( init_database, save_assessment, query_assessments, get_statistics, export_to_csv, export_to_json, ) # ====== 可选模块 (lazy import) ====== face_analyzer = None speech_analyzer = None def get_face_analyzer(): global face_analyzer if face_analyzer is None: try: from face_module.face_emotion import FaceEmotionAnalyzer face_analyzer = FaceEmotionAnalyzer() except Exception: face_analyzer = False return face_analyzer def get_speech_analyzer(): global speech_analyzer if speech_analyzer is None: try: from speech_module.speech_emotion import SpeechEmotionAnalyzer speech_analyzer = SpeechEmotionAnalyzer() except Exception: speech_analyzer = False return speech_analyzer # ====== 初始化 ====== app = FastAPI( title="Emotion Fusion API v2.0", description="多模态情绪识别 REST API", version="2.0.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # ====== 健康检查 ====== @app.get("/") async def root(): return { "service": "Emotion Fusion API v2.0", "status": "running", "endpoints": { "/": "health check", "/health": "health check", "/api/analyze": "POST - multimodal analysis (FormData)", "/api/analyze_json": "POST - text analysis (JSON)", "/api/assessments": "GET - query history", "/api/statistics": "GET - statistics", "/api/export/csv": "GET - export CSV", "/api/export/json": "GET - export JSON", } } @app.get("/health") async def health(): return {"status": "ok", "modules": {"text": "loaded", "face": "lazy", "speech": "lazy"}} # 初始化数据库 DATA_DIR = PROJECT_ROOT / "data" DATA_DIR.mkdir(exist_ok=True) db_conn = init_database(str(DATA_DIR)) # 初始化文本分析器 text_analyzer = TextEmotionAnalyzer() # ============================================================ # 辅助函数 # ============================================================ def _dict_row(row: dict[str, Any]) -> dict[str, Any]: """清洗数据库行,确保 JSON 兼容""" result = {} for k, v in row.items(): if isinstance(v, bytes): # 尝试解析 JSON try: result[k] = json.loads(v.decode("utf-8")) except Exception: result[k] = v.decode("utf-8", errors="replace") else: result[k] = v return result # ============================================================ # 端点 # ============================================================ @app.get("/api/health") async def health(): return {"status": "ok", "version": "2.0.0", "timestamp": datetime.now().isoformat()} @app.post("/api/analyze") async def analyze_emotion( text: str | None = Form(None), face_file: UploadFile | None = File(None), speech_file: UploadFile | None = File(None), ecg_csv_file: UploadFile | None = File(None), ): """多模态情绪融合分析 (FormData 模式)""" return await _do_analyze(text=text, face_file=face_file, speech_file=speech_file, ecg_csv_file=ecg_csv_file) @app.post("/api/analyze_json") async def analyze_emotion_json(request: Request): """多模态情绪融合分析 (JSON 模式)""" try: body = await request.json() except Exception: return JSONResponse(status_code=400, content={"error": "Invalid JSON", "available": False}) text = body.get("text", body.get("content", "")) return await _do_analyze(text=text) async def _do_analyze( text: str | None = None, face_file: UploadFile | None = None, speech_file: UploadFile | None = None, ecg_csv_file: UploadFile | None = None, ): """核心分析逻辑,同时支持 FormData 和 JSON""" raw_results = [] # ====== 1. 文本分析 ====== if text and text.strip(): try: text_result = text_analyzer.analyze(text.strip()) raw_results.append(text_result) except Exception as e: raw_results.append({ "available": False, "modality": "text", "emotion": None, "valence": None, "arousal": None, "confidence": 0, "quality": 0, "evidence": [], "warning": f"文本分析失败: {str(e)}", "error_code": "text_error", }) # ====== 2. 人脸分析 ====== if face_file: try: fa = get_face_analyzer() if fa: from PIL import Image contents = await face_file.read() image = Image.open(std_io.BytesIO(contents)).convert("RGB") face_result = fa.analyze_image(image) raw_results.append(face_result) else: raw_results.append({ "available": False, "modality": "face", "emotion": None, "valence": None, "arousal": None, "confidence": 0, "quality": 0, "evidence": [], "warning": "人脸分析模块未加载", "error_code": "module_error", }) except Exception as e: raw_results.append({ "available": False, "modality": "face", "emotion": None, "valence": None, "arousal": None, "confidence": 0, "quality": 0, "evidence": [], "warning": f"人脸分析失败: {str(e)}", "error_code": "face_error", }) # ====== 3. 语音分析 ====== if speech_file: try: sa = get_speech_analyzer() if sa: contents = await speech_file.read() with tempfile.NamedTemporaryFile(suffix=".wav", delete=True) as tmp: tmp.write(contents) tmp.flush() speech_result = sa.analyze(tmp.name) raw_results.append(speech_result) else: raw_results.append({ "available": False, "modality": "speech", "emotion": None, "valence": None, "arousal": None, "confidence": 0, "quality": 0, "evidence": [], "warning": "语音分析模块未加载", "error_code": "module_error", }) except Exception as e: raw_results.append({ "available": False, "modality": "speech", "emotion": None, "valence": None, "arousal": None, "confidence": 0, "quality": 0, "evidence": [], "warning": f"语音分析失败: {str(e)}", "error_code": "speech_error", }) # ====== 4. ECG 分析 (简化版) ====== if ecg_csv_file: try: from ecg_module.hrv_features import extract_hrv_features from ecg_module.arousal_ml import predict_arousal_from_hrv, load_arousal_model import pandas as pd import numpy as np contents = await ecg_csv_file.read() df = pd.read_csv(std_io.StringIO(contents.decode("utf-8"))) # 尝试从 CSV 中提取 HRV 特征 if "HR" in df.columns or "hr" in df.columns: hr_col = "HR" if "HR" in df.columns else "hr" hr_mean = float(df[hr_col].mean()) hr_std = float(df[hr_col].std()) else: hr_mean, hr_std = 72.0, 8.0 hrv_features = { "hr_mean": hr_mean, "sdnn": 45.0, "rmssd": 38.0, "pnn50": 12.0, "lf_hf_ratio": 1.5, } # 调用增强版 arousal 预测 model_data = load_arousal_model() arousal_result = predict_arousal_from_hrv(hrv_features, model_data, use_extended=True) # 估计 valence from ecg_module.hrv_features import estimate_ecg_valence_from_hrv ecg_valence = estimate_ecg_valence_from_hrv(hrv_features) ecg_result = { "available": True, "modality": "ecg", "emotion": arousal_result.get("arousal_label", "normal_arousal"), "valence": ecg_valence, "arousal": arousal_result.get("arousal_score", 0.5), "confidence": 0.7, "quality": 0.72, "evidence": [ f"HR={hr_mean:.1f} bpm", f"SDNN={hrv_features['sdnn']:.1f}", ], "warning": None, "error_code": None, "features": hrv_features, "arousal_probs": arousal_result.get("arousal_probabilities", {}), } raw_results.append(ecg_result) except Exception as e: raw_results.append({ "available": False, "modality": "ecg", "emotion": None, "valence": None, "arousal": None, "confidence": 0, "quality": 0, "evidence": [], "warning": f"ECG分析失败: {str(e)}", "error_code": "ecg_error", }) # ====== 5. 融合 ====== if not raw_results: return JSONResponse( status_code=400, content={"error": "请提供至少一种模态的数据", "available": False}, ) fusion_result = fuse_multimodal_va(raw_results) # ====== 6. 持久化 ====== try: save_assessment(db_conn, fusion_result, raw_results) except Exception: pass # 构建响应 response = { **fusion_result, "modality_table": fusion_result.get("modality_table", []), "raw_modality_count": len(raw_results), } # 转换所有值为 JSON 兼容类型 return _deep_serialize(response) @app.get("/api/assessments") async def get_assessments( limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0), patient_id: str | None = Query(None), emotion: str | None = Query(None), start_date: str | None = Query(None), end_date: str | None = Query(None), min_confidence: float = Query(0.0, le=1.0), ): """查询历史评估记录""" records, total = query_assessments( db_conn, limit=limit, offset=offset, patient_id=patient_id, emotion_filter=emotion, min_confidence=min_confidence, start_date=start_date, end_date=end_date, ) return { "records": [_dict_row(dict(r)) for r in records], "total": total, "limit": limit, "offset": offset, } @app.get("/api/statistics") async def statistics(): """获取系统统计信息""" try: stats = get_statistics(db_conn) return _deep_serialize(stats) except Exception as e: return {"error": str(e), "total_records": 0} @app.get("/api/export/csv") async def export_csv( patient_id: str | None = Query(None), emotion: str | None = Query(None), start_date: str | None = Query(None), end_date: str | None = Query(None), ): """导出评估记录为 CSV""" import tempfile, csv as csv_module filters = { "patient_id": patient_id, "emotion_filter": emotion, "start_date": start_date, "end_date": end_date, } filters = {k: v for k, v in filters.items() if v is not None} records, _ = query_assessments(db_conn, limit=10000, **filters) output = std_io.StringIO() if records: fieldnames = [ "id", "timestamp", "final_emotion", "valence", "arousal", "confidence", "quality", "modality_count", "uncertainty_level", "suggestion", ] writer = csv_module.DictWriter(output, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() for rec in records: writer.writerow(dict(rec)) csv_content = output.getvalue() if not csv_content.strip(): csv_content = "id,timestamp,final_emotion,valence,arousal,confidence,quality,modality_count,uncertainty_level,suggestion\n" return StreamingResponse( std_io.StringIO(csv_content), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=emotion_records.csv"}, ) @app.get("/api/export/json") async def export_json( patient_id: str | None = Query(None), emotion: str | None = Query(None), start_date: str | None = Query(None), end_date: str | None = Query(None), ): """导出评估记录为 JSON""" filters = { "patient_id": patient_id, "emotion_filter": emotion, "start_date": start_date, "end_date": end_date, } filters = {k: v for k, v in filters.items() if v is not None} records, _ = query_assessments(db_conn, limit=10000, **filters) data = [_dict_row(dict(r)) for r in records] return JSONResponse( content={"records": data, "total": len(data)}, headers={"Content-Disposition": "attachment; filename=emotion_records.json"}, ) # ============================================================ # 工具函数 # ============================================================ def _deep_serialize(obj: Any) -> Any: """递归转换所有值为 JSON 可序列化类型""" import numpy as np if isinstance(obj, dict): return {str(k): _deep_serialize(v) for k, v in obj.items()} elif isinstance(obj, (list, tuple)): return [_deep_serialize(item) for item in obj] elif isinstance(obj, (np.integer,)): return int(obj) elif isinstance(obj, (np.floating,)): return float(obj) elif isinstance(obj, np.ndarray): return obj.tolist() elif isinstance(obj, (datetime,)): return obj.isoformat() elif isinstance(obj, bytes): return obj.decode("utf-8", errors="replace") elif hasattr(obj, "item"): # numpy scalar return obj.item() return obj # ============================================================ # 启动 # ============================================================ if __name__ == "__main__": print("=" * 50) print(" 多模态情绪识别系统 - FastAPI 后端 v2.0") print(" Listening on http://localhost:8088") print(" API docs: http://localhost:8088/docs") print("=" * 50) uvicorn.run(app, host="0.0.0.0", port=8088, log_level="info")