ecmr_client.py

"""Python ECMR Client - SDK for Cargoffer ECMR API"""
import requests
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from datetime import datetime

@dataclass
class ECMR:
    service_code: str
    status: str
    sender: Dict[str, str]
    receiver: Dict[str, str]
    from_address: Dict[str, str]
    to_address: Dict[str, str]
    goods: Dict[str, Any]

class ECMRClient:
    """Client for Cargoffer ECMR API"""
    
    BASE_URL = "https://ecmr.api.cargoffer.com"
    
    def __init__(self, api_key: str, base_url: str = None):
        self.api_key = api_key
        self.base_url = base_url or self.BASE_URL
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Apikey {api_key}",
            "Content-Type": "application/json"
        })
    
    # === ECMR Operations ===
    
    def create_ecmr(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Create a new eCMR"""
        response = self.session.post(f"{self.base_url}/ecmr", json=payload)
        response.raise_for_status()
        return response.json().get("data", {})
    
    def get_ecmr(self, service_code: str) -> Dict[str, Any]:
        """Get eCMR by service code"""
        response = self.session.get(f"{self.base_url}/ecmr/{service_code}")
        response.raise_for_status()
        return response.json().get("data", {})
    
    def update_ecmr(self, service_code: str, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Update eCMR"""
        response = self.session.put(f"{self.base_url}/ecmr/{service_code}", json=payload)
        response.raise_for_status()
        return response.json().get("data", {})
    
    def delete_ecmr(self, service_code: str) -> bool:
        """Delete eCMR"""
        response = self.session.delete(f"{self.base_url}/ecmr/{service_code}")
        return response.status_code == 200
    
    def lock_ecmr(self, service_code: str) -> Dict[str, Any]:
        """Lock eCMR (legally close)"""
        response = self.session.put(f"{self.base_url}/ecmr/lock/{service_code}")
        response.raise_for_status()
        return response.json().get("data", {})
    
    # === Signature Operations ===
    
    def sign_sender(self, service_code: str, signature: str) -> Dict[str, Any]:
        """Sign as sender"""
        response = self.session.put(
            f"{self.base_url}/ecmr/sign/sender/{service_code}",
            json={"signature": signature, "signed_at": datetime.now().isoformat()}
        )
        response.raise_for_status()
        return response.json()
    
    def sign_pickup(self, service_code: str, signature: str) -> Dict[str, Any]:
        """Sign pickup"""
        response = self.session.put(
            f"{self.base_url}/ecmr/sign/pickup/{service_code}",
            json={"signature": signature, "signed_at": datetime.now().isoformat()}
        )
        response.raise_for_status()
        return response.json()
    
    def sign_delivery(self, service_code: str, signature: str) -> Dict[str, Any]:
        """Sign delivery"""
        response = self.session.put(
            f"{self.base_url}/ecmr/sign/delivery/{service_code}",
            json={"signature": signature, "signed_at": datetime.now().isoformat()}
        )
        response.raise_for_status()
        return response.json()
    
    # === QR Operations ===
    
    def get_qr(self, service_code: str) -> str:
        """Get QR code URL"""
        response = self.session.get(f"{self.base_url}/ecmr/qr/{service_code}")
        response.raise_for_status()
        return response.json().get("data", {}).get("qr_url", "")
    
    def validate_qr(self, service_code: str) -> Dict[str, Any]:
        """Validate QR code"""
        response = self.session.get(f"{self.base_url}/ecmr/qr/check/{service_code}")
        response.raise_for_status()
        return response.json().get("data", {})
    
    # === Driver Operations ===
    
    def create_driver(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Create driver"""
        response = self.session.post(f"{self.base_url}/drivers/", json=payload)
        response.raise_for_status()
        return response.json().get("data", {})
    
    def list_drivers(self, limit: int = 50) -> List[Dict[str, Any]]:
        """List drivers"""
        response = self.session.get(f"{self.base_url}/drivers/?limit={limit}")
        response.raise_for_status()
        return response.json().get("data", {}).get("docs", [])
    
    def get_driver(self, driver_id: str) -> Dict[str, Any]:
        """Get driver"""
        response = self.session.get(f"{self.base_url}/drivers/{driver_id}")
        response.raise_for_status()
        return response.json().get("data", {})
    
    # === Vehicle Operations ===
    
    def create_vehicle(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Create vehicle"""
        response = self.session.post(f"{self.base_url}/vehicles/", json=payload)
        response.raise_for_status()
        return response.json().get("data", {})
    
    def list_vehicles(self, limit: int = 50) -> List[Dict[str, Any]]:
        """List vehicles"""
        response = self.session.get(f"{self.base_url}/vehicles/?limit={limit}")
        response.raise_for_status()
        return response.json().get("data", {}).get("docs", [])


if __name__ == "__main__":
    # Example usage
    client = ECMRClient("YOUR_API_KEY")
    
    # Create eCMR
    payload = {
        "sender": {"company_name": "ACME SL", "cif": "B12345678"},
        "receiver": {"company_name": "DEST SL", "cif": "B87654321"},
        "from": {"address": "Calle A 1, Madrid", "postal_code": "28001", "country": "ES"},
        "to": {"address": "Calle B 1, Barcelona", "postal_code": "08001", "country": "ES"},
        "goods": {"description": "Mercancía general", "packages": 2, "weight": 500}
    }
    
    result = client.create_ecmr(payload)
    print(f"Created: {result.get('service_code')}")