api_server.py

"""
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")