Spaces:
Sleeping
Sleeping
File size: 4,406 Bytes
91990f9 | 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 | 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
|