yingqing/core/backend/apps/projects/serializers.py

import uuid

from rest_framework import serializers

from apps.assets.serializers import AssetFileSerializer

from .models import (
    BaseAssetGroup,
    BgmTrack,
    ExportJob,
    Project,
    ProjectStage,
    ScriptSegment,
    ScriptVersion,
    StoryboardFrame,
    StoryboardVersion,
    SubtitleTrack,
    Timeline,
    TimelineClip,
    VideoSegment,
    VideoSegmentVersion,
)


def _asset_preview_url(asset) -> str:
    """资产主文件的可播放/可显示 URL(主图优先,其次首张),内嵌进各阶段序列化,
    让前端缩略图不再依赖(分页 20 条的)团队 assets 列表解析——团队资产 >20 时新生成的图本会丢。"""
    if asset is None:
        return ""
    files = list(asset.files.all())
    primary = next((f for f in files if f.is_primary), files[0] if files else None)
    return AssetFileSerializer().get_preview_url(primary) if primary else ""


class ProjectStageSerializer(serializers.ModelSerializer):
    class Meta:
        model = ProjectStage
        fields = ["id", "stage", "status", "started_at", "completed_at", "error_message", "metadata"]
        read_only_fields = fields


class VideoSegmentSerializer(serializers.ModelSerializer):
    adopted_asset = serializers.SerializerMethodField()
    adopted_asset_url = serializers.SerializerMethodField()
    versions = serializers.SerializerMethodField()

    class Meta:
        model = VideoSegment
        fields = ["id", "sort_order", "target_duration_seconds", "status", "error_message", "adopted_version", "adopted_asset", "adopted_asset_url", "versions"]
        read_only_fields = ["id", "sort_order", "target_duration_seconds", "status", "error_message", "adopted_version"]

    def get_adopted_asset(self, obj):
        # pipeline stage4 缩略图:暴露已采用版本对应的资产 id(供前端在 assets 里解析 preview_url)
        version = obj.adopted_version
        return str(version.asset_id) if version and version.asset_id else None

    def get_adopted_asset_url(self, obj) -> str:
        version = obj.adopted_version
        return _asset_preview_url(version.asset) if version is not None else ""

    def get_versions(self, obj):
        # 视频详情弹窗:历史版本(倒序,带可播放 URL)。模型无 Meta.ordering,这里按 created_at 排
        versions = sorted(obj.versions.all(), key=lambda v: (v.created_at is None, v.created_at), reverse=True)
        return [
            {
                "id": str(v.id),
                "asset": str(v.asset_id) if v.asset_id else None,
                "asset_url": _asset_preview_url(v.asset) if v.asset_id else "",
                "prompt": v.prompt,
                "is_adopted": v.is_adopted,
                "created_at": v.created_at.isoformat() if v.created_at else "",
            }
            for v in versions
        ]


class BaseAssetGroupSerializer(serializers.ModelSerializer):
    candidate_assets = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
    adopted_asset_url = serializers.SerializerMethodField()
    candidate_asset_urls = serializers.SerializerMethodField()
    # 采用资产的火山审核态随项目详情下发,刷新后仍能回显绿/红/审核中徽章(前端不再依赖那次性的轮询)
    adopted_asset_review = serializers.SerializerMethodField()
    adopted_asset_review_error = serializers.SerializerMethodField()

    class Meta:
        model = BaseAssetGroup
        fields = ["id", "kind", "prompt", "adopted_asset", "adopted_asset_url", "candidate_assets", "candidate_asset_urls", "adopted_asset_review", "adopted_asset_review_error", "version", "metadata", "created_at"]
        read_only_fields = fields

    def get_adopted_asset_url(self, obj) -> str:
        return _asset_preview_url(obj.adopted_asset)

    def get_adopted_asset_review(self, obj) -> str:
        return getattr(obj.adopted_asset, "review_status", "") or ""

    def get_adopted_asset_review_error(self, obj) -> str:
        return getattr(obj.adopted_asset, "review_error", "") or ""

    def get_candidate_asset_urls(self, obj) -> dict:
        return {str(asset.id): _asset_preview_url(asset) for asset in obj.candidate_assets.all()}


class StoryboardFrameSerializer(serializers.ModelSerializer):
    asset_url = serializers.SerializerMethodField()
    # 分镜图含人脸,后端自动送火山人像审核;暴露审核态让前端挂盾牌(与角色/场景卡同款)
    review_status = serializers.SerializerMethodField()
    review_error = serializers.SerializerMethodField()

    class Meta:
        model = StoryboardFrame
        fields = ["id", "script_segment", "asset", "asset_url", "sort_order", "prompt", "review_status", "review_error"]
        read_only_fields = fields

    def get_asset_url(self, obj) -> str:
        return _asset_preview_url(obj.asset)

    def get_review_status(self, obj) -> str:
        return obj.asset.review_status if obj.asset_id else ""

    def get_review_error(self, obj) -> str:
        return (obj.asset.review_error or "") if obj.asset_id else ""


class StoryboardVersionSerializer(serializers.ModelSerializer):
    frames = StoryboardFrameSerializer(many=True, read_only=True)

    class Meta:
        model = StoryboardVersion
        fields = ["id", "prompt", "is_adopted", "frames", "created_at", "updated_at"]
        read_only_fields = fields


class VideoSegmentVersionSerializer(serializers.ModelSerializer):
    class Meta:
        model = VideoSegmentVersion
        fields = ["id", "video_segment", "asset", "prompt", "is_adopted", "metadata", "created_at"]
        read_only_fields = fields


class TimelineClipSerializer(serializers.ModelSerializer):
    # 直接内嵌片段资产的可播放 URL + 是否视频,前端播放器无需再依赖(分页的)团队 assets 列表解析
    asset_url = serializers.SerializerMethodField()
    asset_is_video = serializers.SerializerMethodField()

    class Meta:
        model = TimelineClip
        fields = ["id", "asset", "asset_url", "asset_is_video", "sort_order", "start_ms", "duration_ms", "trim_start_ms", "trim_end_ms"]
        read_only_fields = ["id", "asset_url", "asset_is_video"]

    def _primary_file(self, obj):
        asset = obj.asset
        if asset is None:
            return None
        files = list(asset.files.all())
        return next((f for f in files if f.is_primary), files[0] if files else None)

    def get_asset_url(self, obj) -> str:
        f = self._primary_file(obj)
        return AssetFileSerializer().get_preview_url(f) if f else ""

    def get_asset_is_video(self, obj) -> bool:
        asset = obj.asset
        if asset is None:
            return False
        if asset.asset_type == "video":
            return True
        f = self._primary_file(obj)
        return bool(f and "video/" in (f.content_type or ""))


class TimelineExportJobSerializer(serializers.ModelSerializer):
    class Meta:
        model = ExportJob
        fields = ["id", "status", "output_asset", "progress", "error_message", "created_at", "updated_at"]
        read_only_fields = fields


class SubtitleTrackSerializer(serializers.ModelSerializer):
    class Meta:
        model = SubtitleTrack
        fields = ["id", "content", "style", "enabled"]
        read_only_fields = fields


class BgmTrackSerializer(serializers.ModelSerializer):
    asset_url = serializers.SerializerMethodField()
    asset_name = serializers.SerializerMethodField()

    class Meta:
        model = BgmTrack
        fields = ["id", "asset", "asset_url", "asset_name", "volume", "start_ms"]
        read_only_fields = fields

    def get_asset_url(self, obj) -> str:
        return _asset_preview_url(obj.asset)

    def get_asset_name(self, obj) -> str:
        return obj.asset.name if obj.asset_id else ""


class TimelineSerializer(serializers.ModelSerializer):
    clips = TimelineClipSerializer(many=True, read_only=True)
    export_jobs = TimelineExportJobSerializer(many=True, read_only=True)
    subtitle_tracks = SubtitleTrackSerializer(many=True, read_only=True)
    bgm_tracks = BgmTrackSerializer(many=True, read_only=True)
    voiceover = serializers.SerializerMethodField()

    class Meta:
        model = Timeline
        fields = ["id", "name", "aspect_ratio", "resolution", "duration_seconds", "metadata", "voiceover", "clips", "export_jobs", "subtitle_tracks", "bgm_tracks"]
        read_only_fields = ["id", "voiceover", "clips", "export_jobs", "subtitle_tracks", "bgm_tracks"]

    def get_voiceover(self, obj):
        # 旁白配音映射(metadata.voiceover)+ 每段新鲜的可播放 URL(TOS 签名 URL 会过期,不能存死)
        vo = (obj.metadata or {}).get("voiceover")
        if not isinstance(vo, dict) or not vo.get("items"):
            return None
        from apps.assets.models import Asset

        ids = []
        for item in vo["items"]:
            try:
                ids.append(uuid.UUID(str(item.get("asset"))))
            except (TypeError, ValueError):
                continue
        assets = {str(a.id): a for a in Asset.objects.filter(id__in=ids).prefetch_related("files")}
        items = []
        for item in vo["items"]:
            asset = assets.get(str(item.get("asset")))
            items.append({**item, "asset_url": _asset_preview_url(asset)})
        return {
            "enabled": bool(vo.get("enabled")),
            "voice_type": str(vo.get("voice_type") or ""),
            "speed_ratio": vo.get("speed_ratio", 1.0),
            "items": items,
        }


class ExportJobSerializer(serializers.ModelSerializer):
    class Meta:
        model = ExportJob
        fields = ["id", "status", "output_asset", "progress", "error_message", "metadata", "created_at", "updated_at"]
        read_only_fields = fields


class ScriptSegmentSerializer(serializers.ModelSerializer):
    class Meta:
        model = ScriptSegment
        fields = [
            "id", "sort_order", "duration_seconds", "narration", "visual_prompt", "product_points",
            "role", "speaker", "product_exposure", "entity_refs", "dialogue",
        ]
        read_only_fields = fields


class ScriptVersionSerializer(serializers.ModelSerializer):
    segments = ScriptSegmentSerializer(many=True, read_only=True)

    class Meta:
        model = ScriptVersion
        # metadata 携带 ScriptDraft 的 hook/tone/entities,供前端结构化渲染与下游故事板 @图N
        fields = ["id", "title", "content", "source", "is_adopted", "segments", "metadata", "created_at", "updated_at"]
        read_only_fields = fields


class ProjectListSerializer(serializers.ModelSerializer):
    """列表/仪表盘/侧栏用的轻量项目序列化:不嵌套 阶段/片段/故事板/时间线(那些只详情页要)。
    脚本数/镜数走 annotate 计数(见 ProjectViewSet.get_queryset),避免逐项目拉全套关联(原列表 2-3s)。"""

    product_title = serializers.CharField(source="product.title", read_only=True, default="")
    cover_preview_url = serializers.SerializerMethodField()
    script_version_count = serializers.IntegerField(read_only=True, default=0)
    video_segment_count = serializers.IntegerField(read_only=True, default=0)

    class Meta:
        model = Project
        fields = [
            "id", "name", "product", "product_title", "cover_preview_url",
            "status", "current_stage", "script_version_count", "video_segment_count",
            "created_at", "updated_at",
        ]

    def get_cover_preview_url(self, obj) -> str:
        return _asset_preview_url(getattr(obj.product, "cover_asset", None)) if obj.product_id else ""


class ProjectSerializer(serializers.ModelSerializer):
    stages = ProjectStageSerializer(many=True, read_only=True)
    video_segments = VideoSegmentSerializer(many=True, read_only=True)
    script_versions = ScriptVersionSerializer(many=True, read_only=True)
    base_asset_groups = BaseAssetGroupSerializer(many=True, read_only=True)
    storyboard_versions = StoryboardVersionSerializer(many=True, read_only=True)
    timeline = TimelineSerializer(read_only=True)

    class Meta:
        model = Project
        fields = [
            "id",
            "name",
            "product",
            "status",
            "current_stage",
            "budget_limit",
            "failure_reason",
            "metadata",
            "stages",
            "script_versions",
            "base_asset_groups",
            "storyboard_versions",
            "video_segments",
            "timeline",
            "created_at",
            "updated_at",
        ]
        read_only_fields = ["id", "status", "current_stage", "failure_reason", "created_at", "updated_at"]