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