chendandan_homework/app/services/finalization_service.py

from __future__ import annotations

"""定稿渲染服务：将审核通过的行程方案渲染为面向最终用户的Markdown格式旅行方案文本。"""

from app.schemas.itinerary import (
    ItineraryPlan,
)
from app.schemas.review import (
    TripReviewResult,
)
from app.schemas.travel_request import (
    TravelRequest,
)
from app.schemas.validation import (
    PlanValidationResult,
)


ACTIVITY_LABELS = {
    "flight": "航班",
    "hotel_check_in": "酒店入住",
    "hotel_check_out": "酒店退房",
    "attraction": "景点",
    "restaurant": "餐饮",
    "transport": "交通",
    "free_time": "自由活动",
    "other": "其他",
}


class FinalizationService:
    """将已审核的结构化方案渲染为最终Markdown。"""

    def render(
        self,
        *,
        request: TravelRequest,
        plan: ItineraryPlan,
        validation: PlanValidationResult,
        review: TripReviewResult,
        approved: bool,
        planning_attempts: int,
        supervisor_reason: str,
    ) -> str:
        """渲染最终旅行方案：根据请求、方案、校验结果和评审结论，分段生成需求摘要、航班、酒店、每日行程、预算与风险说明。"""
        lines: list[str] = []

        if approved:
            lines.append("# 个性化旅行方案")
        else:
            lines.append(
                "# 个性化旅行方案（存在未解决风险）"
            )

            lines.extend(
                [
                    "",
                    "> 当前方案已达到最大修改次数，"
                    "但尚未完全通过校验或审查。"
                    "请重点阅读文末风险说明。",
                ]
            )

        lines.extend(
            [
                "",
                f"## {plan.title}",
                "",
                plan.overview,
                "",
                "### 需求摘要",
                "",
                (
                    f"- 行程：{request.origin_city}"
                    f" → {request.destination_city}"
                ),
                (
                    f"- 日期：{request.departure_date}"
                    f" 至 {request.return_date}"
                ),
                (
                    f"- 人数：{request.adults}位成人，"
                    f"{request.children}位儿童"
                ),
                (
                    f"- 节奏：{request.pace}"
                ),
                (
                    "- 总预算："
                    + (
                        f"{request.total_budget:.0f}"
                        f" {request.currency}"
                        if request.total_budget
                        is not None
                        else "未设置"
                    )
                ),
            ]
        )

        self._append_flight(lines, plan)
        self._append_hotel(lines, plan)
        self._append_days(lines, plan)
        self._append_budget(
            lines,
            validation,
        )
        self._append_review(
            lines,
            review,
        )

        combined_warnings = self._collect_warnings(
            plan=plan,
            validation=validation,
            review=review,
            approved=approved,
        )

        if combined_warnings:
            lines.extend(
                [
                    "",
                    "### 风险与注意事项",
                    "",
                ]
            )

            for warning in combined_warnings:
                lines.append(f"- {warning}")

        if plan.assumptions:
            lines.extend(
                [
                    "",
                    "### 方案假设",
                    "",
                ]
            )

            for assumption in plan.assumptions:
                lines.append(f"- {assumption}")

        lines.extend(
            [
                "",
                "### 系统执行信息",
                "",
                (
                    f"- Planner执行次数："
                    f"{planning_attempts}"
                ),
                (
                    "- 最终状态："
                    + (
                        "校验与审查通过"
                        if approved
                        else "达到最大修改次数后输出"
                    )
                ),
                (
                    f"- Supervisor说明："
                    f"{supervisor_reason}"
                ),
            ]
        )

        return "\n".join(lines)

    def _append_flight(
        self,
        lines: list[str],
        plan: ItineraryPlan,
    ) -> None:
        """向输出文本追加航班推荐段落。"""
        flight = plan.selected_flight

        outbound_numbers = "、".join(
            flight.outbound_flight_numbers
        ) or "未提供航班号"

        return_numbers = "、".join(
            flight.return_flight_numbers
        ) or "未提供航班号"

        lines.extend(
            [
                "",
                "### 推荐航班",
                "",
                (
                    f"- 方案ID："
                    f"{flight.combination_id or flight.outbound_option_id}"
                ),
                (
                    f"- 去程：{outbound_numbers}"
                ),
                (
                    "- 去程时间："
                    f"{flight.outbound_departure_time or '未知'}"
                    " → "
                    f"{flight.outbound_arrival_time or '未知'}"
                ),
            ]
        )

        if flight.flight_type == "round_trip":
            lines.extend(
                [
                    (
                        f"- 返程：{return_numbers}"
                    ),
                    (
                        "- 返程时间："
                        f"{flight.return_departure_time or '未知'}"
                        " → "
                        f"{flight.return_arrival_time or '未知'}"
                    ),
                ]
            )

        lines.extend(
            [
                (
                    "- 接口报价："
                    + (
                        f"{flight.quoted_price:.0f}"
                        f" {flight.currency}"
                        if flight.quoted_price
                        is not None
                        else "未知"
                    )
                ),
                (
                    f"- 选择原因："
                    f"{flight.selection_reason}"
                ),
            ]
        )

    def _append_hotel(
        self,
        lines: list[str],
        plan: ItineraryPlan,
    ) -> None:
        """向输出文本追加酒店推荐段落。"""
        hotel = plan.selected_hotel

        lines.extend(
            [
                "",
                "### 推荐住宿",
                "",
                f"- 酒店：{hotel.name}",
                f"- 酒店ID：{hotel.hotel_id}",
                (
                    f"- 入住：{hotel.check_in_date}"
                    f" 至 {hotel.check_out_date}"
                    f"，共{hotel.nights}晚"
                ),
                (
                    "- 每晚价格："
                    + (
                        f"{hotel.price_per_night:.0f}"
                        f" {hotel.currency}"
                        if hotel.price_per_night
                        is not None
                        else "未知"
                    )
                ),
                (
                    "- 酒店估算："
                    + (
                        f"{hotel.estimated_total_price:.0f}"
                        f" {hotel.currency}"
                        if hotel.estimated_total_price
                        is not None
                        else "未知"
                    )
                ),
                (
                    f"- 地址："
                    f"{hotel.address or '未提供'}"
                ),
                (
                    f"- 选择原因："
                    f"{hotel.selection_reason}"
                ),
            ]
        )

    def _append_days(
        self,
        lines: list[str],
        plan: ItineraryPlan,
    ) -> None:
        """向输出文本追加每日行程段落。"""
        lines.extend(
            [
                "",
                "### 每日行程",
            ]
        )

        for day in sorted(
            plan.days,
            key=lambda item: item.day_index,
        ):
            lines.extend(
                [
                    "",
                    (
                        f"#### 第{day.day_index}天｜"
                        f"{day.date}｜{day.theme}"
                    ),
                    "",
                ]
            )

            activities = sorted(
                day.activities,
                key=lambda item: (
                    item.start_time,
                    item.sequence,
                ),
            )

            for activity in activities:
                activity_label = (
                    ACTIVITY_LABELS.get(
                        activity.activity_type,
                        activity.activity_type,
                    )
                )

                activity_line = (
                    f"- {activity.start_time.strftime('%H:%M')}"
                    f"–{activity.end_time.strftime('%H:%M')}"
                    f"｜{activity_label}｜"
                    f"{activity.name}"
                )

                lines.append(activity_line)

                details: list[str] = []

                if activity.address:
                    details.append(
                        f"地址：{activity.address}"
                    )

                if (
                    activity.transport_mode
                    and activity.transport_mode
                    != "unknown"
                ):
                    details.append(
                        "交通方式："
                        f"{activity.transport_mode}"
                    )

                if (
                    activity
                    .estimated_transport_minutes
                    is not None
                ):
                    details.append(
                        "预计交通："
                        f"{activity.estimated_transport_minutes}"
                        "分钟"
                    )

                if activity.notes:
                    details.append(
                        f"说明：{activity.notes}"
                    )

                if details:
                    lines.append(
                        "  - " + "；".join(details)
                    )

            for note in day.daily_notes:
                lines.append(f"- 当日提示：{note}")

    def _append_budget(
        self,
        lines: list[str],
        validation: PlanValidationResult,
    ) -> None:
        """向输出文本追加预算估算段落。"""
        budget = validation.budget

        lines.extend(
            [
                "",
                "### 已知预算估算",
                "",
                (
                    "- 航班接口报价："
                    + self._money(
                        budget.flight_quote,
                        budget.currency,
                    )
                ),
                (
                    "- 酒店估算："
                    + self._money(
                        budget.hotel_estimate,
                        budget.currency,
                    )
                ),
                (
                    "- 已明确的活动费用："
                    + self._money(
                        budget.explicit_activity_costs,
                        budget.currency,
                    )
                ),
                (
                    "- 当前已知费用合计："
                    + self._money(
                        budget.known_total,
                        budget.currency,
                    )
                ),
                (
                    "- 用户预算："
                    + self._money(
                        budget.user_budget,
                        budget.currency,
                    )
                ),
                (
                    "- 当前已知预算余额："
                    + self._money(
                        budget.remaining_known_budget,
                        budget.currency,
                    )
                ),
                "",
                f"> {budget.coverage_note}",
            ]
        )

    def _append_review(
        self,
        lines: list[str],
        review: TripReviewResult,
    ) -> None:
        """向输出文本追加独立审查结论段落。"""
        lines.extend(
            [
                "",
                "### 独立审查结论",
                "",
                f"- {review.summary}",
            ]
        )

        if review.strengths:
            lines.append("- 方案优点：")

            for strength in review.strengths:
                lines.append(
                    f"  - {strength}"
                )

        if review.recommendations:
            lines.append("- 出行前建议：")

            for recommendation in (
                review.recommendations
            ):
                lines.append(
                    f"  - {recommendation}"
                )

    def _collect_warnings(
        self,
        *,
        plan: ItineraryPlan,
        validation: PlanValidationResult,
        review: TripReviewResult,
        approved: bool,
    ) -> list[str]:
        """汇总方案、校验和审查三个阶段的警告信息并去重。"""
        warnings: list[str] = []

        warnings.extend(plan.warnings)

        for issue in validation.issues:
            if (
                issue.severity == "warning"
                or not approved
            ):
                warnings.append(issue.message)

        for issue in review.issues:
            if (
                issue.severity == "warning"
                or not approved
            ):
                warnings.append(issue.message)

        if not approved:
            warnings.extend(
                review.revision_feedback
            )

        result: list[str] = []

        for warning in warnings:
            normalized = warning.strip()

            if (
                normalized
                and normalized not in result
            ):
                result.append(normalized)

        return result

    @staticmethod
    def _money(
        value: float | None,
        currency: str,
    ) -> str:
        """把金额（float|None）和货币代码格式化为"数值 货币"的中文展示文本。"""
        if value is None:
            return "未知"

        return f"{value:.0f} {currency}"