video-image-production-build/tests/preprocess/test_production_brief.py

from __future__ import annotations

import json
import copy
import tempfile
import unittest
from pathlib import Path

from production_build_agents.contracts.models import IssueKind, IssueSeverity
from production_build_agents.preprocess.production_brief import (
    preprocess_file,
    preprocess_payload,
)


GOLDEN_DIR = Path(__file__).parents[1] / "golden_case"
INPUT_PATH = (
    Path(__file__).resolve().parents[2]
    / "容易内耗的人其实需要停止悲伤者叙事#自我(690788)_production_final.json"
)
EXPECTED_BRIEF_PATH = GOLDEN_DIR / "production_brief.json"
EXPECTED_REPORT_PATH = GOLDEN_DIR / "preprocess_report.md"


class GoldenCaseTest(unittest.TestCase):
    def test_real_input_matches_golden_outputs(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            output_path = Path(temp_dir) / "production_brief.json"
            report_path = Path(temp_dir) / "preprocess_report.md"

            outcome = preprocess_file(INPUT_PATH, output_path, report_path)

            self.assertTrue(outcome.can_proceed)
            self.assertEqual(
                output_path.read_text(encoding="utf-8"),
                EXPECTED_BRIEF_PATH.read_text(encoding="utf-8"),
            )
            self.assertEqual(
                report_path.read_text(encoding="utf-8"),
                EXPECTED_REPORT_PATH.read_text(encoding="utf-8"),
            )

    def test_main_json_contains_no_report_or_ignored_source_fields(self) -> None:
        brief = json.loads(EXPECTED_BRIEF_PATH.read_text(encoding="utf-8"))

        self.assertEqual(
            set(brief),
            {
                "schema_version",
                "帖子类型",
                "核心制作点",
                "制作表",
                "source_assets",
            },
        )
        serialized = EXPECTED_BRIEF_PATH.read_text(encoding="utf-8")
        for ignored_name in ("帖子ID", "原始文本", "原始视频路径", "report"):
            self.assertNotIn(f'"{ignored_name}"', serialized)

    def test_source_assets_cover_and_deduplicate_full_retained_brief(self) -> None:
        brief = json.loads(EXPECTED_BRIEF_PATH.read_text(encoding="utf-8"))
        source_assets = brief["source_assets"]

        self.assertEqual(len(source_assets), 7)
        self.assertEqual(
            len(
                {
                    (item["artifact_type"], item["source_uri"])
                    for item in source_assets
                }
            ),
            7,
        )
        character = next(
            item
            for item in source_assets
            if item["source_uri"].endswith(
                "d20b24da7f704398a15a46b3e3711858.png"
            )
        )
        self.assertEqual(
            character["source_paths"],
            [
                "$.制作表.段落结构[1]",
                "$.核心制作点[0]",
                "$.核心制作点[14]",
            ],
        )
        self.assertTrue(
            any(
                "$.制作表.形式结果[47]" in item["source_paths"]
                for item in source_assets
            )
        )

    def test_selected_business_structure_only_has_deterministic_normalization(self) -> None:
        source = json.loads(INPUT_PATH.read_text(encoding="utf-8"))
        brief = json.loads(EXPECTED_BRIEF_PATH.read_text(encoding="utf-8"))

        self.assertEqual(brief["帖子类型"], source["帖子类型"])
        self.assertEqual(brief["核心制作点"], source["核心制作点"])
        self.assertEqual(
            brief["制作表"]["段落结构"],
            source["制作表"]["段落结构"],
        )
        self.assertEqual(
            brief["制作表"]["形式结果"],
            source["制作表"]["形式结果"],
        )

        expected_relationships = copy.deepcopy(source["制作表"]["关系结果"])
        original_text = expected_relationships[0]["global_relationships"][4]["自然语言描述"]
        expected_relationships[0]["global_relationships"][4]["自然语言描述"] = (
            original_text.replace("16:9", "9:16")
        )
        self.assertEqual(
            brief["制作表"]["关系结果"],
            expected_relationships,
        )
        self.assertNotIn(
            "16:9的视频画幅",
            EXPECTED_BRIEF_PATH.read_text(encoding="utf-8"),
        )

    def test_golden_case_reports_only_the_aspect_ratio_conflict(self) -> None:
        payload = json.loads(INPUT_PATH.read_text(encoding="utf-8"))
        outcome = preprocess_payload(payload)

        self.assertEqual(len(outcome.issues), 1)
        issue = outcome.issues[0]
        self.assertEqual(issue.kind, IssueKind.CONFLICT)
        self.assertEqual(issue.severity, IssueSeverity.WARNING)
        self.assertEqual([value.value for value in issue.values], ["9:16", "16:9"])

    def test_same_input_produces_identical_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            first_json = root / "first.json"
            first_md = root / "first.md"
            second_json = root / "second.json"
            second_md = root / "second.md"

            preprocess_file(INPUT_PATH, first_json, first_md)
            preprocess_file(INPUT_PATH, second_json, second_md)

            self.assertEqual(first_json.read_bytes(), second_json.read_bytes())
            self.assertEqual(first_md.read_bytes(), second_md.read_bytes())

    def test_time_range_is_normalized_by_code(self) -> None:
        payload = {
            "帖子类型": "video",
            "核心制作点": [],
            "制作表": {
                "段落结构": [
                    {
                        "段落ID": "段落1",
                        "内容范围": {
                            "时间轴": ["0:0:1.5 - 0:0:3.25"],
                        },
                    }
                ],
                "形式结果": [],
                "关系结果": [],
            },
        }

        outcome = preprocess_payload(payload)

        self.assertTrue(outcome.can_proceed)
        self.assertEqual(
            outcome.brief.production_table.segment_structure[0]["内容范围"]["时间轴"][0],
            "00:00:01.500-00:00:03.250",
        )


class ContractErrorTest(unittest.TestCase):
    def test_critical_error_writes_report_but_refuses_brief(self) -> None:
        invalid_payload = {
            "帖子类型": "video",
            "制作表": {
                "段落结构": [],
                "形式结果": [],
                "关系结果": [],
            },
        }

        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            input_path = root / "input.json"
            output_path = root / "production_brief.json"
            report_path = root / "preprocess_report.md"
            input_path.write_text(
                json.dumps(invalid_payload, ensure_ascii=False),
                encoding="utf-8",
            )
            output_path.write_text("上一次成功结果", encoding="utf-8")

            outcome = preprocess_file(input_path, output_path, report_path)

            self.assertFalse(outcome.can_proceed)
            self.assertFalse(output_path.exists())
            self.assertTrue(report_path.exists())
            self.assertIn("不可进入 Planner", report_path.read_text(encoding="utf-8"))

    def test_conflicting_structured_aspect_ratios_block_planner(self) -> None:
        payload = {
            "帖子类型": "video",
            "核心制作点": [],
            "制作表": {
                "段落结构": [],
                "形式结果": [
                    {
                        "段落ID": "段落1",
                        "属性维度": [
                            {"维度名": "画幅比例", "维度值": "9:16"},
                        ],
                    },
                    {
                        "段落ID": "段落2",
                        "属性维度": [
                            {"维度名": "画幅比例", "维度值": "16:9"},
                        ],
                    },
                ],
                "关系结果": [],
            },
        }

        outcome = preprocess_payload(payload)

        self.assertFalse(outcome.can_proceed)
        self.assertTrue(
            any(
                issue.kind == IssueKind.CONFLICT
                and issue.severity == IssueSeverity.ERROR
                and issue.source_path == "$.制作表.形式结果"
                for issue in outcome.issues
            )
        )

    def test_optional_relationship_results_becomes_null_and_does_not_block(self) -> None:
        payload = {
            "帖子类型": "video",
            "核心制作点": [],
            "制作表": {
                "段落结构": [],
                "形式结果": [],
            },
        }

        outcome = preprocess_payload(payload)

        self.assertTrue(outcome.can_proceed)
        self.assertIsNone(outcome.brief.production_table.relationship_results)
        self.assertTrue(
            any(
                issue.kind == IssueKind.MISSING_FIELD
                and issue.severity == IssueSeverity.WARNING
                and issue.source_path == "$.制作表.关系结果"
                for issue in outcome.issues
            )
        )

    def test_critical_field_becomes_null_and_blocks(self) -> None:
        payload = {
            "帖子类型": "video",
            "制作表": {
                "段落结构": [],
                "形式结果": [],
                "关系结果": [],
            },
        }

        outcome = preprocess_payload(payload)

        self.assertFalse(outcome.can_proceed)
        self.assertIsNone(outcome.brief.core_production_points)
        self.assertTrue(
            any(
                issue.kind == IssueKind.MISSING_FIELD
                and issue.severity == IssueSeverity.ERROR
                and issue.source_path == "$.核心制作点"
                for issue in outcome.issues
            )
        )

    def test_ignored_fields_are_never_validated_or_emitted(self) -> None:
        payload = {
            "帖子ID": {"任意": "错误类型也忽略"},
            "原始文本": 123,
            "原始视频路径": ["不是", "路径"],
            "帖子类型": "video",
            "核心制作点": [],
            "制作表": {
                "段落结构": [],
                "形式结果": [],
                "关系结果": [],
            },
        }

        outcome = preprocess_payload(payload)
        serialized = outcome.brief.model_dump()

        self.assertTrue(outcome.can_proceed)
        self.assertNotIn("帖子ID", serialized)
        self.assertNotIn("原始文本", serialized)
        self.assertNotIn("原始视频路径", serialized)


if __name__ == "__main__":
    unittest.main()