Spaces:
Build error
Build error
File size: 3,975 Bytes
a282d4b | 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 | from collections import defaultdict
from sqlalchemy.orm import Session
from app.database.models import Subscription
def analyze_subscriptions(db: Session, user_id: str):
"""
Analyzes subscriptions to detect duplicates, unused memberships, and cancellation opportunities.
"""
subs = db.query(Subscription).filter(
Subscription.user_id == user_id,
Subscription.active == True
).all()
if not subs:
return {
"subscriptions": [],
"duplicates": [],
"unused_subscriptions": [],
"yearly_savings_potential": 0.0,
"risk_analysis": []
}
merchant_map = defaultdict(list)
unused_list = []
cancellation_suggestions = []
yearly_savings = 0.0
risk_analysis = []
for s in subs:
# Standardize merchant name to detect duplicates
clean_merchant = s.merchant.strip().lower()
merchant_map[clean_merchant].append(s)
# Determine cost
m_cost = s.amount if s.billing_cycle.lower() == "monthly" else (s.amount / 12)
y_cost = (s.amount * 12) if s.billing_cycle.lower() == "monthly" else s.amount
# Detect unused (if frequency is 'low' or 'none' in usage detection metadata)
usage = s.ai_usage_detection or {}
freq = str(usage.get("usage_frequency", "medium")).lower()
if freq in ["low", "none", "unused"]:
unused_list.append(s)
cancellation_suggestions.append({
"subscription_id": s.id,
"merchant": s.merchant,
"amount": s.amount,
"billing_cycle": s.billing_cycle,
"reason": f"Usage frequency is flagged as '{freq}'.",
"yearly_savings": round(y_cost, 2)
})
yearly_savings += y_cost
# Detect duplicates
duplicates = []
for merchant, items in merchant_map.items():
if len(items) > 1:
total_cost = sum(x.amount for x in items)
duplicates.append({
"merchant": items[0].merchant,
"count": len(items),
"items": [
{
"id": x.id,
"amount": x.amount,
"billing_cycle": x.billing_cycle
}
for x in items
],
"recommendation": f"You have {len(items)} active subscriptions for {items[0].merchant}. Consolidate to a single account to save."
})
# Risk Analysis (utilities vs entertainment)
essential_categories = ["electricity", "water", "gas", "internet", "phone", "insurance"]
for s in subs:
is_essential = any(k in s.merchant.lower() for k in essential_categories)
if is_essential:
risk_analysis.append({
"merchant": s.merchant,
"risk_level": "high",
"consequences": "Utility interruption, account reactivation fees, or legal service contract breaches."
})
else:
risk_analysis.append({
"merchant": s.merchant,
"risk_level": "low",
"consequences": "Loss of entertainment streaming access only. Service can be reactivated instantly."
})
return {
"subscriptions": [
{
"id": s.id,
"merchant": s.merchant,
"amount": s.amount,
"billing_cycle": s.billing_cycle,
"usage_frequency": s.ai_usage_detection.get("usage_frequency", "medium") if s.ai_usage_detection else "medium"
}
for s in subs
],
"duplicates": duplicates,
"unused_subscriptions": cancellation_suggestions,
"yearly_savings_potential": round(yearly_savings, 2),
"risk_analysis": risk_analysis
}
|