bd44771738
- Updated links in index_old.html, varekob_salg.html, log.html, opportunities.html, detail.html, and various frontend files to point to the new versioned sag URLs. - Modified reminder_notification_service.py to reflect the new sag URL structure in notifications. - Added FedEx shipment management functionality, including API client, service layer, and router for handling FedEx bookings, tracking, and cancellations. - Created database migration for FedEx shipments, including tables for shipments, packages, and tracking events.
88 lines
3.7 KiB
Python
88 lines
3.7 KiB
Python
import logging
|
|
from typing import Any, Dict, List
|
|
|
|
import aiohttp
|
|
|
|
from app.core.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FedExApiClient:
|
|
def __init__(self) -> None:
|
|
self.base_url = (settings.FEDEX_BASE_URL or "").rstrip("/")
|
|
self.timeout_seconds = max(5, int(settings.FEDEX_TIMEOUT_SECONDS or 20))
|
|
|
|
def _headers(self) -> Dict[str, str]:
|
|
return {
|
|
"Content-Type": "application/json",
|
|
"X-API-KEY": settings.FEDEX_API_KEY or "",
|
|
"X-API-SECRET": settings.FEDEX_API_SECRET or "",
|
|
"X-FEDEX-ACCOUNT": settings.FEDEX_ACCOUNT_NUMBER or "",
|
|
}
|
|
|
|
async def create_shipment(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
if not self.base_url:
|
|
raise RuntimeError("FedEx base URL is not configured")
|
|
|
|
url = f"{self.base_url}/shipments"
|
|
logger.info("🚀 FedEx create shipment request sent")
|
|
|
|
timeout = aiohttp.ClientTimeout(total=self.timeout_seconds)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.post(url, json=payload, headers=self._headers()) as response:
|
|
body = await response.text()
|
|
if response.status >= 400:
|
|
logger.error("❌ FedEx create shipment failed (%s): %s", response.status, body)
|
|
raise RuntimeError(f"FedEx create shipment failed: HTTP {response.status}")
|
|
return await response.json()
|
|
|
|
async def get_tracking(self, tracking_number: str) -> Dict[str, Any]:
|
|
if not self.base_url:
|
|
raise RuntimeError("FedEx base URL is not configured")
|
|
|
|
url = f"{self.base_url}/tracking/{tracking_number}"
|
|
timeout = aiohttp.ClientTimeout(total=self.timeout_seconds)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.get(url, headers=self._headers()) as response:
|
|
body = await response.text()
|
|
if response.status >= 400:
|
|
logger.error("❌ FedEx tracking failed (%s): %s", response.status, body)
|
|
raise RuntimeError(f"FedEx tracking failed: HTTP {response.status}")
|
|
return await response.json()
|
|
|
|
async def cancel_shipment(self, tracking_number: str) -> Dict[str, Any]:
|
|
if not self.base_url:
|
|
raise RuntimeError("FedEx base URL is not configured")
|
|
|
|
url = f"{self.base_url}/shipments/{tracking_number}/cancel"
|
|
timeout = aiohttp.ClientTimeout(total=self.timeout_seconds)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.post(url, headers=self._headers()) as response:
|
|
body = await response.text()
|
|
if response.status >= 400:
|
|
logger.error("❌ FedEx cancel failed (%s): %s", response.status, body)
|
|
raise RuntimeError(f"FedEx cancel failed: HTTP {response.status}")
|
|
return await response.json()
|
|
|
|
|
|
def parse_tracking_events(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
raw_events = payload.get("events") or []
|
|
if not isinstance(raw_events, list):
|
|
return []
|
|
|
|
normalized: List[Dict[str, Any]] = []
|
|
for event in raw_events:
|
|
if not isinstance(event, dict):
|
|
continue
|
|
normalized.append(
|
|
{
|
|
"status": str(event.get("status") or "unknown"),
|
|
"description": event.get("description"),
|
|
"event_timestamp": event.get("event_timestamp") or event.get("timestamp"),
|
|
"location_city": event.get("location_city") or event.get("city"),
|
|
"location_country": event.get("location_country") or event.get("country"),
|
|
}
|
|
)
|
|
return normalized
|