Spaces:
Sleeping
Sleeping
File size: 5,860 Bytes
a3f8002 fe2ba41 addc8d1 fe2ba41 c9d3d56 fe2ba41 c9d3d56 fe2ba41 2e6e690 fe2ba41 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 |
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):
pass
@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))
|