""" Centralisation des tarifs d'abonnement (source unique pour admin, API publique, Stripe, checkout). Règle métier : le prix annuel est toujours dérivé du mensuel : prix_annuel = arrondi(mensuel × 12 × YEARLY_DISCOUNT_FACTOR, 2 décimales) avec YEARLY_DISCOUNT_FACTOR = 0,8 (−20 % sur l'équivalent 12 mois au mensuel). Les overrides admin ne peuvent pas fixer un annuel incohérent : le serveur recalcule et persiste. """ from __future__ import annotations import json import logging import os import re from pathlib import Path from typing import Any, Optional from models.subscription import PLANS, PlanType logger = logging.getLogger(__name__) # Toujours relatif à la racine du projet (config.py), pas au cwd du processus — # sinon uvicorn lancé depuis un autre dossier écrit/lit un autre fichier et les # changements admin (ex. Starter 7 €) semblent « ne rien faire ». from config import config as _app_config PRICING_FILE = _app_config.BASE_DIR / "data" / "pricing_overrides.json" # −20 % sur la facturation annuelle vs 12 × prix mensuel YEARLY_DISCOUNT_FACTOR = 0.8 ANNUAL_DISCOUNT_PERCENT = 20 MAX_MONTHLY_PRICE_EUR = 50_000.0 MIN_MONTHLY_PRICE_EUR = 0.01 _STRIPE_PRICE_ID_RE = re.compile(r"^price_[a-zA-Z0-9]+$") _STRIPE_SECRET_RE = re.compile(r"^sk_(test|live)_[a-zA-Z0-9]+$") _STRIPE_PUBLISHABLE_RE = re.compile(r"^pk_(test|live)_[a-zA-Z0-9]+$") _STRIPE_WEBHOOK_SECRET_RE = re.compile(r"^whsec_[a-zA-Z0-9]+$") def compute_yearly_from_monthly(monthly: float) -> float: return round(float(monthly) * 12.0 * YEARLY_DISCOUNT_FACTOR, 2) def validate_monthly_price_eur(value: float) -> None: v = float(value) if v < MIN_MONTHLY_PRICE_EUR or v > MAX_MONTHLY_PRICE_EUR: raise ValueError( f"Le prix mensuel doit être entre {MIN_MONTHLY_PRICE_EUR} et {MAX_MONTHLY_PRICE_EUR} €" ) def validate_stripe_price_id(value: Optional[str], field_label: str) -> str: """Retourne une chaîne normalisée ou lève ValueError si format douteux.""" if value is None: return "" s = str(value).strip() if not s: return "" lower = s.lower() if "xxx" in lower or "placeholder" in lower: raise ValueError(f"{field_label} : identifiant invalide (placeholder interdit).") if not _STRIPE_PRICE_ID_RE.match(s): raise ValueError( f"{field_label} : format attendu price_… (identifiant Stripe Price)." ) return s def validate_stripe_secret_key(value: str) -> None: s = value.strip() if not _STRIPE_SECRET_RE.match(s): raise ValueError( "Clé secrète Stripe invalide (attendu sk_test_… ou sk_live_…)." ) def validate_stripe_publishable_key(value: str) -> None: s = value.strip() if not _STRIPE_PUBLISHABLE_RE.match(s): raise ValueError( "Clé publique Stripe invalide (attendu pk_test_… ou pk_live_…)." ) def validate_stripe_webhook_secret(value: str) -> None: s = value.strip() if not _STRIPE_WEBHOOK_SECRET_RE.match(s): raise ValueError("Secret webhook Stripe invalide (attendu whsec_…).") def load_pricing_overrides() -> dict[str, Any]: try: if PRICING_FILE.exists(): return json.loads(PRICING_FILE.read_text(encoding="utf-8")) except Exception as e: logger.warning("Lecture pricing_overrides impossible: %s", e) return {} def save_pricing_overrides(data: dict[str, Any]) -> None: PRICING_FILE.parent.mkdir(parents=True, exist_ok=True) PRICING_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8") logger.info("pricing_overrides enregistré : %s", PRICING_FILE.resolve()) def reload_dotenv_from_dotenv_file() -> None: """ Recharge le fichier .env dans os.environ (override=True). Nécessaire après mise à jour admin : sans cela le process garde les valeurs du démarrage et il faudrait redémarrer le serveur. """ try: from dotenv import load_dotenv p = _app_config.BASE_DIR / ".env" if p.exists(): load_dotenv(p, override=True) except Exception as e: logger.debug("reload_dotenv_from_dotenv_file: %s", e) def apply_runtime_config_after_admin_write() -> None: """À appeler après sauvegarde admin (JSON +/ou .env).""" reload_dotenv_from_dotenv_file() def stripe_price_id_for_plan( plan_id: str, period: str, overrides: Optional[dict] = None, ) -> str: """period: 'monthly' | 'yearly' — aligné sur STRIPE_PRICE_STARTER_MONTHLY etc.""" ov = (overrides if overrides is not None else load_pricing_overrides()).get( plan_id, {} ) key = f"stripe_price_id_{period}" raw = (ov.get(key) or "").strip() if raw: return raw # Fichier .env modifié par l'admin : recharger depuis le disque (pas le os.environ figé au démarrage). reload_dotenv_from_dotenv_file() env_key = f"STRIPE_PRICE_{plan_id.upper()}_{period.upper()}" env_val = (os.getenv(env_key, "") or "").strip() return env_val def stripe_price_ids_for_plan(plan_id: str) -> tuple[str, str]: ov = load_pricing_overrides() return ( stripe_price_id_for_plan(plan_id, "monthly", ov), stripe_price_id_for_plan(plan_id, "yearly", ov), ) def get_effective_monthly_yearly(plan_id: str) -> tuple[float, float]: """Mensuel depuis override ou PLANS ; annuel dérivé sauf forfaits spéciaux (gratuit / entreprise).""" try: pt = PlanType(plan_id) except ValueError: return (0.0, 0.0) base = PLANS[pt] ov = load_pricing_overrides().get(plan_id, {}) monthly = float(ov.get("price_monthly", base["price_monthly"])) if monthly < 0: return (float(base["price_monthly"]), float(base["price_yearly"])) if monthly == 0.0: return (0.0, 0.0) yearly = compute_yearly_from_monthly(monthly) return (monthly, yearly) def get_subscription_line_amount_eur(plan: PlanType, billing_period: str) -> float: """Montant unitaire pour une période de facturation (checkout fallback price_data).""" plan_id = plan.value monthly, yearly = get_effective_monthly_yearly(plan_id) if monthly < 0: return -1.0 if monthly == 0.0: return 0.0 if billing_period == "yearly": return yearly return monthly def normalize_plan_override_block(plan_id: str, ov: dict[str, Any]) -> dict[str, Any]: """Recalcule price_yearly dans un bloc override à partir de price_monthly.""" try: pt = PlanType(plan_id) except ValueError: return ov base = PLANS[pt] monthly = float(ov.get("price_monthly", base["price_monthly"])) ov = dict(ov) ov["price_monthly"] = monthly ov["price_yearly"] = compute_yearly_from_monthly(monthly) return ov