from fastapi.responses import JSONResponse from pydantic import BaseModel, Field from typing import List, Dict, Any, Optional class MessageText(BaseModel): text: list[str] = Field(default_factory=list) class Message(BaseModel): text: MessageText | None = None payload: dict | None = None class FulfillmentResponse(BaseModel): messages: list[Message] = Field(default_factory=list) class SessionInfo(BaseModel): parameters: dict[str, Any] = Field(default_factory=dict) class DialogflowResponse(BaseModel): fulfillment_response: FulfillmentResponse sessionInfo: Optional[SessionInfo] = Field(default_factory=dict) def DialogFlowResponseAPI( text: list[str] = None, payload: dict | None = None, parameters: dict | None = None, ): messages = [] if text: messages.append(Message(text=MessageText(text=text))) if payload: messages.append(Message(payload=payload)) response_data = DialogflowResponse( fulfillment_response=FulfillmentResponse(messages=messages), sessionInfo=SessionInfo(parameters=parameters or {}), ) return JSONResponse(content=response_data.model_dump()) def dialogflow_response( text: list[str] | None = None, payload: dict | None = None, parameters: dict | None = None, ) -> JSONResponse: messages: List[Message] = [] if text: messages.append(Message(text=MessageText(text=text))) if payload: messages.append(Message(payload=payload)) response = DialogflowResponse( fulfillment_response=FulfillmentResponse(messages=messages), sessionInfo=SessionInfo(parameters=parameters or {}), ) return JSONResponse(content=response.model_dump(mode="json")) def dialogflow_error(message: str) -> JSONResponse: return dialogflow_response(text=[message]) def dialogflow_choice( prompt: str, options: list[str], parameters: dict | None = None ) -> JSONResponse: chips = { "richContent": [[{"type": "chips", "options": [{"text": o} for o in options]}]] } return dialogflow_response(text=[prompt], payload=chips, parameters=parameters) class DialogflowResponseBuilder: def __init__(self): self._messages: list[Message] = [] self._parameters: dict = {} def add_text(self, lines: list[str]) -> "DialogflowResponseBuilder": if lines: self._messages.append(Message(text=MessageText(text=lines))) return self def add_payload(self, payload: dict) -> "DialogflowResponseBuilder": if payload: self._messages.append(Message(payload=payload)) return self def add_choice_chips( self, options: list[str] | list[dict], key: str = None, prompt: str | None = None, ) -> "DialogflowResponseBuilder": if options: chips = { "richContent": [ [ { "type": "chips", "options": [ ( {"text": option.get(key, "")} if isinstance(option, dict) else {"text": option} ) for option in options ], } ] ] } self.add_payload(chips) if prompt: self.add_text([prompt]) return self def set_parameters(self, parameters: dict[str, any]) -> "DialogflowResponseBuilder": if parameters: self._parameters = parameters return self def build(self) -> JSONResponse: response = DialogflowResponse( fulfillment_response=FulfillmentResponse(messages=self._messages), sessionInfo=SessionInfo(parameters=self._parameters), ) return JSONResponse(content=response.model_dump(mode="json")) ## Use Builder return ( # DialogflowResponseBuilder() # .add_choice_chips(["Xem vé", "Đặt vé"], prompt="Bạn muốn làm gì tiếp?") # .set_parameters({"next_step": "booking"}) # .build() # ) class DialogServiceResult(BaseModel): text: list[str] | None = None options: list[str] | list[dict] | None = None parameters: dict | None = None