dreamweaver/backend/app/services/harness/quality_gates.py

"""Deterministic quality gates for generated child-facing content."""

from dataclasses import dataclass
from enum import StrEnum

from app.services.adapters.storybook.primary import Storybook
from app.services.adapters.text.models import StoryOutput
from app.services.harness.types import FailureCategory


class QualityGateCode(StrEnum):
    """Stable issue codes emitted by deterministic quality gates."""

    MISSING_TITLE = "missing_title"
    MISSING_STORY_TEXT = "missing_story_text"
    MISSING_COVER_PROMPT = "missing_cover_prompt"
    MISSING_STORYBOOK_PAGE = "missing_storybook_page"
    INVALID_STORYBOOK_PAGE_NUMBER = "invalid_storybook_page_number"
    MISSING_STORYBOOK_PAGE_TEXT = "missing_storybook_page_text"
    UNSAFE_CHILD_CONTENT = "unsafe_child_content"


@dataclass(frozen=True)
class QualityGateIssue:
    """One deterministic quality gate issue."""

    code: QualityGateCode
    message: str
    failure_category: FailureCategory = FailureCategory.SCHEMA_ERROR
    field: str | None = None

    def to_metadata(self) -> dict:
        """Return a JSON-safe metadata payload."""

        return {
            "code": self.code.value,
            "message": self.message,
            "failure_category": self.failure_category.value,
            "field": self.field,
        }


class QualityGateError(ValueError):
    """Raised when generated content fails deterministic quality gates."""

    def __init__(self, issues: list[QualityGateIssue]):
        self.issues = issues
        message = "；".join(issue.message for issue in issues)
        super().__init__(message)

    def to_metadata(self) -> dict:
        """Return a JSON-safe metadata payload."""

        return {"issues": [issue.to_metadata() for issue in self.issues]}


UNSAFE_CHILD_TERMS = (
    "自杀",
    "自残",
    "血腥",
    "虐待",
    "毒品",
    "色情",
)


def _is_blank(value: str | None) -> bool:
    return not value or not value.strip()


def _unsafe_issue_if_present(text: str, *, field: str) -> QualityGateIssue | None:
    for term in UNSAFE_CHILD_TERMS:
        if term in text:
            return QualityGateIssue(
                code=QualityGateCode.UNSAFE_CHILD_CONTENT,
                message="生成内容包含不适合 3-8 岁儿童的明显风险词。",
                failure_category=FailureCategory.SAFETY_ERROR,
                field=field,
            )
    return None


def validate_story_output(output: StoryOutput) -> None:
    """Validate generated text story output before persistence."""

    issues: list[QualityGateIssue] = []

    if _is_blank(output.title):
        issues.append(
            QualityGateIssue(
                code=QualityGateCode.MISSING_TITLE,
                message="故事标题为空。",
                field="title",
            )
        )

    if _is_blank(output.story_text):
        issues.append(
            QualityGateIssue(
                code=QualityGateCode.MISSING_STORY_TEXT,
                message="故事正文为空。",
                field="story_text",
            )
        )

    if _is_blank(output.cover_prompt_suggestion):
        issues.append(
            QualityGateIssue(
                code=QualityGateCode.MISSING_COVER_PROMPT,
                message="封面提示词为空。",
                field="cover_prompt_suggestion",
            )
        )

    unsafe_issue = _unsafe_issue_if_present(
        " ".join([output.title or "", output.story_text or ""]),
        field="story_text",
    )
    if unsafe_issue is not None:
        issues.append(unsafe_issue)

    if issues:
        raise QualityGateError(issues)


def validate_storybook_output(output: Storybook) -> None:
    """Validate generated storybook output before persistence."""

    issues: list[QualityGateIssue] = []

    if _is_blank(output.title):
        issues.append(
            QualityGateIssue(
                code=QualityGateCode.MISSING_TITLE,
                message="绘本标题为空。",
                field="title",
            )
        )

    if not output.pages:
        issues.append(
            QualityGateIssue(
                code=QualityGateCode.MISSING_STORYBOOK_PAGE,
                message="绘本至少需要一页内容。",
                field="pages",
            )
        )

    seen_page_numbers: set[int] = set()
    page_texts: list[str] = []
    for index, page in enumerate(output.pages, start=1):
        if not isinstance(page.page_number, int) or page.page_number <= 0:
            issues.append(
                QualityGateIssue(
                    code=QualityGateCode.INVALID_STORYBOOK_PAGE_NUMBER,
                    message=f"绘本第 {index} 个页面页码无效。",
                    field=f"pages[{index - 1}].page_number",
                )
            )
        elif page.page_number in seen_page_numbers:
            issues.append(
                QualityGateIssue(
                    code=QualityGateCode.INVALID_STORYBOOK_PAGE_NUMBER,
                    message=f"绘本页码 {page.page_number} 重复。",
                    field=f"pages[{index - 1}].page_number",
                )
            )
        else:
            seen_page_numbers.add(page.page_number)

        if _is_blank(page.text):
            issues.append(
                QualityGateIssue(
                    code=QualityGateCode.MISSING_STORYBOOK_PAGE_TEXT,
                    message=f"绘本第 {index} 页正文为空。",
                    field=f"pages[{index - 1}].text",
                )
            )
        else:
            page_texts.append(page.text)

    unsafe_issue = _unsafe_issue_if_present(
        " ".join([output.title or "", *page_texts]),
        field="pages",
    )
    if unsafe_issue is not None:
        issues.append(unsafe_issue)

    if issues:
        raise QualityGateError(issues)