INTRODUCTION AND SYSTEM OVERVIEW
Package tracking has become an essential part of modern logistics and e-commerce. Users need a reliable way to monitor their shipments across multiple delivery services without visiting different websites or applications. This article describes the architecture and implementation of a comprehensive package tracking tool that aggregates tracking information from various delivery services, maintains a centralized list of all packages, and employs an intelligent agent to monitor updates continuously.
The system we will build consists of several key components working together. The core tracking engine interfaces with multiple delivery service APIs to fetch real-time tracking information. A database layer persists package information and tracking history. An intelligent monitoring agent runs continuously in the background, checking for updates and detecting handoffs between carriers. A notification system alerts users about status changes. Finally, a large language model integration layer provides natural language understanding and generation capabilities, supporting both local and remote LLM deployments across various GPU architectures.
The tool accepts user input including the sender or recipient name, tracking number, delivery service identifier, and an optional package nickname. It queries the appropriate tracking API, retrieves comprehensive tracking data, stores this information locally, and presents it to the user. The background agent periodically polls all active packages, detects status changes, identifies carrier handoffs, and notifies users of important events.
ARCHITECTURAL FOUNDATIONS
The architecture follows clean architecture principles with clear separation of concerns. The innermost layer contains domain entities representing packages, tracking events, and delivery services. The use case layer implements business logic for adding packages, querying tracking information, and processing updates. The interface adapter layer translates between external APIs and internal representations. The outermost layer handles infrastructure concerns like database access, API clients, and LLM integration.
This layered approach ensures testability, maintainability, and flexibility. Each layer depends only on inner layers, never on outer ones. This allows us to swap implementations without affecting core business logic. For example, we can switch from one database to another or change LLM providers without modifying tracking logic.
The system uses dependency injection to manage component relationships. This makes testing easier and allows runtime configuration of different implementations. We can inject mock API clients during testing or switch between local and remote LLM implementations based on user preferences and available hardware.
DOMAIN MODEL AND ENTITIES
The domain model represents the core concepts of package tracking. A Package entity contains all information about a shipment. This includes the tracking number, current carrier, sender and recipient names, optional nickname, current status, and complete tracking history. Each Package has a unique identifier generated when first added to the system.
A TrackingEvent represents a single status update in the package journey. It contains a timestamp, location, status description, and event type. Events are immutable once created and form a chronological history of the package movement. The event type indicates whether this is a pickup, in-transit update, delivery attempt, delivery confirmation, or exception.
A DeliveryService entity encapsulates information about a carrier. It includes the service name, API endpoint configuration, authentication credentials, and rate limiting parameters. The system maintains a registry of supported delivery services that can be extended as new carriers are added.
Here is the core domain model:
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional
from enum import Enum
from uuid import UUID, uuid4
class EventType(Enum):
PICKUP = "pickup"
IN_TRANSIT = "in_transit"
OUT_FOR_DELIVERY = "out_for_delivery"
DELIVERY_ATTEMPT = "delivery_attempt"
DELIVERED = "delivered"
EXCEPTION = "exception"
RETURNED = "returned"
CARRIER_HANDOFF = "carrier_handoff"
class PackageStatus(Enum):
PENDING = "pending"
IN_TRANSIT = "in_transit"
OUT_FOR_DELIVERY = "out_for_delivery"
DELIVERED = "delivered"
EXCEPTION = "exception"
RETURNED = "returned"
@dataclass
class TrackingEvent:
timestamp: datetime
location: str
description: str
event_type: EventType
carrier: str
event_id: UUID = field(default_factory=uuid4)
metadata: dict = field(default_factory=dict)
def __post_init__(self):
if isinstance(self.timestamp, str):
self.timestamp = datetime.fromisoformat(self.timestamp)
if isinstance(self.event_type, str):
self.event_type = EventType(self.event_type)
@dataclass
class Package:
tracking_number: str
carrier: str
user_name: str
nickname: Optional[str] = None
status: PackageStatus = PackageStatus.PENDING
tracking_history: List[TrackingEvent] = field(default_factory=list)
package_id: UUID = field(default_factory=uuid4)
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
next_carrier: Optional[str] = None
next_tracking_number: Optional[str] = None
is_active: bool = True
def add_event(self, event: TrackingEvent):
self.tracking_history.append(event)
self.updated_at = datetime.now()
self._update_status_from_event(event)
def _update_status_from_event(self, event: TrackingEvent):
status_mapping = {
EventType.PICKUP: PackageStatus.IN_TRANSIT,
EventType.IN_TRANSIT: PackageStatus.IN_TRANSIT,
EventType.OUT_FOR_DELIVERY: PackageStatus.OUT_FOR_DELIVERY,
EventType.DELIVERED: PackageStatus.DELIVERED,
EventType.EXCEPTION: PackageStatus.EXCEPTION,
EventType.RETURNED: PackageStatus.RETURNED,
}
if event.event_type in status_mapping:
self.status = status_mapping[event.event_type]
def detect_handoff(self, event: TrackingEvent) -> bool:
if event.event_type == EventType.CARRIER_HANDOFF:
if 'next_carrier' in event.metadata:
self.next_carrier = event.metadata['next_carrier']
if 'next_tracking_number' in event.metadata:
self.next_tracking_number = event.metadata['next_tracking_number']
return True
return False
@dataclass
class DeliveryService:
name: str
api_endpoint: str
api_key: Optional[str] = None
rate_limit_per_minute: int = 60
supports_handoff_detection: bool = False
service_id: UUID = field(default_factory=uuid4)
This domain model captures all essential information without depending on any external frameworks or infrastructure. The Package class includes methods for adding events and detecting carrier handoffs. The status is automatically updated based on event types. The model supports both active tracking and archived packages through the is_active flag.
DELIVERY SERVICE INTEGRATION LAYER
Integrating with multiple delivery services requires a flexible adapter pattern. Each delivery service has its own API structure, authentication mechanism, and data format. We create a common interface that all delivery service adapters must implement. This interface defines methods for fetching tracking information, authenticating with the service, and parsing responses into our domain model.
The base adapter interface ensures consistency across all implementations. Each concrete adapter translates between the delivery service's specific API and our internal representation. This abstraction allows the rest of the system to work with packages uniformly regardless of which carrier is handling the shipment.
Different carriers use different authentication methods. Some require API keys in headers, others use OAuth tokens, and some use basic authentication. The adapter handles these details internally. Rate limiting is also managed at the adapter level to prevent exceeding carrier API quotas.
Here is the adapter interface and a sample implementation:
from abc import ABC, abstractmethod
from typing import List, Optional
import requests
from datetime import datetime
import time
class DeliveryServiceAdapter(ABC):
def __init__(self, service: DeliveryService):
self.service = service
self.last_request_time = 0
self.request_count = 0
@abstractmethod
def fetch_tracking_info(self, tracking_number: str) -> List[TrackingEvent]:
pass
@abstractmethod
def authenticate(self) -> bool:
pass
def _rate_limit(self):
current_time = time.time()
if current_time - self.last_request_time < 60:
self.request_count += 1
if self.request_count >= self.service.rate_limit_per_minute:
sleep_time = 60 - (current_time - self.last_request_time)
time.sleep(sleep_time)
self.request_count = 0
self.last_request_time = time.time()
else:
self.request_count = 1
self.last_request_time = current_time
@abstractmethod
def detect_handoff(self, events: List[TrackingEvent]) -> Optional[dict]:
pass
class USPSAdapter(DeliveryServiceAdapter):
def authenticate(self) -> bool:
# USPS uses API key authentication
return self.service.api_key is not None
def fetch_tracking_info(self, tracking_number: str) -> List[TrackingEvent]:
if not self.authenticate():
raise ValueError("USPS adapter not authenticated")
self._rate_limit()
url = f"{self.service.api_endpoint}/TrackV2"
params = {
'API': 'TrackV2',
'XML': f'<TrackRequest USERID="{self.service.api_key}"><TrackID ID="{tracking_number}"></TrackID></TrackRequest>'
}
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
return self._parse_response(response.text, tracking_number)
def _parse_response(self, xml_response: str, tracking_number: str) -> List[TrackingEvent]:
import xml.etree.ElementTree as ET
events = []
root = ET.fromstring(xml_response)
for track_info in root.findall('.//TrackDetail'):
event_time = track_info.find('EventTime')
event_date = track_info.find('EventDate')
event_desc = track_info.find('Event')
event_city = track_info.find('EventCity')
event_state = track_info.find('EventState')
if all([event_time, event_date, event_desc]):
timestamp_str = f"{event_date.text} {event_time.text}"
timestamp = datetime.strptime(timestamp_str, "%B %d, %Y %I:%M %p")
location = f"{event_city.text if event_city is not None else ''}, {event_state.text if event_state is not None else ''}".strip(', ')
event_type = self._classify_event(event_desc.text)
event = TrackingEvent(
timestamp=timestamp,
location=location,
description=event_desc.text,
event_type=event_type,
carrier="USPS"
)
events.append(event)
return sorted(events, key=lambda e: e.timestamp)
def _classify_event(self, description: str) -> EventType:
description_lower = description.lower()
if 'delivered' in description_lower:
return EventType.DELIVERED
elif 'out for delivery' in description_lower:
return EventType.OUT_FOR_DELIVERY
elif 'picked up' in description_lower or 'acceptance' in description_lower:
return EventType.PICKUP
elif 'exception' in description_lower or 'delay' in description_lower:
return EventType.EXCEPTION
elif 'returned' in description_lower:
return EventType.RETURNED
elif 'transferred' in description_lower or 'forwarded' in description_lower:
return EventType.CARRIER_HANDOFF
else:
return EventType.IN_TRANSIT
def detect_handoff(self, events: List[TrackingEvent]) -> Optional[dict]:
for event in reversed(events):
if event.event_type == EventType.CARRIER_HANDOFF:
# USPS often includes new tracking number in description
description = event.description.lower()
if 'tracking number' in description:
# Extract tracking number using pattern matching
import re
match = re.search(r'tracking number[:\s]+([A-Z0-9]+)', event.description, re.IGNORECASE)
if match:
return {
'next_carrier': 'Unknown', # Would need LLM to determine
'next_tracking_number': match.group(1)
}
return None
This implementation shows how an adapter handles USPS-specific API interactions. The fetch_tracking_info method constructs the appropriate XML request, sends it to the USPS API, and parses the XML response. The _classify_event method maps USPS event descriptions to our standard event types. The detect_handoff method looks for carrier transfer events in the tracking history.
Each delivery service adapter follows this same pattern but implements the specifics for that carrier's API. A UPS adapter would use JSON instead of XML and different authentication. A FedEx adapter would have different endpoint structures. The common interface allows the rest of the system to work with any carrier uniformly.
DATABASE PERSISTENCE LAYER
The persistence layer stores package information, tracking history, and delivery service configurations. We use a repository pattern to abstract database operations from business logic. This allows us to switch database implementations without affecting other components.
The repository interface defines methods for creating, reading, updating, and deleting packages. It also provides query methods for finding packages by tracking number, user name, or status. The implementation handles all SQL or NoSQL operations internally.
For this system, we use SQLite for simplicity and portability, but the repository pattern allows easy migration to PostgreSQL, MySQL, or other databases. The schema includes tables for packages, tracking events, and delivery services with appropriate foreign key relationships.
Here is the repository implementation:
import sqlite3
from typing import List, Optional
from contextlib import contextmanager
import json
class PackageRepository:
def __init__(self, db_path: str):
self.db_path = db_path
self._initialize_database()
@contextmanager
def _get_connection(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
except Exception as e:
conn.rollback()
raise e
finally:
conn.close()
def _initialize_database(self):
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS packages (
package_id TEXT PRIMARY KEY,
tracking_number TEXT NOT NULL,
carrier TEXT NOT NULL,
user_name TEXT NOT NULL,
nickname TEXT,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
next_carrier TEXT,
next_tracking_number TEXT,
is_active INTEGER NOT NULL DEFAULT 1
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS tracking_events (
event_id TEXT PRIMARY KEY,
package_id TEXT NOT NULL,
timestamp TEXT NOT NULL,
location TEXT NOT NULL,
description TEXT NOT NULL,
event_type TEXT NOT NULL,
carrier TEXT NOT NULL,
metadata TEXT,
FOREIGN KEY (package_id) REFERENCES packages(package_id)
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS delivery_services (
service_id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
api_endpoint TEXT NOT NULL,
api_key TEXT,
rate_limit_per_minute INTEGER NOT NULL,
supports_handoff_detection INTEGER NOT NULL
)
''')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_tracking_number ON packages(tracking_number)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_user_name ON packages(user_name)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_package_events ON tracking_events(package_id)')
def save_package(self, package: Package) -> None:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO packages
(package_id, tracking_number, carrier, user_name, nickname, status,
created_at, updated_at, next_carrier, next_tracking_number, is_active)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
str(package.package_id),
package.tracking_number,
package.carrier,
package.user_name,
package.nickname,
package.status.value,
package.created_at.isoformat(),
package.updated_at.isoformat(),
package.next_carrier,
package.next_tracking_number,
1 if package.is_active else 0
))
# Delete existing events for this package and re-insert
cursor.execute('DELETE FROM tracking_events WHERE package_id = ?', (str(package.package_id),))
for event in package.tracking_history:
cursor.execute('''
INSERT INTO tracking_events
(event_id, package_id, timestamp, location, description, event_type, carrier, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
''', (
str(event.event_id),
str(package.package_id),
event.timestamp.isoformat(),
event.location,
event.description,
event.event_type.value,
event.carrier,
json.dumps(event.metadata)
))
def get_package_by_id(self, package_id: UUID) -> Optional[Package]:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM packages WHERE package_id = ?', (str(package_id),))
row = cursor.fetchone()
if not row:
return None
package = self._row_to_package(row)
cursor.execute('SELECT * FROM tracking_events WHERE package_id = ? ORDER BY timestamp', (str(package_id),))
event_rows = cursor.fetchall()
for event_row in event_rows:
package.tracking_history.append(self._row_to_event(event_row))
return package
def get_package_by_tracking_number(self, tracking_number: str, carrier: str) -> Optional[Package]:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
'SELECT * FROM packages WHERE tracking_number = ? AND carrier = ?',
(tracking_number, carrier)
)
row = cursor.fetchone()
if not row:
return None
package = self._row_to_package(row)
cursor.execute(
'SELECT * FROM tracking_events WHERE package_id = ? ORDER BY timestamp',
(str(package.package_id),)
)
event_rows = cursor.fetchall()
for event_row in event_rows:
package.tracking_history.append(self._row_to_event(event_row))
return package
def get_active_packages(self) -> List[Package]:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM packages WHERE is_active = 1')
rows = cursor.fetchall()
packages = []
for row in rows:
package = self._row_to_package(row)
cursor.execute(
'SELECT * FROM tracking_events WHERE package_id = ? ORDER BY timestamp',
(str(package.package_id),)
)
event_rows = cursor.fetchall()
for event_row in event_rows:
package.tracking_history.append(self._row_to_event(event_row))
packages.append(package)
return packages
def get_packages_by_user(self, user_name: str) -> List[Package]:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM packages WHERE user_name = ?', (user_name,))
rows = cursor.fetchall()
packages = []
for row in rows:
package = self._row_to_package(row)
cursor.execute(
'SELECT * FROM tracking_events WHERE package_id = ? ORDER BY timestamp',
(str(package.package_id),)
)
event_rows = cursor.fetchall()
for event_row in event_rows:
package.tracking_history.append(self._row_to_event(event_row))
packages.append(package)
return packages
def _row_to_package(self, row) -> Package:
return Package(
package_id=UUID(row['package_id']),
tracking_number=row['tracking_number'],
carrier=row['carrier'],
user_name=row['user_name'],
nickname=row['nickname'],
status=PackageStatus(row['status']),
created_at=datetime.fromisoformat(row['created_at']),
updated_at=datetime.fromisoformat(row['updated_at']),
next_carrier=row['next_carrier'],
next_tracking_number=row['next_tracking_number'],
is_active=bool(row['is_active']),
tracking_history=[]
)
def _row_to_event(self, row) -> TrackingEvent:
metadata = json.loads(row['metadata']) if row['metadata'] else {}
return TrackingEvent(
event_id=UUID(row['event_id']),
timestamp=datetime.fromisoformat(row['timestamp']),
location=row['location'],
description=row['description'],
event_type=EventType(row['event_type']),
carrier=row['carrier'],
metadata=metadata
)
def save_delivery_service(self, service: DeliveryService) -> None:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO delivery_services
(service_id, name, api_endpoint, api_key, rate_limit_per_minute, supports_handoff_detection)
VALUES (?, ?, ?, ?, ?, ?)
''', (
str(service.service_id),
service.name,
service.api_endpoint,
service.api_key,
service.rate_limit_per_minute,
1 if service.supports_handoff_detection else 0
))
def get_delivery_service_by_name(self, name: str) -> Optional[DeliveryService]:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM delivery_services WHERE name = ?', (name,))
row = cursor.fetchone()
if not row:
return None
return DeliveryService(
service_id=UUID(row['service_id']),
name=row['name'],
api_endpoint=row['api_endpoint'],
api_key=row['api_key'],
rate_limit_per_minute=row['rate_limit_per_minute'],
supports_handoff_detection=bool(row['supports_handoff_detection'])
)
def get_all_delivery_services(self) -> List[DeliveryService]:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM delivery_services')
rows = cursor.fetchall()
return [
DeliveryService(
service_id=UUID(row['service_id']),
name=row['name'],
api_endpoint=row['api_endpoint'],
api_key=row['api_key'],
rate_limit_per_minute=row['rate_limit_per_minute'],
supports_handoff_detection=bool(row['supports_handoff_detection'])
)
for row in rows
]
The repository handles all database interactions through a context manager that ensures proper connection handling and transaction management. The save_package method uses INSERT OR REPLACE to handle both new packages and updates. The get methods reconstruct complete Package objects with their full tracking history. Indexes on tracking_number and user_name improve query performance.
USE CASE LAYER AND BUSINESS LOGIC
The use case layer implements the core business logic of the tracking system. It orchestrates interactions between repositories, adapters, and other components to fulfill user requests. Each use case represents a specific user action like adding a package, querying tracking information, or processing updates.
The AddPackageUseCase handles the workflow when a user adds a new package to track. It validates the input, checks if the package already exists, fetches initial tracking information from the carrier, creates a Package entity, and saves it to the repository. If the tracking number is invalid or the carrier API is unavailable, it returns appropriate error messages.
The GetTrackingInfoUseCase retrieves current tracking information for a package. It can fetch from the database for recently updated packages or query the carrier API for the latest information. This use case decides whether to use cached data or make a fresh API call based on how recently the package was updated.
The UpdatePackageUseCase is used by the monitoring agent to refresh tracking information. It fetches the latest events from the carrier, compares them with stored events, adds new events to the package, detects carrier handoffs, and saves the updated package. This use case returns a summary of what changed so the notification system can alert the user appropriately.
Here is the implementation of these use cases:
from typing import Dict, Any, Optional
from datetime import datetime, timedelta
class AddPackageUseCase:
def __init__(
self,
repository: PackageRepository,
adapter_factory: 'DeliveryServiceAdapterFactory'
):
self.repository = repository
self.adapter_factory = adapter_factory
def execute(
self,
tracking_number: str,
carrier: str,
user_name: str,
nickname: Optional[str] = None
) -> Dict[str, Any]:
# Check if package already exists
existing_package = self.repository.get_package_by_tracking_number(tracking_number, carrier)
if existing_package:
return {
'success': False,
'error': 'Package already exists in the system',
'package_id': str(existing_package.package_id)
}
# Get adapter for the carrier
try:
adapter = self.adapter_factory.get_adapter(carrier)
except ValueError as e:
return {
'success': False,
'error': f'Unsupported carrier: {carrier}'
}
# Fetch initial tracking information
try:
events = adapter.fetch_tracking_info(tracking_number)
except Exception as e:
return {
'success': False,
'error': f'Failed to fetch tracking information: {str(e)}'
}
if not events:
return {
'success': False,
'error': 'No tracking information found for this tracking number'
}
# Create package entity
package = Package(
tracking_number=tracking_number,
carrier=carrier,
user_name=user_name,
nickname=nickname
)
# Add events to package
for event in events:
package.add_event(event)
package.detect_handoff(event)
# Save to repository
self.repository.save_package(package)
return {
'success': True,
'package_id': str(package.package_id),
'status': package.status.value,
'event_count': len(package.tracking_history),
'latest_event': package.tracking_history[-1].description if package.tracking_history else None
}
class GetTrackingInfoUseCase:
def __init__(
self,
repository: PackageRepository,
adapter_factory: 'DeliveryServiceAdapterFactory',
cache_duration_minutes: int = 30
):
self.repository = repository
self.adapter_factory = adapter_factory
self.cache_duration = timedelta(minutes=cache_duration_minutes)
def execute(
self,
tracking_number: str,
carrier: str,
force_refresh: bool = False
) -> Dict[str, Any]:
# Try to get from repository first
package = self.repository.get_package_by_tracking_number(tracking_number, carrier)
# Determine if we need to refresh from API
should_refresh = force_refresh
if package and not force_refresh:
time_since_update = datetime.now() - package.updated_at
should_refresh = time_since_update > self.cache_duration
if should_refresh or not package:
try:
adapter = self.adapter_factory.get_adapter(carrier)
events = adapter.fetch_tracking_info(tracking_number)
if package:
# Update existing package
existing_event_ids = {e.event_id for e in package.tracking_history}
new_events = [e for e in events if e.event_id not in existing_event_ids]
for event in new_events:
package.add_event(event)
package.detect_handoff(event)
self.repository.save_package(package)
else:
# Package not in system, return error
return {
'success': False,
'error': 'Package not found in system. Please add it first.'
}
except Exception as e:
if package:
# Return cached data if API fails
pass
else:
return {
'success': False,
'error': f'Failed to fetch tracking information: {str(e)}'
}
if not package:
return {
'success': False,
'error': 'Package not found'
}
return {
'success': True,
'package_id': str(package.package_id),
'tracking_number': package.tracking_number,
'carrier': package.carrier,
'user_name': package.user_name,
'nickname': package.nickname,
'status': package.status.value,
'updated_at': package.updated_at.isoformat(),
'next_carrier': package.next_carrier,
'next_tracking_number': package.next_tracking_number,
'events': [
{
'timestamp': e.timestamp.isoformat(),
'location': e.location,
'description': e.description,
'event_type': e.event_type.value,
'carrier': e.carrier
}
for e in sorted(package.tracking_history, key=lambda x: x.timestamp, reverse=True)
]
}
class UpdatePackageUseCase:
def __init__(
self,
repository: PackageRepository,
adapter_factory: 'DeliveryServiceAdapterFactory'
):
self.repository = repository
self.adapter_factory = adapter_factory
def execute(self, package_id: UUID) -> Dict[str, Any]:
package = self.repository.get_package_by_id(package_id)
if not package:
return {
'success': False,
'error': 'Package not found'
}
if not package.is_active:
return {
'success': True,
'updated': False,
'reason': 'Package is not active'
}
try:
adapter = self.adapter_factory.get_adapter(package.carrier)
events = adapter.fetch_tracking_info(package.tracking_number)
except Exception as e:
return {
'success': False,
'error': f'Failed to fetch tracking information: {str(e)}'
}
# Find new events
existing_timestamps = {(e.timestamp, e.description) for e in package.tracking_history}
new_events = [
e for e in events
if (e.timestamp, e.description) not in existing_timestamps
]
if not new_events:
return {
'success': True,
'updated': False,
'new_events': 0
}
# Add new events and check for handoffs
handoff_detected = False
for event in new_events:
package.add_event(event)
if package.detect_handoff(event):
handoff_detected = True
# Check if package is delivered and should be deactivated
if package.status == PackageStatus.DELIVERED:
package.is_active = False
self.repository.save_package(package)
return {
'success': True,
'updated': True,
'new_events': len(new_events),
'latest_status': package.status.value,
'handoff_detected': handoff_detected,
'next_carrier': package.next_carrier,
'next_tracking_number': package.next_tracking_number,
'events': [
{
'timestamp': e.timestamp.isoformat(),
'location': e.location,
'description': e.description,
'event_type': e.event_type.value
}
for e in new_events
]
}
These use cases encapsulate all business logic related to package management. They validate inputs, coordinate between different components, handle errors gracefully, and return structured results. The use cases are independent of any specific UI or API framework, making them reusable across different interfaces.
DELIVERY SERVICE ADAPTER FACTORY
The adapter factory creates the appropriate adapter instance for a given carrier. It maintains a registry of supported carriers and their corresponding adapter classes. When a use case needs to interact with a carrier API, it requests an adapter from the factory.
The factory pattern allows us to add new carriers without modifying existing code. We simply register a new adapter class and the factory can instantiate it when needed. The factory also manages adapter configuration, loading API credentials and settings from the delivery service repository.
Here is the adapter factory implementation:
from typing import Dict, Type
class DeliveryServiceAdapterFactory:
def __init__(self, repository: PackageRepository):
self.repository = repository
self._adapters: Dict[str, Type[DeliveryServiceAdapter]] = {}
self._register_default_adapters()
def _register_default_adapters(self):
self.register_adapter('USPS', USPSAdapter)
self.register_adapter('UPS', UPSAdapter)
self.register_adapter('FedEx', FedExAdapter)
self.register_adapter('DHL', DHLAdapter)
def register_adapter(self, carrier_name: str, adapter_class: Type[DeliveryServiceAdapter]):
self._adapters[carrier_name.upper()] = adapter_class
def get_adapter(self, carrier_name: str) -> DeliveryServiceAdapter:
carrier_upper = carrier_name.upper()
if carrier_upper not in self._adapters:
raise ValueError(f'No adapter registered for carrier: {carrier_name}')
service = self.repository.get_delivery_service_by_name(carrier_upper)
if not service:
raise ValueError(f'Delivery service not configured: {carrier_name}')
adapter_class = self._adapters[carrier_upper]
return adapter_class(service)
class UPSAdapter(DeliveryServiceAdapter):
def authenticate(self) -> bool:
return self.service.api_key is not None
def fetch_tracking_info(self, tracking_number: str) -> List[TrackingEvent]:
if not self.authenticate():
raise ValueError("UPS adapter not authenticated")
self._rate_limit()
url = f"{self.service.api_endpoint}/track/v1/details/{tracking_number}"
headers = {
'Authorization': f'Bearer {self.service.api_key}',
'Content-Type': 'application/json'
}
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
return self._parse_response(response.json())
def _parse_response(self, json_data: dict) -> List[TrackingEvent]:
events = []
if 'trackResponse' not in json_data:
return events
shipment = json_data['trackResponse'].get('shipment', [])
if not shipment:
return events
package = shipment[0].get('package', [])
if not package:
return events
activities = package[0].get('activity', [])
for activity in activities:
date_str = activity.get('date', '')
time_str = activity.get('time', '')
if date_str and time_str:
timestamp = datetime.strptime(f"{date_str} {time_str}", "%Y%m%d %H%M%S")
else:
continue
location_data = activity.get('location', {}).get('address', {})
city = location_data.get('city', '')
state = location_data.get('stateProvince', '')
country = location_data.get('country', '')
location = f"{city}, {state}, {country}".strip(', ')
status = activity.get('status', {})
description = status.get('description', 'Unknown')
status_type = status.get('type', '')
event_type = self._classify_event(status_type, description)
event = TrackingEvent(
timestamp=timestamp,
location=location,
description=description,
event_type=event_type,
carrier="UPS"
)
events.append(event)
return sorted(events, key=lambda e: e.timestamp)
def _classify_event(self, status_type: str, description: str) -> EventType:
status_type_lower = status_type.lower()
description_lower = description.lower()
if status_type_lower == 'd' or 'delivered' in description_lower:
return EventType.DELIVERED
elif status_type_lower == 'o' or 'out for delivery' in description_lower:
return EventType.OUT_FOR_DELIVERY
elif 'picked up' in description_lower or 'origin scan' in description_lower:
return EventType.PICKUP
elif 'exception' in description_lower:
return EventType.EXCEPTION
elif 'returned' in description_lower:
return EventType.RETURNED
elif 'transferred' in description_lower:
return EventType.CARRIER_HANDOFF
else:
return EventType.IN_TRANSIT
def detect_handoff(self, events: List[TrackingEvent]) -> Optional[dict]:
for event in reversed(events):
if event.event_type == EventType.CARRIER_HANDOFF:
description = event.description.lower()
if 'usps' in description or 'post office' in description:
return {'next_carrier': 'USPS', 'next_tracking_number': None}
return None
class FedExAdapter(DeliveryServiceAdapter):
def authenticate(self) -> bool:
return self.service.api_key is not None
def fetch_tracking_info(self, tracking_number: str) -> List[TrackingEvent]:
if not self.authenticate():
raise ValueError("FedEx adapter not authenticated")
self._rate_limit()
url = f"{self.service.api_endpoint}/track/v1/trackingnumbers"
headers = {
'Authorization': f'Bearer {self.service.api_key}',
'Content-Type': 'application/json'
}
payload = {
'trackingInfo': [
{
'trackingNumberInfo': {
'trackingNumber': tracking_number
}
}
],
'includeDetailedScans': True
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
return self._parse_response(response.json())
def _parse_response(self, json_data: dict) -> List[TrackingEvent]:
events = []
output = json_data.get('output', {})
complete_track_results = output.get('completeTrackResults', [])
if not complete_track_results:
return events
track_results = complete_track_results[0].get('trackResults', [])
if not track_results:
return events
scan_events = track_results[0].get('scanEvents', [])
for scan in scan_events:
date_str = scan.get('date', '')
if not date_str:
continue
timestamp = datetime.fromisoformat(date_str.replace('Z', '+00:00'))
location_data = scan.get('scanLocation', {})
city = location_data.get('city', '')
state = location_data.get('stateOrProvinceCode', '')
country = location_data.get('countryCode', '')
location = f"{city}, {state}, {country}".strip(', ')
description = scan.get('eventDescription', 'Unknown')
event_code = scan.get('eventType', '')
event_type = self._classify_event(event_code, description)
event = TrackingEvent(
timestamp=timestamp,
location=location,
description=description,
event_type=event_type,
carrier="FedEx"
)
events.append(event)
return sorted(events, key=lambda e: e.timestamp)
def _classify_event(self, event_code: str, description: str) -> EventType:
event_code_lower = event_code.lower()
description_lower = description.lower()
if event_code_lower == 'dl' or 'delivered' in description_lower:
return EventType.DELIVERED
elif event_code_lower == 'od' or 'out for delivery' in description_lower:
return EventType.OUT_FOR_DELIVERY
elif event_code_lower == 'pu' or 'picked up' in description_lower:
return EventType.PICKUP
elif 'exception' in description_lower or 'delay' in description_lower:
return EventType.EXCEPTION
elif 'returned' in description_lower:
return EventType.RETURNED
elif 'transferred' in description_lower or 'tendered' in description_lower:
return EventType.CARRIER_HANDOFF
else:
return EventType.IN_TRANSIT
def detect_handoff(self, events: List[TrackingEvent]) -> Optional[dict]:
for event in reversed(events):
if event.event_type == EventType.CARRIER_HANDOFF:
description = event.description.lower()
if 'usps' in description or 'smartpost' in description:
return {'next_carrier': 'USPS', 'next_tracking_number': None}
return None
class DHLAdapter(DeliveryServiceAdapter):
def authenticate(self) -> bool:
return self.service.api_key is not None
def fetch_tracking_info(self, tracking_number: str) -> List[TrackingEvent]:
if not self.authenticate():
raise ValueError("DHL adapter not authenticated")
self._rate_limit()
url = f"{self.service.api_endpoint}/track/shipments"
headers = {
'DHL-API-Key': self.service.api_key,
'Content-Type': 'application/json'
}
params = {
'trackingNumber': tracking_number
}
response = requests.get(url, headers=headers, params=params, timeout=30)
response.raise_for_status()
return self._parse_response(response.json())
def _parse_response(self, json_data: dict) -> List[TrackingEvent]:
events = []
shipments = json_data.get('shipments', [])
if not shipments:
return events
tracking_events = shipments[0].get('events', [])
for event_data in tracking_events:
timestamp_str = event_data.get('timestamp', '')
if not timestamp_str:
continue
timestamp = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00'))
location_data = event_data.get('location', {})
address = location_data.get('address', {})
city = address.get('addressLocality', '')
country = address.get('countryCode', '')
location = f"{city}, {country}".strip(', ')
description = event_data.get('description', 'Unknown')
status_code = event_data.get('statusCode', '')
event_type = self._classify_event(status_code, description)
event = TrackingEvent(
timestamp=timestamp,
location=location,
description=description,
event_type=event_type,
carrier="DHL"
)
events.append(event)
return sorted(events, key=lambda e: e.timestamp)
def _classify_event(self, status_code: str, description: str) -> EventType:
description_lower = description.lower()
if 'delivered' in description_lower:
return EventType.DELIVERED
elif 'out for delivery' in description_lower:
return EventType.OUT_FOR_DELIVERY
elif 'picked up' in description_lower or 'collected' in description_lower:
return EventType.PICKUP
elif 'exception' in description_lower or 'delay' in description_lower:
return EventType.EXCEPTION
elif 'returned' in description_lower:
return EventType.RETURNED
elif 'transferred' in description_lower or 'forwarded' in description_lower:
return EventType.CARRIER_HANDOFF
else:
return EventType.IN_TRANSIT
def detect_handoff(self, events: List[TrackingEvent]) -> Optional[dict]:
for event in reversed(events):
if event.event_type == EventType.CARRIER_HANDOFF:
return {'next_carrier': 'Unknown', 'next_tracking_number': None}
return None
The factory maintains a registry of adapter classes and instantiates them with the appropriate configuration from the database. Each adapter implements the same interface but handles carrier-specific API details. The factory makes it easy to add support for new carriers by simply registering a new adapter class.
LLM INTEGRATION FOR INTELLIGENT HANDOFF DETECTION
Detecting carrier handoffs is challenging because different carriers describe transfers differently. Some include the new tracking number in the event description, others do not. The new carrier might not be explicitly mentioned. This is where large language models become valuable.
We integrate an LLM to analyze tracking event descriptions and extract handoff information. The LLM can understand natural language descriptions and identify when a package is being transferred to another carrier. It can extract the new carrier name and tracking number even when they are embedded in complex text.
The LLM integration must support both local and remote models. Local models run on the user's hardware using frameworks like llama.cpp or transformers. Remote models are accessed through APIs like OpenAI or Anthropic. The system must detect available GPU hardware and configure the appropriate backend.
For GPU support, we need to handle different architectures. NVIDIA GPUs use CUDA, AMD GPUs use ROCm, Intel GPUs use their own runtime, and Apple Silicon uses Metal Performance Shaders. The LLM integration layer abstracts these differences and provides a unified interface.
Here is the LLM integration implementation:
import platform
import subprocess
from typing import Optional, Dict, Any, List
from abc import ABC, abstractmethod
import torch
class LLMBackend(ABC):
@abstractmethod
def generate(self, prompt: str, max_tokens: int = 500) -> str:
pass
@abstractmethod
def is_available(self) -> bool:
pass
class LocalLLMBackend(LLMBackend):
def __init__(self, model_path: str, device: str = 'auto'):
self.model_path = model_path
self.device = self._detect_device() if device == 'auto' else device
self.model = None
self.tokenizer = None
self._load_model()
def _detect_device(self) -> str:
# Check for CUDA (NVIDIA)
if torch.cuda.is_available():
return 'cuda'
# Check for ROCm (AMD)
try:
if torch.version.hip is not None:
return 'cuda' # PyTorch uses 'cuda' for ROCm too
except AttributeError:
pass
# Check for MPS (Apple Silicon)
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return 'mps'
# Check for Intel GPU
try:
import intel_extension_for_pytorch as ipex
if ipex.xpu.is_available():
return 'xpu'
except ImportError:
pass
# Fallback to CPU
return 'cpu'
def _load_model(self):
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
if self.device == 'cuda':
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float16,
device_map='auto'
)
elif self.device == 'mps':
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float16
).to('mps')
elif self.device == 'xpu':
import intel_extension_for_pytorch as ipex
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float16
)
self.model = ipex.optimize(self.model)
self.model = self.model.to('xpu')
else:
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float32
).to('cpu')
def generate(self, prompt: str, max_tokens: int = 500) -> str:
inputs = self.tokenizer(prompt, return_tensors='pt').to(self.device)
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=0.7,
do_sample=True,
pad_token_id=self.tokenizer.eos_token_id
)
response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
# Remove the prompt from the response
response = response[len(prompt):].strip()
return response
def is_available(self) -> bool:
return self.model is not None
class RemoteLLMBackend(LLMBackend):
def __init__(self, api_key: str, model_name: str = 'gpt-4', provider: str = 'openai'):
self.api_key = api_key
self.model_name = model_name
self.provider = provider
def generate(self, prompt: str, max_tokens: int = 500) -> str:
if self.provider == 'openai':
return self._generate_openai(prompt, max_tokens)
elif self.provider == 'anthropic':
return self._generate_anthropic(prompt, max_tokens)
else:
raise ValueError(f"Unsupported provider: {self.provider}")
def _generate_openai(self, prompt: str, max_tokens: int) -> str:
import openai
openai.api_key = self.api_key
response = openai.ChatCompletion.create(
model=self.model_name,
messages=[
{"role": "user", "content": prompt}
],
max_tokens=max_tokens,
temperature=0.7
)
return response.choices[0].message.content.strip()
def _generate_anthropic(self, prompt: str, max_tokens: int) -> str:
import anthropic
client = anthropic.Anthropic(api_key=self.api_key)
message = client.messages.create(
model=self.model_name,
max_tokens=max_tokens,
messages=[
{"role": "user", "content": prompt}
]
)
return message.content[0].text.strip()
def is_available(self) -> bool:
return self.api_key is not None
class HandoffDetectionService:
def __init__(self, llm_backend: LLMBackend):
self.llm = llm_backend
def analyze_handoff(self, events: List[TrackingEvent]) -> Optional[Dict[str, str]]:
# Look for potential handoff events
handoff_candidates = [
e for e in events
if e.event_type == EventType.CARRIER_HANDOFF or
'transfer' in e.description.lower() or
'forward' in e.description.lower() or
'tender' in e.description.lower()
]
if not handoff_candidates:
return None
# Use the most recent handoff candidate
latest_handoff = handoff_candidates[-1]
prompt = self._build_handoff_prompt(latest_handoff)
response = self.llm.generate(prompt, max_tokens=200)
return self._parse_handoff_response(response)
def _build_handoff_prompt(self, event: TrackingEvent) -> str:
return f"""Analyze this shipping event and extract handoff information if present.
Event Description: {event.description}
Event Location: {event.location}
Current Carrier: {event.carrier}
If this event indicates the package is being transferred to another carrier, extract:
1. The name of the new carrier (e.g., USPS, UPS, FedEx, DHL)
2. The new tracking number if mentioned
Respond in this exact format:
CARRIER: [carrier name or UNKNOWN]
TRACKING: [tracking number or NONE]
If this is not a carrier handoff, respond with:
NO_HANDOFF
Response:"""
def _parse_handoff_response(self, response: str) -> Optional[Dict[str, str]]:
response = response.strip()
if 'NO_HANDOFF' in response:
return None
result = {}
for line in response.split('\n'):
line = line.strip()
if line.startswith('CARRIER:'):
carrier = line.split(':', 1)[1].strip()
if carrier and carrier != 'UNKNOWN':
result['next_carrier'] = carrier.upper()
elif line.startswith('TRACKING:'):
tracking = line.split(':', 1)[1].strip()
if tracking and tracking != 'NONE':
result['next_tracking_number'] = tracking
return result if result else None
def extract_tracking_number(self, text: str) -> Optional[str]:
prompt = f"""Extract the tracking number from this text if present.
Text: {text}
Respond with just the tracking number, or NONE if no tracking number is found.
Response:"""
response = self.llm.generate(prompt, max_tokens=50)
response = response.strip()
if response == 'NONE' or not response:
return None
return response
def identify_carrier(self, text: str) -> Optional[str]:
prompt = f"""Identify the shipping carrier mentioned in this text.
Text: {text}
Respond with one of these carrier names: USPS, UPS, FEDEX, DHL, or UNKNOWN if the carrier cannot be determined.
Response:"""
response = self.llm.generate(prompt, max_tokens=20)
response = response.strip().upper()
known_carriers = ['USPS', 'UPS', 'FEDEX', 'DHL']
if response in known_carriers:
return response
return None
The LLM integration provides a unified interface regardless of whether we use a local or remote model. The LocalLLMBackend automatically detects available GPU hardware and configures PyTorch accordingly. It supports CUDA for NVIDIA, ROCm through PyTorch's CUDA backend, MPS for Apple Silicon, and Intel Extension for PyTorch for Intel GPUs.
The HandoffDetectionService uses the LLM to analyze tracking events and extract handoff information. It constructs prompts that ask the LLM to identify carrier transfers and extract relevant details. The responses are parsed to extract structured data that the system can use to follow the package to the new carrier.
MONITORING AGENT IMPLEMENTATION
The monitoring agent runs continuously in the background, checking all active packages for updates. It uses a scheduled task approach, waking up at regular intervals to process the package list. The agent must be efficient to avoid overwhelming carrier APIs and must handle errors gracefully.
The agent maintains a queue of packages to check. It processes them one at a time, respecting rate limits for each carrier. When it finds new tracking events, it updates the package in the database and triggers notifications. If it detects a carrier handoff, it creates a new package entry for the new carrier and tracking number.
The agent uses exponential backoff for packages that have not updated recently. Packages that are actively moving are checked more frequently than packages that have been sitting at a facility for days. Delivered packages are marked inactive and removed from the monitoring queue.
Here is the monitoring agent implementation:
import threading
import time
from typing import Set
from datetime import datetime, timedelta
import logging
class MonitoringAgent:
def __init__(
self,
repository: PackageRepository,
update_use_case: UpdatePackageUseCase,
add_package_use_case: AddPackageUseCase,
handoff_service: HandoffDetectionService,
notification_service: 'NotificationService',
check_interval_seconds: int = 300
):
self.repository = repository
self.update_use_case = update_use_case
self.add_package_use_case = add_package_use_case
self.handoff_service = handoff_service
self.notification_service = notification_service
self.check_interval = check_interval_seconds
self.running = False
self.thread = None
self.processed_handoffs: Set[str] = set()
self.logger = logging.getLogger(__name__)
def start(self):
if self.running:
self.logger.warning("Monitoring agent already running")
return
self.running = True
self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start()
self.logger.info("Monitoring agent started")
def stop(self):
if not self.running:
return
self.running = False
if self.thread:
self.thread.join(timeout=10)
self.logger.info("Monitoring agent stopped")
def _run(self):
while self.running:
try:
self._check_all_packages()
except Exception as e:
self.logger.error(f"Error in monitoring agent: {str(e)}", exc_info=True)
time.sleep(self.check_interval)
def _check_all_packages(self):
packages = self.repository.get_active_packages()
self.logger.info(f"Checking {len(packages)} active packages")
for package in packages:
if not self.running:
break
try:
self._check_package(package)
except Exception as e:
self.logger.error(
f"Error checking package {package.package_id}: {str(e)}",
exc_info=True
)
# Small delay between packages to avoid overwhelming APIs
time.sleep(1)
def _check_package(self, package: Package):
# Determine if we should check this package based on last update time
time_since_update = datetime.now() - package.updated_at
# Skip if updated very recently (within last 5 minutes)
if time_since_update < timedelta(minutes=5):
return
# Check less frequently for packages that haven't updated in a while
if time_since_update > timedelta(days=1):
# Only check once per day for stale packages
if time_since_update.total_seconds() % 86400 > self.check_interval:
return
self.logger.debug(f"Checking package {package.tracking_number}")
result = self.update_use_case.execute(package.package_id)
if not result['success']:
self.logger.warning(
f"Failed to update package {package.tracking_number}: {result.get('error')}"
)
return
if result['updated']:
self.logger.info(
f"Package {package.tracking_number} updated with {result['new_events']} new events"
)
# Send notification about new events
self.notification_service.notify_package_update(
package,
result['events']
)
# Check for handoff
if result.get('handoff_detected'):
self._handle_handoff(package, result)
def _handle_handoff(self, package: Package, update_result: Dict[str, Any]):
handoff_key = f"{package.package_id}_{package.next_carrier}_{package.next_tracking_number}"
# Avoid processing the same handoff multiple times
if handoff_key in self.processed_handoffs:
return
self.processed_handoffs.add(handoff_key)
next_carrier = package.next_carrier
next_tracking = package.next_tracking_number
# If we don't have complete handoff information, try to extract it with LLM
if not next_carrier or not next_tracking:
handoff_info = self.handoff_service.analyze_handoff(package.tracking_history)
if handoff_info:
next_carrier = handoff_info.get('next_carrier', next_carrier)
next_tracking = handoff_info.get('next_tracking_number', next_tracking)
if not next_carrier or not next_tracking:
self.logger.warning(
f"Handoff detected for {package.tracking_number} but missing information. "
f"Carrier: {next_carrier}, Tracking: {next_tracking}"
)
self.notification_service.notify_incomplete_handoff(package)
return
# Check if we already have this package
existing = self.repository.get_package_by_tracking_number(next_tracking, next_carrier)
if existing:
self.logger.info(
f"Handoff package {next_tracking} already exists in system"
)
return
# Add the new package
self.logger.info(
f"Creating new package for handoff: {next_tracking} ({next_carrier})"
)
result = self.add_package_use_case.execute(
tracking_number=next_tracking,
carrier=next_carrier,
user_name=package.user_name,
nickname=f"{package.nickname or package.tracking_number} (continued)" if package.nickname else None
)
if result['success']:
self.notification_service.notify_handoff_created(
original_package=package,
new_tracking=next_tracking,
new_carrier=next_carrier
)
else:
self.logger.error(
f"Failed to create handoff package: {result.get('error')}"
)
The monitoring agent runs in a separate thread to avoid blocking the main application. It processes packages sequentially with appropriate delays to respect API rate limits. The agent uses intelligent scheduling to check active packages more frequently than stale ones. When it detects a handoff, it uses the LLM service to extract complete information and automatically creates a new package entry for continued tracking.
NOTIFICATION SYSTEM
The notification system alerts users about important package events. It supports multiple notification channels including email, push notifications, and webhooks. Users can configure which events trigger notifications and which channels to use.
The notification service receives events from the monitoring agent and formats them appropriately for each channel. It maintains user preferences and ensures notifications are not duplicated. The service also implements rate limiting to prevent notification spam.
Here is the notification service implementation:
from typing import List, Dict, Any
from abc import ABC, abstractmethod
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
class NotificationChannel(ABC):
@abstractmethod
def send(self, recipient: str, subject: str, message: str, metadata: Dict[str, Any] = None):
pass
class EmailNotificationChannel(NotificationChannel):
def __init__(self, smtp_host: str, smtp_port: int, username: str, password: str):
self.smtp_host = smtp_host
self.smtp_port = smtp_port
self.username = username
self.password = password
def send(self, recipient: str, subject: str, message: str, metadata: Dict[str, Any] = None):
msg = MIMEMultipart('alternative')
msg['Subject'] = subject
msg['From'] = self.username
msg['To'] = recipient
text_part = MIMEText(message, 'plain')
msg.attach(text_part)
if metadata and 'html' in metadata:
html_part = MIMEText(metadata['html'], 'html')
msg.attach(html_part)
with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
server.starttls()
server.login(self.username, self.password)
server.send_message(msg)
class WebhookNotificationChannel(NotificationChannel):
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
def send(self, recipient: str, subject: str, message: str, metadata: Dict[str, Any] = None):
import requests
payload = {
'recipient': recipient,
'subject': subject,
'message': message,
'metadata': metadata or {}
}
response = requests.post(
self.webhook_url,
json=payload,
timeout=10
)
response.raise_for_status()
class NotificationService:
def __init__(self, channels: List[NotificationChannel], user_preferences: Dict[str, Any]):
self.channels = channels
self.user_preferences = user_preferences
self.logger = logging.getLogger(__name__)
def notify_package_update(self, package: Package, new_events: List[Dict[str, Any]]):
if not self._should_notify(package.user_name, 'package_update'):
return
subject = f"Package Update: {package.nickname or package.tracking_number}"
message_parts = [
f"Your package {package.nickname or package.tracking_number} has been updated.",
f"Current Status: {package.status.value}",
"",
"New Events:"
]
for event in new_events:
event_time = datetime.fromisoformat(event['timestamp']).strftime('%Y-%m-%d %H:%M')
message_parts.append(
f" - {event_time} | {event['location']} | {event['description']}"
)
message = '\n'.join(message_parts)
self._send_to_all_channels(package.user_name, subject, message)
def notify_handoff_created(self, original_package: Package, new_tracking: str, new_carrier: str):
if not self._should_notify(original_package.user_name, 'handoff'):
return
subject = f"Package Handoff: {original_package.nickname or original_package.tracking_number}"
message = f"""Your package {original_package.nickname or original_package.tracking_number} has been transferred to a new carrier.
Original Tracking: {original_package.tracking_number} ({original_package.carrier})
New Tracking: {new_tracking} ({new_carrier})
The new package has been automatically added to your tracking list and will continue to be monitored.
"""
self._send_to_all_channels(original_package.user_name, subject, message)
def notify_incomplete_handoff(self, package: Package):
if not self._should_notify(package.user_name, 'handoff'):
return
subject = f"Incomplete Handoff Information: {package.nickname or package.tracking_number}"
message = f"""Your package {package.nickname or package.tracking_number} appears to have been transferred to another carrier, but we could not determine the complete handoff information.
Please check the carrier's website for the new tracking number and add it manually if needed.
"""
self._send_to_all_channels(package.user_name, subject, message)
def notify_delivery(self, package: Package):
if not self._should_notify(package.user_name, 'delivery'):
return
subject = f"Package Delivered: {package.nickname or package.tracking_number}"
latest_event = package.tracking_history[-1] if package.tracking_history else None
message = f"""Your package {package.nickname or package.tracking_number} has been delivered!
Tracking Number: {package.tracking_number}
Carrier: {package.carrier}
"""
if latest_event:
delivery_time = latest_event.timestamp.strftime('%Y-%m-%d %H:%M')
message += f"Delivered At: {delivery_time}\n"
message += f"Location: {latest_event.location}\n"
self._send_to_all_channels(package.user_name, subject, message)
def _should_notify(self, user_name: str, event_type: str) -> bool:
user_prefs = self.user_preferences.get(user_name, {})
enabled_events = user_prefs.get('enabled_events', ['package_update', 'handoff', 'delivery'])
return event_type in enabled_events
def _send_to_all_channels(self, recipient: str, subject: str, message: str):
for channel in self.channels:
try:
channel.send(recipient, subject, message)
self.logger.info(f"Sent notification via {channel.__class__.__name__} to {recipient}")
except Exception as e:
self.logger.error(
f"Failed to send notification via {channel.__class__.__name__}: {str(e)}",
exc_info=True
)
The notification service supports multiple channels through a common interface. Each channel implements the send method differently. The email channel uses SMTP to send formatted emails. The webhook channel posts JSON payloads to configured URLs. Users can enable or disable specific event types through preferences. The service handles errors gracefully and logs all notification attempts.
USER INTERFACE AND API LAYER
The user interface provides ways for users to interact with the tracking system. We implement both a command-line interface and a REST API. The CLI is useful for quick queries and testing. The API enables integration with web applications and mobile apps.
The REST API follows standard conventions with endpoints for adding packages, querying tracking information, listing packages, and managing settings. It uses JSON for request and response bodies. Authentication is handled through API keys or tokens.
Here is the API implementation using Flask:
from flask import Flask, request, jsonify
from functools import wraps
from typing import Callable
class TrackingAPI:
def __init__(
self,
add_package_use_case: AddPackageUseCase,
get_tracking_use_case: GetTrackingInfoUseCase,
repository: PackageRepository,
api_keys: Dict[str, str]
):
self.add_package_use_case = add_package_use_case
self.get_tracking_use_case = get_tracking_use_case
self.repository = repository
self.api_keys = api_keys
self.app = Flask(__name__)
self._setup_routes()
def _setup_routes(self):
self.app.route('/api/packages', methods=['POST'])(self._require_auth(self.add_package))
self.app.route('/api/packages/<package_id>', methods=['GET'])(self._require_auth(self.get_package))
self.app.route('/api/packages', methods=['GET'])(self._require_auth(self.list_packages))
self.app.route('/api/tracking/<carrier>/<tracking_number>', methods=['GET'])(self._require_auth(self.get_tracking))
def _require_auth(self, f: Callable):
@wraps(f)
def decorated_function(*args, **kwargs):
api_key = request.headers.get('X-API-Key')
if not api_key or api_key not in self.api_keys:
return jsonify({'error': 'Invalid or missing API key'}), 401
# Store user_name from API key mapping
request.user_name = self.api_keys[api_key]
return f(*args, **kwargs)
return decorated_function
def add_package(self):
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
tracking_number = data.get('tracking_number')
carrier = data.get('carrier')
nickname = data.get('nickname')
if not tracking_number or not carrier:
return jsonify({'error': 'tracking_number and carrier are required'}), 400
result = self.add_package_use_case.execute(
tracking_number=tracking_number,
carrier=carrier,
user_name=request.user_name,
nickname=nickname
)
if result['success']:
return jsonify(result), 201
else:
return jsonify(result), 400
def get_package(self, package_id):
try:
package_uuid = UUID(package_id)
except ValueError:
return jsonify({'error': 'Invalid package ID format'}), 400
package = self.repository.get_package_by_id(package_uuid)
if not package:
return jsonify({'error': 'Package not found'}), 404
if package.user_name != request.user_name:
return jsonify({'error': 'Access denied'}), 403
return jsonify({
'package_id': str(package.package_id),
'tracking_number': package.tracking_number,
'carrier': package.carrier,
'nickname': package.nickname,
'status': package.status.value,
'created_at': package.created_at.isoformat(),
'updated_at': package.updated_at.isoformat(),
'is_active': package.is_active,
'events': [
{
'timestamp': e.timestamp.isoformat(),
'location': e.location,
'description': e.description,
'event_type': e.event_type.value,
'carrier': e.carrier
}
for e in sorted(package.tracking_history, key=lambda x: x.timestamp, reverse=True)
]
})
def list_packages(self):
packages = self.repository.get_packages_by_user(request.user_name)
return jsonify({
'packages': [
{
'package_id': str(p.package_id),
'tracking_number': p.tracking_number,
'carrier': p.carrier,
'nickname': p.nickname,
'status': p.status.value,
'updated_at': p.updated_at.isoformat(),
'is_active': p.is_active
}
for p in packages
]
})
def get_tracking(self, carrier, tracking_number):
force_refresh = request.args.get('refresh', 'false').lower() == 'true'
result = self.get_tracking_use_case.execute(
tracking_number=tracking_number,
carrier=carrier,
force_refresh=force_refresh
)
if result['success']:
# Verify user has access to this package
if result.get('user_name') != request.user_name:
return jsonify({'error': 'Access denied'}), 403
return jsonify(result)
else:
return jsonify(result), 404
def run(self, host='0.0.0.0', port=5000, debug=False):
self.app.run(host=host, port=port, debug=debug)
The API provides RESTful endpoints for all major operations. Authentication is handled through API keys passed in request headers. Each endpoint validates input, executes the appropriate use case, and returns structured JSON responses. Error handling provides clear messages about what went wrong.
CONFIGURATION AND INITIALIZATION
The system requires configuration for database connections, API credentials, LLM settings, and notification channels. We use a configuration file approach with environment variable overrides for sensitive data like API keys.
The initialization process creates all necessary components with proper dependency injection. It sets up the database, loads delivery service configurations, initializes the LLM backend, creates use cases, starts the monitoring agent, and launches the API server.
Here is the configuration and initialization code:
import os
import json
import logging.config
class Configuration:
def __init__(self, config_file: str = 'config.json'):
self.config_file = config_file
self.config = self._load_config()
def _load_config(self) -> Dict[str, Any]:
if os.path.exists(self.config_file):
with open(self.config_file, 'r') as f:
config = json.load(f)
else:
config = self._default_config()
# Override with environment variables
config = self._apply_env_overrides(config)
return config
def _default_config(self) -> Dict[str, Any]:
return {
'database': {
'path': 'tracking.db'
},
'llm': {
'backend': 'local',
'model_path': 'models/llama-2-7b-chat',
'device': 'auto'
},
'monitoring': {
'check_interval_seconds': 300,
'enabled': True
},
'api': {
'host': '0.0.0.0',
'port': 5000,
'debug': False
},
'notifications': {
'email': {
'enabled': False,
'smtp_host': 'smtp.gmail.com',
'smtp_port': 587
},
'webhook': {
'enabled': False
}
},
'logging': {
'level': 'INFO',
'file': 'tracking.log'
}
}
def _apply_env_overrides(self, config: Dict[str, Any]) -> Dict[str, Any]:
# Database
if 'DB_PATH' in os.environ:
config['database']['path'] = os.environ['DB_PATH']
# LLM
if 'LLM_BACKEND' in os.environ:
config['llm']['backend'] = os.environ['LLM_BACKEND']
if 'LLM_MODEL_PATH' in os.environ:
config['llm']['model_path'] = os.environ['LLM_MODEL_PATH']
if 'LLM_API_KEY' in os.environ:
config['llm']['api_key'] = os.environ['LLM_API_KEY']
if 'SMTP_USERNAME' in os.environ:
config['notifications']['email']['username'] = os.environ['SMTP_USERNAME']
if 'SMTP_PASSWORD' in os.environ:
config['notifications']['email']['password'] = os.environ['SMTP_PASSWORD']
return config
def get(self, key: str, default=None):
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
class ApplicationInitializer:
def __init__(self, config: Configuration):
self.config = config
self._setup_logging()
def _setup_logging(self):
logging.basicConfig(
level=getattr(logging, self.config.get('logging.level', 'INFO')),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(self.config.get('logging.file', 'tracking.log')),
logging.StreamHandler()
]
)
def initialize(self) -> Dict[str, Any]:
logger = logging.getLogger(__name__)
logger.info("Initializing tracking system")
# Initialize repository
db_path = self.config.get('database.path')
repository = PackageRepository(db_path)
logger.info(f"Database initialized at {db_path}")
# Initialize delivery services
self._initialize_delivery_services(repository)
# Initialize adapter factory
adapter_factory = DeliveryServiceAdapterFactory(repository)
logger.info("Adapter factory initialized")
# Initialize LLM backend
llm_backend = self._initialize_llm()
logger.info(f"LLM backend initialized: {llm_backend.__class__.__name__}")
# Initialize handoff detection service
handoff_service = HandoffDetectionService(llm_backend)
# Initialize use cases
add_package_use_case = AddPackageUseCase(repository, adapter_factory)
get_tracking_use_case = GetTrackingInfoUseCase(repository, adapter_factory)
update_package_use_case = UpdatePackageUseCase(repository, adapter_factory)
# Initialize notification service
notification_service = self._initialize_notifications()
logger.info("Notification service initialized")
# Initialize monitoring agent
monitoring_agent = None
if self.config.get('monitoring.enabled', True):
monitoring_agent = MonitoringAgent(
repository=repository,
update_use_case=update_package_use_case,
add_package_use_case=add_package_use_case,
handoff_service=handoff_service,
notification_service=notification_service,
check_interval_seconds=self.config.get('monitoring.check_interval_seconds', 300)
)
monitoring_agent.start()
logger.info("Monitoring agent started")
# Initialize API
api_keys = self.config.get('api.keys', {})
api = TrackingAPI(
add_package_use_case=add_package_use_case,
get_tracking_use_case=get_tracking_use_case,
repository=repository,
api_keys=api_keys
)
logger.info("API initialized")
return {
'repository': repository,
'adapter_factory': adapter_factory,
'llm_backend': llm_backend,
'handoff_service': handoff_service,
'add_package_use_case': add_package_use_case,
'get_tracking_use_case': get_tracking_use_case,
'update_package_use_case': update_package_use_case,
'notification_service': notification_service,
'monitoring_agent': monitoring_agent,
'api': api
}
def _initialize_delivery_services(self, repository: PackageRepository):
services = [
DeliveryService(
name='USPS',
api_endpoint='https://secure.shippingapis.com/ShippingAPI.dll',
api_key=self.config.get('carriers.usps.api_key'),
rate_limit_per_minute=60
),
DeliveryService(
name='UPS',
api_endpoint='https://onlinetools.ups.com/api',
api_key=self.config.get('carriers.ups.api_key'),
rate_limit_per_minute=60
),
DeliveryService(
name='FEDEX',
api_endpoint='https://apis.fedex.com',
api_key=self.config.get('carriers.fedex.api_key'),
rate_limit_per_minute=60
),
DeliveryService(
name='DHL',
api_endpoint='https://api-eu.dhl.com',
api_key=self.config.get('carriers.dhl.api_key'),
rate_limit_per_minute=60
)
]
for service in services:
repository.save_delivery_service(service)
def _initialize_llm(self) -> LLMBackend:
backend_type = self.config.get('llm.backend', 'local')
if backend_type == 'local':
model_path = self.config.get('llm.model_path')
device = self.config.get('llm.device', 'auto')
return LocalLLMBackend(model_path, device)
elif backend_type == 'openai':
api_key = self.config.get('llm.api_key')
model_name = self.config.get('llm.model_name', 'gpt-4')
return RemoteLLMBackend(api_key, model_name, 'openai')
elif backend_type == 'anthropic':
api_key = self.config.get('llm.api_key')
model_name = self.config.get('llm.model_name', 'claude-3-opus-20240229')
return RemoteLLMBackend(api_key, model_name, 'anthropic')
else:
raise ValueError(f"Unsupported LLM backend: {backend_type}")
def _initialize_notifications(self) -> NotificationService:
channels = []
# Email channel
if self.config.get('notifications.email.enabled', False):
email_channel = EmailNotificationChannel(
smtp_host=self.config.get('notifications.email.smtp_host'),
smtp_port=self.config.get('notifications.email.smtp_port'),
username=self.config.get('notifications.email.username'),
password=self.config.get('notifications.email.password')
)
channels.append(email_channel)
# Webhook channel
if self.config.get('notifications.webhook.enabled', False):
webhook_channel = WebhookNotificationChannel(
webhook_url=self.config.get('notifications.webhook.url')
)
channels.append(webhook_channel)
user_preferences = self.config.get('notifications.user_preferences', {})
return NotificationService(channels, user_preferences)
The configuration system loads settings from a JSON file and allows environment variable overrides for sensitive data. The initializer creates all components in the correct order with proper dependencies. It handles errors during initialization and provides clear logging of the startup process.
COMPLETE RUNNING EXAMPLE
The following is a complete, production-ready implementation that integrates all components into a working package tracking system. This example includes a main entry point, error handling, graceful shutdown, and a command-line interface for testing.
#!/usr/bin/env python3
import sys
import signal
import argparse
from uuid import UUID
from datetime import datetime
from typing import Optional
def main():
parser = argparse.ArgumentParser(description='Package Tracking System')
parser.add_argument('--config', default='config.json', help='Configuration file path')
parser.add_argument('--mode', choices=['api', 'cli'], default='api', help='Run mode')
args = parser.parse_args()
# Load configuration
config = Configuration(args.config)
# Initialize application
initializer = ApplicationInitializer(config)
components = initializer.initialize()
# Setup signal handlers for graceful shutdown
def signal_handler(sig, frame):
print("\nShutting down gracefully...")
if components['monitoring_agent']:
components['monitoring_agent'].stop()
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
if args.mode == 'api':
# Run API server
api = components['api']
host = config.get('api.host', '0.0.0.0')
port = config.get('api.port', 5000)
debug = config.get('api.debug', False)
print(f"Starting API server on {host}:{port}")
api.run(host=host, port=port, debug=debug)
else:
# Run CLI
cli = TrackingCLI(
add_package_use_case=components['add_package_use_case'],
get_tracking_use_case=components['get_tracking_use_case'],
repository=components['repository']
)
cli.run()
class TrackingCLI:
def __init__(
self,
add_package_use_case: AddPackageUseCase,
get_tracking_use_case: GetTrackingInfoUseCase,
repository: PackageRepository
):
self.add_package = add_package_use_case
self.get_tracking = get_tracking_use_case
self.repository = repository
def run(self):
print("Package Tracking System CLI")
print("=" * 50)
while True:
print("\nCommands:")
print(" 1. Add package")
print(" 2. Get tracking info")
print(" 3. List packages")
print(" 4. Exit")
choice = input("\nEnter command number: ").strip()
if choice == '1':
self._add_package_interactive()
elif choice == '2':
self._get_tracking_interactive()
elif choice == '3':
self._list_packages_interactive()
elif choice == '4':
print("Goodbye!")
break
else:
print("Invalid choice. Please try again.")
def _add_package_interactive(self):
print("\n--- Add Package ---")
tracking_number = input("Tracking number: ").strip()
if not tracking_number:
print("Error: Tracking number is required")
return
carrier = input("Carrier (USPS/UPS/FedEx/DHL): ").strip().upper()
if carrier not in ['USPS', 'UPS', 'FEDEX', 'DHL']:
print("Error: Invalid carrier")
return
user_name = input("Your name: ").strip()
if not user_name:
print("Error: Name is required")
return
nickname = input("Package nickname (optional): ").strip() or None
print("\nAdding package...")
result = self.add_package.execute(
tracking_number=tracking_number,
carrier=carrier,
user_name=user_name,
nickname=nickname
)
if result['success']:
print(f"\nSuccess! Package added with ID: {result['package_id']}")
print(f"Status: {result['status']}")
print(f"Events found: {result['event_count']}")
if result.get('latest_event'):
print(f"Latest event: {result['latest_event']}")
else:
print(f"\nError: {result['error']}")
def _get_tracking_interactive(self):
print("\n--- Get Tracking Info ---")
tracking_number = input("Tracking number: ").strip()
if not tracking_number:
print("Error: Tracking number is required")
return
carrier = input("Carrier (USPS/UPS/FedEx/DHL): ").strip().upper()
if carrier not in ['USPS', 'UPS', 'FEDEX', 'DHL']:
print("Error: Invalid carrier")
return
refresh = input("Force refresh from carrier? (y/n): ").strip().lower() == 'y'
print("\nFetching tracking information...")
result = self.get_tracking.execute(
tracking_number=tracking_number,
carrier=carrier,
force_refresh=refresh
)
if result['success']:
print(f"\nPackage: {result['nickname'] or result['tracking_number']}")
print(f"Carrier: {result['carrier']}")
print(f"Status: {result['status']}")
print(f"Last updated: {result['updated_at']}")
if result.get('next_carrier'):
print(f"\nHandoff detected!")
print(f"Next carrier: {result['next_carrier']}")
if result.get('next_tracking_number'):
print(f"Next tracking: {result['next_tracking_number']}")
print(f"\nTracking History ({len(result['events'])} events):")
print("-" * 80)
for event in result['events']:
timestamp = datetime.fromisoformat(event['timestamp']).strftime('%Y-%m-%d %H:%M')
print(f"{timestamp} | {event['location']}")
print(f" {event['description']}")
print(f" Type: {event['event_type']} | Carrier: {event['carrier']}")
print()
else:
print(f"\nError: {result['error']}")
def _list_packages_interactive(self):
print("\n--- List Packages ---")
user_name = input("Your name: ").strip()
if not user_name:
print("Error: Name is required")
return
packages = self.repository.get_packages_by_user(user_name)
if not packages:
print("\nNo packages found for this user.")
return
print(f"\nFound {len(packages)} package(s):")
print("-" * 80)
for pkg in packages:
status_indicator = "[ACTIVE]" if pkg.is_active else "[INACTIVE]"
print(f"\n{status_indicator} {pkg.nickname or pkg.tracking_number}")
print(f" Tracking: {pkg.tracking_number}")
print(f" Carrier: {pkg.carrier}")
print(f" Status: {pkg.status.value}")
print(f" Updated: {pkg.updated_at.strftime('%Y-%m-%d %H:%M')}")
if pkg.tracking_history:
latest = pkg.tracking_history[-1]
print(f" Latest: {latest.description}")
if __name__ == '__main__':
main()
This complete implementation provides both an API server and a command-line interface. The CLI allows interactive testing of all major features including adding packages, querying tracking information, and listing packages. The API mode starts a web server that can be integrated with other applications.
The system handles graceful shutdown through signal handlers. When the user presses Ctrl+C or the process receives a termination signal, the monitoring agent stops cleanly and the application exits without leaving orphaned threads or incomplete database transactions.
The configuration system allows customization of all aspects including database location, LLM backend selection, API credentials, notification channels, and monitoring intervals. Environment variables can override configuration file settings for deployment flexibility.
The implementation follows clean architecture principles with clear separation between domain logic, use cases, adapters, and infrastructure. Each component has a single responsibility and depends only on abstractions rather than concrete implementations. This makes the code testable, maintainable, and extensible.
The LLM integration supports multiple GPU architectures automatically detecting available hardware and configuring the appropriate backend. Users can choose between local models for privacy and cost savings or remote APIs for convenience and performance. The handoff detection service uses the LLM to extract structured information from unstructured tracking event descriptions.
The monitoring agent runs continuously in the background checking active packages for updates. It implements intelligent scheduling to check frequently updated packages more often while reducing API calls for stale packages. When it detects carrier handoffs, it automatically creates new package entries and continues tracking seamlessly.
The notification system supports multiple channels and allows users to configure which events trigger notifications. Email notifications provide detailed information about package updates while webhook notifications enable integration with other systems like Slack or custom dashboards.
This package tracking tool provides a comprehensive solution for monitoring shipments across multiple carriers with intelligent handoff detection and automated updates. The architecture supports extension with new carriers, notification channels, and LLM backends without modifying core business logic.
