Spaces:
Running
Running
| import os | |
| from enum import Enum | |
| from clerk_backend_api import Clerk | |
| from core.subscription.tracker import get_used_tokens, get_wizard_iterations | |
| class SubscriptionTier(Enum): | |
| FREE = "free" | |
| PRO = "pro" | |
| BUSINESS = "business" | |
| def get_subscription_limits(tier: SubscriptionTier): | |
| limits = { | |
| SubscriptionTier.FREE: { | |
| "max_wizard_iterations": 15, | |
| "max_tokens_monthly": 100000, | |
| "grants_search_limit": 5, | |
| "export_pdf": False, | |
| "max_documents_per_project": 5, | |
| }, | |
| SubscriptionTier.PRO: { | |
| "max_wizard_iterations": 50, | |
| "max_tokens_monthly": 500000, | |
| "grants_search_limit": 20, | |
| "export_pdf": True, | |
| "max_documents_per_project": 20, | |
| }, | |
| SubscriptionTier.BUSINESS: { | |
| "max_wizard_iterations": 9999, | |
| "max_tokens_monthly": 2000000, | |
| "grants_search_limit": 100, | |
| "export_pdf": True, | |
| "max_documents_per_project": 9999, | |
| }, | |
| } | |
| return limits.get(tier, limits[SubscriptionTier.FREE]) | |
| clerk_secret = os.getenv("CLERK_SECRET_KEY") | |
| clerk = Clerk(bearer_auth=clerk_secret) if clerk_secret else None | |
| class SubscriptionChecker: | |
| def __init__(self, user_id: str): | |
| self.user_id = user_id | |
| def _fetch_tier_from_clerk_with_db_fallback(self) -> SubscriptionTier: | |
| if self.user_id == "anonymous" or self.user_id == "test_dev_user": | |
| return SubscriptionTier.FREE | |
| tier_str = None | |
| # 1. Clerk to jedyne źródło prawdy oznaczające autoryzacje subskrypcji | |
| if clerk: | |
| try: | |
| user = clerk.users.get(user_id=self.user_id) | |
| if getattr(user, "public_metadata", None): | |
| tier_str = user.public_metadata.get("stripe_subscription") | |
| except Exception as e: | |
| from core.audit_logger import audit_log | |
| try: | |
| audit_log("ERROR", f"Clerk HTTP fetch failed: {e}") | |
| except Exception: | |
| pass | |
| # 2. Jeśli Clerk nie odpowiedział lub public_metadata puste, fallback do Postgresa | |
| if not tier_str: | |
| from core.subscription.db import SessionLocal | |
| from core.subscription.models import User | |
| db = SessionLocal() | |
| try: | |
| user = db.query(User).filter(User.clerk_id == self.user_id).first() | |
| if user and user.tier: | |
| tier_str = user.tier | |
| except Exception: | |
| pass | |
| finally: | |
| db.close() | |
| tier_str = tier_str or "free" | |
| try: | |
| return SubscriptionTier(tier_str.lower()) | |
| except ValueError: | |
| return SubscriptionTier.FREE | |
| def can_start_iteration(self) -> bool: | |
| tier = self._fetch_tier_from_clerk_with_db_fallback() | |
| limits = get_subscription_limits(tier) | |
| used = get_wizard_iterations(self.user_id) | |
| return used < limits["max_wizard_iterations"] | |
| def can_use_tokens(self) -> bool: | |
| tier = self._fetch_tier_from_clerk_with_db_fallback() | |
| limits = get_subscription_limits(tier) | |
| used = get_used_tokens(self.user_id) | |
| return used < limits["max_tokens_monthly"] | |
| def get_tier(self) -> SubscriptionTier: | |
| return self._fetch_tier_from_clerk_with_db_fallback() | |
| def get_current_limits(self): | |
| tier = self.get_tier() | |
| return get_subscription_limits(tier) | |