from fastapi import FastAPI, HTTPException,Query from supabase import create_client, Client from typing import List, Dict from statistics import mean from collections import Counter import os from datetime import datetime import json app = FastAPI() # Initialize Supabase client url: str = os.getenv('SUPABASE_URL') key: str = os.getenv('SUPABASE_KEY') supabase: Client = create_client(url, key) @app.get('/get_travel_data') async def get_user_travel_data(hushh_id:str): resp = supabase.table("receipt_radar_structured_data_duplicate").select("metadata,message_id,logo,company").eq("brand_category","Travel").eq('user_id',hushh_id).execute() print(resp.data) data = {} arrival_location_frequency = {} for msg in resp.data: print(msg) date = msg.get('Date') if msg.get('metadata') is not None and msg.get('metadata') != 'null': ds = json.loads(msg.get('metadata')) arrival_date = ds.get('arrival_date') or date travel_type = ds.get('travel_type',None) if ds.get('travel_type') == 'null' or travel_type is None: continue departure_destination = ds.get('departure_destination') if ds.get('departure_destination') is not None else None arrival_destination = ds.get('arrival_destination') if ds.get('arrival_destination') is not None else None arrival_city = ds.get('arrival_city') if ds.get('arrival_city') is not None else None if 'travel_history' not in data: data['travel_history'] = [] data['travel_history'].append({"message_id":msg.get('message_id'),"domain":msg.get('company'),"logo":msg.get('logo'),'arrival_date':arrival_date,'travel_type':travel_type,'departure_destination':departure_destination,'arrival_destination':arrival_destination}) print('Data') print(data) # Increment arrival_location_frequency count for the current arrival_destination if arrival_city: if arrival_city in arrival_location_frequency: arrival_location_frequency[arrival_city] += 1 else: arrival_location_frequency[arrival_city] = 1 sorted_arrival_location_frequency = dict( sorted(arrival_location_frequency.items(), key=lambda item: item[1], reverse=True) ) data['arrival_location_frequency'] = sorted_arrival_location_frequency return data @app.get("/travel-analytics") async def get_travel_analytics(user_id: str = Query(..., description="User's hush ID")): try: # Fetch data from Supabase response = supabase.table("receipt_radar_structured_data_duplicate_duplicate").select("metadata, total_cost, brand_category").eq('user_id',user_id).execute() # Extract metadata from the response metadata_list: List[Dict] = [json.loads(row['metadata']) for row in response.data if row['metadata']] total_costs = [float(row['total_cost']) for row in response.data if row.get('total_cost')] print(metadata_list) print(total_costs) # Initialize variables for analytics total_trips = len(metadata_list) trip_durations = [] domestic_trips = 0 international_trips = 0 destination_types = Counter() booking_lead_times = [] accommodation_spending = [] transport_spending = [] activities_spending = [] domestic_departure_countries = Counter() international_departure_countries = Counter() domestic_arrival_countries = Counter() international_arrival_countries = Counter() all_dates = [] # Process each metadata entry for metadata in metadata_list: # Trip duration if 'check_in_date' in metadata and 'check_out_date' in metadata: print("inside 1st") check_in_date = metadata['check_in_date'] check_out_date = metadata['check_out_date'] duration = (check_out_date - check_in_date).days trip_durations.append(duration) all_dates.append(check_in_date) all_dates.append(check_out_date) # Domestic vs International if 'departure_country' in metadata and 'arrival_country' in metadata: print("inside 2nd") departure_country = metadata['departure_country'] arrival_country = metadata['arrival_country'] if departure_country == arrival_country: print("inside 2nd") domestic_trips += 1 domestic_departure_countries[departure_country] += 1 domestic_arrival_countries[arrival_country] += 1 else: international_trips += 1 international_departure_countries[departure_country] += 1 international_arrival_countries[arrival_country] += 1 # Destination type (simplified) if 'destination_type' in metadata: print("inside 3rd") destination_types[metadata['destination_type']] += 1 # Booking lead time if 'date_of_purchase' in metadata and 'departure_date' in metadata: print("inside 4th") lead_time = (datetime.strptime(metadata['departure_date'],"%d-%m-%Y") - datetime.strptime(metadata['date_of_purchase'],"%d-%m-%Y")).days booking_lead_times.append(lead_time) # Spending if 'accommodation_cost' in metadata: accommodation_spending.append(metadata['accommodation_cost']) if 'transport_cost' in metadata: transport_spending.append(metadata['transport_cost']) if 'activities_cost' in metadata: activities_spending.append(metadata['activities_cost']) # Calculate the number of years covered by the data if all_dates: min_date = min(all_dates) max_date = max(all_dates) date_range_years = (max_date - min_date).days / 365.25 else: date_range_years = 1 # Default to 1 year if no dates are available # Calculate analytics analytics = { "travel_frequency": { "trips_per_year": total_trips / date_range_years, "average_trip_duration": mean(trip_durations) if trip_durations else None, "domestic_vs_international": f"{domestic_trips}:{international_trips}", "domestic_departure_countries": dict(domestic_departure_countries), "international_departure_countries": dict(international_departure_countries), "domestic_arrival_countries": dict(domestic_arrival_countries), "international_arrival_countries": dict(international_arrival_countries), }, # "destination_preferences": { # "popular_types": dict(destination_types.most_common(5)) # }, "booking_patterns": { "average_lead_time": mean(booking_lead_times) if booking_lead_times else None }, "travel_expenditure": { "average_accommodation_cost": mean(accommodation_spending) if accommodation_spending else None, "average_transport_cost": mean(transport_spending) if transport_spending else None, "average_activities_cost": mean(activities_spending) if activities_spending else None } } return analytics except Exception as e: raise HTTPException(status_code=500, detail=str(e))