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}"