whFalsa / app /tools /handlers.py
codeBOKER's picture
Allow drivers to publish trips in their own style; add reply_instruction for no-match results; improve privacy prompts
c925062
Raw
History Blame Contribute Delete
42.7 kB
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 {}
# For unregistered drivers, phone_number may contain multiple numbers separated by /
driver_phone_raw = driver_customer.get("phone_number")
if driver_phone_raw and "/" in driver_phone_raw:
# Multiple phone numbers - each on its own line
driver_phone = driver_phone_raw.replace("/", "\n")
driver_recipient = driver_phone_raw.split("/")[0]
else:
# Single phone number - use existing logic
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: # noqa: BLE001
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"
"يرجى التواصل مع العميل للاتفاق على الحجز"
)