| import logging |
| from decimal import Decimal, InvalidOperation |
| from typing import Any |
|
|
| logger = logging.getLogger(__name__) |
|
|
| _NO_DRIVER_ERROR = ( |
| "No driver account for this WhatsApp number. " |
| "Ask the sender to register with create_driver_account first." |
| ) |
|
|
| from app.database.supabase import SupabaseRepository |
| from app.models.domain import ToolResult |
| from app.services.embedding_service import JinaEmbeddingService |
| from app.services.trip_indexing import index_trip |
| from app.utils.departure import ( |
| _parse_date_value, |
| normalize_departure_bucket, |
| parse_departure_request, |
| parse_requested_clock_time, |
| trip_departure_bucket, |
| trip_departure_date, |
| trip_satisfies_departure_request, |
| ) |
| from app.utils.emoji import count_emojis |
| from app.whatsapp.client import WhatsAppClient, WhatsAppClientError |
| from app.whatsapp.trip_selection import format_trip_card |
|
|
|
|
| class FalzhToolHandlers: |
| def __init__( |
| self, |
| *, |
| repository: SupabaseRepository, |
| embeddings: JinaEmbeddingService, |
| whatsapp: WhatsAppClient, |
| customer: dict[str, Any], |
| remoteJid: str, |
| embedding_model: str, |
| current_message: dict[str, Any] | None = None, |
| ) -> None: |
| self.repository = repository |
| self.embeddings = embeddings |
| self.whatsapp = whatsapp |
| self.customer = customer |
| self.remoteJid = remoteJid |
| self.embedding_model = embedding_model |
| self.current_message = current_message |
|
|
| async def _resolve_trip_id_from_reply(self) -> str | None: |
| if not self.current_message: |
| return None |
| metadata = self.current_message.get("metadata") or {} |
| context_message_id = metadata.get("context_message_id") |
| if not context_message_id: |
| return None |
| original = await self.repository.get_message_by_whatsapp_id(context_message_id) |
| if not original: |
| return None |
| original_meta = original.get("metadata") or {} |
| return original_meta.get("trip_id") |
|
|
| async def about_falzh(self, arguments: dict[str, Any]) -> ToolResult: |
| query = str(arguments.get("query") or "").strip() |
| if not query: |
| return ToolResult(ok=False, data={}, error="query is required") |
|
|
| query_embedding = await self.embeddings.embed_query(query) |
| matches = await self.repository.search_info_chunks_by_vector( |
| query_embedding=query_embedding, |
| match_count=5, |
| ) |
| if not matches: |
| return ToolResult( |
| ok=True, |
| data={ |
| "answer": "No matching FALZH policy or FAQ content was found.", |
| "sources": [], |
| "reply_instruction": ( |
| "Apologize briefly. Do not guess or invent an answer." |
| ), |
| }, |
| ) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "answer_context": [ |
| { |
| "text": match.get("chunk_text") |
| or match.get("metadata", {}).get("chunk_text"), |
| "source": match.get("source") or match.get("metadata", {}).get("source"), |
| "score": match.get("score") or match.get("similarity"), |
| } |
| for match in matches |
| ], |
| "reply_instruction": ( |
| "Answer only from this context. If it does not clearly answer the user's " |
| "question, apologize briefly and say there is no confirmed information." |
| ), |
| }, |
| ) |
|
|
| async def search_trips(self, arguments: dict[str, Any]) -> ToolResult: |
| departure = _optional_string(arguments.get("departure")) |
| destination = _optional_string(arguments.get("destination")) |
| driver_name = _optional_string(arguments.get("driver_name")) |
| travel_date = _optional_string(arguments.get("travel_date")) |
| travel_time = _optional_string(arguments.get("travel_time")) |
| travel_time_exact = _optional_string(arguments.get("travel_time_exact")) |
| travel_datetime = _optional_string(arguments.get("travel_datetime")) |
| seats = _optional_int(arguments.get("seats")) or 1 |
| vehicle_type = _optional_string(arguments.get("vehicle_type")) |
| vector_query_text = _optional_string(arguments.get("vector_query_text")) |
| departure_request = parse_departure_request( |
| travel_date=travel_date, |
| travel_time=travel_time, |
| travel_datetime=travel_datetime, |
| ) |
| requested_time = parse_requested_clock_time( |
| travel_time=travel_time, |
| travel_datetime=travel_datetime, |
| exact_time=travel_time_exact, |
| ) |
|
|
| if not departure and not destination and not driver_name: |
| return ToolResult( |
| ok=False, |
| data={}, |
| error=( |
| "At least departure, destination, or driver name" |
| " is required before searching trips" |
| ), |
| ) |
|
|
| query = vector_query_text or _trip_vector_query_text( |
| departure=departure, |
| destination=destination, |
| driver_name=driver_name, |
| travel_date=travel_date, |
| travel_time=travel_time, |
| travel_time_exact=travel_time_exact, |
| travel_datetime=travel_datetime, |
| seats=seats, |
| vehicle_type=vehicle_type, |
| ) |
| query_embedding = await self.embeddings.embed_query(query) |
| trips = await self.repository.search_trips_by_vector( |
| query_embedding=query_embedding, |
| departure=departure, |
| destination=destination, |
| driver_name=driver_name, |
| departure_date=departure_request.departure_date, |
| departure_time=departure_request.departure_time, |
| requested_time=requested_time, |
| seats=seats, |
| vehicle_type=vehicle_type, |
| match_count=10, |
| ) |
| if not trips: |
| trips = await self.repository.search_active_trips( |
| departure=departure, |
| destination=destination, |
| driver_name=driver_name, |
| seats=seats, |
| vehicle_type=vehicle_type, |
| departure_request=departure_request, |
| ) |
|
|
| alternate_alert = _alternate_time_alert(trips) |
| filtered = _sort_trip_summaries([ |
| _trip_summary(trip) |
| for trip in trips |
| if _is_trip_match( |
| trip, |
| departure=departure, |
| destination=destination, |
| driver_name=driver_name, |
| seats=seats, |
| vehicle_type=vehicle_type, |
| departure_request=departure_request, |
| ) |
| ]) |
|
|
| top_trips = filtered[:5] |
|
|
| if top_trips: |
| for trip_summary in top_trips: |
| trip_id = trip_summary["trip_id"] |
| selection_count = await self.repository.count_trip_selections(trip_id) |
| trip_summary["selection_count"] = selection_count |
| trip = next((t for t in trips if (t.get("trip_id") or t.get("id")) == trip_id), {}) |
| trip["selection_count"] = selection_count |
| card = format_trip_card(trip) |
| try: |
| resp = await self.whatsapp.send_text(self.remoteJid, card) |
| wam_id = resp.get("messages", [{}])[0].get("id") |
| if wam_id: |
| await self.repository.create_message( |
| customer_id=str(self.customer["id"]), |
| sender_type="assistant", |
| message=card, |
| whatsapp_message_id=wam_id, |
| metadata={"trip_id": trip_id, "type": "trip_card"}, |
| ) |
| except WhatsAppClientError: |
| logger.warning("Failed to send trip card for trip %s", trip_id) |
|
|
| prompt = "يرجى الرد على إحدى بطاقات الرحلات أعلاه لاختيار رحلتك" |
| await self.whatsapp.send_text(self.remoteJid, prompt) |
| await self.repository.create_message( |
| customer_id=str(self.customer["id"]), |
| sender_type="assistant", |
| message=prompt, |
| metadata={"type": "trip_selection_prompt"}, |
| ) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "count": len(top_trips), |
| "matches": top_trips, |
| "alternate_alert": alternate_alert, |
| "sent_as_messages": True, |
| "note": "Cards sent. No text reply needed.", |
| }, |
| suppress_llm_reply=True, |
| ) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "count": 0, |
| "matches": [], |
| "alternate_alert": alternate_alert, |
| "sent_as_messages": False, |
| "note": "No active matching trips were found.", |
| "reply_instruction": _no_trip_reply_instruction(departure_request), |
| }, |
| ) |
|
|
| async def select_trip(self, arguments: dict[str, Any]) -> ToolResult: |
| trip_id = _optional_string(arguments.get("trip_id")) |
| requested_seats = _optional_int(arguments.get("requested_seats")) or 1 |
| notes = _optional_string(arguments.get("notes")) |
|
|
| if not trip_id: |
| trip_id = await self._resolve_trip_id_from_reply() |
| if not trip_id: |
| return ToolResult(ok=False, data={}, error="trip_id is required. Ask the user to reply to a trip card message or provide the trip ID.") |
| if requested_seats < 1: |
| return ToolResult(ok=False, data={}, error="requested_seats must be at least 1") |
|
|
| trip = await self.repository.get_trip_by_id(trip_id) |
| if not trip: |
| return ToolResult(ok=False, data={}, error="Trip was not found") |
| if trip.get("status") != "active": |
| return ToolResult(ok=False, data={}, error="Trip is not active") |
| available_seats = trip.get("available_seats") |
| if available_seats is not None and int(available_seats) < requested_seats: |
| return ToolResult( |
| ok=False, |
| data={"available_seats": trip.get("available_seats")}, |
| error="Not enough available seats", |
| ) |
|
|
| selection = await self.repository.create_trip_selection( |
| customer_id=str(self.customer["id"]), |
| trip_id=trip_id, |
| requested_seats=requested_seats, |
| notes=notes, |
| ) |
|
|
| driver_record = _first_or_dict(trip.get("drivers")) or {} |
| driver_customer = driver_record.get("customers") or {} |
|
|
| |
| driver_phone_raw = driver_customer.get("phone_number") |
| if driver_phone_raw and "/" in driver_phone_raw: |
| |
| driver_phone = driver_phone_raw.replace("/", "\n") |
| driver_recipient = driver_phone_raw.split("/")[0] |
| else: |
| |
| driver_recipient = driver_phone_raw or driver_customer.get("remoteJid") or driver_record.get("remoteJid") |
| driver_phone = driver_recipient.split("@")[0] if driver_recipient else None |
|
|
| notification_status = "not_sent" |
| notification_error = None |
|
|
| is_registered = _driver_is_registered(trip) |
|
|
| if is_registered and driver_recipient: |
| try: |
| await self.whatsapp.send_text( |
| driver_recipient, |
| _driver_notification_text( |
| customer=self.customer, |
| trip=trip, |
| requested_seats=requested_seats, |
| notes=notes, |
| ), |
| ) |
| notification_status = "sent" |
| except Exception as exc: |
| notification_status = "failed" |
| notification_error = str(exc) |
|
|
| await self.repository.update_selection_notification( |
| selection_id=str(selection["id"]), |
| status=notification_status, |
| metadata={"error": notification_error} if notification_error else None, |
| ) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "selection_id": selection["id"], |
| "status": "pending", |
| "driver_notification_status": notification_status, |
| "driver_notification_error": notification_error, |
| "driver_phone": driver_phone, |
| "message": "Customer interest recorded. Share driver number so the two parties can coordinate directly.", |
| }, |
| ) |
|
|
| async def create_driver_account(self, arguments: dict[str, Any]) -> ToolResult: |
| name = _optional_string(arguments.get("name")) |
| if not name: |
| return ToolResult(ok=False, data={}, error="name is required") |
|
|
| remote_jid = self.remoteJid |
| existing = await self.repository.get_driver_by_remoteJid(remote_jid) |
| if existing: |
| return ToolResult( |
| ok=False, |
| data={"driver_id": existing["id"]}, |
| error="Driver account already exists for this WhatsApp number", |
| ) |
|
|
| driver = await self.repository.create_driver(customer_id=str(self.customer["id"])) |
| return ToolResult( |
| ok=True, |
| data={ |
| "driver_id": driver["id"], |
| "name": self.customer.get("name"), |
| "remoteJid": self.customer.get("remoteJid"), |
| "message": "Driver account created successfully.", |
| "reply_instruction": ( |
| "Tell the driver in Arabic that they can publish each trip in their own " |
| "format/style and FALZH will extract the trip details automatically. Also " |
| "tell them personal data such as phone numbers will be removed from trip " |
| "ads/cards and shared with passengers only after a passenger selects the " |
| "trip and wants to contact them." |
| ), |
| }, |
| ) |
|
|
| async def _summarize_trips(self, trips: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| return [ |
| _trip_summary(trip, trip_number=index + 1) |
| for index, trip in enumerate(trips) |
| ] |
|
|
| async def check_driver_info(self, arguments: dict[str, Any]) -> ToolResult: |
| driver = await self.repository.get_driver_by_remoteJid(self.remoteJid) |
| if not driver: |
| return ToolResult( |
| ok=False, |
| data={"action": "create_driver_account"}, |
| error=_NO_DRIVER_ERROR, |
| ) |
|
|
| customer_info = driver.get("customers", {}) |
| cars = await self.repository.list_driver_cars(str(driver["id"])) |
| upcoming_trips = await self.repository.list_driver_trips(str(driver["id"])) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "driver_id": driver["id"], |
| "name": customer_info.get("name") or driver.get("name"), |
| "remoteJid": customer_info.get("remoteJid") or driver.get("remoteJid"), |
| "status": driver.get("status"), |
| "vehicle_count": len(cars), |
| "active_trip_count": len(upcoming_trips), |
| "vehicles": [ |
| { |
| "car_id": str(car.get("id")), |
| "name": car.get("car_type"), |
| "plate_number": car.get("plate_number"), |
| "seat_count": car.get("seat_count"), |
| } |
| for car in cars |
| ], |
| "active_trips": await self._summarize_trips(upcoming_trips), |
| }, |
| ) |
|
|
| async def check_driver_trips(self, arguments: dict[str, Any]) -> ToolResult: |
| driver = await self.repository.get_driver_by_remoteJid(self.remoteJid) |
| if not driver: |
| return ToolResult( |
| ok=False, |
| data={"action": "create_driver_account"}, |
| error=_NO_DRIVER_ERROR, |
| ) |
|
|
| trips = await self.repository.list_driver_trips(str(driver["id"])) |
|
|
| if not trips: |
| return ToolResult( |
| ok=True, |
| data={ |
| "count": 0, |
| "message": "No upcoming active trips found.", |
| }, |
| ) |
|
|
| for trip in trips[:5]: |
| trip_id = trip.get("trip_id") or trip.get("id") |
| selection_count = await self.repository.count_trip_selections(trip_id) |
| trip["selection_count"] = selection_count |
| card = format_trip_card(trip) |
| try: |
| resp = await self.whatsapp.send_text(self.remoteJid, card) |
| wam_id = resp.get("messages", [{}])[0].get("id") |
| if wam_id: |
| await self.repository.create_message( |
| customer_id=str(self.customer["id"]), |
| sender_type="assistant", |
| message=card, |
| whatsapp_message_id=wam_id, |
| metadata={"trip_id": trip_id, "type": "driver_trip_card"}, |
| ) |
| except WhatsAppClientError: |
| logger.warning("Failed to send trip card for trip %s", trip_id) |
|
|
| prompt = "هذه قائمة رحلاتك المسجلة" |
| await self.whatsapp.send_text(self.remoteJid, prompt) |
| await self.repository.create_message( |
| customer_id=str(self.customer["id"]), |
| sender_type="assistant", |
| message=prompt, |
| metadata={"type": "driver_trip_list_prompt"}, |
| ) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "count": len(trips), |
| "sent_as_messages": True, |
| "note": "Trip cards sent. No text reply needed.", |
| }, |
| suppress_llm_reply=True, |
| ) |
|
|
| async def add_driver_car(self, arguments: dict[str, Any]) -> ToolResult: |
| driver = await self.repository.get_driver_by_remoteJid(self.remoteJid) |
| if not driver: |
| return ToolResult( |
| ok=False, |
| data={"action": "create_driver_account"}, |
| error=_NO_DRIVER_ERROR, |
| ) |
|
|
| car_type = _optional_string(arguments.get("name")) |
| if not car_type: |
| return ToolResult(ok=False, data={}, error="name is required") |
|
|
| plate_number = _optional_string(arguments.get("plate_number")) |
| seat_count = _optional_int(arguments.get("seat_count")) |
| if seat_count is not None and seat_count < 1: |
| return ToolResult(ok=False, data={}, error="seat_count must be at least 1") |
|
|
| car = await self.repository.create_driver_car( |
| driver_id=str(driver["id"]), |
| car_type=car_type, |
| plate_number=plate_number, |
| seat_count=seat_count, |
| ) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "car_id": car.get("id"), |
| "name": car.get("car_type"), |
| "plate_number": car.get("plate_number"), |
| "seat_count": car.get("seat_count"), |
| "message": "Driver vehicle registered successfully.", |
| }, |
| ) |
|
|
| async def add_trip_by_driver(self, arguments: dict[str, Any]) -> ToolResult: |
| driver = await self.repository.get_driver_by_remoteJid(self.remoteJid) |
| if not driver: |
| return ToolResult( |
| ok=False, |
| data={"action": "create_driver_account"}, |
| error=_NO_DRIVER_ERROR, |
| ) |
|
|
| departure = _optional_string(arguments.get("departure")) |
| destination = _optional_string(arguments.get("destination")) |
| if not departure or not destination: |
| return ToolResult( |
| ok=False, |
| data={}, |
| error="departure and destination are required", |
| ) |
|
|
| parsed_date = _parse_date_value(arguments.get("departure_date")) |
| if not parsed_date: |
| return ToolResult( |
| ok=False, |
| data={}, |
| error="departure_date is required as YYYY-MM-DD", |
| ) |
|
|
| departure_time = normalize_departure_bucket(arguments.get("departure_time")) |
| if not departure_time: |
| return ToolResult( |
| ok=False, |
| data={}, |
| error="departure_time must be morning, noon, night, or Arabic صباح / ظهر / ليل", |
| ) |
|
|
| existing = await self.repository.get_driver_trip_by_datetime( |
| driver_id=str(driver["id"]), |
| departure_date=parsed_date, |
| departure_time=departure_time, |
| ) |
| if existing: |
| return ToolResult( |
| ok=False, |
| data={"existing_trip_id": existing.get("id")}, |
| error=( |
| "You already have a trip scheduled at this date and time. " |
| "Cancel or modify the existing trip before creating a new one." |
| ), |
| ) |
|
|
| latest_trip = await self.repository.get_driver_latest_trip(str(driver["id"])) |
| cars = await self.repository.list_driver_cars(str(driver["id"])) |
|
|
| vehicle_type = _optional_string(arguments.get("vehicle_type")) |
| matched_car = _resolve_driver_car(cars, vehicle_type=vehicle_type) |
| if matched_car is None and latest_trip: |
| matched_car = _resolve_driver_car( |
| cars, |
| vehicle_type=None, |
| car_id=_optional_string(latest_trip.get("car_id")), |
| ) |
| if matched_car is None and len(cars) == 1: |
| matched_car = cars[0] |
|
|
| car_id = str(matched_car["id"]) if matched_car else None |
|
|
| available_seats = _optional_int(arguments.get("available_seats")) |
| if available_seats is None and latest_trip is not None: |
| available_seats = _optional_int(latest_trip.get("available_seats")) |
|
|
| total_seats = _optional_int(arguments.get("total_seats")) |
| if total_seats is None and latest_trip is not None: |
| total_seats = _optional_int(latest_trip.get("total_seats")) |
|
|
| if total_seats is None and matched_car is not None: |
| total_seats = _optional_int(matched_car.get("seat_count")) |
|
|
| price = _optional_price(arguments.get("price")) |
| if price is None and latest_trip is not None: |
| price = _optional_price(latest_trip.get("price")) |
|
|
| if price is None: |
| price = 0 |
|
|
| if not matched_car: |
| if vehicle_type: |
| return ToolResult( |
| ok=False, |
| data={}, |
| error=( |
| f"No registered vehicle matches '{vehicle_type}'. " |
| "Ask the driver to use the exact car type or plate from their account." |
| ), |
| ) |
| return ToolResult( |
| ok=False, |
| data={}, |
| error="No registered vehicle found for this driver", |
| ) |
|
|
| if total_seats is not None and total_seats < 1: |
| return ToolResult(ok=False, data={}, error="total_seats must be at least 1") |
| if available_seats is not None and available_seats < 0: |
| return ToolResult(ok=False, data={}, error="available_seats must be at least 0") |
| if total_seats is not None and available_seats is not None and available_seats > total_seats: |
| return ToolResult( |
| ok=False, |
| data={}, |
| error="available_seats cannot exceed total_seats", |
| ) |
| if price is None or price < 0: |
| return ToolResult(ok=False, data={}, error="price must be zero or greater") |
|
|
| driver_message: str | None = None |
| use_driver_message = False |
| if self.current_message: |
| raw_msg = self.current_message.get("message") or "" |
| llm_message = _optional_string(arguments.get("driver_message")) |
| if llm_message or count_emojis(raw_msg) > 2: |
| driver_message = llm_message or raw_msg |
| use_driver_message = True |
|
|
| trip = await self.repository.create_driver_trip( |
| driver_id=str(driver["id"]), |
| car_id=car_id, |
| departure=departure, |
| destination=destination, |
| departure_date=parsed_date, |
| departure_time=departure_time, |
| available_seats=available_seats, |
| total_seats=total_seats, |
| price=price, |
| driver_message=driver_message, |
| use_driver_message=use_driver_message, |
| ) |
|
|
| await index_trip( |
| repository=self.repository, |
| embeddings=self.embeddings, |
| embedding_model=self.embedding_model, |
| trip=trip, |
| ) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "trip_id": trip.get("id"), |
| "departure": trip.get("departure"), |
| "destination": trip.get("destination"), |
| "departure_date": parsed_date.isoformat(), |
| "departure_time": departure_time, |
| "available_seats": available_seats, |
| "total_seats": total_seats, |
| "price": price, |
| "car_id": car_id, |
| "indexed": True, |
| "message": "Trip created and indexed for search.", |
| }, |
| ) |
|
|
| async def initiate_trip_action(self, arguments: dict[str, Any]) -> ToolResult: |
| action_type = _optional_string(arguments.get("action_type")) |
| if action_type not in {"DELETE", "MODIFY"}: |
| return ToolResult( |
| ok=False, |
| data={}, |
| error="action_type must be DELETE or MODIFY", |
| ) |
|
|
| driver = await self.repository.get_driver_by_remoteJid(self.remoteJid) |
| if not driver: |
| return ToolResult( |
| ok=False, |
| data={"action": "create_driver_account"}, |
| error=_NO_DRIVER_ERROR, |
| ) |
|
|
| trips = await self.repository.list_driver_trips(str(driver["id"])) |
| travel_date = _optional_string(arguments.get("travel_date")) |
| travel_time = _optional_string(arguments.get("travel_time")) |
| departure_request = parse_departure_request( |
| travel_date=travel_date, |
| travel_time=travel_time, |
| ) |
| if departure_request.departure_date or departure_request.departure_time: |
| trips = [ |
| trip |
| for trip in trips |
| if trip_satisfies_departure_request(trip, departure_request) |
| ] |
|
|
| if not trips: |
| return ToolResult( |
| ok=True, |
| data={ |
| "count": 0, |
| "message": "No trips found. Ask the user for clarification.", |
| }, |
| ) |
|
|
| action_label = "حذفها" if action_type == "DELETE" else "تعديلها" |
| for trip in trips[:5]: |
| trip_id = trip.get("trip_id") or trip.get("id") |
| selection_count = await self.repository.count_trip_selections(trip_id) |
| trip["selection_count"] = selection_count |
| card = format_trip_card(trip) |
| try: |
| resp = await self.whatsapp.send_text(self.remoteJid, card) |
| wam_id = resp.get("messages", [{}])[0].get("id") |
| if wam_id: |
| await self.repository.create_message( |
| customer_id=str(self.customer["id"]), |
| sender_type="assistant", |
| message=card, |
| whatsapp_message_id=wam_id, |
| metadata={"trip_id": trip_id, "type": "driver_trip_card", "action": action_type}, |
| ) |
| except WhatsAppClientError: |
| logger.warning("Failed to send trip card for trip %s", trip_id) |
|
|
| prompt = f"قم بالرد على بطاقة الرحلة التي تريد {action_label}" |
| await self.whatsapp.send_text(self.remoteJid, prompt) |
| await self.repository.create_message( |
| customer_id=str(self.customer["id"]), |
| sender_type="assistant", |
| message=prompt, |
| metadata={"type": "driver_trip_selection_prompt"}, |
| ) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "count": len(trips), |
| "action_type": action_type, |
| "sent_as_messages": True, |
| "note": "Trip cards sent. No text reply needed.", |
| }, |
| suppress_llm_reply=True, |
| ) |
|
|
| async def update_trip_field(self, arguments: dict[str, Any]) -> ToolResult: |
| driver = await self.repository.get_driver_by_remoteJid(self.remoteJid) |
| if not driver: |
| return ToolResult( |
| ok=False, |
| data={"action": "create_driver_account"}, |
| error=_NO_DRIVER_ERROR, |
| ) |
|
|
| session = await self.repository.get_customer_session(str(self.customer["id"])) |
| trip_id = _optional_string(session.get("active_edit_trip_id")) |
| if not trip_id: |
| return ToolResult( |
| ok=False, |
| data={}, |
| error=( |
| "No trip is selected for editing. " |
| "Use initiate_trip_action with MODIFY first." |
| ), |
| ) |
|
|
| trip = await self.repository.get_trip_by_id(trip_id) |
| if not trip: |
| return ToolResult(ok=False, data={}, error="Trip was not found") |
| if str(trip.get("driver_id")) != str(driver["id"]): |
| return ToolResult(ok=False, data={}, error="Trip does not belong to this driver") |
| if trip.get("status") != "active": |
| return ToolResult(ok=False, data={}, error="Trip is not active") |
|
|
| field = _optional_string(arguments.get("field")) |
| value = _optional_string(arguments.get("value")) |
| if not field or value is None: |
| return ToolResult(ok=False, data={}, error="field and value are required") |
|
|
| updates, error = await self._build_trip_field_update( |
| driver_id=str(driver["id"]), |
| field=field, |
| value=value, |
| ) |
| if error: |
| return ToolResult(ok=False, data={}, error=error) |
|
|
| updated_trip = await self.repository.update_driver_trip(trip_id, updates) |
| await index_trip( |
| repository=self.repository, |
| embeddings=self.embeddings, |
| embedding_model=self.embedding_model, |
| trip=updated_trip, |
| ) |
| await self.repository.clear_customer_session_field( |
| customer_id=str(self.customer["id"]), |
| key="active_edit_trip_id", |
| ) |
|
|
| return ToolResult( |
| ok=True, |
| data={ |
| "trip_id": trip_id, |
| "field": field, |
| "value": value, |
| "trip": _trip_summary(updated_trip), |
| "message": "Trip updated successfully.", |
| }, |
| ) |
|
|
| async def _build_trip_field_update( |
| self, |
| *, |
| driver_id: str, |
| field: str, |
| value: str, |
| ) -> tuple[dict[str, Any], str | None]: |
| if field in {"departure", "destination"}: |
| return {field: value}, None |
|
|
| if field in {"departure_time", "pickup_time"}: |
| bucket = normalize_departure_bucket(value) |
| if not bucket: |
| return {}, "departure_time must be morning, noon, night, or HH:MM" |
| return {"departure_time": bucket}, None |
|
|
| if field == "vehicle_type": |
| cars = await self.repository.list_driver_cars(driver_id) |
| matched_car = _resolve_driver_car(cars, vehicle_type=value) |
| if not matched_car: |
| return {}, f"No registered vehicle matches '{value}'" |
| return {"car_id": str(matched_car["id"])}, None |
|
|
| if field in {"available_seats", "total_seats"}: |
| seats = _optional_int(value) |
| if seats is None: |
| return {}, f"{field} must be an integer" |
| if field == "available_seats" and seats < 0: |
| return {}, "available_seats must be at least 0" |
| if field == "total_seats" and seats < 1: |
| return {}, "total_seats must be at least 1" |
| return {field: seats}, None |
|
|
| if field == "price": |
| price = _optional_price(value) |
| if price is None or price < 0: |
| return {}, "price must be zero or greater" |
| return {"price": price}, None |
|
|
| return {}, f"Unsupported field: {field}" |
|
|
| async def switch_to_driver(self, arguments: dict[str, Any]) -> ToolResult: |
| driver = await self.repository.get_driver_by_remoteJid(self.remoteJid) |
| if not driver: |
| return ToolResult( |
| ok=False, |
| data={"action": "create_driver_account"}, |
| error=( |
| "No driver account for this WhatsApp number. " |
| "Use create_driver_account first, then switch_to_driver." |
| ), |
| ) |
|
|
| await self.repository.update_customer_user_mode( |
| customer_id=str(self.customer["id"]), |
| user_mode="driver", |
| ) |
| self.customer["user_mode"] = "driver" |
| return ToolResult( |
| ok=True, |
| data={ |
| "user_mode": "driver", |
| "driver_id": driver["id"], |
| "message": "Switched to driver mode.", |
| }, |
| ) |
|
|
| async def switch_to_passenger(self, arguments: dict[str, Any]) -> ToolResult: |
| name = _optional_string(arguments.get("name")) |
| if name: |
| await self.repository.update_customer_name( |
| customer_id=str(self.customer["id"]), |
| name=name, |
| ) |
| self.customer["name"] = name |
|
|
| await self.repository.update_customer_user_mode( |
| customer_id=str(self.customer["id"]), |
| user_mode="passenger", |
| ) |
| self.customer["user_mode"] = "passenger" |
| return ToolResult( |
| ok=True, |
| data={ |
| "user_mode": "passenger", |
| "name": self.customer.get("name"), |
| "message": "Switched to passenger mode.", |
| }, |
| ) |
|
|
|
|
| def _optional_string(value: Any) -> str | None: |
| if value is None: |
| return None |
| text = str(value).strip() |
| return text or None |
|
|
|
|
| def _optional_int(value: Any) -> int | None: |
| if value is None or value == "": |
| return None |
| return int(value) |
|
|
|
|
| def _optional_price(value: Any) -> float | None: |
| if value is None or value == "": |
| return None |
| try: |
| return float(Decimal(str(value))) |
| except (InvalidOperation, ValueError): |
| return None |
|
|
|
|
| def _resolve_driver_car( |
| cars: list[dict[str, Any]], |
| *, |
| vehicle_type: str | None = None, |
| car_id: str | None = None, |
| ) -> dict[str, Any] | None: |
| if car_id: |
| for car in cars: |
| if str(car.get("id")) == car_id: |
| return car |
| return None |
|
|
| if not vehicle_type: |
| return None |
|
|
| query = vehicle_type.lower() |
| matches = [ |
| car |
| for car in cars |
| if query in str(car.get("car_type") or "").lower() |
| or query in str(car.get("plate_number") or "").lower() |
| ] |
| if len(matches) == 1: |
| return matches[0] |
| if len(matches) > 1: |
| return None |
| return None |
|
|
|
|
| def _trip_vector_query_text( |
| *, |
| departure: str | None, |
| destination: str | None, |
| driver_name: str | None, |
| travel_date: str | None, |
| travel_time: str | None, |
| travel_time_exact: str | None, |
| travel_datetime: str | None, |
| seats: int, |
| vehicle_type: str | None, |
| ) -> str: |
| return " ".join( |
| part |
| for part in [ |
| departure, |
| destination, |
| driver_name, |
| travel_date, |
| travel_time_exact, |
| travel_time, |
| travel_datetime, |
| f"{seats} seats", |
| vehicle_type, |
| ] |
| if part |
| ) |
|
|
|
|
| def _is_trip_match( |
| trip: dict[str, Any], |
| *, |
| departure: str | None, |
| destination: str | None, |
| driver_name: str | None, |
| seats: int, |
| vehicle_type: str | None, |
| departure_request: Any, |
| ) -> bool: |
| if trip.get("status") != "active": |
| return False |
| if trip.get("available_seats") is not None and int(trip["available_seats"]) < seats: |
| return False |
| if departure and departure.lower() not in str(trip.get("departure") or "").lower(): |
| return False |
| if destination and destination.lower() not in str(trip.get("destination") or "").lower(): |
| return False |
| if driver_name: |
| trip_driver_name = str(trip.get("driver_name") or "").lower() |
| if driver_name.lower() not in trip_driver_name: |
| return False |
| if vehicle_type: |
| car = _first_or_dict(trip.get("driver_cars")) or {} |
| car_type = car.get("car_type") or trip.get("car_type") |
| if vehicle_type.lower() not in str(car_type or "").lower(): |
| return False |
| if not trip_satisfies_departure_request(trip, departure_request): |
| return False |
| return True |
|
|
|
|
| def _trip_summary(trip: dict[str, Any], *, trip_number: int | None = None) -> dict[str, Any]: |
| driver = _first_or_dict(trip.get("drivers")) or {} |
| car = _first_or_dict(trip.get("driver_cars")) or {} |
| summary = { |
| "trip_id": trip.get("trip_id") or trip.get("id"), |
| "departure": trip.get("departure"), |
| "destination": trip.get("destination"), |
| "departure_date": ( |
| parsed_date.isoformat() if (parsed_date := trip_departure_date(trip)) else None |
| ), |
| "departure_time": trip.get("departure_time"), |
| "departure_time_type": trip_departure_bucket(trip), |
| "available_seats": trip.get("available_seats"), |
| "total_seats": trip.get("total_seats"), |
| "price": trip.get("price"), |
| "driver_name": driver.get("name") or trip.get("driver_name"), |
| "car_type": car.get("car_type") or trip.get("car_type"), |
| "status": trip.get("status"), |
| "similarity": trip.get("similarity"), |
| "time_difference_minutes": trip.get("time_difference_minutes"), |
| "selection_count": trip.get("selection_count", 0), |
| "registered": _driver_is_registered(trip), |
| } |
| if trip_number is not None: |
| summary["trip_number"] = trip_number |
| return summary |
|
|
|
|
| def _sort_trip_summaries(trips: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| bucket_order = {"morning": 0, "noon": 1, "night": 2} |
| return sorted( |
| trips, |
| key=lambda trip: ( |
| not trip.get("registered", False), |
| str(trip.get("departure_date") or ""), |
| bucket_order.get(str(trip.get("departure_time_type") or ""), 99), |
| ), |
| ) |
|
|
|
|
| def _first_or_dict(value: Any) -> dict[str, Any] | None: |
| if isinstance(value, list): |
| return value[0] if value else None |
| if isinstance(value, dict): |
| return value |
| return None |
|
|
|
|
| def _driver_is_registered(trip: dict[str, Any]) -> bool: |
| drivers = trip.get("drivers") |
| if isinstance(drivers, list): |
| drivers = drivers[0] if drivers else {} |
| if not isinstance(drivers, dict): |
| return False |
| customer = drivers.get("customers") |
| if isinstance(customer, list): |
| customer = customer[0] if customer else {} |
| if not isinstance(customer, dict): |
| return False |
| return bool(customer.get("registered", False)) |
|
|
|
|
| def _alternate_time_alert(trips: list[dict[str, Any]]) -> str | None: |
| if not trips: |
| return None |
| raw_difference = trips[0].get("time_difference_minutes") |
| if raw_difference is None: |
| return None |
| try: |
| difference = int(raw_difference) |
| except (TypeError, ValueError): |
| return None |
| if difference <= 60: |
| return None |
| return ( |
| "The closest available trip is more than 60 minutes away from the requested time. " |
| "Mention that it is an alternate time before listing the options." |
| ) |
|
|
|
|
| def _no_trip_reply_instruction(departure_request: Any) -> str: |
| instruction = ( |
| "Tell the passenger briefly that no active matching trip is currently available. " |
| "Do not promise to notify them when a future matching trip is added." |
| ) |
| if departure_request.departure_date or departure_request.departure_time: |
| instruction += ( |
| " Do not call search_trips again with a narrower/specific time for the same " |
| "route unless the passenger provides new route/date information." |
| ) |
| else: |
| instruction += ( |
| " This was already a general search for the provided trip details; do not call " |
| "search_trips again only to add a specific time for the same trip." |
| ) |
| return instruction |
|
|
|
|
| def _trip_search_note(alternate_alert: str | None) -> str: |
| if alternate_alert: |
| return alternate_alert |
| return "Trips are available for handoff only; seats are not reserved yet." |
|
|
|
|
| def _driver_notification_text( |
| *, |
| customer: dict[str, Any], |
| trip: dict[str, Any], |
| requested_seats: int, |
| notes: str | None, |
| ) -> str: |
| return ( |
| "🔔 اهتمام جديد برحلة في فلزة\n" |
| f"العميل: {customer.get('name') or 'عميل جديد'}\n" |
| f"رقم العميل: {customer.get('phone_number') or 'غير متوفر'}\n" |
| f"الرحلة: {trip.get('departure')} ← {trip.get('destination')}\n" |
| f"التاريخ: {trip_departure_date(trip)} {trip_departure_bucket(trip)}\n" |
| f"المقاعد المطلوبة: {requested_seats}\n" |
| f"ملاحظات: {notes or 'لا يوجد'}\n" |
| "يرجى التواصل مع العميل للاتفاق على الحجز" |
| ) |
|
|