video-gen/video-gen-api/app/services/admin_upload/service.py

from __future__ import annotations

from typing import Any

from fastapi import UploadFile
from sqlalchemy.ext.asyncio import AsyncSession

from app.enums.admin_upload import AdminUploadLogEventEnum, AdminUploadResourceTypeEnum, AdminUploadSceneEnum
from app.enums.common import LogEventStatusEnum, LogSourceEnum
from app.enums.upload_resource import (
    UploadResourceBindStatusEnum,
    UploadResourceCreatedByEnum,
    UploadResourceDeletePolicyEnum,
    UploadResourceDurationSourceEnum,
    UploadResourceModuleEnum,
)
from app.models.user import User
from app.schemas.admin_upload import AdminUploadFileOut, AdminUploadMediaReferenceOut
from app.services.admin_upload.storage import save_upload_file
from app.services.operation_log_service import build_exception_detail, log_operation_event
from app.services.upload_resource.core_service import record_external_upload_resource


def _duration(value: float | None) -> float | None:
    if value is None:
        return None
    try:
        v = float(value)
        return round(v, 3) if v >= 0 else None
    except Exception:
        return None


def _module_for_scene(scene: AdminUploadSceneEnum) -> str:
    if scene == AdminUploadSceneEnum.HOME_MATERIAL_REFERENCE:
        return UploadResourceModuleEnum.HOME_MATERIAL.value
    return UploadResourceModuleEnum.ADMIN_UPLOAD.value


def _role_for_type(resource_type: AdminUploadResourceTypeEnum) -> str | None:
    if resource_type == AdminUploadResourceTypeEnum.IMAGE:
        return "reference_image"
    if resource_type == AdminUploadResourceTypeEnum.VIDEO:
        return "reference_video"
    if resource_type == AdminUploadResourceTypeEnum.AUDIO:
        return "reference_audio"
    return None


def _media_reference(*, url: str, resource_type: AdminUploadResourceTypeEnum, name: str | None, duration_seconds: float | None, resource_id: str) -> AdminUploadMediaReferenceOut:
    return AdminUploadMediaReferenceOut(
        url=url,
        type=resource_type.value,
        name=name,
        duration=duration_seconds,
        source="admin_upload",
        upload_resource_id=resource_id,
        display_url=url,
        preview_url=url,
        role=_role_for_type(resource_type),
    )


def _log(event_type: str, *, admin: User, scene: str, resource_type: str, status: str = LogEventStatusEnum.SUCCESS.value, message: str | None = None, detail: dict[str, Any] | None = None, error: str | None = None) -> None:
    log_operation_event(
        domain="admin_upload",
        module="admin_upload",
        event_type=event_type,
        event_status=status,
        source=LogSourceEnum.API.value,
        user_id=admin.id,
        message=message,
        detail={"scene": scene, "resource_type": resource_type, **(detail or {})},
        error=error,
    )


class AdminUploadService:
    async def upload_file(
        self,
        db: AsyncSession,
        *,
        file: UploadFile,
        scene: AdminUploadSceneEnum,
        resource_type: AdminUploadResourceTypeEnum,
        admin: User,
        duration_seconds: float | None = None,
    ) -> AdminUploadFileOut:
        duration = _duration(duration_seconds)
        _log(
            AdminUploadLogEventEnum.VALIDATE_STARTED.value,
            admin=admin,
            scene=scene.value,
            resource_type=resource_type.value,
            status=LogEventStatusEnum.STARTED.value,
            detail={"filename": file.filename, "content_type": file.content_type},
        )
        stored = None
        try:
            _log(
                AdminUploadLogEventEnum.VALIDATE_SUCCESS.value,
                admin=admin,
                scene=scene.value,
                resource_type=resource_type.value,
                detail={"filename": file.filename, "duration_seconds": duration},
            )
            _log(
                AdminUploadLogEventEnum.SAVE_STARTED.value,
                admin=admin,
                scene=scene.value,
                resource_type=resource_type.value,
                status=LogEventStatusEnum.STARTED.value,
                detail={"filename": file.filename},
            )
            stored = await save_upload_file(file, scene=scene, resource_type=resource_type, admin_id=admin.id)
            _log(
                AdminUploadLogEventEnum.SAVE_SUCCESS.value,
                admin=admin,
                scene=scene.value,
                resource_type=resource_type.value,
                detail={"url": stored.file_url, "storage_path": stored.storage_path, "file_size_bytes": stored.file_size_bytes},
            )
            module = _module_for_scene(scene)
            resource = await record_external_upload_resource(
                db,
                user_id=admin.id,
                module=module,
                resource_type=resource_type.value,
                resource_url=stored.file_url,
                storage_path=stored.storage_path,
                file_size_bytes=stored.file_size_bytes,
                file_name=stored.file_name,
                mime_type=file.content_type,
                duration_seconds=duration,
                duration_source=UploadResourceDurationSourceEnum.CLIENT.value if duration is not None else None,
                bind_status=UploadResourceBindStatusEnum.PENDING.value,
                delete_policy=UploadResourceDeletePolicyEnum.USER_DELETABLE.value,
                created_by=UploadResourceCreatedByEnum.API.value,
                metadata={
                    "scene": scene.value,
                    "original_filename": stored.original_file_name,
                    "client_duration_seconds": duration_seconds,
                    "source": "admin_upload",
                },
            )
            await db.refresh(resource)
            _log(
                AdminUploadLogEventEnum.DB_RECORD_SUCCESS.value,
                admin=admin,
                scene=scene.value,
                resource_type=resource_type.value,
                detail={"resource_id": resource.id, "url": stored.file_url, "module": module},
            )
            media_reference = _media_reference(
                url=stored.file_url,
                resource_type=resource_type,
                name=stored.original_file_name or stored.file_name,
                duration_seconds=resource.duration_seconds,
                resource_id=resource.id,
            )
            return AdminUploadFileOut(
                resource_id=resource.id,
                scene=scene,
                module=module,
                resource_type=resource_type,
                url=stored.file_url,
                file_name=stored.file_name,
                original_file_name=stored.original_file_name,
                file_size_bytes=stored.file_size_bytes,
                duration_seconds=resource.duration_seconds,
                media_reference=media_reference,
            )
        except Exception as exc:
            _log(
                AdminUploadLogEventEnum.FAILED.value,
                admin=admin,
                scene=scene.value,
                resource_type=resource_type.value,
                status=LogEventStatusEnum.FAILED.value,
                detail=build_exception_detail(exc, {"filename": file.filename}),
                error=str(exc),
            )
            raise


admin_upload_service = AdminUploadService()