# /opt/docker/dev/service_finder/backend/app/services/subscription_service.py """ Subscription & Entitlement Service — Premium költségkategória láthatóság. A szolgáltatás a felhasználó előfizetési szintje (tier) alapján szabályozza, hogy mely költségkategóriák (CostCategory) érhetők el. A tier-ek hierarchikusak: free < premium < enterprise. Hierarchia: - free: alap kategóriák (üzemanyag, szerviz, gumik, biztosítás, adók) - premium: free + útdíj/parkolás, ápolás, egyéb - enterprise: premium + minden (nincs korlátozás) P0 Feature Flag System: - A get_user_tier() most már valós adatbázis lekérdezést végez a UserSubscription és OrganizationSubscription táblákból. - A get_user_feature_flags() metódus visszaadja az összes feature flag állapotát a felhasználó tier-je alapján. - A get_org_subscription_details() metódus visszaadja a szervezet előfizetési adatait a frontend számára. """ from __future__ import annotations import logging from datetime import datetime from typing import Optional, List, Dict, Any from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from sqlalchemy.orm import selectinload from app.models.fleet_finance import CostCategory from app.models.core_logic import SubscriptionTier, UserSubscription, OrganizationSubscription from app.models.identity import User from app.models.marketplace.organization import Organization logger = logging.getLogger(__name__) # ── Subscription Feature Matrix ────────────────────────────────────────────── # Minden funkció/kategória minimális tier szintje. # A tier-ek: 0=free, 1=premium, 2=enterprise # A TIER_HIERARCHY a SubscriptionTier.tier_level mezőből származik. # A SUBSCRIPTION_FEATURES itt a minimális tier_level értéket tárolja (int). SUBSCRIPTION_FEATURES: dict[str, int] = { # Feature -> minimum required tier_level (0=free, 1=premium, 2=enterprise) "cost_category:fuel": 0, "cost_category:service": 0, "cost_category:tires": 0, "cost_category:insurance": 0, "cost_category:taxes": 0, "cost_category:toll_parking": 1, "cost_category:cleaning": 1, "cost_category:other": 1, "cost_category:all": 2, # Feature flags "export_csv": 0, "analytics_tco": 1, "analytics_detailed": 2, "api_access": 0, "multi_vehicle": 0, "unlimited_vehicles": 1, # P0 Frontend feature flags "subscription_management": 0, "vehicle_tracking": 0, "cost_tracking": 0, "service_booking": 0, "advanced_reports": 1, "team_management": 1, "priority_support": 1, "white_label": 2, "api_webhooks": 2, "custom_integrations": 2, } class SubscriptionService: """ Entitlement Mátrix — előfizetési szintek kezelése. Metódusok: get_user_tier(db, user_id) -> str get_user_feature_flags(db, user_id) -> dict[str, bool] get_org_subscription_details(db, org_id) -> dict | None can_access_feature(tier, feature_key) -> bool get_visible_categories(db, tier) -> list[CostCategory] require_tier(tier, required_feature_or_min_tier) -> bool P0 AUDIT FIX: A tier feloldás már a SubscriptionTier.tier_level mezőt használja ahelyett, hogy a tier nevét string-matchinggel próbálná meg kitalálni. A tier_level egy integer: 0=free, 1=premium, 2=enterprise. """ # ── Tier level → display name mapping ────────────────────────────── # This is the canonical mapping from DB tier_level to the human-readable # entitlement tier name used in feature flag resolution. TIER_LEVEL_NAMES: dict[int, str] = { 0: "free", 1: "premium", 2: "enterprise", } @staticmethod def _tier_level_to_name(level: int) -> str: """Convert a numeric tier_level to the canonical tier name.""" return SubscriptionService.TIER_LEVEL_NAMES.get(level, "free") @staticmethod async def _get_default_fallback_tier(db: AsyncSession) -> Optional[SubscriptionTier]: """ Lekérdezi az egyetlen csomagot, ahol is_default_fallback == True. Ha nincs ilyen, None-t ad vissza. """ stmt = select(SubscriptionTier).where( SubscriptionTier.is_default_fallback == True ) result = await db.execute(stmt) return result.scalar_one_or_none() @staticmethod async def get_user_tier(db: AsyncSession, user_id: int) -> str: """ Visszaadja a felhasználó aktuális előfizetési szintjét valós adatbázis lekérdezéssel. P0 Feature Flag: Valós tier feloldás a UserSubscription és OrganizationSubscription táblákból, valamint a User.subscription_plan mezőből. Feloldási sorrend: 1. Aktív UserSubscription lekérdezése (finance.user_subscriptions) - Ha van aktív és nem járt le, a tier.tier_level alapján 2. Ha nincs user subscription, de a user aktív szervezethez tartozik: - OrganizationSubscription lekérdezése 3. Ha egyik sincs, visszaesés a User.subscription_plan mezőre 4. Ha az is üres, ellenőrizzük a default fallback csomagot 5. Alapértelmezett 'free' P0 AUDIT FIX: A tier_level mezőt használjuk a SubscriptionTier modellből, nem pedig a tier name string-matchingjét. P0 SINGLETON FALLBACK: Ha a subscription lejárt (expired), dinamikusan lekérdezzük a is_default_fallback==True csomagot és annak feature-jeit adjuk vissza. """ # ── 1. UserSubscription lekérdezése ── us_stmt = ( select(UserSubscription) .options(selectinload(UserSubscription.tier)) .where( UserSubscription.user_id == user_id, UserSubscription.is_active == True, ) .order_by(UserSubscription.id.desc()) ) us_result = await db.execute(us_stmt) user_sub = us_result.scalars().first() if user_sub and user_sub.tier: # Ellenőrizzük a lejáratot if user_sub.valid_until is None or user_sub.valid_until > datetime.utcnow(): # P0 FIX: Use tier_level from DB instead of hardcoded string matching level = user_sub.tier.tier_level return SubscriptionService._tier_level_to_name(level) else: # ── P0 SINGLETON FALLBACK: subscription lejárt → default fallback tier ── logger.info( f"UserSubscription {user_sub.id} for user {user_id} has expired " f"(valid_until={user_sub.valid_until}). Falling back to default fallback tier." ) fallback = await SubscriptionService._get_default_fallback_tier(db) if fallback: level = fallback.tier_level return SubscriptionService._tier_level_to_name(level) # ── 2. Aktív szervezet subscription ── user_stmt = select(User).where(User.id == user_id) user_result = await db.execute(user_stmt) user = user_result.scalar_one_or_none() # active_organization_id is stored in the User.scope_id column (String) # and set on current_user by the auth dependency from the JWT token. # When querying from DB directly, we use scope_id. active_org_id = None if user and user.scope_id: try: active_org_id = int(user.scope_id) except (ValueError, TypeError): pass if active_org_id: org_stmt = ( select(OrganizationSubscription) .options(selectinload(OrganizationSubscription.tier)) .where( OrganizationSubscription.org_id == active_org_id, OrganizationSubscription.is_active == True, ) .order_by(OrganizationSubscription.id.desc()) ) org_result = await db.execute(org_stmt) org_sub = org_result.scalars().first() if org_sub and org_sub.tier: if org_sub.valid_until is None or org_sub.valid_until > datetime.utcnow(): # P0 FIX: Use tier_level from DB instead of hardcoded string matching level = org_sub.tier.tier_level return SubscriptionService._tier_level_to_name(level) else: # ── P0 SINGLETON FALLBACK: org subscription lejárt ── logger.info( f"OrganizationSubscription {org_sub.id} for org {active_org_id} has expired " f"(valid_until={org_sub.valid_until}). Falling back to default fallback tier." ) fallback = await SubscriptionService._get_default_fallback_tier(db) if fallback: level = fallback.tier_level return SubscriptionService._tier_level_to_name(level) # ── 3. Fallback: User.subscription_plan ── if user and user.subscription_plan: plan = user.subscription_plan.lower() # For the fallback plan field, we still need a simple heuristic # since there's no SubscriptionTier object. We use a minimal mapping. if "enterprise" in plan or "vip" in plan: return "enterprise" if "premium" in plan or "pro" in plan: return "premium" return "free" # ── 4. Default fallback tier ── fallback = await SubscriptionService._get_default_fallback_tier(db) if fallback: level = fallback.tier_level return SubscriptionService._tier_level_to_name(level) # ── 5. Alapértelmezett ── return "free" @staticmethod async def get_user_feature_flags( db: AsyncSession, user_id: int, ) -> Dict[str, Any]: """ Visszaadja az összes feature flag állapotát a felhasználó tier-je alapján. P0 Feature: Ez a metódus a frontend useFeatureFlag() composable számára biztosítja a backend által validált feature flag-eket. Returns: Dict a következő struktúrával: { "tier": "premium", "features": { "export_csv": True, "analytics_tco": True, ... }, "expires_at": "2026-07-24T12:00:00Z" | None } """ user_tier = await SubscriptionService.get_user_tier(db, user_id) # Lekérdezzük a lejárati időt is expires_at = None us_stmt = ( select(UserSubscription) .where( UserSubscription.user_id == user_id, UserSubscription.is_active == True, ) .order_by(UserSubscription.id.desc()) ) us_result = await db.execute(us_stmt) user_sub = us_result.scalars().first() if user_sub and user_sub.valid_until: expires_at = user_sub.valid_until.isoformat() # Összes feature flag feloldása features = {} for feature_key, required_tier in SUBSCRIPTION_FEATURES.items(): features[feature_key] = SubscriptionService.can_access_feature(user_tier, feature_key) return { "tier": user_tier, "features": features, "expires_at": expires_at, } @staticmethod async def get_org_subscription_details( db: AsyncSession, org_id: int, ) -> Optional[Dict[str, Any]]: """ Visszaadja a szervezet előfizetési adatait a frontend számára. P0 Feature: Ez a metódus a GET /subscriptions/my végpont számára biztosítja a szervezet előfizetési adatait. Returns: Dict a következő struktúrával: { "tier_id": 16, "tier_name": "corp_premium_v1", "display_name": "Céges Prémium", "valid_from": "2026-06-01T00:00:00Z", "valid_until": "2026-07-01T00:00:00Z", "is_active": True, "allowances": { ... }, "pricing": { ... }, "feature_capabilities": { ... }, } vagy None, ha nincs aktív előfizetés. """ stmt = ( select(OrganizationSubscription) .options(selectinload(OrganizationSubscription.tier)) .where( OrganizationSubscription.org_id == org_id, OrganizationSubscription.is_active == True, ) .order_by(OrganizationSubscription.id.desc()) ) result = await db.execute(stmt) org_sub = result.scalars().first() if not org_sub or not org_sub.tier: return None tier = org_sub.tier rules = tier.rules or {} return { "tier_id": tier.id, "tier_name": tier.name, "display_name": rules.get("display_name", tier.name), "valid_from": org_sub.valid_from.isoformat() if org_sub.valid_from else None, "valid_until": org_sub.valid_until.isoformat() if org_sub.valid_until else None, "is_active": org_sub.is_active, "allowances": rules.get("allowances", {}), "pricing": rules.get("pricing", {}), "duration": rules.get("duration", {}), "ad_policy": rules.get("ad_policy", {}), "entitlements": rules.get("entitlements", []), "feature_capabilities": tier.feature_capabilities or {}, "extra_allowances": org_sub.extra_allowances or {}, } @staticmethod async def get_user_subscription_details( db: AsyncSession, user_id: int, ) -> Optional[Dict[str, Any]]: """ Visszaadja a felhasználó személyes előfizetési adatait. P0 Feature: Ez a metódus a GET /subscriptions/my végpont számára biztosítja a felhasználói előfizetés adatait (ha nincs org subscription). Returns: Dict a következő struktúrával: { "tier_id": 5, "tier_name": "private_pro_v1", "display_name": "Privát Pro", "valid_from": "...", "valid_until": "...", "is_active": True, "allowances": { ... }, "pricing": { ... }, } vagy None, ha nincs aktív előfizetés. """ stmt = ( select(UserSubscription) .options(selectinload(UserSubscription.tier)) .where( UserSubscription.user_id == user_id, UserSubscription.is_active == True, ) .order_by(UserSubscription.id.desc()) ) result = await db.execute(stmt) user_sub = result.scalars().first() if not user_sub or not user_sub.tier: return None tier = user_sub.tier rules = tier.rules or {} return { "tier_id": tier.id, "tier_name": tier.name, "display_name": rules.get("display_name", tier.name), "valid_from": user_sub.valid_from.isoformat() if user_sub.valid_from else None, "valid_until": user_sub.valid_until.isoformat() if user_sub.valid_until else None, "is_active": user_sub.is_active, "allowances": rules.get("allowances", {}), "pricing": rules.get("pricing", {}), "duration": rules.get("duration", {}), "ad_policy": rules.get("ad_policy", {}), "entitlements": rules.get("entitlements", []), "feature_capabilities": tier.feature_capabilities or {}, } @staticmethod def _get_user_level(user_tier: str) -> int: """ Convert a user tier name to its numeric level. Uses the canonical TIER_LEVEL_NAMES mapping. """ reverse_map = {v: k for k, v in SubscriptionService.TIER_LEVEL_NAMES.items()} return reverse_map.get(user_tier.lower(), 0) @staticmethod def can_access_feature(user_tier: str, feature_key: str) -> bool: """ Ellenőrzi, hogy a felhasználó hozzáfér-e egy adott funkcióhoz. P0 AUDIT FIX: A SUBSCRIPTION_FEATURES most már integer tier_level értékeket tartalmaz (0=free, 1=premium, 2=enterprise), így nincs szükség string konverzióra. Args: user_tier: A felhasználó tier szintje (free/premium/enterprise) feature_key: A funkció kulcsa (pl. 'cost_category:fuel', 'analytics_tco') Returns: bool: True ha hozzáfér, False ha nem """ required_level = SUBSCRIPTION_FEATURES.get(feature_key) if required_level is None: # Ismeretlen feature — alapértelmezetten tiltva logger.warning(f"Unknown feature key: {feature_key}") return False user_level = SubscriptionService._get_user_level(user_tier) return user_level >= required_level @staticmethod async def get_visible_categories( db: AsyncSession, user_tier: str ) -> List[CostCategory]: """ Visszaadja azokat a CostCategory-ket, amelyek a user tier-jének megfelelnek. Szűrési logika: - A kategória min_tier mezője alapján szűrünk. - Ha a kategória min_tier értéke kisebb vagy egyenlő, mint a user tier, akkor látható. - Ha a min_tier nincs beállítva (None), akkor 'free'-ként kezeljük. Args: db: Adatbázis session user_tier: A felhasználó tier szintje Returns: List[CostCategory]: A látható kategóriák listája """ user_level = SubscriptionService._get_user_level(user_tier) # Lekérjük az összes kategóriát stmt = select(CostCategory).order_by(CostCategory.id) result = await db.execute(stmt) all_categories = result.scalars().all() # Szűrés a min_tier alapján visible = [] for cat in all_categories: cat_tier = getattr(cat, "min_tier", "free") or "free" cat_level = SubscriptionService._get_user_level(cat_tier) if user_level >= cat_level: visible.append(cat) return visible @staticmethod def require_tier(user_tier: str, required_feature_or_min_tier: str) -> bool: """ Egyszerű tier ellenőrzés — dependency check-hez. Args: user_tier: A felhasználó tier szintje required_feature_or_min_tier: Vagy egy feature kulcs, vagy egy tier név Returns: bool: True ha megfelel, False ha nem """ # Először feature kulcsként próbáljuk if required_feature_or_min_tier in SUBSCRIPTION_FEATURES: return SubscriptionService.can_access_feature( user_tier, required_feature_or_min_tier ) # Ha nem feature kulcs, akkor tier névként kezeljük user_level = SubscriptionService._get_user_level(user_tier) required_level = SubscriptionService._get_user_level(required_feature_or_min_tier) return user_level >= required_level @staticmethod async def assign_default_trial( db: AsyncSession, org_id: int, user_id: int, ) -> Optional[Dict[str, Any]]: """ Új regisztrációnál automatikusan hozzárendeli a próbaidős csomagot. P0 SINGLETON TRIAL: Lekérdezi a trial_days_on_signup > 0 csomagot. Ha van ilyen, létrehoz egy OrganizationSubscription rekordot a megadott org_id-hoz, ami a trial_days_on_signup nap múlva jár le. Args: db: Adatbázis session org_id: A szervezet ID-ja, amelyhez a trial subscription tartozik user_id: A felhasználó ID-ja (naplózáshoz) Returns: Dict a subscription adatokkal, vagy None ha nincs trial csomag beállítva. """ # Lekérdezzük a trial csomagot (ahol trial_days_on_signup > 0) stmt = select(SubscriptionTier).where( SubscriptionTier.trial_days_on_signup > 0 ) result = await db.execute(stmt) trial_tier = result.scalar_one_or_none() if not trial_tier: logger.info( f"No trial tier configured (no tier with trial_days_on_signup > 0). " f"Skipping trial assignment for org {org_id}, user {user_id}." ) return None from datetime import timedelta # Számoljuk a lejárati időt trial_days = trial_tier.trial_days_on_signup now = datetime.utcnow() valid_until = now + timedelta(days=trial_days) # Létrehozzuk a subscription rekordot org_sub = OrganizationSubscription( org_id=org_id, tier_id=trial_tier.id, valid_from=now, valid_until=valid_until, is_active=True, ) db.add(org_sub) await db.commit() await db.refresh(org_sub) logger.info( f"Assigned trial subscription for org {org_id}, user {user_id}: " f"tier='{trial_tier.name}' (id={trial_tier.id}), " f"trial_days={trial_days}, valid_until={valid_until}" ) return { "tier_id": trial_tier.id, "tier_name": trial_tier.name, "valid_from": now.isoformat(), "valid_until": valid_until.isoformat(), "is_active": True, "trial_days": trial_days, }