輸送管理データテンプレート

import requests
import pandas as pd
import datetime
import time

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># Replace with your actual project44 API Key
API_KEY = class="hl-string">'YOUR_API_KEY'

class=class="hl-string">"hl-comment"># Define the date range for the extraction (ISO 8601 format)
START_DATE = class="hl-string">'2023-01-01T00:00:00Z'
END_DATE = class="hl-string">'2023-06-30T23:59:59Z'

class=class="hl-string">"hl-comment"># project44 API Endpoint
BASE_URL = class="hl-string">'https://api.project44.com'

class=class="hl-string">"hl-comment"># Pagination settings
PAGE_SIZE = 100

class=class="hl-string">"hl-comment"># --- Mappings ---
class=class="hl-string">"hl-comment"># Maps project44 event reason codes to the required ActivityName
class=class="hl-string">"hl-comment"># This mapping is critical and may need to be customized based on your specific project44 event configuration.
EVENT_CODE_MAP = {
    class="hl-string">'AS': class="hl-string">'Carrier Assigned',
    class="hl-string">'PU': class="hl-string">'Goods Picked Up',
    class="hl-string">'DP': class="hl-string">'Goods Picked Up', class=class="hl-string">"hl-comment"># Departure from origin often signifies pickup
    class="hl-string">'ETA': class="hl-string">'ETA Updated',
    class="hl-string">'EDD': class="hl-string">'ETA Updated', class=class="hl-string">"hl-comment"># Estimated Delivery Date update
    class="hl-string">'DELAYED': class="hl-string">'Shipment Delayed',
    class="hl-string">'AB': class="hl-string">'At Border Crossing',
    class="hl-string">'CC': class="hl-string">'Customs Cleared',
    class="hl-string">'AR': class="hl-string">'Arrival At Destination',
    class="hl-string">'DL': class="hl-string">'Goods Delivered',
    class="hl-string">'POD': class="hl-string">'Proof of Delivery Received',
    class=class="hl-string">"hl-comment"># The following events are less common in standard visibility data and may require custom events in project44.
    class=class="hl-string">"hl-comment"># They are included here as placeholders for a complete mapping.
    class="hl-string">'PICKUP_SCHEDULED': class="hl-string">'Pickup Scheduled',
    class="hl-string">'DELIVERY_SCHEDULED': class="hl-string">'Delivery Scheduled',
    class="hl-string">'FREIGHT_BILL_AUDITED': class="hl-string">'Freight Bill Audited',
    class="hl-string">'PAYMENT_PROCESSED': class="hl-string">'Payment Processed'
}

def get_shipments(page_number):
    class="hl-string">"""Fetches a single page of shipments from the project44 API."""
    headers = {
        class="hl-string">'Authorization': fclass="hl-string">'ApiKey {API_KEY}',
        class="hl-string">'Content-Type': class="hl-string">'application/json'
    }
    params = {
        class="hl-string">'createdSince': START_DATE,
        class="hl-string">'createdUntil': END_DATE,
        class="hl-string">'size': PAGE_SIZE,
        class="hl-string">'page': page_number
    }
    url = fclass="hl-string">'{BASE_URL}/api/v4/shipments'
    print(fclass="hl-string">"Fetching page {page_number}...")
    try:
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status() class=class="hl-string">"hl-comment"># Raises an HTTPError for bad responses (4xx or 5xx)
        return response.json()
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"Error fetching data on page {page_number}: {e}")
        return None

def process_shipment_data(shipment):
    class="hl-string">"""Processes a single shipment and extracts its events."""
    events = []
    shipment_id = shipment.get(class="hl-string">'shipmentId')
    last_data_update = datetime.datetime.utcnow().isoformat() + class="hl-string">'Z'

    class=class="hl-string">"hl-comment"># Common attributes for all events of this shipment
    base_attributes = {
        class="hl-string">'Shipment': shipment_id,
        class="hl-string">'SourceSystem': class="hl-string">'project44',
        class="hl-string">'LastDataUpdate': last_data_update,
        class="hl-string">'CarrierName': shipment.get(class="hl-string">'carrierName'),
        class="hl-string">'ModeOfTransport': shipment.get(class="hl-string">'mode'),
        class="hl-string">'RequestedDeliveryDate': shipment.get(class="hl-string">'requestedDeliveryDate'),
        class="hl-string">'OriginCountry': shipment.get(class="hl-string">'origin', {}).get(class="hl-string">'address', {}).get(class="hl-string">'country'),
        class="hl-string">'DestinationCountry': shipment.get(class="hl-string">'destination', {}).get(class="hl-string">'address', {}).get(class="hl-string">'country'),
        class=class="hl-string">"hl-comment"># TransportationCost is often not available in visibility data. Sourced from TMS/ERP.
        class="hl-string">'TransportationCost': shipment.get(class="hl-string">'cost', {}).get(class="hl-string">'total'), 
        class="hl-string">'DelayReason': None class=class="hl-string">"hl-comment"># To be populated by delay events
    }

    class=class="hl-string">"hl-comment"># 1. Activity: Shipment Created
    creation_time = shipment.get(class="hl-string">'createdDate')
    if creation_time:
        events.append({
            **base_attributes,
            class="hl-string">'ActivityName': class="hl-string">'Shipment Created',
            class="hl-string">'EventTime': creation_time
        })

    class=class="hl-string">"hl-comment"># 2. Process all other activities from the tracking history
    tracking_history = shipment.get(class="hl-string">'trackingHistory', [])
    for event in tracking_history:
        event_code = event.get(class="hl-string">'reason', {}).get(class="hl-string">'code')
        activity_name = EVENT_CODE_MAP.get(event_code)
        
        if activity_name:
            event_attributes = base_attributes.copy()
            event_attributes[class="hl-string">'ActivityName'] = activity_name
            event_attributes[class="hl-string">'EventTime'] = event.get(class="hl-string">'eventTimestamp')
            
            if activity_name == class="hl-string">'Shipment Delayed':
                event_attributes[class="hl-string">'DelayReason'] = event.get(class="hl-string">'reason', {}).get(class="hl-string">'description')

            events.append(event_attributes)

    return events

class=class="hl-string">"hl-comment"># --- Main Execution ---
all_events = []
current_page = 0

while True:
    data = get_shipments(current_page)
    if not data or not data.get(class="hl-string">'content'):
        print(class="hl-string">"No more data to fetch or API error.")
        break

    shipments = data.get(class="hl-string">'content', [])
    for shipment in shipments:
        all_events.extend(process_shipment_data(shipment))
    
    if data.get(class="hl-string">'last', True):
        print(class="hl-string">"Reached the last page.")
        break

    current_page += 1
    time.sleep(1) class=class="hl-string">"hl-comment"># Be respectful of API rate limits

class=class="hl-string">"hl-comment"># Convert to DataFrame and save to CSV
if all_events:
    df = pd.DataFrame(all_events)
    class=class="hl-string">"hl-comment"># Reorder columns for clarity
    columns_order = [
        class="hl-string">'Shipment', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'SourceSystem', class="hl-string">'LastDataUpdate',
        class="hl-string">'CarrierName', class="hl-string">'DelayReason', class="hl-string">'ModeOfTransport', class="hl-string">'RequestedDeliveryDate',
        class="hl-string">'OriginCountry', class="hl-string">'DestinationCountry', class="hl-string">'TransportationCost'
    ]
    class=class="hl-string">"hl-comment"># Ensure all required columns exist, even if empty
    for col in columns_order:
        if col not in df.columns:
            df[col] = None
    
    df = df[columns_order]
    
    output_filename = class="hl-string">'project44_event_log.csv'
    df.to_csv(output_filename, index=False, date_format=class="hl-string">'%Y-%m-%dT%H:%M:%SZ')
    print(fclass="hl-string">"Extraction complete. Event log saved to {output_filename}")
else:
    print(class="hl-string">"No events were extracted. Please check your configuration and API response.")

query getShipmentEvents($cursor: String, $updatedSince: String!) {
  shipments(first: 100, after: $cursor, filter: {updatedSince: $updatedSince}) {
    pageInfo {
      hasNextPage
      endCursor
    }
    nodes {
      id
      createdAt
      carrierName
      mode
      requestedDeliveryDate
      origin {
        country
      }
      destination {
        country
      }
      events {
        timestamp
        code
        description
        reason
      }
    }
  }
}