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

from django.conf import settings
from rest_framework import serializers

from .models import Asset, AssetFile, Model
from .storage import TosStorage


_tos_storage = None


def _tos():
    global _tos_storage
    if _tos_storage is None:
        _tos_storage = TosStorage()
    return _tos_storage


class AssetFileSerializer(serializers.ModelSerializer):
    preview_url = serializers.SerializerMethodField()

    class Meta:
        model = AssetFile
        fields = [
            "id",
            "object_key",
            "bucket",
            "content_type",
            "size_bytes",
            "width",
            "height",
            "duration_ms",
            "preview_url",
            "is_primary",
        ]
        read_only_fields = [
            "id",
            "object_key",
            "bucket",
            "content_type",
            "size_bytes",
            "width",
            "height",
            "duration_ms",
            "is_primary",
        ]

    def get_preview_url(self, obj):
        # 存储字段优先(如外部已写入绝对 URL);否则用 object_key 拼 TOS 公读直链。
        # TOS 桶公读,直链(虚拟主机式)免签名、且稳定可缓存——不再逐图签发预签名 URL
        # (签名是纯本地计算但每次都变、1h 过期会打不到浏览器/CDN 缓存;直链一劳永逸)。
        if obj.preview_url:
            return obj.preview_url
        if not obj.object_key or not settings.TOS.get("endpoint"):
            return ""
        try:
            return _tos().public_url(object_key=obj.object_key, bucket=obj.bucket or None)
        except Exception:
            return ""


class AssetSerializer(serializers.ModelSerializer):
    files = AssetFileSerializer(many=True, read_only=True)
    # 资产归属商品(只读):商品详情页据此只展示「该商品」的 AI 素材,而非全团队
    product = serializers.SerializerMethodField()

    def get_product(self, obj):
        """解析资产所属商品:
        1) 独立生图(图片创作/模特图/平台套图):生图时已写入 metadata.product_id;
        2) 项目内生成(基础资产/分镜图等):回溯 origin_task → project → product。
        两条路都拿不到则归属 None(如纯手动上传到素材库、与任何商品无关的图)。"""
        pid = (obj.metadata or {}).get("product_id")
        if pid:
            return str(pid)
        task = obj.origin_task
        if task and task.project_id and task.project.product_id:
            return str(task.project.product_id)
        return None

    class Meta:
        model = Asset
        fields = [
            "id",
            "name",
            "asset_type",
            "source",
            "category",
            "description",
            "metadata",
            "is_deleted",
            "in_library",
            "origin_task",
            "product",
            "files",
            "review_status",
            "review_error",
            "created_at",
            "updated_at",
        ]
        # in_library 只读:增删资产库走专用 set-library 接口,不允许普通 PATCH 直接改
        read_only_fields = ["id", "in_library", "review_status", "review_error", "created_at", "updated_at"]


class AssetUploadSerializer(serializers.Serializer):
    file = serializers.FileField()
    name = serializers.CharField(max_length=255, required=False, allow_blank=True)
    asset_type = serializers.ChoiceField(choices=Asset.Type.choices)
    category = serializers.ChoiceField(choices=Asset.Category.choices, default=Asset.Category.UPLOAD)
    description = serializers.CharField(required=False, allow_blank=True)


def _asset_preview(asset) -> str:
    """取资产主文件的公读直链(模特卡缩略图用)。无文件返回 ""。"""
    if asset is None:
        return ""
    f = asset.files.filter(is_primary=True).first() or asset.files.first()
    if f is None:
        return ""
    if f.preview_url:
        return f.preview_url
    if not f.object_key or not settings.TOS.get("endpoint"):
        return ""
    try:
        return _tos().public_url(object_key=f.object_key, bucket=f.bucket or None)
    except Exception:
        return ""


class ModelLibrarySerializer(serializers.ModelSerializer):
    """模特库卡片:实体字段 + 形象图/三视图缩略图直链 + 是否已挂声线。"""

    portrait = serializers.SerializerMethodField()
    triview = serializers.SerializerMethodField()
    has_voice = serializers.SerializerMethodField()

    class Meta:
        model = Model
        fields = [
            "id",
            "name",
            "is_official",
            "source",
            "description",
            "metadata",
            "is_deleted",
            "purged_at",
            "portrait_asset",
            "triview_asset",
            "voice_asset",
            "portrait",
            "triview",
            "has_voice",
            "created_at",
            "updated_at",
        ]
        read_only_fields = ["id", "is_official", "source", "is_deleted", "purged_at", "created_at", "updated_at"]

    def get_portrait(self, obj):
        return _asset_preview(obj.portrait_asset)

    def get_triview(self, obj):
        return _asset_preview(obj.triview_asset)

    def get_has_voice(self, obj):
        return bool(obj.voice_asset_id)

    def validate(self, attrs):
        """防越权:引用的形象图/三视图/声线资产必须属于当前团队。"""
        from apps.common.api import get_current_team

        request = self.context.get("request")
        if request is None:
            return attrs
        team = get_current_team(request.user)
        for field in ("portrait_asset", "triview_asset", "voice_asset"):
            asset = attrs.get(field)
            if asset is not None and asset.team_id != team.id:
                raise serializers.ValidationError({field: "资产不属于当前团队"})
        return attrs