SupplyAgent/api/schemas/demand_feedback.py

from __future__ import annotations

from typing import Literal

from pydantic import BaseModel, Field, field_validator, model_validator

FeedbackTargetType = Literal["demand", "video", "hit_content"]
FeedbackAction = Literal["support", "oppose", "correct", "supplement"]

REASON_CODES_BY_TARGET: dict[str, set[str]] = {
    "demand": {
        "not_a_demand",
        "wrong_grade",
        "wrong_category",
        "unclear_wording",
        "wrong_reason",
        "missing_demand",
        "other",
    },
    "video": {
        "irrelevant",
        "weak_relevance",
        "unavailable",
        "duplicate",
        "wrong_info",
        "missing_video",
        "other",
    },
    "hit_content": {
        "not_hit",
        "wrong_point_type",
        "inaccurate_wording",
        "wrong_reason",
        "duplicate",
        "missing_content",
        "other",
    },
}


class CreateDemandFeedbackBody(BaseModel):
    client_request_id: str = Field(min_length=8, max_length=64)
    target_type: FeedbackTargetType
    demand_grade_id: int = Field(gt=0)
    video_id: str | None = Field(default=None, max_length=64)
    demand_video_expansion_id: int | None = Field(default=None, gt=0)
    feedback_action: FeedbackAction
    reason_code: str | None = Field(default=None, max_length=64)
    content: str | None = Field(default=None, max_length=1000)

    @field_validator("client_request_id", "video_id", "reason_code", "content")
    @classmethod
    def strip_text(cls, value: str | None) -> str | None:
        if value is None:
            return None
        stripped = value.strip()
        return stripped or None

    @model_validator(mode="after")
    def validate_target_and_content(self) -> "CreateDemandFeedbackBody":
        if self.target_type == "demand":
            if self.video_id is not None or self.demand_video_expansion_id is not None:
                raise ValueError("需求反馈不能包含视频或命中内容 ID")
        elif self.target_type == "video":
            if not self.video_id or self.demand_video_expansion_id is not None:
                raise ValueError("视频反馈必须且只能包含视频 ID")
        elif not self.video_id or self.demand_video_expansion_id is None:
            raise ValueError("命中内容反馈必须包含视频 ID 和命中内容 ID")

        if self.feedback_action == "oppose" and not self.reason_code:
            raise ValueError("不认可反馈必须选择问题原因")
        if self.feedback_action in {"correct", "supplement"} and not self.content:
            raise ValueError("纠错或补充反馈必须填写说明")
        if (
            self.reason_code is not None
            and self.reason_code not in REASON_CODES_BY_TARGET[self.target_type]
        ):
            raise ValueError("问题原因不适用于当前反馈对象")
        return self