forked from erp-dev/erp
99 lines
3.8 KiB
Python
99 lines
3.8 KiB
Python
from rest_framework import serializers
|
|
|
|
|
|
def _normalize_user_ids(values: list[str] | None) -> list[str]:
|
|
normalized = []
|
|
for value in values or []:
|
|
item = str(value or "").strip()
|
|
if not item:
|
|
raise serializers.ValidationError("user_ids 中不能包含空值")
|
|
normalized.append(item)
|
|
return normalized
|
|
|
|
|
|
class MessageTextRequestSerializer(serializers.Serializer):
|
|
agent_id = serializers.IntegerField(min_value=1)
|
|
content = serializers.CharField(allow_blank=False, trim_whitespace=True)
|
|
user_ids = serializers.ListField(
|
|
child=serializers.CharField(allow_blank=False, trim_whitespace=True),
|
|
required=False,
|
|
allow_empty=True,
|
|
)
|
|
|
|
def validate_content(self, value: str) -> str:
|
|
content = (value or "").strip()
|
|
if not content:
|
|
raise serializers.ValidationError("content 不能为空")
|
|
if len(content.encode("utf-8")) > 2048:
|
|
raise serializers.ValidationError("content 不能超过 2048 字节")
|
|
return content
|
|
|
|
def validate_user_ids(self, value):
|
|
return _normalize_user_ids(value)
|
|
|
|
|
|
class MessageNewsRequestSerializer(serializers.Serializer):
|
|
agent_ids = serializers.ListField(
|
|
child=serializers.IntegerField(min_value=1),
|
|
allow_empty=False,
|
|
)
|
|
title = serializers.CharField(allow_blank=False, trim_whitespace=True, max_length=128)
|
|
description = serializers.CharField(allow_blank=False, trim_whitespace=True, max_length=512)
|
|
url = serializers.URLField(allow_blank=False)
|
|
image_url = serializers.URLField(allow_blank=False)
|
|
user_ids = serializers.ListField(
|
|
child=serializers.CharField(allow_blank=False, trim_whitespace=True),
|
|
required=False,
|
|
allow_empty=True,
|
|
)
|
|
|
|
def validate_agent_ids(self, value):
|
|
return list(dict.fromkeys(value or []))
|
|
|
|
def validate_user_ids(self, value):
|
|
return _normalize_user_ids(value)
|
|
|
|
|
|
class MessageTemplatePayloadSerializer(serializers.Serializer):
|
|
msgtype = serializers.ChoiceField(choices=["text", "news"])
|
|
agent_id = serializers.IntegerField(min_value=1, required=False)
|
|
agent_ids = serializers.ListField(
|
|
child=serializers.IntegerField(min_value=1),
|
|
required=False,
|
|
allow_empty=False,
|
|
)
|
|
content = serializers.CharField(required=False, allow_blank=False, trim_whitespace=True)
|
|
title = serializers.CharField(required=False, allow_blank=False, trim_whitespace=True, max_length=128)
|
|
description = serializers.CharField(required=False, allow_blank=False, trim_whitespace=True, max_length=512)
|
|
url = serializers.URLField(required=False)
|
|
image_url = serializers.URLField(required=False)
|
|
user_ids = serializers.ListField(
|
|
child=serializers.CharField(allow_blank=False, trim_whitespace=True),
|
|
required=False,
|
|
allow_empty=True,
|
|
)
|
|
|
|
def validate_user_ids(self, value):
|
|
return _normalize_user_ids(value)
|
|
|
|
def validate(self, attrs):
|
|
msgtype = attrs["msgtype"]
|
|
if msgtype == "text":
|
|
if not attrs.get("content"):
|
|
raise serializers.ValidationError({"content": "text 类型必须提供 content"})
|
|
if "agent_ids" in attrs:
|
|
raise serializers.ValidationError({"agent_ids": "text 类型不支持 agent_ids"})
|
|
return attrs
|
|
|
|
required = ["title", "description", "url", "image_url"]
|
|
errors = {}
|
|
for field in required:
|
|
if not attrs.get(field):
|
|
errors[field] = f"news 类型必须提供 {field}"
|
|
if "agent_id" in attrs:
|
|
errors["agent_id"] = "news 类型不支持 agent_id"
|
|
if "content" in attrs:
|
|
errors["content"] = "news 类型不支持 content"
|
|
if errors:
|
|
raise serializers.ValidationError(errors)
|
|
return attrs |