happy-up/apps/api/app/services/analysis.py

from datetime import datetime, timezone

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.ai.pose_analyzer import analyze_task
from app.config import settings
from app.db.models import AnalysisTask, Report, Video
from app.mock_data import MOCK_MOVEMENT_REPORT, MOCK_REPORT
from app.queue.analysis_queue import get_analysis_queue
from app.schemas.models import AnalysisTaskCreateRequest, TaskStatus, TaskType
from app.services import oss as oss_service
from app.services.videos import verify_child_owner

DISCLAIMER_SCREENING = "本报告用于健康管理建议，不构成医疗诊断。"
DISCLAIMER_MOVEMENT = "本报告用于运动训练反馈，不构成医疗诊断。"


def task_to_dict(task: AnalysisTask) -> dict:
    report_id = None
    if task.result and isinstance(task.result, dict):
        report_id = task.result.get("reportId")
    return {
        "id": task.id,
        "status": task.status,
        "progress": task.progress,
        "errorCode": task.error_code,
        "reportId": report_id,
        "retryCount": task.retry_count,
    }


def build_screening_report_payload(child_id: int, task_id: int) -> dict:
    data = {**MOCK_REPORT, "childId": child_id, "taskId": task_id}
    return data


def build_movement_report_payload(child_id: int, task_id: int) -> dict:
    return {**MOCK_MOVEMENT_REPORT, "childId": child_id, "taskId": task_id}


def create_report_for_task(db: Session, task: AnalysisTask, analysis: dict) -> Report:
    report_data = analysis.get("report", {})
    if task.task_type == TaskType.movement_scoring.value:
        payload = {**MOCK_MOVEMENT_REPORT, **report_data}
        report = Report(
            child_id=task.child_id,
            task_id=task.id,
            report_type="movement_scoring",
            risk_level="medium",
            summary="跟练打卡报告已生成",
            metrics=[payload],
            recommendations=["继续保持训练节奏", "28 天后安排复测"],
            disclaimer=DISCLAIMER_MOVEMENT,
            status="published",
        )
    else:
        payload = {**MOCK_REPORT, **report_data, "childId": task.child_id, "taskId": task.id}
        risk_level = payload.get("riskLevel", "medium")
        report_status = "pending_review" if risk_level in ("high", "review_required") else "published"
        report = Report(
            child_id=task.child_id,
            task_id=task.id,
            report_type="posture_screening",
            risk_level=risk_level,
            summary=payload.get("summary", MOCK_REPORT["summary"]),
            metrics=payload.get("metrics", MOCK_REPORT["metrics"]),
            recommendations=payload.get("recommendations", MOCK_REPORT["recommendations"]),
            disclaimer=payload.get("disclaimer", DISCLAIMER_SCREENING),
            status=report_status,
        )
    db.add(report)
    db.flush()
    return report


def process_analysis_task(db: Session, task_id: int) -> AnalysisTask | None:
    task = db.get(AnalysisTask, task_id)
    if not task or task.status in (TaskStatus.SUCCEEDED.value, TaskStatus.CANCELLED.value):
        return task

    task.status = TaskStatus.PROCESSING.value
    task.progress = 35
    task.started_at = datetime.now(timezone.utc)
    db.commit()

    video = db.get(Video, task.video_id)
    video_path = None
    if video:
        video_path = oss_service.resolve_local_path(video.object_key)

    analysis = analyze_task(task.task_type, video_path)
    task.model_version = analysis.get("modelVersion", "mock-v1")
    task.progress = 80
    db.commit()

    report = create_report_for_task(db, task, analysis)
    task.status = TaskStatus.SUCCEEDED.value
    task.progress = 100
    task.finished_at = datetime.now(timezone.utc)
    task.result = {
        "reportId": report.id,
        "engine": analysis.get("engine"),
        "confidence": analysis.get("confidence"),
    }
    db.commit()
    db.refresh(task)
    return task


def enqueue_task(task_id: int, db: Session | None = None) -> None:
    queue = get_analysis_queue()
    queue.push(task_id)
    if not settings.analysis_inline_process:
        return
    if db is not None:
        process_analysis_task(db, task_id)
        return
    from app.db.session import SessionLocal

    local_db = SessionLocal()
    try:
        process_analysis_task(local_db, task_id)
    finally:
        local_db.close()


def create_analysis_task(
    db: Session,
    user_id: int,
    body: AnalysisTaskCreateRequest,
    idempotency_key: str | None,
) -> tuple[AnalysisTask | None, str | None, AnalysisTask | None]:
    child = verify_child_owner(db, user_id, body.child_id)
    if not child:
        return None, "child_not_found", None

    video = db.get(Video, body.video_id)
    if not video or video.child_id != body.child_id:
        return None, "video_not_found", None

    if idempotency_key:
        existing = db.scalar(
            select(AnalysisTask).where(
                AnalysisTask.child_id == body.child_id,
                AnalysisTask.video_id == body.video_id,
                AnalysisTask.task_type == body.task_type.value,
                AnalysisTask.idempotency_key == idempotency_key,
            )
        )
        if existing:
            return existing, "duplicate", existing

    task = AnalysisTask(
        child_id=body.child_id,
        video_id=body.video_id,
        task_type=body.task_type.value,
        status=TaskStatus.QUEUED.value,
        progress=0,
        idempotency_key=idempotency_key,
    )
    db.add(task)
    db.commit()
    db.refresh(task)
    enqueue_task(task.id, db=db)
    db.refresh(task)
    return task, None, None


def get_task_for_user(db: Session, user_id: int, task_id: int) -> AnalysisTask | None:
    task = db.get(AnalysisTask, task_id)
    if not task:
        return None
    child = verify_child_owner(db, user_id, task.child_id)
    if not child:
        return None
    return task


def cancel_task(db: Session, user_id: int, task_id: int) -> AnalysisTask | None:
    task = get_task_for_user(db, user_id, task_id)
    if not task:
        return None
    if task.status in (TaskStatus.SUCCEEDED.value, TaskStatus.CANCELLED.value):
        return task
    task.status = TaskStatus.CANCELLED.value
    task.finished_at = datetime.now(timezone.utc)
    db.commit()
    db.refresh(task)
    return task


def handle_webhook(db: Session, task_id: int, status: str) -> None:
    task = db.get(AnalysisTask, task_id)
    if not task:
        return
    if status == TaskStatus.SUCCEEDED.value:
        process_analysis_task(db, task_id)
    elif status == TaskStatus.FAILED.value:
        task.status = TaskStatus.FAILED.value
        task.error_code = "worker_failed"
        task.finished_at = datetime.now(timezone.utc)
        db.commit()