flighty_to_airtrail.py

import requests
import argparse
import csv
from urllib.parse import urljoin
from datetime import datetime
import airportsdata

airports = airportsdata.load("IATA")

def get_icao_code(iata_code):
    """Retrieve ICAO code from IATA code."""
    airport = airports.get(iata_code.upper())
    return airport["icao"] if airport else "X"

def get_value(row, key):
    """Retrieve value from row, returning None if empty."""
    return row[key].strip() or None

def parse_datetime(row, actual_key, scheduled_key):
    """Parse datetime, prioritizing actual over scheduled."""
    try:
        return datetime.fromisoformat(row[actual_key]) if row[actual_key] else datetime.fromisoformat(row[scheduled_key])
    except (ValueError, KeyError, TypeError):
        return None  # Handle invalid or missing data

def format_record(row, user_id):
    """Format flight record into structured JSON format."""
    
    # Normalize seat-related fields
    for field in ["Seat Type", "Cabin Class", "Flight Reason"]:
        if field in row and row[field]:
            row[field] = row[field].lower()

    departure = parse_datetime(row, "Gate Departure (Actual)", "Gate Departure (Scheduled)")
    arrival = parse_datetime(row, "Gate Arrival (Actual)", "Gate Arrival (Scheduled)")

    if not departure or not arrival:
        print(f"Skipping record due to missing datetime: {row}")
        return None  # Skip invalid records

    return {
        "from": get_icao_code(row["From"]),
        "to": get_icao_code(row["To"]),
        "departure": departure.strftime("%Y-%m-%dT00:00:00Z"),
        "departureTime": departure.strftime("%H:%M"),
        "arrival": arrival.strftime("%Y-%m-%dT00:00:00Z"),
        "arrivalTime": arrival.strftime("%H:%M"),
        "seats": [
            {
                "userId": user_id,
                "guestName": None,
                "seat": get_value(row, "Seat Type"),
                "seatNumber": get_value(row, "Seat"),
                "seatClass": get_value(row, "Cabin Class"),
            }
        ],
        "airline": row["Airline"],
        "flightNumber": f"{row['Airline']}{row['Flight']}",
        "aircraft": None,  # row.get("Aircraft Type Name", ""),
        "aircraftReg": get_value(row, "Tail Number"),
        "flightReason": get_value(row, "Flight Reason"),
        "notes": get_value(row, "Notes"),
    }

def submit_record(url, token, record):
    """Submit formatted flight record to API."""
    headers = {"Authorization": f"Bearer {token}"}
    try:
        response = requests.post(urljoin(url, "/api/flight/save"), json=record, headers=headers, timeout=10)
        return response.status_code, response.text
    except requests.RequestException as e:
        return None, str(e)

def process_csv(file_path, url, token, user_id):
    """Read CSV file and submit records to API."""
    try:
        with open(file_path, mode="r", encoding="utf-8") as file:
            reader = csv.DictReader(file)
            for row in reader:
                formatted_record = format_record(row, user_id)
                if not formatted_record:
                    continue  # Skip invalid records

                status_code, response_text = submit_record(url, token, formatted_record)
                if status_code:
                    print(
                        f"Submitted {formatted_record['departure']} - {formatted_record['flightNumber']}, Status: {status_code}"
                    )
                else:
                    print(f"Failed to submit record: {response_text}")

    except FileNotFoundError:
        print(f"Error: File '{file_path}' not found.")
    except Exception as e:
        print(f"Unexpected error processing CSV: {e}")

def main():
    """Parse arguments and process CSV file."""
    parser = argparse.ArgumentParser(description="Submit flight records to API from CSV file.")
    parser.add_argument("--url", required=True, help="API endpoint URL")
    parser.add_argument("--token", required=True, help="Bearer token for authentication")
    parser.add_argument("--csv", required=True, help="Path to the CSV file containing flight records")
    parser.add_argument("--user-id", required=True, help="User ID for flight records")
    args = parser.parse_args()

    process_csv(args.csv, args.url, args.token, args.user_id)

if __name__ == "__main__":
    main()