338 lines
14 KiB
Python
338 lines
14 KiB
Python
"""Order Agent - Handles order intake, validation, and categorization."""
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Dict, List, Any, Optional
|
|
|
|
from core.agent import SpecializedAgent
|
|
from core.types import (
|
|
AgentTask, MessageType, Priority, OrderStatus, ZoneType
|
|
)
|
|
from core.logger import logger
|
|
from core.http_client import api_get, api_post, api_patch
|
|
from config.system_config import GO_API_BASE_URL
|
|
|
|
|
|
class OrderAgent(SpecializedAgent):
|
|
"""
|
|
Order Agent - Manages the entire order lifecycle from intake to validation.
|
|
|
|
Writes go to the Doormile Go backend (POST /api/v1/admin/crmbooking).
|
|
Reads and status updates also go through the Go API.
|
|
"""
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
agent_id="ORDER_AGENT",
|
|
domain="order_management",
|
|
description="Handles order intake, validation, and categorization"
|
|
)
|
|
self._validation_rules = self._init_validation_rules()
|
|
self._pincode_zones = self._init_pincode_mapping()
|
|
|
|
def _init_validation_rules(self) -> Dict[str, Any]:
|
|
return {
|
|
"required_fields": ["customer_name", "customer_phone", "pickup_address", "delivery_address", "items"],
|
|
"phone_pattern": r'^\+?[1-9]\d{9,14}$',
|
|
"email_pattern": r'^[\w\.-]+@[\w\.-]+\.\w+$',
|
|
"pincode_length": 6,
|
|
"max_items_per_order": 100,
|
|
"max_weight_kg": 500,
|
|
}
|
|
|
|
def _init_pincode_mapping(self) -> Dict[str, Any]:
|
|
return {
|
|
"100": {"zone": "north", "region": "delhi_ncr", "hub_prefix": "DL"},
|
|
"101": {"zone": "north", "region": "delhi_ncr", "hub_prefix": "DL"},
|
|
"200": {"zone": "south", "region": "hyderabad", "hub_prefix": "HY"},
|
|
"201": {"zone": "south", "region": "bangalore", "hub_prefix": "BL"},
|
|
"400": {"zone": "west", "region": "mumbai", "hub_prefix": "MU"},
|
|
"401": {"zone": "west", "region": "pune", "hub_prefix": "PU"},
|
|
"500": {"zone": "central", "region": "bhopal", "hub_prefix": "BH"},
|
|
"600": {"zone": "east", "region": "kolkata", "hub_prefix": "KL"},
|
|
"700": {"zone": "northeast", "region": "guwahati", "hub_prefix": "GH"},
|
|
"800": {"zone": "northwest", "region": "jaipur", "hub_prefix": "JP"},
|
|
}
|
|
|
|
async def handle_task(self, task: AgentTask) -> Dict[str, Any]:
|
|
handlers = {
|
|
"receive_order": self._receive_order,
|
|
"validate_order": self._validate_order,
|
|
"categorize_order": self._categorize_order,
|
|
"update_status": self._update_status,
|
|
"cancel_order": self._cancel_order,
|
|
"get_order": self._get_order,
|
|
"list_orders": self._list_orders,
|
|
}
|
|
handler = handlers.get(task.task_type, self._unknown_task)
|
|
return await handler(task)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Go API helpers (use shared session + retry from http_client) #
|
|
# ------------------------------------------------------------------ #
|
|
|
|
async def _api_post(self, path: str, payload: Dict) -> Optional[Dict]:
|
|
result = await api_post(f"{GO_API_BASE_URL}{path}", json=payload)
|
|
if result is None:
|
|
logger.error(f"Go API POST {path} returned no response")
|
|
return result
|
|
|
|
async def _api_get(self, path: str, params: Dict = None) -> Optional[Dict]:
|
|
result = await api_get(f"{GO_API_BASE_URL}{path}", params=params)
|
|
if result is None:
|
|
logger.error(f"Go API GET {path} returned no response")
|
|
return result
|
|
|
|
async def _api_patch(self, path: str, payload: Dict) -> Optional[Dict]:
|
|
result = await api_patch(f"{GO_API_BASE_URL}{path}", json=payload)
|
|
if result is None:
|
|
logger.error(f"Go API PATCH {path} returned no response")
|
|
return result
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Task handlers #
|
|
# ------------------------------------------------------------------ #
|
|
|
|
async def _receive_order(self, task: AgentTask) -> Dict[str, Any]:
|
|
order_data = task.data.get("order", {})
|
|
|
|
order_id = f"ORD-{datetime.now().strftime('%Y%m%d')}-{uuid.uuid4().hex[:8].upper()}"
|
|
order_data["order_id"] = order_id
|
|
order_data["status"] = OrderStatus.RECEIVED.value
|
|
order_data["created_at"] = datetime.now().isoformat()
|
|
|
|
logger.info(f"Order Agent: Received order {order_id} | customer={order_data.get('customer_name')} items={len(order_data.get('items', []))}")
|
|
|
|
validation_result = await self._validate_order_data(order_data)
|
|
order_data["validation"] = validation_result
|
|
|
|
category = self._categorize_order_data(order_data)
|
|
order_data["category"] = category
|
|
|
|
api_result = await self._api_post("/api/v1/admin/crmbooking", order_data)
|
|
if api_result is None:
|
|
logger.warning(f"Failed to persist order {order_id} to Go backend")
|
|
else:
|
|
go_id = (api_result.get("data") or {}).get("bookingid")
|
|
if go_id:
|
|
order_id = str(go_id)
|
|
order_data["order_id"] = order_id
|
|
order_data["booking_id"] = order_id
|
|
logger.info(f"Go API booking ID: {order_id}")
|
|
|
|
await self.send_message(
|
|
recipient="JARVIS",
|
|
message_type=MessageType.ORDER_RECEIVED,
|
|
payload={
|
|
"order_id": order_id,
|
|
"booking_id": order_id,
|
|
"customer": order_data.get("customer_name"),
|
|
"priority": category.get("priority"),
|
|
"zone": category.get("zone"),
|
|
"validation_status": validation_result.get("status"),
|
|
},
|
|
correlation_id=order_id,
|
|
)
|
|
|
|
return {
|
|
"status": "received",
|
|
"order_id": order_id,
|
|
"validation": validation_result,
|
|
"category": category,
|
|
}
|
|
|
|
async def _validate_order(self, task: AgentTask) -> Dict[str, Any]:
|
|
order_id = task.data.get("order_id")
|
|
|
|
order_data = await self._api_get(f"/api/v1/admin/crmbooking/{order_id}")
|
|
if not order_data:
|
|
return {"status": "error", "message": f"Order {order_id} not found"}
|
|
|
|
validation = await self._validate_order_data(order_data)
|
|
|
|
if validation.get("status") == "valid":
|
|
await self._api_patch(
|
|
f"/api/v1/admin/crmbooking/{order_id}",
|
|
{"status": OrderStatus.VALIDATED.value, "validation": validation},
|
|
)
|
|
|
|
return validation
|
|
|
|
async def _categorize_order(self, task: AgentTask) -> Dict[str, Any]:
|
|
order_id = task.data.get("order_id")
|
|
order_data = await self._api_get(f"/api/v1/admin/crmbooking/{order_id}")
|
|
if not order_data:
|
|
return {"status": "error", "message": f"Order {order_id} not found"}
|
|
return self._categorize_order_data(order_data)
|
|
|
|
async def _update_status(self, task: AgentTask) -> Dict[str, Any]:
|
|
order_id = task.data.get("order_id")
|
|
new_status = task.data.get("status")
|
|
|
|
result = await self._api_patch(
|
|
f"/api/v1/admin/crmbooking/{order_id}",
|
|
{"status": new_status, "status_updated_at": datetime.now().isoformat()},
|
|
)
|
|
if not result:
|
|
return {"status": "error", "message": f"Failed to update order {order_id}"}
|
|
|
|
logger.info(f"Order {order_id} status -> {new_status}")
|
|
|
|
await self.send_message(
|
|
recipient="CUSTOMER_AGENT",
|
|
message_type=MessageType.ORDER_STATUS_UPDATE,
|
|
payload={"order_id": order_id, "new_status": new_status},
|
|
)
|
|
|
|
return {"status": "updated", "order_id": order_id, "new_status": new_status}
|
|
|
|
async def _cancel_order(self, task: AgentTask) -> Dict[str, Any]:
|
|
order_id = task.data.get("order_id")
|
|
reason = task.data.get("reason", "Customer requested cancellation")
|
|
|
|
result = await self._api_patch(
|
|
f"/api/v1/admin/crmbooking/{order_id}",
|
|
{
|
|
"status": OrderStatus.CANCELLED.value,
|
|
"cancellation_reason": reason,
|
|
"cancelled_at": datetime.now().isoformat(),
|
|
},
|
|
)
|
|
if not result:
|
|
return {"status": "error", "message": f"Failed to cancel order {order_id}"}
|
|
|
|
logger.info(f"Order {order_id} cancelled: {reason}")
|
|
|
|
await self.broadcast(
|
|
message_type=MessageType.ORDER_CANCELLED,
|
|
payload={"order_id": order_id, "reason": reason},
|
|
correlation_id=order_id,
|
|
)
|
|
|
|
return {"status": "cancelled", "order_id": order_id, "reason": reason}
|
|
|
|
async def _get_order(self, task: AgentTask) -> Dict[str, Any]:
|
|
order_id = task.data.get("order_id")
|
|
order_data = await self._api_get(f"/api/v1/admin/crmbooking/{order_id}")
|
|
if not order_data:
|
|
return {"status": "error", "message": f"Order {order_id} not found"}
|
|
return order_data
|
|
|
|
async def _list_orders(self, task: AgentTask) -> Dict[str, Any]:
|
|
params = {}
|
|
if task.data.get("status"):
|
|
params["status"] = task.data["status"]
|
|
if task.data.get("priority"):
|
|
params["priority"] = task.data["priority"]
|
|
|
|
result = await self._api_get("/api/v1/admin/crmbooking", params=params)
|
|
if not result:
|
|
return {"total": 0, "orders": []}
|
|
|
|
orders = result if isinstance(result, list) else result.get("data", [])
|
|
return {"total": len(orders), "orders": orders}
|
|
|
|
async def _unknown_task(self, task: AgentTask) -> Dict[str, Any]:
|
|
return {"status": "error", "message": f"Unknown task type: {task.task_type}"}
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Validation / categorization (local, no DB) #
|
|
# ------------------------------------------------------------------ #
|
|
|
|
async def _validate_order_data(self, order_data: Dict) -> Dict[str, Any]:
|
|
errors = []
|
|
warnings = []
|
|
|
|
for field in self._validation_rules["required_fields"]:
|
|
if field not in order_data or not order_data[field]:
|
|
errors.append(f"Missing required field: {field}")
|
|
|
|
phone = order_data.get("customer_phone", "")
|
|
if phone and not self._is_valid_phone(phone):
|
|
errors.append("Invalid phone number format")
|
|
|
|
email = order_data.get("customer_email")
|
|
if email and not self._is_valid_email(email):
|
|
warnings.append("Invalid email format")
|
|
|
|
pickup_pincode = order_data.get("pickup_address", {}).get("pincode", "")
|
|
if not self._is_valid_pincode(pickup_pincode):
|
|
errors.append("Invalid pickup pincode")
|
|
|
|
delivery_pincode = order_data.get("delivery_address", {}).get("pincode", "")
|
|
if not self._is_valid_pincode(delivery_pincode):
|
|
errors.append("Invalid delivery pincode")
|
|
|
|
items = order_data.get("items", [])
|
|
if not items:
|
|
errors.append("No items in order")
|
|
elif len(items) > self._validation_rules["max_items_per_order"]:
|
|
errors.append(f"Too many items (max {self._validation_rules['max_items_per_order']})")
|
|
|
|
total_weight = sum(item.get("weight", 0) for item in items)
|
|
if total_weight > self._validation_rules["max_weight_kg"]:
|
|
warnings.append(f"Heavy order ({total_weight}kg) - may require special handling")
|
|
|
|
return {
|
|
"status": "valid" if not errors else "invalid",
|
|
"errors": errors,
|
|
"warnings": warnings,
|
|
"total_weight_kg": total_weight,
|
|
}
|
|
|
|
def _categorize_order_data(self, order_data: Dict) -> Dict[str, Any]:
|
|
pickup_pincode = order_data.get("pickup_address", {}).get("pincode", "")[:3]
|
|
delivery_pincode = order_data.get("delivery_address", {}).get("pincode", "")[:3]
|
|
|
|
if pickup_pincode == delivery_pincode:
|
|
zone_type = ZoneType.LAST_MILE.value
|
|
else:
|
|
pickup_zone = self._pincode_zones.get(pickup_pincode, {})
|
|
delivery_zone = self._pincode_zones.get(delivery_pincode, {})
|
|
if pickup_zone.get("zone") == delivery_zone.get("zone"):
|
|
zone_type = ZoneType.HUB_TO_SPOKE.value
|
|
else:
|
|
zone_type = ZoneType.HUB_TO_HUB.value
|
|
|
|
priority = order_data.get("priority", Priority.MEDIUM.value)
|
|
total_weight = sum(item.get("weight", 0) for item in order_data.get("items", []))
|
|
if total_weight > 50 or order_data.get("special_instructions"):
|
|
priority = Priority.HIGH.value
|
|
|
|
pickup_info = self._pincode_zones.get(pickup_pincode, {"zone": "unknown", "region": "unknown"})
|
|
|
|
return {
|
|
"priority": priority,
|
|
"zone_type": zone_type,
|
|
"pickup_zone": pickup_info.get("zone"),
|
|
"delivery_zone": self._pincode_zones.get(delivery_pincode, {}).get("zone"),
|
|
"pickup_region": pickup_info.get("region"),
|
|
"estimated_hubs": self._get_nearest_hubs(pickup_pincode),
|
|
"requires_cold_chain": self._check_cold_chain_requirement(order_data),
|
|
"is_fragile": self._check_fragile_items(order_data),
|
|
}
|
|
|
|
def _get_nearest_hubs(self, pincode: str) -> List[str]:
|
|
zone = self._pincode_zones.get(pincode[:3], {})
|
|
hub_prefix = zone.get("hub_prefix", "XX")
|
|
return [f"{hub_prefix}-HUB-01", f"{hub_prefix}-HUB-02"]
|
|
|
|
def _check_cold_chain_requirement(self, order_data: Dict) -> bool:
|
|
return any(item.get("requires_cold_chain") for item in order_data.get("items", []))
|
|
|
|
def _check_fragile_items(self, order_data: Dict) -> bool:
|
|
return any(item.get("fragile") for item in order_data.get("items", []))
|
|
|
|
def _is_valid_phone(self, phone: str) -> bool:
|
|
import re
|
|
return bool(re.match(self._validation_rules["phone_pattern"], phone))
|
|
|
|
def _is_valid_email(self, email: str) -> bool:
|
|
import re
|
|
return bool(re.match(self._validation_rules["email_pattern"], email))
|
|
|
|
def _is_valid_pincode(self, pincode: str) -> bool:
|
|
return len(pincode) == self._validation_rules["pincode_length"] and pincode.isdigit()
|
|
|
|
async def think(self, context: str, options: List[str] = None) -> str:
|
|
return f"[ORDER_AGENT reasoning]: {context}"
|