Incremental Updates of Polygon Events

 
#!/usr/bin/env python3
import os
import logging
import requests
import pandas as pd
from datetime import datetime, timedelta

# --------------------------------------------------------------------
# 1️⃣  Configuration
# ------------------------------------------------------------------
TOKEN = "INSERT MY TOKEN" # ← Replace with your token
HEADERS = {"Authorization": f"Bearer {TOKEN}"}
BASE_URL = "https://apihub.axsmarine.com/global/events/v1"

# Example: only retrieve "shipyard" events.
POLYGON_TYPES = ["shipyard"]
ENTRY_START_DATE = "2025-01-01"     # ISO‑8601
CSV_FILE = "2025_shipyard_events.csv"

# ------------------------------------------------------------------
# 2️⃣  Load cache (if present)
# ------------------------------------------------------------------
last_execution_time: str | None = None
data = pd.DataFrame()

if os.path.exists(CSV_FILE):
    # Ensure we don't miss events updated the same day.
    last_exec_ts = datetime.fromtimestamp(os.path.getmtime(CSV_FILE)) - timedelta(days=1)
    last_execution_time = last_exec_ts.strftime("%Y-%m-%d")
    data = pd.read_csv(CSV_FILE)

# ------------------------------------------------------------------
# 3️⃣  Generic GraphQL query helper
# ------------------------------------------------------------------
def gql_query(query: str, variables: dict) -> dict:
    """Execute a GraphQL request and return the JSON payload."""
    logging.debug(f"Request variables: {variables}")
    resp = requests.post(BASE_URL, json={"query": query, "variables": variables}, headers=HEADERS)
    resp.raise_for_status()
    return resp.json()["data"]

# ------------------------------------------------------------------
# 4️⃣  Retrieve active polygon events
# ------------------------------------------------------------------
def fetch_polygon_events() -> pd.DataFrame:
    """Fetch active events, paginating until exhaustion."""
    page_size = 5000
    variables = {
        "pageSize": page_size,
        "polygonTypes": POLYGON_TYPES,
        "entryDate": {"from": ENTRY_START_DATE},
        "lastUpdated": {"from": last_execution_time} if last_execution_time else None,
    }

    query = """
    query polygonEvents(
        $pageSize: Int,
        $afterCursor: String,
        $polygonTypes: [String],
        $entryDate: RangeDate,
        $lastUpdated: RangeDate
    ) {
        polygonEvents(
            first: $pageSize,
            after: $afterCursor,
            polygonTypes: $polygonTypes,
            entryDate: $entryDate,
            lastUpdated: $lastUpdated
        ) {
            pageInfo { endCursor }
            edges {
                node {
                    _id
                    polygon { _id, name }
                    vessel { _id, imo, name, type }
                    entryAis { time }
                    outAis   { time }
                    lastUpdated
                }
            }
        }
    }
    """

    all_rows = []
    cursor = None
    while True:
        variables["afterCursor"] = cursor
        payload = gql_query(query, variables)
        edges = payload["polygonEvents"]["edges"]
        rows = pd.json_normalize([e["node"] for e in edges])
        all_rows.append(rows)

        if len(edges) < page_size:
            break
        cursor = payload["polygonEvents"]["pageInfo"]["endCursor"]

    if all_rows:
        df = pd.concat(all_rows, ignore_index=True)
        return df
    return pd.DataFrame()

# ------------------------------------------------------------------
# 5️⃣  Remove deleted events
# ------------------------------------------------------------------
def purge_deleted_events(df: pd.DataFrame) -> pd.DataFrame:
    """Delete rows that match events reported as deleted."""
    if last_execution_time is None:
        # First run → nothing to purge
        return df

    page_size = 1000
    variables = {
        "pageSize": page_size,
        "polygonTypes": POLYGON_TYPES,
        "deletedAt": {"from": last_execution_time},
    }

    query = """
    query deletedPolygonEvents(
        $pageSize: Int,
        $afterCursor: String,
        $polygonTypes: [String],
        $deletedAt: RangeDate
    ) {
        deletedPolygonEvents(
            first: $pageSize,
            after: $afterCursor,
            polygonTypes: $polygonTypes,
            deletedAt: $deletedAt
        ) {
            pageInfo { endCursor }
            edges {
                node { _id }
            }
        }
    }
    """

    cursor = None
    deleted_ids = set()
    while True:
        variables["afterCursor"] = cursor
        payload = gql_query(query, variables)
        edges = payload["deletedPolygonEvents"]["edges"]
        deleted_ids.update(e["node"]["_id"] for e in edges)

        if len(edges) < page_size:
            break
        cursor = payload["deletedPolygonEvents"]["pageInfo"]["endCursor"]

    if deleted_ids:
        df = df[~df["_id"].isin(deleted_ids)].reset_index(drop=True)
    return df

# ------------------------------------------------------------------
# 6️⃣  Main integration
# ------------------------------------------------------------------
def main(data):
    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

    # 6.1 Fetch new/updated events
    new_events = fetch_polygon_events()
    logging.info(f"Fetched {len(new_events)} new/updated events")

    # 6.2 Merge with existing cache
    if not data.empty:
        data = pd.concat([data, new_events], ignore_index=True)
        data = data.drop_duplicates(subset=["_id"], keep="last")
    else:
        data = new_events

    # 6.3 Remove events that were deleted
    data = purge_deleted_events(data)
    logging.info(f"Dataset after purge: {len(data)} records")

    # 6.4 Persist
    data.to_csv(CSV_FILE, index=False)
    logging.info(f"Persisted {CSV_FILE}")

if __name__ == "__main__":
    main(data)