Multi-shipment BOL support

{
  "cost": {
    "input_tokens": 46,
    "estimated_usd": 0.086463,
    "output_tokens": 5755
  },
  "model": "claude-sonnet-4-20250514",
  "status": "success",
  "transport": "anthropic_api",
  "latency_ms": 66102,
  "response_text": "I'll help you implement a comprehensive multi-shipment Bill of Lading system that handles split/consolidation logic, line-item tracking, and multi-container references.\n\n```python\nfrom dataclasses import dataclass, field\nfrom typing import Dict, List, Optional, Set, Tuple\nfrom enum import Enum\nfrom datetime import datetime\nfrom decimal import Decimal\nimport uuid\n\nclass BOLType(Enum):\n    CONSOLIDATED = \"consolidated\"  # Multiple shipments into one BOL\n    SPLIT = \"split\"               # One shipment split into multiple BOLs\n    STANDARD = \"standard\"         # One-to-one shipment-BOL\n\nclass ShipmentStatus(Enum):\n    PENDING = \"pending\"\n    IN_TRANSIT = \"in_transit\"\n    DELIVERED = \"delivered\"\n    EXCEPTION = \"exception\"\n\n@dataclass\nclass LineItem:\n    \"\"\"Individual line item within a shipment\"\"\"\n    id: str = field(default_factory=lambda: str(uuid.uuid4()))\n    sku: str = \"\"\n    description: str = \"\"\n    quantity: int = 0\n    unit_weight: Decimal = Decimal('0')\n    unit_dimensions: Dict[str, Decimal] = field(default_factory=dict)\n    hazmat: bool = False\n    commodity_class: Optional[str] = None\n    value: Decimal = Decimal('0')\n    \n    @property\n    def total_weight(self) -> Decimal:\n        return self.unit_weight * self.quantity\n    \n    @property\n    def total_volume(self) -> Decimal:\n        if all(k in self.unit_dimensions for k in ['length', 'width', 'height']):\n            volume = (self.unit_dimensions['length'] * \n                     self.unit_dimensions['width'] * \n                     self.unit_dimensions['height'])\n            return volume * self.quantity\n        return Decimal('0')\n\n@dataclass\nclass Container:\n    \"\"\"Container information for multi-container shipments\"\"\"\n    id: str = field(default_factory=lambda: str(uuid.uuid4()))\n    container_number: str = \"\"\n    container_type: str = \"\"  # 20ft, 40ft, etc.\n    seal_number: str = \"\"\n    weight: Decimal = Decimal('0')\n    line_items: List[LineItem] = field(default_factory=list)\n    \n    @property\n    def total_line_items_weight(self) -> Decimal:\n        return sum(item.total_weight for item in self.line_items)\n\n@dataclass\nclass ShipmentReference:\n    \"\"\"Reference to original shipment and its components\"\"\"\n    shipment_id: str\n    order_number: str = \"\"\n    po_number: str = \"\"\n    tracking_number: str = \"\"\n    line_items: List[LineItem] = field(default_factory=list)\n    containers: List[Container] = field(default_factory=list)\n    status: ShipmentStatus = ShipmentStatus.PENDING\n    \n    @property\n    def total_weight(self) -> Decimal:\n        return sum(item.total_weight for item in self.line_items)\n    \n    @property\n    def total_pieces(self) -> int:\n        return sum(item.quantity for item in self.line_items)\n\nclass MultiShipmentBOL:\n    \"\"\"Multi-shipment Bill of Lading with consolidation and split support\"\"\"\n    \n    def __init__(self, bol_number: str):\n        self.bol_number = bol_number\n        self.bol_type = BOLType.STANDARD\n        self.created_date = datetime.now()\n        self.shipments: Dict[str, ShipmentReference] = {}\n        self.parent_bol_id: Optional[str] = None  # For split BOLs\n        self.child_bol_ids: Set[str] = set()      # For consolidated BOLs that are split\n        \n        # Shipping details\n        self.shipper: Dict[str, str] = {}\n        self.consignee: Dict[str, str] = {}\n        self.carrier: Dict[str, str] = {}\n        self.pickup_date: Optional[datetime] = None\n        self.delivery_date: Optional[datetime] = None\n        \n        # Tracking\n        self._line_item_index: Dict[str, Tuple[str, str]] = {}  # line_item_id -> (shipment_id, container_id)\n        self._container_index: Dict[str, str] = {}  # container_id -> shipment_id\n    \n    def add_shipment(self, shipment: ShipmentReference) -> bool:\n        \"\"\"Add a shipment to the BOL\"\"\"\n        if shipment.shipment_id in self.shipments:\n            return False\n        \n        self.shipments[shipment.shipment_id] = shipment\n        \n        # Update indices\n        for container in shipment.containers:\n            self._container_index[container.id] = shipment.shipment_id\n            for line_item in container.line_items:\n                self._line_item_index[line_item.id] = (shipment.shipment_id, container.id)\n        \n        # Update BOL type\n        self._update_bol_type()\n        return True\n    \n    def remove_shipment(self, shipment_id: str) -> bool:\n        \"\"\"Remove a shipment from the BOL\"\"\"\n        if shipment_id not in self.shipments:\n            return False\n        \n        shipment = self.shipments[shipment_id]\n        \n        # Remove from indices\n        for container in shipment.containers:\n            if container.id in self._container_index:\n                del self._container_index[container.id]\n            for line_item in container.line_items:\n                if line_item.id in self._line_item_index:\n                    del self._line_item_index[line_item.id]\n        \n        del self.shipments[shipment_id]\n        self._update_bol_type()\n        return True\n    \n    def consolidate_shipments(self, shipment_ids: List[str], new_bol_number: str) -> 'MultiShipmentBOL':\n        \"\"\"Consolidate multiple shipments into a new BOL\"\"\"\n        consolidated_bol = MultiShipmentBOL(new_bol_number)\n        consolidated_bol.bol_type = BOLType.CONSOLIDATED\n        \n        # Copy shipping details\n        consolidated_bol.shipper = self.shipper.copy()\n        consolidated_bol.consignee = self.consignee.copy()\n        consolidated_bol.carrier = self.carrier.copy()\n        \n        # Move shipments to consolidated BOL\n        for shipment_id in shipment_ids:\n            if shipment_id in self.shipments:\n                shipment = self.shipments[shipment_id]\n                consolidated_bol.add_shipment(shipment)\n                self.remove_shipment(shipment_id)\n        \n        return consolidated_bol\n    \n    def split_bol(self, split_config: Dict[str, List[str]]) -> List['MultiShipmentBOL']:\n        \"\"\"\n        Split current BOL into multiple BOLs\n        split_config: {new_bol_number: [shipment_ids]}\n        \"\"\"\n        split_bols = []\n        \n        for new_bol_number, shipment_ids in split_config.items():\n            split_bol = MultiShipmentBOL(new_bol_number)\n            split_bol.bol_type = BOLType.SPLIT\n            split_bol.parent_bol_id = self.bol_number\n            \n            # Copy shipping details\n            split_bol.shipper = self.shipper.copy()\n            split_bol.consignee = self.consignee.copy()\n            split_bol.carrier = self.carrier.copy()\n            \n            # Move specified shipments\n            for shipment_id in shipment_ids:\n                if shipment_id in self.shipments:\n                    shipment = self.shipments[shipment_id]\n                    split_bol.add_shipment(shipment)\n            \n            split_bols.append(split_bol)\n            self.child_bol_ids.add(new_bol_number)\n        \n        return split_bols\n    \n    def track_line_item(self, line_item_id: str) -> Optional[Dict]:\n        \"\"\"Track a specific line item across shipments and containers\"\"\"\n        if line_item_id not in self._line_item_index:\n            return None\n        \n        shipment_id, container_id = self._line_item_index[line_item_id]\n        shipment = self.shipments[shipment_id]\n        \n        # Find the container and line item\n        container = next((c for c in shipment.containers if c.id == container_id), None)\n        line_item = next((li for li in container.line_items if li.id == line_item_id), None)\n        \n        return {\n            'bol_number': self.bol_number,\n            'shipment_id': shipment_id,\n            'container_id': container_id,\n            'container_number': container.container_number,\n            'tracking_number': shipment.tracking_number,\n            'line_item': line_item,\n            'status': shipment.status.value\n        }\n    \n    def get_container_contents(self, container_id: str) -> Optional[Dict]:\n        \"\"\"Get all contents of a specific container\"\"\"\n        if container_id not in self._container_index:\n            return None\n        \n        shipment_id = self._container_index[container_id]\n        shipment = self.shipments[shipment_id]\n        container = next((c for c in shipment.containers if c.id == container_id), None)\n        \n        return {\n            'container_id': container_id,\n            'container_number': container.container_number,\n            'shipment_id': shipment_id,\n            'line_items': container.line_items,\n            'total_weight': container.total_line_items_weight,\n            'total_pieces': sum(item.quantity for item in container.line_items)\n        }\n    \n    def validate_consolidation(self, shipment_ids: List[str]) -> Dict[str, List[str]]:\n        \"\"\"Validate if shipments can be consolidated\"\"\"\n        issues = []\n        warnings = []\n        \n        # Check if all shipments exist\n        for shipment_id in shipment_ids:\n            if shipment_id not in self.shipments:\n                issues.append(f\"Shipment {shipment_id} not found\")\n        \n        if issues:\n            return {'issues': issues, 'warnings': warnings}\n        \n        shipments = [self.shipments[sid] for sid in shipment_ids]\n        \n        # Check for conflicting delivery addresses\n        consignees = set()\n        for shipment in shipments:\n            # Assuming consignee info is stored in shipment or BOL level\n            consignee_key = f\"{self.consignee.get('company', '')}-{self.consignee.get('address', '')}\"\n            consignees.add(consignee_key)\n        \n        if len(consignees) > 1:\n            warnings.append(\"Multiple consignees detected in consolidation\")\n        \n        # Check for hazmat compatibility\n        hazmat_items = []\n        for shipment in shipments:\n            for line_item in shipment.line_items:\n                if line_item.hazmat:\n                    hazmat_items.append(f\"{shipment.shipment_id}:{line_item.sku}\")\n        \n        if hazmat_items:\n            warnings.append(f\"Hazmat items detected: {', '.join(hazmat_items)}\")\n        \n        # Check total weight/volume constraints\n        total_weight = sum(shipment.total_weight for shipment in shipments)\n        if total_weight > Decimal('80000'):  # Example weight limit\n            warnings.append(f\"Total weight ({total_weight} lbs) may exceed carrier limits\")\n        \n        return {'issues': issues, 'warnings': warnings}\n    \n    def _update_bol_type(self):\n        \"\"\"Update BOL type based on current shipments\"\"\"\n        shipment_count = len(self.shipments)\n        if shipment_count == 0:\n            self.bol_type = BOLType.STANDARD\n        elif shipment_count == 1:\n            self.bol_type = BOLType.STANDARD\n        else:\n            if self.parent_bol_id:\n                self.bol_type = BOLType.SPLIT\n            else:\n                self.bol_type = BOLType.CONSOLIDATED\n    \n    @property\n    def total_weight(self) -> Decimal:\n        \"\"\"Calculate total weight across all shipments\"\"\"\n        return sum(shipment.total_weight for shipment in self.shipments.values())\n    \n    @property\n    def total_pieces(self) -> int:\n        \"\"\"Calculate total pieces across all shipments\"\"\"\n        return sum(shipment.total_pieces for shipment in self.shipments.values())\n    \n    @property\n    def all_containers(self) -> List[Container]:\n        \"\"\"Get all containers across all shipments\"\"\"\n        containers = []\n        for shipment in self.shipments.values():\n            containers.extend(shipment.containers)\n        return containers\n    \n    def get_summary(self) -> Dict:\n        \"\"\"Get BOL summary information\"\"\"\n        return {\n            'bol_number': self.bol_number,\n            'bol_type': self.bol_type.value,\n            'created_date': self.created_date.isoformat(),\n            'shipment_count': len(self.shipments),\n            'container_count': len(self.all_containers),\n            'total_weight': str(self.total_weight),\n            'total_pieces': self.total_pieces,\n            'shipment_ids': list(self.shipments.keys()),\n            'parent_bol': self.parent_bol_id,\n            'child_bols': list(self.child_bol_ids)\n        }\n\nclass BOLManager:\n    \"\"\"Manager for handling multiple BOLs and their relationships\"\"\"\n    \n    def __init__(self):\n        self.bols: Dict[str, MultiShipmentBOL] = {}\n        self.shipment_to_bol_mapping: Dict[str, str] = {}\n    \n    def create_bol(self, bol_number: str) -> MultiShipmentBOL:\n        \"\"\"Create a new BOL\"\"\"\n        bol = MultiShipmentBOL(bol_number)\n        self.bols[bol_number] = bol\n        return bol\n    \n    def add_bol(self, bol: MultiShipmentBOL) -> bool:\n        \"\"\"Add an existing BOL to management\"\"\"\n        if bol.bol_number in self.bols:\n            return False\n        \n        self.bols[bol.bol_number] = bol\n        \n        # Update shipment mapping\n        for shipment_id in bol.shipments.keys():\n            self.shipment_to_bol_mapping[shipment_id] = bol.bol_number\n        \n        return True\n    \n    def find_bol_by_shipment(self, shipment_id: str) -> Optional[MultiShipmentBOL]:\n        \"\"\"Find BOL containing a specific shipment\"\"\"\n        bol_number = self.shipment_to_bol_mapping.get(shipment_id)\n        return self.bols.get(bol_number) if bol_number else None\n    \n    def consolidate_across_bols(self, source_bol_numbers: List[str], \n                               target_bol_number: str) -> Optional[MultiShipmentBOL]:\n        \"\"\"Consolidate shipments from multiple BOLs into a new BOL\"\"\"\n        consolidated_bol = self.create_bol(target_bol_number)\n        consolidated_bol.bol_type = BOLType.CONSOLIDATED\n        \n        for bol_number in source_bol_numbers:\n            if bol_number in self.bols:\n                source_bol = self.bols[bol_number]\n                \n                # Move all shipments to consolidated BOL\n                shipment_ids = list(source_bol.shipments.keys())\n                for shipment_id in shipment_ids:\n                    shipment = source_bol.shipments[shipment_id]\n                    source_bol.remove_shipment(shipment_id)\n                    consolidated_bol.add_shipment(shipment)\n                    self.shipment_to_bol_mapping[shipment_id] = target_bol_number\n        \n        return consolidated_bol\n    \n    def track_shipment_history(self, shipment_id: str) -> List[Dict]:\n        \"\"\"Track the history of a shipment across BOL changes\"\"\"\n        # This would typically query a database for historical records\n        # For now, return current state\n        current_bol = self.find_bol_by_shipment(shipment_id)\n        if not current_bol:\n            return []\n        \n        return [{\n            'timestamp': datetime.now().isoformat(),\n            'action': 'current_state',\n            'bol_number': current_bol.bol_number,\n            'bol_type': current_bol.bol_type.value\n        }]\n    \n    def get_related_bols(self, bol_number: str) -> Dict[str, List[str]]:\n        \"\"\"Get all BOLs related to the given BOL (parent/children)\"\"\"\n        if bol_number not in self.bols:\n            return {'parent': [], 'children': []}\n        \n        bol = self.bols[bol_number]\n        parent = [bol.parent_bol_id] if bol.parent_bol_id else []\n        children = list(bol.child_bol_ids)\n        \n        return {'parent': parent, 'children': children}\n\n# Example usage and testing\ndef example_usage():\n    \"\"\"Demonstrate the multi-shipment BOL system\"\"\"\n    \n    # Create BOL manager\n    manager = BOLManager()\n    \n    # Create line items\n    line_item1 = LineItem(\n        sku=\"WIDGET001\",\n        description=\"Blue Widget\",\n        quantity=100,\n        unit_weight=Decimal('2.5'),\n        unit_dimensions={'length': Decimal('10'), 'width': Decimal('8'), 'height': Decimal('6')},\n        value=Decimal('1000')\n    )\n    \n    line_item2 = LineItem(\n        sku=\"GADGET002\",\n        description=\"Red Gadget\",\n        quantity=50,\n        unit_weight=Decimal('1.2'),\n        hazmat=True,\n        value=Decimal('500')\n    )\n    \n    # Create containers\n    container1 = Container(\n        container_number=\"CONT001\",\n        container_type=\"20ft\",\n        seal_number=\"SEAL123\",\n        line_items=[line_item1]\n    )\n    \n    container2 = Container(\n        container_number=\"CONT002\", \n        container_type=\"20ft\",\n        seal_number=\"SEAL456\",\n        line_items=[line_item2]\n    )\n    \n    # Create shipments\n    shipment1 = ShipmentReference(\n        shipment_id=\"SHIP001\",\n        order_number=\"ORD001\",\n        tracking_number=\"TRACK001\",\n        containers=[container1],\n        line_items=[line_item1]\n    )\n    \n    shipment2 = ShipmentReference(\n        shipment_id=\"SHIP002\",\n        order_number=\"ORD002\", \n        tracking_number=\"TRACK002\",\n        containers=[container2],\n        line_items=[line_item2]\n    )\n    \n    # Create initial BOLs\n    bol1 = manager.create_bol(\"BOL001\")\n    bol1.add_shipment(shipment1)\n    \n    bol2 = manager.create_bol(\"BOL002\")\n    bol2.add_shipment(shipment2)\n    \n    print(\"Initial BOLs created:\")\n    print(f\"BOL1 Summary: {bol1.get_summary()}\")\n    print(f\"BOL2 Summary: {bol2.get_summary()}\")\n    \n    # Consolidate shipments\n    print(\"\\nConsolidating shipments...\")\n    validation = bol1.validate_consolidation([\"SHIP001\", \"SHIP002\"])\n    print(f\"Validation result: {validation}\")\n    \n    consolidated_bol = manager.consolidate_across_bols([\"BOL001\", \"BOL002\"], \"BOL_CONSOLIDATED\")\n    print(f\"Consolidated BOL Summary: {consolidated_bol.get_summary()}\")\n    \n    # Track line items\n    print(f\"\\nTracking line item {line_item1.id}:\")\n    tracking_info = consolidated_bol.track_line_item(line_item1.id)\n    print(f\"Tracking info: {tracking_info}\")\n    \n    # Split consolidated BOL\n    print(\"\\nSplitting consolidated BOL...\")\n    split_config = {\n        \"BOL_SPLIT_1\": [\"SHIP001\"],\n        \"BOL_SPLIT_2\": [\"SHIP002\"]\n    }\n    split_bols = consolidated_bol.split_bol(split_config)\n    \n    for split_bol in split_bols:\n        manager.add_bol(split_bol)\n        print(f\"Split BOL Summary: {split_bol.get_summary()}\")\n    \n    # Get container contents\n    print(f\"\\nContainer {container1.id} contents:\")\n    container_contents = consolidated_bol.get_container_contents(container1.id)\n    if container_contents:\n        print(f\"Container contents: {container_contents}\")\n\nif __name__ == \"__main__\":\n    example_usage()\n```\n\nThis comprehensive multi-shipment BOL system provides:\n\n## Key Features:\n\n1. **Multi-Shipment Support**: Single BOL can contain multiple shipments with proper tracking\n\n2. **Split/Consolidation Logic**: \n   - Consolidate multiple BOLs into one\n   - Split single BOL into multiple BOLs\n   - Validation for consolidation compatibility\n\n3. **Line-Item Tracking**: Track individual items across shipments and containers\n\n4. **Multi-Container References**: Support for multiple containers per shipment\n\n5. **BOL Types**: Handles consolidated, split, and standard BOL types\n\n## Additional Features:\n\n- **Relationship Tracking**: Parent/child BOL relationships\n- **Validation System**: Checks for hazmat, weight limits, address conflicts\n- **Index Management**: Fast lookups for line items and containers\n- **History Tracking**: Track shipment movement across BOLs\n- **Weight/Volume Calculations**: Automatic totaling across all levels\n\nThe system is designed to handle complex logistics scenarios while maintaining data integrity and providing comprehensive tracking capabilities."
}