Spaces:
Sleeping
Sleeping
from fastapi import FastAPI, HTTPException | |
from supabase import create_client, Client | |
from typing import List, Dict | |
from datetime import datetime | |
import os | |
from dotenv import load_dotenv | |
import json | |
load_dotenv() | |
import logging | |
app = FastAPI() | |
# Initialize Supabase client | |
url: str = os.getenv('SUPABASE_URL') | |
key: str = os.getenv('SUPABASE_KEY') | |
supabase: Client = create_client(url, key) | |
async def get_insurance_analytics(): | |
# Fetch data from Supabase | |
response = supabase.table("insurance_data").select("*").eq("hushh_id","4b7fa719-244e-4ae8-9025-651a9a41e43f").execute() | |
receipts = response.data | |
print(receipts) | |
analytics = { | |
"policy_overview": { | |
"types_of_policies": {}, | |
"total_coverage_amount": 0, | |
"premium_payments": { | |
"total": 0, | |
"frequency_distribution": {} | |
} | |
}, | |
"claim_patterns": { | |
"total_claims": 0, | |
"total_claim_amount": 0 | |
}, | |
"policy_renewal": { | |
"total_policies": len(receipts), | |
"active_policies": 0, | |
"lapsed_policies": 0 | |
} | |
} | |
current_date = datetime.now() | |
for receipt in receipts: | |
metadata = json.loads(receipt['metadata']) | |
print("printing metadata") | |
print(metadata) | |
# Policy overview | |
policy_type = metadata.get('policy_type') | |
analytics['policy_overview']['types_of_policies'][policy_type] = analytics['policy_overview']['types_of_policies'].get(policy_type, 0) + 1 | |
coverage_amount = float(metadata.get('coverage_amount', 0)) | |
analytics['policy_overview']['total_coverage_amount'] += coverage_amount | |
premium_amount = float(metadata.get('premium_amount', 0)) | |
analytics['policy_overview']['premium_payments']['total'] += premium_amount | |
premium_frequency = metadata.get('premium_frequency') | |
analytics['policy_overview']['premium_payments']['frequency_distribution'][premium_frequency] = analytics['policy_overview']['premium_payments']['frequency_distribution'].get(premium_frequency, 0) + 1 | |
# Claim patterns | |
if metadata.get('claim_number'): | |
analytics['claim_patterns']['total_claims'] += 1 | |
claim_amount = metadata.get('claim_amount') | |
if claim_amount is not None: | |
try: | |
claim_amount = float(claim_amount) | |
analytics['claim_patterns']['total_claim_amount'] += claim_amount | |
except: | |
logging.warning(f"Invalid claim amount: {claim_amount}") | |
# Policy renewal | |
# Policy renewal | |
try: | |
policy_end_date = datetime.strptime(metadata.get('policy_end_date'), "%m-%d-%Y") | |
if policy_end_date > current_date: | |
analytics['policy_renewal']['active_policies'] += 1 | |
else: | |
analytics['policy_renewal']['lapsed_policies'] += 1 | |
except (ValueError, TypeError): | |
logging.error(f"Invalid date format for policy_end_date: {metadata.get('policy_end_date')}") | |
# Calculate percentage distribution for premium frequency | |
total_policies = sum(analytics['policy_overview']['premium_payments']['frequency_distribution'].values()) | |
if total_policies > 0: | |
analytics['policy_overview']['premium_payments']['frequency_percentage'] = { | |
freq: (count / total_policies) * 100 | |
for freq, count in analytics['policy_overview']['premium_payments']['frequency_distribution'].items() | |
} | |
else: | |
logging.warning("No valid policies found for frequency percentage calculation") | |
logging.debug(f"Final analytics: {analytics}") | |
return analytics | |