82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
from rest_framework import serializers
|
|
|
|
from .models import Notification
|
|
|
|
|
|
class NotificationSerializer(serializers.ModelSerializer):
|
|
type = serializers.CharField(source="notification_type", read_only=True)
|
|
unread = serializers.SerializerMethodField()
|
|
project_name = serializers.CharField(source="project.name", read_only=True)
|
|
brief = serializers.SerializerMethodField()
|
|
body = serializers.SerializerMethodField()
|
|
metadata = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = Notification
|
|
fields = [
|
|
"id",
|
|
"type",
|
|
"notification_type",
|
|
"priority",
|
|
"title",
|
|
"brief",
|
|
"body",
|
|
"source",
|
|
"project",
|
|
"project_name",
|
|
"stage",
|
|
"owner_label",
|
|
"cost_label",
|
|
"related_url",
|
|
"is_read",
|
|
"unread",
|
|
"read_at",
|
|
"archived_at",
|
|
"metadata",
|
|
"created_at",
|
|
"updated_at",
|
|
]
|
|
read_only_fields = [
|
|
"id",
|
|
"type",
|
|
"project_name",
|
|
"read_at",
|
|
"archived_at",
|
|
"created_at",
|
|
"updated_at",
|
|
]
|
|
|
|
def get_unread(self, obj):
|
|
return not obj.is_read
|
|
|
|
def _current_generation_message(self, obj):
|
|
metadata = obj.metadata if isinstance(obj.metadata, dict) else {}
|
|
error = metadata.get("generation_error")
|
|
if not isinstance(error, dict) or error.get("domain") != "generation":
|
|
return ""
|
|
from apps.ai.generation_errors import public_error_for_code
|
|
|
|
return public_error_for_code(
|
|
str(error.get("code") or "unknown"),
|
|
operation=str(error.get("operation") or "image_generate"),
|
|
reference_id=str(error.get("reference_id") or "") or None,
|
|
).fallback_message
|
|
|
|
def get_brief(self, obj):
|
|
return self._current_generation_message(obj) or obj.brief or ""
|
|
|
|
def get_body(self, obj):
|
|
"""兼容早期 AI 失败通知：不再向普通用户回传嵌入正文的上游原始异常。"""
|
|
current = self._current_generation_message(obj)
|
|
if current:
|
|
return current
|
|
body = obj.body or ""
|
|
marker = "—— 第三方服务商 API 返回的原始报错 ——"
|
|
return body.split(marker, 1)[0].strip()
|
|
|
|
def get_metadata(self, obj):
|
|
"""旧通知的 api_error 也必须从普通用户 API 中移除。"""
|
|
metadata = dict(obj.metadata or {})
|
|
metadata.pop("api_error", None)
|
|
return metadata
|