from datetime import date from fastapi import HTTPException from config.database import get_supabase_admin async def check_quota(user_id: str, quota_type: str): """ Checks if a user has exceeded their daily quota for a given action. quota_type can be 'analysis' or 'ai_input_converter'. """ db = get_supabase_admin() # 1. Get user's role/tier user_res = db.table("users").select("role, daily_analyses_count, last_analysis_date, daily_ai_input_count, last_ai_input_date").eq("id", user_id).single().execute() if not user_res.data: raise HTTPException(status_code=404, detail="User not found") user = user_res.data role = user.get("role", "free") # 2. Master Admin Bypass if role == "admin": return True today = str(date.today()) # 2. Get site-wide limits from site_settings settings_res = db.table("site_settings").select("value").eq("key", "quotas").single().execute() # Standardized limits as requested: Guest 3, Free 5, Pro 100 default_limits = { "analysis": {"guest": 3, "free": 5, "pro": 100}, "ai_input_converter": {"guest": 3, "free": 5, "pro": 25} } # Safe fetch from DB with fallback limits = settings_res.data.get("value", default_limits) if (settings_res.data and "value" in settings_res.data) else default_limits # Determine the specific limit for this action and user role type_limits = limits.get(quota_type, default_limits.get(quota_type, {})) limit = type_limits.get(role, type_limits.get("free", 5)) # 3. Check and reset counter if it's a new day counter_field = "daily_analyses_count" if quota_type == "analysis" else "daily_ai_input_count" date_field = "last_analysis_date" if quota_type == "analysis" else "last_ai_input_date" current_count = user.get(counter_field, 0) last_date = user.get(date_field) if last_date != today: current_count = 0 db.table("users").update({counter_field: 0, date_field: today}).eq("id", user_id).execute() if limit != -1 and current_count >= limit: raise HTTPException( status_code=429, detail=f"Daily {quota_type} limit ({limit}) reached for {role} tier. Please upgrade to Pro for more access." ) # 4. Increment count db.table("users").update({counter_field: current_count + 1, date_field: today}).eq("id", user_id).execute() return True async def check_guest_quota(client_ip: str, quota_type: str): """ Checks guest quota based on IP address. Standard limits: 3 analyses per day. """ db = get_supabase_admin() today = str(date.today()) # Get guest limits from DB or fallback settings_res = db.table("site_settings").select("value").eq("key", "quotas").single().execute() default_guest_limit = 3 limit = default_guest_limit if settings_res.data and "value" in settings_res.data: limit = settings_res.data["value"].get(quota_type, {}).get("guest", default_guest_limit) # Block immediately if limit is 0 — before any DB operations # This ensures the setting takes effect even if guest_usage table is missing if limit == 0: raise HTTPException(status_code=429, detail=f"Guest access to {quota_type} is currently disabled. Please sign in for access.") # Check IP usage in guest_usage table try: log_res = db.table("guest_usage").select("count").eq("ip", client_ip).eq("action_type", quota_type).eq("usage_date", today).single().execute() count = 0 if log_res.data: count = log_res.data["count"] if count >= limit: raise HTTPException(status_code=429, detail=f"Guest limit of {limit} daily {quota_type}s reached. Please sign in for higher limits.") db.table("guest_usage").update({"count": count + 1}).eq("ip", client_ip).eq("action_type", quota_type).eq("usage_date", today).execute() else: db.table("guest_usage").insert({ "ip": client_ip, "action_type": quota_type, "usage_date": today, "count": 1 }).execute() except HTTPException: raise except Exception: # Ignore if the guest_usage table doesn't exist pass return True