from dotenv import load_dotenv from pathlib import Path import os import logging ROOT_DIR = Path(__file__).parent # Essayer de charger .env depuis plusieurs emplacements (dev local, Docker, NAS) _ENV_CANDIDATES = [ ROOT_DIR / '.env', Path('/app/.env'), Path('/app/backend/.env'), Path('/share/smartpanier/.env'), Path.cwd() / '.env', ] _loaded_env_path = None for _p in _ENV_CANDIDATES: try: if _p.exists(): load_dotenv(_p, override=False) _loaded_env_path = str(_p) break except Exception: pass # Charger aussi depuis l'environnement Docker (variables injectées) load_dotenv(override=False) print(f"[SmartPanier] .env chargé depuis: {_loaded_env_path or 'aucun fichier trouvé (variables Docker/env utilisées)'}", flush=True) from fastapi import FastAPI, APIRouter, HTTPException, Request, Response from starlette.middleware.cors import CORSMiddleware import aiomysql import bcrypt import jwt import secrets import uuid import httpx from datetime import datetime, timezone, timedelta from pydantic import BaseModel from typing import Optional, Dict import aiosmtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import random # MySQL config MYSQL_CONFIG = { "host": os.environ.get("MYSQL_HOST", "host.docker.internal"), "port": int(os.environ.get("MYSQL_PORT", "3306")), "user": os.environ.get("MYSQL_USER", "smartpanier"), "password": os.environ.get("MYSQL_PASSWORD", "yXuwt-Zb-e*n,aWu4JN4KWdA"), "db": os.environ.get("MYSQL_DB", "smartpanier"), "autocommit": True, } pool: aiomysql.Pool = None async def get_db(): global pool if pool is None: pool = await aiomysql.create_pool(**MYSQL_CONFIG, minsize=2, maxsize=10) return pool async def fetchone(query, params=None): p = await get_db() async with p.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cur: await cur.execute(query, params or ()) return await cur.fetchone() async def fetchall(query, params=None): p = await get_db() async with p.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cur: await cur.execute(query, params or ()) return await cur.fetchall() async def execute(query, params=None): p = await get_db() async with p.acquire() as conn: async with conn.cursor() as cur: await cur.execute(query, params or ()) return cur.lastrowid # JWT config JWT_ALGORITHM = "HS256" _JWT_SECRET_CACHE = None def get_jwt_secret(): global _JWT_SECRET_CACHE if _JWT_SECRET_CACHE: return _JWT_SECRET_CACHE secret = os.environ.get("JWT_SECRET") or os.environ.get("JWT_SECRET_KEY") if not secret: # Fallback: générer un secret temporaire pour éviter le crash # IMPORTANT: définir JWT_SECRET dans le .env pour la production ! print("[SmartPanier] ⚠️ ATTENTION: JWT_SECRET non défini ! Utilisation d'un secret temporaire.", flush=True) print("[SmartPanier] ⚠️ Les tokens seront invalidés à chaque redémarrage du serveur.", flush=True) print("[SmartPanier] ⚠️ Ajoutez JWT_SECRET= dans le .env ou variables Docker.", flush=True) secret = secrets.token_urlsafe(64) _JWT_SECRET_CACHE = secret return secret # Subscription plans SUBSCRIPTION_PLANS = { "solo": {"name": "Solo", "price": 0.0, "max_lists": 3, "description": "Gratuit - 3 listes maximum"}, "duo": {"name": "Duo", "price": 2.99, "max_lists": 10, "description": "2,99€/mois - 10 listes maximum"}, "family": {"name": "Famille", "price": 6.99, "max_lists": 999999, "description": "6,99€/mois - Listes illimitées"}, } ITEM_CATEGORIES = [ {"id": "fruits", "name": "Fruits", "icon": "🍎"}, {"id": "vegetables", "name": "Légumes", "icon": "🥦"}, {"id": "dairy", "name": "Produits laitiers", "icon": "🧀"}, {"id": "meat", "name": "Viandes & Poissons", "icon": "🥩"}, {"id": "bakery", "name": "Boulangerie", "icon": "🍞"}, {"id": "beverages", "name": "Boissons", "icon": "🥤"}, {"id": "frozen", "name": "Surgelés", "icon": "🧊"}, {"id": "grocery", "name": "Épicerie", "icon": "🍝"}, {"id": "condiments", "name": "Condiments & Sauces", "icon": "🧂"}, {"id": "hygiene", "name": "Hygiène", "icon": "🧴"}, {"id": "household", "name": "Ménage", "icon": "🧹"}, {"id": "pets", "name": "Animaux", "icon": "🐶"}, {"id": "other", "name": "Autres", "icon": "📦"}, ] CATEGORY_IDS = [c["id"] for c in ITEM_CATEGORIES] STORE_TYPES = { "discount": {"label": "Discount", "examples": ["Lidl", "Aldi", "Netto", "Leader Price", "Action", "Normal"], "price_factor": 0.85}, "standard": {"label": "Standard", "examples": ["Carrefour", "Leclerc", "Intermarché", "Super U", "Auchan", "Cora", "G20", "Match", "Colruyt", "Carrefour Market", "Carrefour City"], "price_factor": 1.0}, "premium": {"label": "Premium", "examples": ["Monoprix", "Casino", "Franprix", "Picard"], "price_factor": 1.15}, "bio": {"label": "Bio", "examples": ["Biocoop", "Naturalia", "La Vie Claire", "Bio c' Bon"], "price_factor": 1.35}, } def classify_store_type(store_name): name_lower = store_name.lower() for stype, info in STORE_TYPES.items(): for ex in info["examples"]: if ex.lower() in name_lower: return stype return "standard" app = FastAPI() api_router = APIRouter(prefix="/api") logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ===== Email Helper ===== async def send_email(to_email: str, subject: str, html_content: str): smtp_host = os.environ.get("SMTP_HOST", "ssl0.ovh.net") smtp_port = int(os.environ.get("SMTP_PORT", "465")) smtp_user = os.environ.get("SMTP_USER", "contact@smartpanier.fr") smtp_pass = os.environ.get("SMTP_PASSWORD", "") smtp_from = os.environ.get("SMTP_FROM", "contact@smartpanier.fr") msg = MIMEMultipart("alternative") msg["From"] = f"SmartPanier <{smtp_from}>" msg["To"] = to_email msg["Subject"] = subject msg.attach(MIMEText(html_content, "html")) try: await aiosmtplib.send(msg, hostname=smtp_host, port=smtp_port, username=smtp_user, password=smtp_pass, use_tls=True) logger.info(f"Email sent to {to_email}") except Exception as e: logger.error(f"Email send error: {e}") raise def generate_reset_email(code: str) -> str: return f"""

SmartPanier

Réinitialisation de mot de passe

Bonjour,

Vous avez demandé la réinitialisation de votre mot de passe. Voici votre code de vérification :

{code}

Ce code est valable 15 minutes.

Si vous n'avez pas fait cette demande, ignorez cet email.

SmartPanier - Vos courses, simplifiées

contact@smartpanier.fr

""" # ===== Password Hashing ===== def hash_password(password): return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(plain, hashed): return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) # ===== JWT ===== def create_access_token(user_id, email): return jwt.encode({"sub": user_id, "email": email, "exp": datetime.now(timezone.utc) + timedelta(minutes=60), "type": "access"}, get_jwt_secret(), algorithm=JWT_ALGORITHM) def create_refresh_token(user_id): return jwt.encode({"sub": user_id, "exp": datetime.now(timezone.utc) + timedelta(days=7), "type": "refresh"}, get_jwt_secret(), algorithm=JWT_ALGORITHM) # ===== Auth Helper ===== async def get_current_user(request): token = request.cookies.get("access_token") if not token: auth_header = request.headers.get("Authorization", "") if auth_header.startswith("Bearer "): token = auth_header[7:] session_token = request.cookies.get("session_token") if not token and session_token: session = await fetchone("SELECT * FROM user_sessions WHERE session_token=%s AND expires_at > NOW()", (session_token,)) if session: user = await fetchone("SELECT * FROM users WHERE user_id=%s", (session["user_id"],)) if user: user.pop("password_hash", None) user.pop("id", None) user["trial_used"] = bool(user.get("trial_used")) return user raise HTTPException(status_code=401, detail="Session expired") if not token: raise HTTPException(status_code=401, detail="Not authenticated") try: payload = jwt.decode(token, get_jwt_secret(), algorithms=[JWT_ALGORITHM]) if payload.get("type") != "access": raise HTTPException(status_code=401, detail="Invalid token type") user = await fetchone("SELECT * FROM users WHERE user_id=%s", (payload["sub"],)) if not user: raise HTTPException(status_code=401, detail="User not found") user.pop("password_hash", None) user.pop("id", None) user["trial_used"] = bool(user.get("trial_used")) return user except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Token expired") except jwt.InvalidTokenError: raise HTTPException(status_code=401, detail="Invalid token") async def get_admin_user(request): user = await get_current_user(request) if user.get("role") != "admin": raise HTTPException(status_code=403, detail="Admin access required") return user # ===== Brute Force ===== async def check_brute_force(identifier): attempt = await fetchone("SELECT * FROM login_attempts WHERE identifier=%s", (identifier,)) if attempt and attempt.get("count", 0) >= 5: locked = attempt.get("locked_until") if locked and locked > datetime.now(): raise HTTPException(status_code=429, detail="Too many failed attempts. Try again in 15 minutes.") elif locked and locked <= datetime.now(): await execute("DELETE FROM login_attempts WHERE identifier=%s", (identifier,)) async def record_failed_attempt(identifier): attempt = await fetchone("SELECT * FROM login_attempts WHERE identifier=%s", (identifier,)) if attempt: new_count = attempt["count"] + 1 if new_count >= 5: await execute("UPDATE login_attempts SET count=%s, locked_until=%s WHERE identifier=%s", (new_count, datetime.now() + timedelta(minutes=15), identifier)) else: await execute("UPDATE login_attempts SET count=%s WHERE identifier=%s", (new_count, identifier)) else: await execute("INSERT INTO login_attempts (identifier, count, created_at) VALUES (%s, 1, %s)", (identifier, datetime.now())) # ===== Pydantic Models ===== class RegisterRequest(BaseModel): email: str password: str name: str referral_code: Optional[str] = None class LoginRequest(BaseModel): email: str password: str class ListCreate(BaseModel): name: str class ListUpdate(BaseModel): name: Optional[str] = None class ItemCreate(BaseModel): name: str quantity: Optional[str] = "1" category: Optional[str] = "other" class ItemUpdate(BaseModel): name: Optional[str] = None quantity: Optional[str] = None category: Optional[str] = None checked: Optional[bool] = None class ShareListRequest(BaseModel): email: str class CheckoutRequest(BaseModel): plan_id: str origin_url: str promo_code: Optional[str] = None class ReferralRedeemRequest(BaseModel): code: str # ===== Helper: build list with items and shares ===== async def build_list_response(lst): if not lst: return None list_id = lst["list_id"] items = await fetchall("SELECT * FROM list_items WHERE list_id=%s ORDER BY position ASC, created_at ASC", (list_id,)) shares = await fetchall("SELECT shared_email FROM list_shares WHERE list_id=%s", (list_id,)) result = {k: v for k, v in lst.items() if k != "id"} result["items"] = [] for it in items: result["items"].append({ "item_id": it["item_id"], "name": it["name"], "quantity": it["quantity"], "category": it["category"], "checked": bool(it["checked"]), "added_by": it["added_by"], "created_at": it["created_at"], }) result["shared_with"] = [s["shared_email"] for s in shares] return result # ===== Notifications Helper ===== async def notify_list_members(list_id, actor_email, item_name, action="added"): lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s", (list_id,)) if not lst: return recipients = set() owner_id = lst["owner_id"] owner = await fetchone("SELECT * FROM users WHERE user_id=%s", (owner_id,)) if owner and owner["email"] != actor_email: recipients.add(owner_id) shares = await fetchall("SELECT shared_email FROM list_shares WHERE list_id=%s", (list_id,)) for s in shares: if s["shared_email"] != actor_email: target = await fetchone("SELECT user_id FROM users WHERE email=%s", (s["shared_email"],)) if target: recipients.add(target["user_id"]) actor_user = await fetchone("SELECT name FROM users WHERE email=%s", (actor_email,)) actor_name = actor_user["name"] if actor_user else actor_email.split("@")[0] list_name = lst.get("name", "une liste") messages = {"added": f'{actor_name} a ajouté "{item_name}" dans "{list_name}"', "checked": f'{actor_name} a coché "{item_name}" dans "{list_name}"', "removed": f'{actor_name} a supprimé "{item_name}" de "{list_name}"'} message = messages.get(action, f'{actor_name} a modifié "{list_name}"') for uid in recipients: settings = await fetchone("SELECT * FROM notification_settings WHERE user_id=%s", (uid,)) if settings and (not settings.get("enabled") or not settings.get("list_updates")): continue await execute("INSERT INTO notifications (notif_id,user_id,type,message,list_id,read_status,created_at) VALUES (%s,%s,%s,%s,%s,0,%s)", (f"notif_{uuid.uuid4().hex[:12]}", uid, "list_update", message, list_id, datetime.now(timezone.utc).isoformat())) # ===== AUTH ENDPOINTS ===== @api_router.post("/auth/register") async def register(req: RegisterRequest, response: Response): email = req.email.lower().strip() existing = await fetchone("SELECT user_id FROM users WHERE email=%s", (email,)) if existing: raise HTTPException(status_code=400, detail="Email already registered") user_id = f"user_{uuid.uuid4().hex[:12]}" now = datetime.now(timezone.utc).isoformat() await execute("INSERT INTO users (user_id,email,name,password_hash,role,subscription_plan,subscription_status,created_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", (user_id, email, req.name, hash_password(req.password), "user", "solo", "active", now)) # Handle referral code at registration if hasattr(req, 'referral_code') and req.referral_code: ref_code = req.referral_code.upper().strip() referral = await fetchone("SELECT * FROM referrals WHERE code=%s", (ref_code,)) if referral and referral["owner_id"] != user_id: existing_redeem = await fetchone("SELECT id FROM referral_redemptions WHERE redeemed_by=%s", (user_id,)) if not existing_redeem: await execute("INSERT INTO referral_redemptions (redemption_id,referral_code,referral_owner_id,redeemed_by,created_at) VALUES (%s,%s,%s,%s,%s)", (f"rdm_{uuid.uuid4().hex[:12]}", ref_code, referral["owner_id"], user_id, now)) await execute("UPDATE referrals SET uses=uses+1, rewarded_months=rewarded_months+1 WHERE code=%s", (ref_code,)) trial_end = (datetime.now(timezone.utc) + timedelta(days=30)).isoformat() await execute("UPDATE users SET subscription_plan='duo', subscription_status='referral_reward', trial_end_date=%s WHERE user_id=%s", (trial_end, user_id)) token = create_access_token(user_id, email) refresh = create_refresh_token(user_id) response.set_cookie(key="access_token", value=token, httponly=True, secure=False, samesite="lax", max_age=3600, path="/") response.set_cookie(key="refresh_token", value=refresh, httponly=True, secure=False, samesite="lax", max_age=604800, path="/") return {"user_id": user_id, "email": email, "name": req.name, "role": "user", "subscription_plan": "solo", "token": token} @api_router.post("/auth/login") async def login(req: LoginRequest, response: Response, request: Request): email = req.email.lower().strip() ip = request.client.host if request.client else "unknown" identifier = f"{ip}:{email}" await check_brute_force(identifier) user = await fetchone("SELECT * FROM users WHERE email=%s", (email,)) if not user or not user.get("password_hash"): await record_failed_attempt(identifier) raise HTTPException(status_code=401, detail="Invalid email or password") if not verify_password(req.password, user["password_hash"]): await record_failed_attempt(identifier) raise HTTPException(status_code=401, detail="Invalid email or password") await execute("DELETE FROM login_attempts WHERE identifier=%s", (identifier,)) token = create_access_token(user["user_id"], email) refresh = create_refresh_token(user["user_id"]) response.set_cookie(key="access_token", value=token, httponly=True, secure=False, samesite="lax", max_age=3600, path="/") response.set_cookie(key="refresh_token", value=refresh, httponly=True, secure=False, samesite="lax", max_age=604800, path="/") return {"user_id": user["user_id"], "email": user["email"], "name": user["name"], "role": user.get("role", "user"), "subscription_plan": user.get("subscription_plan", "solo"), "token": token} @api_router.post("/auth/logout") async def logout(response: Response): response.delete_cookie("access_token", path="/") response.delete_cookie("refresh_token", path="/") response.delete_cookie("session_token", path="/") return {"message": "Logged out"} @api_router.get("/auth/me") async def get_me(request: Request): return await get_current_user(request) @api_router.post("/auth/refresh") async def refresh_token(request: Request, response: Response): token = request.cookies.get("refresh_token") if not token: raise HTTPException(status_code=401, detail="No refresh token") try: payload = jwt.decode(token, get_jwt_secret(), algorithms=[JWT_ALGORITHM]) if payload.get("type") != "refresh": raise HTTPException(status_code=401, detail="Invalid token type") user = await fetchone("SELECT * FROM users WHERE user_id=%s", (payload["sub"],)) if not user: raise HTTPException(status_code=401, detail="User not found") new_token = create_access_token(user["user_id"], user["email"]) response.set_cookie(key="access_token", value=new_token, httponly=True, secure=False, samesite="lax", max_age=3600, path="/") return {"token": new_token} except jwt.InvalidTokenError: raise HTTPException(status_code=401, detail="Invalid refresh token") # ===== ACCOUNT SETTINGS ===== @api_router.put("/auth/change-password") async def change_password(request: Request): user = await get_current_user(request) body = await request.json() current = body.get("current_password", "") new_pwd = body.get("new_password", "") if not current or not new_pwd: raise HTTPException(status_code=400, detail="Mot de passe actuel et nouveau requis") if len(new_pwd) < 6: raise HTTPException(status_code=400, detail="Le nouveau mot de passe doit contenir au moins 6 caractères") full_user = await fetchone("SELECT * FROM users WHERE user_id=%s", (user["user_id"],)) if not full_user or not full_user.get("password_hash"): raise HTTPException(status_code=400, detail="Changement de mot de passe non disponible pour les comptes Google") if not verify_password(current, full_user["password_hash"]): raise HTTPException(status_code=401, detail="Mot de passe actuel incorrect") await execute("UPDATE users SET password_hash=%s WHERE user_id=%s", (hash_password(new_pwd), user["user_id"])) return {"message": "Mot de passe modifié avec succès"} @api_router.put("/auth/change-email") async def change_email(request: Request): user = await get_current_user(request) body = await request.json() new_email = body.get("new_email", "").lower().strip() password = body.get("password", "") if not new_email or not password: raise HTTPException(status_code=400, detail="Nouvel email et mot de passe requis") if "@" not in new_email: raise HTTPException(status_code=400, detail="Email invalide") existing = await fetchone("SELECT user_id FROM users WHERE email=%s", (new_email,)) if existing: raise HTTPException(status_code=400, detail="Cet email est déjà utilisé") full_user = await fetchone("SELECT * FROM users WHERE user_id=%s", (user["user_id"],)) if not full_user or not full_user.get("password_hash"): raise HTTPException(status_code=400, detail="Changement d'email non disponible pour les comptes Google") if not verify_password(password, full_user["password_hash"]): raise HTTPException(status_code=401, detail="Mot de passe incorrect") old_email = full_user["email"] await execute("UPDATE users SET email=%s WHERE user_id=%s", (new_email, user["user_id"])) await execute("UPDATE shopping_lists SET owner_email=%s WHERE owner_id=%s", (new_email, user["user_id"])) await execute("UPDATE list_shares SET shared_email=%s WHERE shared_email=%s", (new_email, old_email)) return {"message": "Email modifié avec succès", "new_email": new_email} @api_router.put("/auth/change-name") async def change_name(request: Request): user = await get_current_user(request) body = await request.json() new_name = body.get("name", "").strip() if not new_name: raise HTTPException(status_code=400, detail="Nom requis") await execute("UPDATE users SET name=%s WHERE user_id=%s", (new_name, user["user_id"])) return {"message": "Nom modifié", "name": new_name} # ===== PASSWORD RESET ===== @api_router.post("/auth/forgot-password") async def forgot_password(request: Request): body = await request.json() email = body.get("email", "").lower().strip() if not email: raise HTTPException(status_code=400, detail="Email requis") user = await fetchone("SELECT user_id FROM users WHERE email=%s", (email,)) if not user: return {"message": "Si cet email existe, un code a été envoyé"} code = str(random.randint(100000, 999999)) expires = datetime.utcnow() + timedelta(minutes=15) await execute("DELETE FROM password_reset_codes WHERE email=%s", (email,)) await execute("INSERT INTO password_reset_codes (email,code,expires_at,used,created_at) VALUES (%s,%s,%s,0,%s)", (email, code, expires, datetime.utcnow())) try: html = generate_reset_email(code) await send_email(email, "SmartPanier - Code de réinitialisation", html) except Exception as e: logger.error(f"Failed to send reset email: {e}") raise HTTPException(status_code=500, detail="Erreur d'envoi de l'email") return {"message": "Code envoyé par email"} @api_router.post("/auth/verify-reset-code") async def verify_reset_code(request: Request): body = await request.json() email = body.get("email", "").lower().strip() code = body.get("code", "").strip() if not email or not code: raise HTTPException(status_code=400, detail="Email et code requis") record = await fetchone("SELECT * FROM password_reset_codes WHERE email=%s AND code=%s AND used=0 AND expires_at > UTC_TIMESTAMP()", (email, code)) if not record: raise HTTPException(status_code=400, detail="Code invalide ou expiré") return {"message": "Code vérifié", "valid": True} @api_router.post("/auth/reset-password") async def reset_password(request: Request): body = await request.json() email = body.get("email", "").lower().strip() code = body.get("code", "").strip() new_password = body.get("new_password", "") if not email or not code or not new_password: raise HTTPException(status_code=400, detail="Tous les champs sont requis") if len(new_password) < 6: raise HTTPException(status_code=400, detail="6 caractères minimum") record = await fetchone("SELECT * FROM password_reset_codes WHERE email=%s AND code=%s AND used=0 AND expires_at > UTC_TIMESTAMP()", (email, code)) if not record: raise HTTPException(status_code=400, detail="Code invalide ou expiré") await execute("UPDATE users SET password_hash=%s WHERE email=%s", (hash_password(new_password), email)) await execute("UPDATE password_reset_codes SET used=1 WHERE email=%s AND code=%s", (email, code)) return {"message": "Mot de passe réinitialisé avec succès"} @api_router.post("/auth/session") async def handle_google_session_endpoint(request: Request, response: Response): body = await request.json() session_id = body.get("session_id") if not session_id: raise HTTPException(status_code=400, detail="Missing session_id") async with httpx.AsyncClient() as http_client: resp = await http_client.get("https://demobackend.emergentagent.com/auth/v1/env/oauth/session-data", headers={"X-Session-ID": session_id}) if resp.status_code != 200: raise HTTPException(status_code=401, detail="Invalid session") google_data = resp.json() email = google_data["email"].lower() name = google_data.get("name", email.split("@")[0]) picture = google_data.get("picture", "") session_token = google_data.get("session_token", secrets.token_urlsafe(32)) user = await fetchone("SELECT * FROM users WHERE email=%s", (email,)) if not user: user_id = f"user_{uuid.uuid4().hex[:12]}" await execute("INSERT INTO users (user_id,email,name,picture,role,subscription_plan,subscription_status,created_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", (user_id, email, name, picture, "user", "solo", "active", datetime.now(timezone.utc).isoformat())) else: user_id = user["user_id"] await execute("UPDATE users SET name=%s, picture=%s WHERE email=%s", (name, picture, email)) await execute("INSERT INTO user_sessions (user_id,session_token,expires_at,created_at) VALUES (%s,%s,%s,%s)", (user_id, session_token, datetime.now(timezone.utc) + timedelta(days=7), datetime.now(timezone.utc))) response.set_cookie(key="session_token", value=session_token, httponly=True, secure=True, samesite="none", max_age=604800, path="/") user_data = await fetchone("SELECT * FROM users WHERE user_id=%s", (user_id,)) user_data.pop("password_hash", None) user_data.pop("id", None) return {**user_data, "session_token": session_token} # ===== GROUPS (Famille/Partenaire) ===== GROUP_MAX_MEMBERS = {"solo": 0, "duo": 1, "family": 5} async def get_effective_plan(user: dict) -> str: """Retourne le meilleur plan entre le plan perso et celui d'un groupe où l'user est membre.""" personal = user.get("subscription_plan", "solo") # 1. Si user est owner d'un groupe actif, utiliser son propre plan # 2. Sinon, s'il est membre d'un groupe, utiliser le plan du owner member_row = await fetchone(""" SELECT ug.plan_type, u.subscription_status, u.subscription_plan, u.subscription_end_date FROM group_members gm JOIN user_groups ug ON ug.group_id = gm.group_id JOIN users u ON u.user_id = ug.owner_id WHERE gm.email = %s AND u.subscription_status = 'active' """, (user["email"],)) if member_row: group_plan = member_row.get("plan_type", "solo") # Order: family > duo > solo order = {"family": 3, "duo": 2, "solo": 1} if order.get(group_plan, 1) > order.get(personal, 1): return group_plan return personal async def ensure_user_group(user: dict) -> Optional[dict]: """Crée ou récupère le groupe d'un user en fonction de son plan.""" plan = user.get("subscription_plan", "solo") if plan not in ("duo", "family"): return None existing = await fetchone("SELECT * FROM user_groups WHERE owner_id=%s", (user["user_id"],)) if existing: # Update plan if user's plan changed if existing["plan_type"] != plan: await execute("UPDATE user_groups SET plan_type=%s WHERE group_id=%s", (plan, existing["group_id"])) existing["plan_type"] = plan return existing group_id = f"grp_{uuid.uuid4().hex[:12]}" await execute("INSERT INTO user_groups (group_id,owner_id,plan_type,created_at) VALUES (%s,%s,%s,%s)", (group_id, user["user_id"], plan, datetime.now(timezone.utc).isoformat())) return {"group_id": group_id, "owner_id": user["user_id"], "plan_type": plan} @api_router.get("/group") async def get_my_group(request: Request): """Retourne le groupe de l'utilisateur (soit en tant qu'owner, soit en tant que member).""" user = await get_current_user(request) plan = user.get("subscription_plan", "solo") # Am I an owner? owned = await fetchone("SELECT * FROM user_groups WHERE owner_id=%s", (user["user_id"],)) # Si user Duo/Family sans groupe → créer automatiquement if not owned and plan in ("duo", "family"): owned = await ensure_user_group(user) if owned: members = await fetchall(""" SELECT gm.email, gm.added_at, u.name, u.user_id, u.subscription_plan FROM group_members gm LEFT JOIN users u ON u.email = gm.email WHERE gm.group_id = %s ORDER BY gm.added_at """, (owned["group_id"],)) return { "role": "owner", "group_id": owned["group_id"], "plan_type": owned["plan_type"], "max_members": GROUP_MAX_MEMBERS.get(owned["plan_type"], 0), "members": members or [], } # Am I a member? member = await fetchone(""" SELECT ug.group_id, ug.plan_type, ug.owner_id, u.name AS owner_name, u.email AS owner_email FROM group_members gm JOIN user_groups ug ON ug.group_id = gm.group_id LEFT JOIN users u ON u.user_id = ug.owner_id WHERE gm.email = %s LIMIT 1 """, (user["email"],)) if member: return { "role": "member", "group_id": member["group_id"], "plan_type": member["plan_type"], "owner_name": member.get("owner_name"), "owner_email": member.get("owner_email"), } return {"role": "none", "plan_type": plan} @api_router.post("/group/members") async def add_group_member(request: Request): user = await get_current_user(request) body = await request.json() email = (body.get("email") or "").lower().strip() if not email or "@" not in email: raise HTTPException(status_code=400, detail="Email invalide") if email == user["email"]: raise HTTPException(status_code=400, detail="Vous ne pouvez pas vous ajouter vous-même") plan = user.get("subscription_plan", "solo") if plan not in ("duo", "family"): raise HTTPException(status_code=403, detail="Un abonnement Duo ou Famille est requis pour créer un groupe") group = await ensure_user_group(user) if not group: raise HTTPException(status_code=400, detail="Impossible de créer le groupe") # Check already member somewhere existing_any = await fetchone("SELECT group_id FROM group_members WHERE email=%s", (email,)) if existing_any: raise HTTPException(status_code=400, detail="Cet email est déjà dans un autre groupe") # Check limit count = await fetchone("SELECT COUNT(*) AS cnt FROM group_members WHERE group_id=%s", (group["group_id"],)) max_m = GROUP_MAX_MEMBERS.get(plan, 0) if count["cnt"] >= max_m: raise HTTPException(status_code=403, detail=f"Limite atteinte : {max_m} membre(s) maximum pour le plan {plan.upper()}") target = await fetchone("SELECT user_id FROM users WHERE email=%s", (email,)) target_user_id = target["user_id"] if target else None await execute("INSERT INTO group_members (group_id,email,user_id,added_at) VALUES (%s,%s,%s,%s)", (group["group_id"], email, target_user_id, datetime.now(timezone.utc).isoformat())) return {"message": f"{email} ajouté à votre groupe", "email": email} @api_router.delete("/group/members/{email}") async def remove_group_member(email: str, request: Request): user = await get_current_user(request) group = await fetchone("SELECT * FROM user_groups WHERE owner_id=%s", (user["user_id"],)) if not group: raise HTTPException(status_code=404, detail="Vous n'avez pas de groupe") email = email.lower().strip() await execute("DELETE FROM group_members WHERE group_id=%s AND email=%s", (group["group_id"], email)) return {"message": "Membre retiré"} @api_router.post("/group/leave") async def leave_group(request: Request): user = await get_current_user(request) await execute("DELETE FROM group_members WHERE email=%s", (user["email"],)) return {"message": "Vous avez quitté le groupe"} # ===== LISTS ===== @api_router.get("/lists") async def get_lists(request: Request, status: str = "active"): user = await get_current_user(request) if status == "active": own = await fetchall("SELECT * FROM shopping_lists WHERE owner_id=%s AND (status IS NULL OR status != 'completed') ORDER BY position ASC, created_at DESC", (user["user_id"],)) shared_ids = await fetchall("SELECT list_id FROM list_shares WHERE shared_email=%s", (user["email"],)) shared = [] for s in shared_ids: lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s AND (status IS NULL OR status != 'completed')", (s["list_id"],)) if lst: shared.append(lst) else: own = await fetchall("SELECT * FROM shopping_lists WHERE owner_id=%s", (user["user_id"],)) shared = [] result = [] for lst in own: built = await build_list_response(lst) if built: result.append(built) for lst in shared: built = await build_list_response(lst) if built: built["is_shared"] = True result.append(built) return result @api_router.post("/lists") async def create_list(req: ListCreate, request: Request): user = await get_current_user(request) plan = await get_effective_plan(user) max_lists = SUBSCRIPTION_PLANS.get(plan, SUBSCRIPTION_PLANS["solo"])["max_lists"] count_row = await fetchone("SELECT COUNT(*) as cnt FROM shopping_lists WHERE owner_id=%s AND (status IS NULL OR status != 'completed')", (user["user_id"],)) if count_row["cnt"] >= max_lists: raise HTTPException(status_code=403, detail=f"Limite de listes atteinte ({max_lists}).") list_id = f"list_{uuid.uuid4().hex[:12]}" now = datetime.now(timezone.utc).isoformat() await execute("INSERT INTO shopping_lists (list_id,name,owner_id,owner_email,status,created_at,updated_at) VALUES (%s,%s,%s,%s,%s,%s,%s)", (list_id, req.name, user["user_id"], user["email"], "active", now, now)) await execute("UPDATE app_stats SET stat_value=stat_value+1 WHERE stat_key='total_lists_created'") lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s", (list_id,)) return await build_list_response(lst) @api_router.get("/lists/history/completed") async def get_history(request: Request): user = await get_current_user(request) completed = await fetchall("SELECT * FROM shopping_lists WHERE owner_id=%s AND status='completed' ORDER BY completed_at DESC", (user["user_id"],)) result = [] for lst in completed: built = await build_list_response(lst) if built: result.append(built) return result @api_router.put("/lists/reorder") async def reorder_lists(request: Request): """Réordonner les listes : body = {ordered_ids: [list_id1, list_id2, ...]}""" user = await get_current_user(request) body = await request.json() ids = body.get("ordered_ids", []) if not isinstance(ids, list): raise HTTPException(status_code=400, detail="ordered_ids must be a list") for pos, lid in enumerate(ids): await execute("UPDATE shopping_lists SET position=%s WHERE list_id=%s AND owner_id=%s", (pos, lid, user["user_id"])) return {"message": "Reordered", "count": len(ids)} @api_router.get("/lists/{list_id}") async def get_list(list_id: str, request: Request): user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s", (list_id,)) if not lst: raise HTTPException(status_code=404, detail="List not found") is_shared = await fetchone("SELECT id FROM list_shares WHERE list_id=%s AND shared_email=%s", (list_id, user["email"])) if lst["owner_id"] != user["user_id"] and not is_shared: raise HTTPException(status_code=404, detail="List not found") return await build_list_response(lst) @api_router.put("/lists/{list_id}") async def update_list(list_id: str, req: ListUpdate, request: Request): user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s AND owner_id=%s", (list_id, user["user_id"])) if not lst: raise HTTPException(status_code=404, detail="List not found") if req.name: await execute("UPDATE shopping_lists SET name=%s, updated_at=%s WHERE list_id=%s", (req.name, datetime.now(timezone.utc).isoformat(), list_id)) return {"message": "Updated"} @api_router.delete("/lists/{list_id}") async def delete_list(list_id: str, request: Request): user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s AND owner_id=%s", (list_id, user["user_id"])) if not lst: raise HTTPException(status_code=404, detail="List not found") await execute("DELETE FROM list_items WHERE list_id=%s", (list_id,)) await execute("DELETE FROM list_shares WHERE list_id=%s", (list_id,)) await execute("DELETE FROM shopping_lists WHERE list_id=%s", (list_id,)) return {"message": "Deleted"} @api_router.post("/lists/{list_id}/complete") async def complete_list(list_id: str, request: Request): user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s AND owner_id=%s", (list_id, user["user_id"])) if not lst: raise HTTPException(status_code=404, detail="List not found") await execute("UPDATE shopping_lists SET status='completed', completed_at=%s WHERE list_id=%s", (datetime.now(timezone.utc).isoformat(), list_id)) return {"message": "List completed"} @api_router.post("/lists/{list_id}/reactivate") async def reactivate_list(list_id: str, request: Request): """Reprend une liste archivée : la remet active SANS ajouter '(copie)' au nom. Réinitialise tous les items à non-cochés et supprime la version archivée précédente.""" user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s AND owner_id=%s", (list_id, user["user_id"])) if not lst: raise HTTPException(status_code=404, detail="List not found") # Décocher tous les items pour repartir à zéro await execute("UPDATE list_items SET checked=0 WHERE list_id=%s", (list_id,)) await execute("UPDATE shopping_lists SET status='active', completed_at=NULL, updated_at=%s WHERE list_id=%s", (datetime.now(timezone.utc).isoformat(), list_id)) return {"message": "List reactivated"} @api_router.patch("/lists/{list_id}") async def rename_list(list_id: str, request: Request): """Renommer une liste.""" user = await get_current_user(request) body = await request.json() new_name = (body.get("name") or "").strip() if not new_name: raise HTTPException(status_code=400, detail="Le nom ne peut pas être vide") if len(new_name) > 100: raise HTTPException(status_code=400, detail="Nom trop long (max 100 caractères)") lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s AND owner_id=%s", (list_id, user["user_id"])) if not lst: raise HTTPException(status_code=404, detail="List not found ou vous n'êtes pas l'owner") await execute("UPDATE shopping_lists SET name=%s, updated_at=%s WHERE list_id=%s", (new_name, datetime.now(timezone.utc).isoformat(), list_id)) await notify_list_update(list_id, "list_renamed", {"name": new_name}, user["user_id"]) return {"message": "List renamed", "name": new_name} @api_router.put("/lists/{list_id}/items/reorder") async def reorder_items(list_id: str, request: Request): """Réordonner les items d'une liste : body = {ordered_ids: [item_id1, item_id2, ...]}""" user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s", (list_id,)) if not lst: raise HTTPException(status_code=404, detail="List not found") is_shared = await fetchone("SELECT id FROM list_shares WHERE list_id=%s AND shared_email=%s", (list_id, user["email"])) if lst["owner_id"] != user["user_id"] and not is_shared: raise HTTPException(status_code=403, detail="No access") body = await request.json() ids = body.get("ordered_ids", []) for pos, iid in enumerate(ids): await execute("UPDATE list_items SET position=%s WHERE item_id=%s AND list_id=%s", (pos, iid, list_id)) return {"message": "Items reordered", "count": len(ids)} @api_router.post("/lists/{list_id}/duplicate") async def duplicate_list(list_id: str, request: Request): user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s", (list_id,)) if not lst: raise HTTPException(status_code=404, detail="List not found") is_shared = await fetchone("SELECT id FROM list_shares WHERE list_id=%s AND shared_email=%s", (list_id, user["email"])) if lst["owner_id"] != user["user_id"] and not is_shared: raise HTTPException(status_code=404, detail="List not found") new_id = f"list_{uuid.uuid4().hex[:12]}" now = datetime.now(timezone.utc).isoformat() await execute("INSERT INTO shopping_lists (list_id,name,owner_id,owner_email,status,created_at,updated_at) VALUES (%s,%s,%s,%s,%s,%s,%s)", (new_id, f"{lst['name']} (copie)", user["user_id"], user["email"], "active", now, now)) items = await fetchall("SELECT * FROM list_items WHERE list_id=%s", (list_id,)) for item in items: new_item_id = f"item_{uuid.uuid4().hex[:12]}" await execute("INSERT INTO list_items (item_id,list_id,name,quantity,category,checked,added_by,created_at) VALUES (%s,%s,%s,%s,%s,0,%s,%s)", (new_item_id, new_id, item["name"], item["quantity"], item["category"], user["email"], now)) new_lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s", (new_id,)) return await build_list_response(new_lst) # ===== LIST ITEMS ===== @api_router.post("/lists/{list_id}/items") async def add_item(list_id: str, req: ItemCreate, request: Request): user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s", (list_id,)) if not lst: raise HTTPException(status_code=404, detail="List not found") is_shared = await fetchone("SELECT id FROM list_shares WHERE list_id=%s AND shared_email=%s", (list_id, user["email"])) if lst["owner_id"] != user["user_id"] and not is_shared: raise HTTPException(status_code=404, detail="List not found") item_id = f"item_{uuid.uuid4().hex[:12]}" now = datetime.now(timezone.utc).isoformat() await execute("INSERT INTO list_items (item_id,list_id,name,quantity,category,checked,added_by,created_at) VALUES (%s,%s,%s,%s,%s,0,%s,%s)", (item_id, list_id, req.name, req.quantity or "1", req.category or "other", user["email"], now)) await execute("UPDATE shopping_lists SET updated_at=%s WHERE list_id=%s", (now, list_id)) shares = await fetchall("SELECT shared_email FROM list_shares WHERE list_id=%s", (list_id,)) if shares or lst["owner_id"] != user["user_id"]: await notify_list_members(list_id, user["email"], req.name, "added") await notify_list_update(list_id, "item_added", {"item_id": item_id, "name": req.name}, user["user_id"]) return {"item_id": item_id, "name": req.name, "quantity": req.quantity, "category": req.category, "checked": False, "added_by": user["email"], "created_at": now} @api_router.put("/lists/{list_id}/items/{item_id}") async def update_item(list_id: str, item_id: str, req: ItemUpdate, request: Request): user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s", (list_id,)) if not lst: raise HTTPException(status_code=404, detail="List not found") updates, params = [], [] if req.name is not None: updates.append("name=%s"); params.append(req.name) if req.quantity is not None: updates.append("quantity=%s"); params.append(req.quantity) if req.category is not None: updates.append("category=%s"); params.append(req.category) if req.checked is not None: updates.append("checked=%s"); params.append(1 if req.checked else 0) if updates: params.append(item_id) await execute(f"UPDATE list_items SET {','.join(updates)} WHERE item_id=%s", params) await execute("UPDATE shopping_lists SET updated_at=%s WHERE list_id=%s", (datetime.now(timezone.utc).isoformat(), list_id)) if req.checked is not None: item = await fetchone("SELECT name FROM list_items WHERE item_id=%s", (item_id,)) if item: await notify_list_members(list_id, user["email"], item["name"], "checked" if req.checked else "added") await notify_list_update(list_id, "item_updated", {"item_id": item_id}, user["user_id"]) return {"message": "Updated"} @api_router.delete("/lists/{list_id}/items/{item_id}") async def delete_item(list_id: str, item_id: str, request: Request): user = await get_current_user(request) item = await fetchone("SELECT name FROM list_items WHERE item_id=%s AND list_id=%s", (item_id, list_id)) await execute("DELETE FROM list_items WHERE item_id=%s AND list_id=%s", (item_id, list_id)) await execute("UPDATE shopping_lists SET updated_at=%s WHERE list_id=%s", (datetime.now(timezone.utc).isoformat(), list_id)) if item: await notify_list_members(list_id, user["email"], item["name"], "removed") await notify_list_update(list_id, "item_deleted", {"item_id": item_id}, user["user_id"]) return {"message": "Deleted"} # ===== SHARE ===== @api_router.post("/lists/{list_id}/share") async def share_list(list_id: str, req: ShareListRequest, request: Request): user = await get_current_user(request) lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s AND owner_id=%s", (list_id, user["user_id"])) if not lst: raise HTTPException(status_code=404, detail="List not found or not owner") target_email = req.email.lower().strip() if target_email == user["email"]: raise HTTPException(status_code=400, detail="Cannot share with yourself") target = await fetchone("SELECT user_id FROM users WHERE email=%s", (target_email,)) if not target: raise HTTPException(status_code=404, detail="User not found") existing = await fetchone("SELECT id FROM list_shares WHERE list_id=%s AND shared_email=%s", (list_id, target_email)) if existing: raise HTTPException(status_code=400, detail="Already shared") await execute("INSERT INTO list_shares (list_id,shared_email) VALUES (%s,%s)", (list_id, target_email)) return {"message": f"List shared with {target_email}"} @api_router.delete("/lists/{list_id}/share/{email}") async def unshare_list(list_id: str, email: str, request: Request): user = await get_current_user(request) await execute("DELETE FROM list_shares WHERE list_id=%s AND shared_email=%s", (list_id, email.lower().strip())) return {"message": "Unshared"} # ===== CATEGORIES ===== @api_router.get("/categories") async def get_categories(): return ITEM_CATEGORIES # ===== STT ===== @api_router.post("/stt/transcribe") async def transcribe_audio(request: Request): user = await get_current_user(request) if user.get("subscription_plan") != "family": raise HTTPException(status_code=403, detail="La dictée vocale est réservée au Pack Famille") form = await request.form() audio_file = form.get("audio") if not audio_file: raise HTTPException(status_code=400, detail="No audio file") import tempfile contents = await audio_file.read() with tempfile.NamedTemporaryFile(suffix=".m4a", delete=False) as tmp: tmp.write(contents) tmp_path = tmp.name try: from emergentintegrations.llm.openai import OpenAISpeechToText stt = OpenAISpeechToText(api_key=os.environ["EMERGENT_LLM_KEY"]) with open(tmp_path, "rb") as f: response = await stt.transcribe(file=f, model="whisper-1", response_format="json", language="fr") return {"text": response.text} finally: try: os.unlink(tmp_path) except: pass # ===== AI CATEGORIZATION ===== @api_router.post("/items/categorize") async def categorize_items(request: Request): user = await get_current_user(request) if user.get("subscription_plan") != "family": raise HTTPException(status_code=403, detail="Réservé au Pack Famille") body = await request.json() item_names = body.get("items", []) if not item_names: return {"categorized": []} categories_desc = ", ".join([f"{c['id']} ({c['name']})" for c in ITEM_CATEGORIES]) prompt = f"""Classifie chaque article dans une catégorie. JSON uniquement. Catégories: {categories_desc} Articles: {', '.join(item_names)} Format: [{{"name":"article","category":"id"}}]""" try: from emergentintegrations.llm.chat import LlmChat, UserMessage import json as json_module chat = LlmChat(api_key=os.environ["EMERGENT_LLM_KEY"], session_id=f"cat_{uuid.uuid4().hex[:8]}", system_message="Classifie des articles de supermarché. JSON uniquement.") chat.with_model("openai", "gpt-4.1-mini") response = await chat.send_message(UserMessage(text=prompt)) text = response.strip() if text.startswith("```"): text = text.split("\n", 1)[1].rsplit("```", 1)[0] parsed = json_module.loads(text) for item in parsed: if item.get("category") not in CATEGORY_IDS: item["category"] = "other" return {"categorized": parsed} except Exception as e: return {"categorized": [{"name": n, "category": "other"} for n in item_names]} # ===== SUGGESTIONS ===== @api_router.get("/suggestions") async def get_suggestions(request: Request, q: str = ""): user = await get_current_user(request) if q: items = await fetchall("SELECT name, category, COUNT(*) as cnt FROM list_items li JOIN shopping_lists sl ON li.list_id=sl.list_id WHERE (sl.owner_id=%s OR li.list_id IN (SELECT list_id FROM list_shares WHERE shared_email=%s)) AND li.name LIKE %s GROUP BY name, category ORDER BY cnt DESC LIMIT 20", (user["user_id"], user["email"], f"%{q}%")) else: items = await fetchall("SELECT name, category, COUNT(*) as cnt FROM list_items li JOIN shopping_lists sl ON li.list_id=sl.list_id WHERE sl.owner_id=%s OR li.list_id IN (SELECT list_id FROM list_shares WHERE shared_email=%s) GROUP BY name, category ORDER BY cnt DESC LIMIT 20", (user["user_id"], user["email"])) return [{"name": i["name"], "category": i["category"], "count": i["cnt"]} for i in items] # ===== SUBSCRIPTIONS ===== @api_router.get("/subscriptions/plans") async def get_plans(): return SUBSCRIPTION_PLANS @api_router.post("/subscriptions/validate-promo") async def validate_promo_code(request: Request): """Valide un code promo côté user et retourne la remise applicable (preview).""" user = await get_current_user(request) body = await request.json() code = (body.get("code") or "").strip().upper() plan_id = body.get("plan_id") if not code: raise HTTPException(status_code=400, detail="Code requis") promo = await fetchone("SELECT * FROM promo_codes WHERE UPPER(code)=%s AND active=1", (code,)) if not promo: raise HTTPException(status_code=404, detail="Code invalide ou inactif") # Check date range now_iso = datetime.now(timezone.utc).isoformat() if promo.get("start_date") and promo["start_date"] > now_iso: raise HTTPException(status_code=400, detail="Code pas encore actif") if promo.get("end_date") and promo["end_date"] < now_iso: raise HTTPException(status_code=400, detail="Code expiré") # Check max uses if promo.get("max_uses", 0) > 0 and promo.get("current_uses", 0) >= promo["max_uses"]: raise HTTPException(status_code=400, detail="Code épuisé (limite atteinte)") # Check if user already used this promo code (any type) already = await fetchone("SELECT 1 FROM promo_redemptions WHERE code_id=%s AND user_id=%s", (promo["code_id"], user["user_id"])) if already: raise HTTPException(status_code=400, detail="Vous avez déjà utilisé ce code") plan = SUBSCRIPTION_PLANS.get(plan_id) if plan_id else None base_price = float(plan["price"]) if plan else 0 discount_type = promo.get("discount_type", "free_months") discount_value = promo.get("discount_value", 0) final_price = base_price description = "" if discount_type == "free_months": final_price = 0 # Bypass Stripe — upgrade direct description = f"{discount_value} mois {promo.get('plan_upgrade','').upper()} offerts" elif discount_type == "percent": final_price = round(base_price * (1 - discount_value / 100.0), 2) description = f"-{discount_value}% sur votre abonnement" elif discount_type == "fixed_amount": final_price = max(0.0, round(base_price - discount_value, 2)) description = f"-{discount_value}€ sur votre abonnement" return { "valid": True, "code": promo["code"], "discount_type": discount_type, "discount_value": discount_value, "plan_upgrade": promo.get("plan_upgrade"), "original_price": base_price, "final_price": final_price, "savings": round(base_price - final_price, 2), "description": description, } @api_router.post("/subscriptions/checkout") async def create_checkout(req: CheckoutRequest, request: Request): user = await get_current_user(request) plan = SUBSCRIPTION_PLANS.get(req.plan_id) if not plan: raise HTTPException(status_code=400, detail="Invalid plan") # Empêcher le downgrade via cet endpoint (sauf solo→gratuit explicite) plan_rank = {"solo": 1, "duo": 2, "family": 3} current_rank = plan_rank.get(user.get("subscription_plan", "solo"), 1) target_rank = plan_rank.get(req.plan_id, 1) if target_rank < current_rank and req.plan_id != "solo": raise HTTPException(status_code=400, detail=f"Vous êtes déjà sur un plan supérieur ({user.get('subscription_plan', 'solo').upper()}). Impossible de souscrire à un plan inférieur.") # Apply promo code if provided final_price = float(plan["price"]) applied_promo = None if req.promo_code: code = req.promo_code.strip().upper() promo = await fetchone("SELECT * FROM promo_codes WHERE UPPER(code)=%s AND active=1", (code,)) if not promo: raise HTTPException(status_code=400, detail="Code promo invalide") now_iso = datetime.now(timezone.utc).isoformat() if promo.get("start_date") and promo["start_date"] > now_iso: raise HTTPException(status_code=400, detail="Code pas encore actif") if promo.get("end_date") and promo["end_date"] < now_iso: raise HTTPException(status_code=400, detail="Code expiré") if promo.get("max_uses", 0) > 0 and promo.get("current_uses", 0) >= promo["max_uses"]: raise HTTPException(status_code=400, detail="Code épuisé") already = await fetchone("SELECT 1 FROM promo_redemptions WHERE code_id=%s AND user_id=%s", (promo["code_id"], user["user_id"])) if already: raise HTTPException(status_code=400, detail="Vous avez déjà utilisé ce code") dt = promo.get("discount_type", "free_months") dv = promo.get("discount_value", 0) if dt == "free_months": # Upgrade direct sans Stripe months = int(dv) plan_upgrade = promo.get("plan_upgrade") or req.plan_id end_date = (datetime.now(timezone.utc) + timedelta(days=30 * months)).isoformat() await execute("UPDATE users SET subscription_plan=%s, subscription_status='active', subscription_end_date=%s WHERE user_id=%s", (plan_upgrade, end_date, user["user_id"])) await execute("UPDATE promo_codes SET current_uses=current_uses+1 WHERE code_id=%s", (promo["code_id"],)) await execute("INSERT INTO promo_redemptions (redemption_id,code_id,promo_code,user_id,created_at) VALUES (%s,%s,%s,%s,%s)", (f"rdm_{uuid.uuid4().hex[:12]}", promo["code_id"], promo["code"], user["user_id"], now_iso)) return {"direct_upgrade": True, "plan": plan_upgrade, "months": months, "message": f"{months} mois {plan_upgrade.upper()} offerts !"} elif dt == "percent": final_price = max(0.5, round(final_price * (1 - dv / 100.0), 2)) elif dt == "fixed_amount": final_price = max(0.5, round(final_price - dv, 2)) applied_promo = promo if final_price == 0: await execute("UPDATE users SET subscription_plan=%s, subscription_status='active' WHERE user_id=%s", (req.plan_id, user["user_id"])) return {"message": "Switched to free plan", "plan": req.plan_id} from emergentintegrations.payments.stripe.checkout import StripeCheckout, CheckoutSessionRequest host_url = str(request.base_url).rstrip("/") stripe_checkout = StripeCheckout(api_key=os.environ["STRIPE_API_KEY"], webhook_url=f"{host_url}api/webhook/stripe") origin = req.origin_url.rstrip("/") metadata = {"user_id": user["user_id"], "plan_id": req.plan_id, "email": user["email"]} if applied_promo: metadata["promo_code_id"] = applied_promo["code_id"] metadata["promo_code"] = applied_promo["code"] checkout_req = CheckoutSessionRequest(amount=final_price, currency="eur", success_url=f"{origin}/subscription?session_id={{CHECKOUT_SESSION_ID}}", cancel_url=f"{origin}/subscription", metadata=metadata) session = await stripe_checkout.create_checkout_session(checkout_req) await execute("INSERT INTO payment_transactions (transaction_id,session_id,user_id,email,plan_id,amount,currency,payment_status,created_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", (f"txn_{uuid.uuid4().hex[:12]}", session.session_id, user["user_id"], user["email"], req.plan_id, final_price, "eur", "pending", datetime.now(timezone.utc).isoformat())) # Mark promo as redeemed immediately (before payment confirmation) to prevent double-use if applied_promo: await execute("UPDATE promo_codes SET current_uses=current_uses+1 WHERE code_id=%s", (applied_promo["code_id"],)) await execute("INSERT INTO promo_redemptions (redemption_id,code_id,promo_code,user_id,created_at) VALUES (%s,%s,%s,%s,%s)", (f"rdm_{uuid.uuid4().hex[:12]}", applied_promo["code_id"], applied_promo["code"], user["user_id"], datetime.now(timezone.utc).isoformat())) return {"url": session.url, "session_id": session.session_id, "final_price": final_price} @api_router.get("/subscriptions/status/{session_id}") async def check_payment_status(session_id: str, request: Request): user = await get_current_user(request) txn = await fetchone("SELECT * FROM payment_transactions WHERE session_id=%s AND user_id=%s", (session_id, user["user_id"])) if not txn: raise HTTPException(status_code=404, detail="Transaction not found") if txn.get("payment_status") == "paid": return {"status": "paid", "plan_id": txn["plan_id"]} from emergentintegrations.payments.stripe.checkout import StripeCheckout host_url = str(request.base_url).rstrip("/") stripe_checkout = StripeCheckout(api_key=os.environ["STRIPE_API_KEY"], webhook_url=f"{host_url}api/webhook/stripe") status = await stripe_checkout.get_checkout_status(session_id) await execute("UPDATE payment_transactions SET payment_status=%s, updated_at=%s WHERE session_id=%s", (status.payment_status, datetime.now(timezone.utc).isoformat(), session_id)) if status.payment_status == "paid": await execute("UPDATE users SET subscription_plan=%s, subscription_status='active' WHERE user_id=%s", (txn["plan_id"], user["user_id"])) return {"status": status.payment_status, "plan_id": txn["plan_id"]} # ===== REFERRAL ===== @api_router.get("/referral/code") async def get_referral_code(request: Request): user = await get_current_user(request) existing = await fetchone("SELECT * FROM referrals WHERE owner_id=%s", (user["user_id"],)) if existing: existing.pop("id", None) return existing code = f"PP-{secrets.token_urlsafe(6).upper()}" ref_id = f"ref_{uuid.uuid4().hex[:12]}" await execute("INSERT INTO referrals (referral_id,owner_id,owner_email,code,uses,rewarded_months,created_at) VALUES (%s,%s,%s,%s,0,0,%s)", (ref_id, user["user_id"], user["email"], code, datetime.now(timezone.utc).isoformat())) return {"referral_id": ref_id, "owner_id": user["user_id"], "owner_email": user["email"], "code": code, "uses": 0, "rewarded_months": 0} @api_router.post("/referral/redeem") async def redeem_referral(req: ReferralRedeemRequest, request: Request): user = await get_current_user(request) existing = await fetchone("SELECT id FROM referral_redemptions WHERE redeemed_by=%s", (user["user_id"],)) if existing: raise HTTPException(status_code=400, detail="Déjà utilisé un code") referral = await fetchone("SELECT * FROM referrals WHERE code=%s", (req.code.upper().strip(),)) if not referral: raise HTTPException(status_code=404, detail="Code invalide") if referral["owner_id"] == user["user_id"]: raise HTTPException(status_code=400, detail="Votre propre code") await execute("INSERT INTO referral_redemptions (redemption_id,referral_code,referral_owner_id,redeemed_by,created_at) VALUES (%s,%s,%s,%s,%s)", (f"rdm_{uuid.uuid4().hex[:12]}", req.code.upper().strip(), referral["owner_id"], user["user_id"], datetime.now(timezone.utc).isoformat())) await execute("UPDATE referrals SET uses=uses+1, rewarded_months=rewarded_months+1 WHERE code=%s", (req.code.upper().strip(),)) if user.get("subscription_plan") == "solo": trial_end = (datetime.now(timezone.utc) + timedelta(days=30)).isoformat() await execute("UPDATE users SET subscription_plan='duo', subscription_status='referral_reward', trial_end_date=%s WHERE user_id=%s", (trial_end, user["user_id"])) return {"message": "Code activé ! 1 mois Duo gratuit."} @api_router.get("/referral/stats") async def get_referral_stats(request: Request): user = await get_current_user(request) ref = await fetchone("SELECT * FROM referrals WHERE owner_id=%s", (user["user_id"],)) if not ref: return {"code": None, "uses": 0, "rewarded_months": 0, "referred_users": []} ref.pop("id", None) # Fetch referred users referred = await fetchall(""" SELECT u.name, u.email, u.subscription_plan, rr.created_at FROM referral_redemptions rr JOIN users u ON u.user_id = rr.redeemed_by WHERE rr.referral_owner_id = %s ORDER BY rr.created_at DESC """, (user["user_id"],)) ref["referred_users"] = referred or [] return ref # ===== TRIAL ===== @api_router.post("/trial/activate") async def activate_trial(request: Request): user = await get_current_user(request) # Vérifie si l'essai gratuit est activé par l'admin trial_enabled = await get_setting("trial_enabled", "1") == "1" if not trial_enabled: raise HTTPException(status_code=400, detail="L'essai gratuit est actuellement désactivé") trial_days = int(await get_setting("trial_duration_days", "7")) if user.get("trial_used"): raise HTTPException(status_code=400, detail="Déjà utilisé") if user.get("subscription_plan") != "solo": raise HTTPException(status_code=400, detail="Réservé aux Solo") trial_end = (datetime.now(timezone.utc) + timedelta(days=trial_days)).isoformat() await execute("UPDATE users SET subscription_plan='family', subscription_status='trial', trial_end_date=%s, trial_used=1 WHERE user_id=%s", (trial_end, user["user_id"])) return {"message": f"Essai {trial_days} jours activé !", "trial_end_date": trial_end, "duration_days": trial_days} @api_router.get("/trial/status") async def get_trial_status(request: Request): user = await get_current_user(request) trial_end = user.get("trial_end_date") if not trial_end: return {"active": False, "trial_used": bool(user.get("trial_used", False))} try: end_date = datetime.fromisoformat(trial_end).replace(tzinfo=timezone.utc) if datetime.fromisoformat(trial_end).tzinfo is None else datetime.fromisoformat(trial_end) remaining = (end_date - datetime.now(timezone.utc)).total_seconds() if remaining > 0: return {"active": True, "trial_end_date": trial_end, "remaining_days": int(remaining / 86400), "trial_used": True} else: if user.get("subscription_status") == "trial": await execute("UPDATE users SET subscription_plan='solo', subscription_status='active' WHERE user_id=%s", (user["user_id"],)) return {"active": False, "expired": True, "trial_used": True} except: return {"active": False, "trial_used": bool(user.get("trial_used", False))} # ===== NOTIFICATIONS ===== @api_router.post("/notifications/register") async def register_push_token(request: Request): user = await get_current_user(request) body = await request.json() push_token = body.get("push_token") if not push_token: raise HTTPException(status_code=400, detail="Missing push_token") existing = await fetchone("SELECT id FROM push_tokens WHERE user_id=%s", (user["user_id"],)) if existing: await execute("UPDATE push_tokens SET push_token=%s, updated_at=%s WHERE user_id=%s", (push_token, datetime.now(timezone.utc).isoformat(), user["user_id"])) else: await execute("INSERT INTO push_tokens (user_id,push_token,updated_at) VALUES (%s,%s,%s)", (user["user_id"], push_token, datetime.now(timezone.utc).isoformat())) return {"message": "Registered"} @api_router.get("/notifications") async def get_notifications(request: Request): user = await get_current_user(request) notifs = await fetchall("SELECT * FROM notifications WHERE user_id=%s ORDER BY created_at DESC LIMIT 50", (user["user_id"],)) return [{"notif_id": n["notif_id"], "user_id": n["user_id"], "type": n["type"], "message": n["message"], "list_id": n["list_id"], "read": bool(n["read_status"]), "created_at": n["created_at"]} for n in notifs] @api_router.put("/notifications/{notif_id}/read") async def mark_notification_read(notif_id: str, request: Request): user = await get_current_user(request) await execute("UPDATE notifications SET read_status=1 WHERE notif_id=%s AND user_id=%s", (notif_id, user["user_id"])) return {"message": "Read"} @api_router.get("/notifications/settings") async def get_notification_settings(request: Request): user = await get_current_user(request) s = await fetchone("SELECT * FROM notification_settings WHERE user_id=%s", (user["user_id"],)) if not s: return {"user_id": user["user_id"], "enabled": True, "list_updates": True} return {"user_id": s["user_id"], "enabled": bool(s["enabled"]), "list_updates": bool(s["list_updates"])} @api_router.put("/notifications/settings") async def update_notification_settings(request: Request): user = await get_current_user(request) body = await request.json() existing = await fetchone("SELECT id FROM notification_settings WHERE user_id=%s", (user["user_id"],)) if existing: await execute("UPDATE notification_settings SET enabled=%s, list_updates=%s, updated_at=%s WHERE user_id=%s", (1 if body.get("enabled", True) else 0, 1 if body.get("list_updates", True) else 0, datetime.now(timezone.utc).isoformat(), user["user_id"])) else: await execute("INSERT INTO notification_settings (user_id,enabled,list_updates,updated_at) VALUES (%s,%s,%s,%s)", (user["user_id"], 1 if body.get("enabled", True) else 0, 1 if body.get("list_updates", True) else 0, datetime.now(timezone.utc).isoformat())) return {"message": "Updated"} @api_router.delete("/notifications/clear") async def clear_notifications(request: Request): user = await get_current_user(request) await execute("DELETE FROM notifications WHERE user_id=%s", (user["user_id"],)) return {"message": "Cleared"} # ===== STORES & PRICES ===== @api_router.post("/stores/nearby") async def find_nearby_stores(request: Request): await get_current_user(request) body = await request.json() lat, lon = body.get("lat"), body.get("lon") radius = body.get("radius", 3000) if not lat or not lon: raise HTTPException(status_code=400, detail="Missing lat/lon") query = f'[out:json][timeout:15];(node["shop"="supermarket"](around:{radius},{lat},{lon});way["shop"="supermarket"](around:{radius},{lat},{lon}););out center 20;' try: async with httpx.AsyncClient() as c: resp = await c.post("https://overpass-api.de/api/interpreter", data={"data": query}, timeout=20) if resp.status_code != 200: return {"stores": []} data = resp.json() import math stores = [] for el in data.get("elements", []): tags = el.get("tags", {}) name = tags.get("name", tags.get("brand", "Supermarché")) s_lat = el.get("lat") or el.get("center", {}).get("lat") s_lon = el.get("lon") or el.get("center", {}).get("lon") if not s_lat or not s_lon: continue dist = math.sqrt((float(s_lat) - lat)**2 + (float(s_lon) - lon)**2) * 111000 stores.append({"name": name, "brand": tags.get("brand", ""), "lat": s_lat, "lon": s_lon, "distance_m": round(dist), "address": tags.get("addr:street", ""), "store_type": classify_store_type(name)}) stores.sort(key=lambda s: s["distance_m"]) return {"stores": stores[:15]} except Exception as e: return {"stores": []} @api_router.get("/stores") async def get_user_stores(request: Request): user = await get_current_user(request) stores = await fetchall("SELECT * FROM user_stores WHERE user_id=%s", (user["user_id"],)) return [{k: v for k, v in s.items() if k != "id"} for s in stores] @api_router.post("/stores") async def add_user_store(request: Request): user = await get_current_user(request) body = await request.json() store_id = f"store_{uuid.uuid4().hex[:12]}" await execute("INSERT INTO user_stores (store_id,user_id,name,brand,store_type,lat,lon,address,created_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", (store_id, user["user_id"], body.get("name", "Magasin"), body.get("brand", ""), body.get("store_type", "standard"), body.get("lat"), body.get("lon"), body.get("address", ""), datetime.now(timezone.utc).isoformat())) return {"store_id": store_id, "name": body.get("name"), "store_type": body.get("store_type", "standard")} @api_router.delete("/stores/{store_id}") async def remove_user_store(store_id: str, request: Request): user = await get_current_user(request) await execute("DELETE FROM user_stores WHERE store_id=%s AND user_id=%s", (store_id, user["user_id"])) return {"message": "Deleted"} @api_router.post("/prices/compare") async def compare_prices(request: Request): user = await get_current_user(request) body = await request.json() items, stores = body.get("items", []), body.get("stores", []) if not items or not stores: raise HTTPException(status_code=400, detail="Missing items or stores") store_names = [s.get("name", "Magasin") for s in stores] store_types = [s.get("store_type", "standard") for s in stores] # Open Food Facts real prices real_prices = {} async with httpx.AsyncClient(timeout=10) as http_client: for item_name in items: try: resp = await http_client.get("https://prices.openfoodfacts.org/api/v1/products", params={"product_name__like": item_name, "page_size": 5}) if resp.status_code == 200: for prod in resp.json().get("items", []): if prod.get("price_count", 0) > 0: pr = await http_client.get("https://prices.openfoodfacts.org/api/v1/prices", params={"product_code": prod["code"], "currency": "EUR", "order_by": "-date", "page_size": 10}) if pr.status_code == 200: pi = pr.json().get("items", []) if pi: real_prices[item_name.lower()] = {"avg_price": round(sum(p.get("price", 0) for p in pi) / len(pi), 2), "product_name": prod.get("product_name", item_name), "count": len(pi), "source": "Open Food Facts"} break except: pass real_info = "" if real_prices: real_info = "\nPrix réels (Open Food Facts):\n" + "\n".join([f"- {n}: {d['avg_price']}€" for n, d in real_prices.items()]) prompt = f"""Estime les prix en euros. Discount=-15%, premium=+15%, bio=+35%.{real_info} Articles: {', '.join(items)} Magasins: {', '.join([f'{n} ({t})' for n, t in zip(store_names, store_types)])} JSON: {{"comparisons":[{{"item":"x","prices":[{{"store":"y","price":1.5,"unit":"pièce"}}]}}]}}""" try: from emergentintegrations.llm.chat import LlmChat, UserMessage import json as json_module chat = LlmChat(api_key=os.environ["EMERGENT_LLM_KEY"], session_id=f"p_{uuid.uuid4().hex[:8]}", system_message="Expert prix supermarchés France. JSON uniquement.") chat.with_model("openai", "gpt-4.1-mini") response = await chat.send_message(UserMessage(text=prompt)) text = response.strip() if text.startswith("```"): text = text.split("\n", 1)[1].rsplit("```", 1)[0] parsed = json_module.loads(text) totals_lower = {n.lower(): 0.0 for n in store_names} name_map = {n.lower(): n for n in store_names} for comp in parsed.get("comparisons", []): for p in comp.get("prices", []): sk = p.get("store", "").lower() if sk in totals_lower: totals_lower[sk] += p.get("price", 0); p["store"] = name_map.get(sk, p["store"]) ik = comp.get("item", "").lower() comp["has_real_data"] = ik in real_prices if ik in real_prices: comp["real_avg_price"] = real_prices[ik]["avg_price"]; comp["source"] = "Open Food Facts" else: comp["source"] = "Estimation IA" parsed["totals"] = sorted([{"store": name_map.get(k, k), "total": round(v, 2)} for k, v in totals_lower.items()], key=lambda x: x["total"]) real_count = sum(1 for k in items if k.lower() in real_prices) parsed["data_quality"] = {"total_items": len(items), "items_with_real_prices": real_count, "items_estimated": len(items) - real_count} return parsed except Exception as e: raise HTTPException(status_code=500, detail="Erreur comparaison") # ===== ADMIN ===== @api_router.get("/admin/stats") async def admin_stats(request: Request): await get_admin_user(request) total = (await fetchone("SELECT COUNT(*) as c FROM users"))["c"] solo = (await fetchone("SELECT COUNT(*) as c FROM users WHERE subscription_plan='solo'"))["c"] duo = (await fetchone("SELECT COUNT(*) as c FROM users WHERE subscription_plan='duo'"))["c"] family = (await fetchone("SELECT COUNT(*) as c FROM users WHERE subscription_plan='family'"))["c"] active_lists = (await fetchone("SELECT COUNT(*) as c FROM shopping_lists WHERE status IS NULL OR status != 'completed'"))["c"] completed_lists = (await fetchone("SELECT COUNT(*) as c FROM shopping_lists WHERE status='completed'"))["c"] total_lists = (await fetchone("SELECT COUNT(*) as c FROM shopping_lists"))["c"] stat_row = await fetchone("SELECT stat_value FROM app_stats WHERE stat_key='total_lists_created'") total_created = stat_row["stat_value"] if stat_row else total_lists txns = (await fetchone("SELECT COUNT(*) as c FROM payment_transactions"))["c"] paid = (await fetchone("SELECT COUNT(*) as c FROM payment_transactions WHERE payment_status='paid'"))["c"] return {"total_users": total, "solo_users": solo, "duo_users": duo, "family_users": family, "active_lists": active_lists, "completed_lists": completed_lists, "total_lists": total_lists, "total_lists_created": max(total_created, total_lists), "total_transactions": txns, "paid_transactions": paid, "estimated_monthly_revenue": round(duo * 2.99 + family * 6.99, 2)} @api_router.get("/admin/users") async def admin_users(request: Request): await get_admin_user(request) users = await fetchall("SELECT user_id,email,name,role,subscription_plan,subscription_status,created_at,picture,trial_used,trial_end_date FROM users") for u in users: u["trial_used"] = bool(u.get("trial_used")) return users @api_router.put("/admin/users/{user_id}/subscription") async def admin_update_subscription(user_id: str, request: Request): await get_admin_user(request) body = await request.json() plan = body.get("plan") if plan not in SUBSCRIPTION_PLANS: raise HTTPException(status_code=400, detail="Invalid plan") await execute("UPDATE users SET subscription_plan=%s, subscription_status='active' WHERE user_id=%s", (plan, user_id)) return {"message": "Updated"} @api_router.put("/admin/users/{user_id}/role") async def admin_update_role(user_id: str, request: Request): await get_admin_user(request) body = await request.json() role = body.get("role") if role not in ["user", "admin"]: raise HTTPException(status_code=400, detail="Invalid role") await execute("UPDATE users SET role=%s WHERE user_id=%s", (role, user_id)) return {"message": "Updated"} # ===== ADMIN PROMO CODES ===== @api_router.get("/admin/promos") async def admin_get_promos(request: Request): await get_admin_user(request) promos = await fetchall("SELECT * FROM promo_codes ORDER BY created_at DESC") return [{k: v for k, v in p.items() if k != "id"} for p in promos] @api_router.post("/admin/promos") async def admin_create_promo(request: Request): admin = await get_admin_user(request) body = await request.json() code = body.get("code", "").upper().strip() if not code: raise HTTPException(status_code=400, detail="Code requis") existing = await fetchone("SELECT id FROM promo_codes WHERE code=%s", (code,)) if existing: raise HTTPException(status_code=400, detail="Ce code existe déjà") code_id = f"promo_{uuid.uuid4().hex[:12]}" await execute("INSERT INTO promo_codes (code_id,code,description,discount_type,discount_value,plan_upgrade,start_date,end_date,max_uses,current_uses,active,created_by,created_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,0,%s,%s,%s)", (code_id, code, body.get("description", ""), body.get("discount_type", "free_months"), body.get("discount_value", 1), body.get("plan_upgrade", "duo"), body.get("start_date", ""), body.get("end_date", ""), body.get("max_uses", 0), 1 if body.get("active", True) else 0, admin["user_id"], datetime.now(timezone.utc).isoformat())) return {"code_id": code_id, "code": code, "message": "Code promo créé"} @api_router.put("/admin/promos/{code_id}") async def admin_update_promo(code_id: str, request: Request): await get_admin_user(request) body = await request.json() updates, params = [], [] for field in ["description", "discount_type", "discount_value", "plan_upgrade", "start_date", "end_date", "max_uses"]: if field in body: updates.append(f"{field}=%s"); params.append(body[field]) if "active" in body: updates.append("active=%s"); params.append(1 if body["active"] else 0) if updates: params.append(code_id) await execute(f"UPDATE promo_codes SET {','.join(updates)} WHERE code_id=%s", params) return {"message": "Code promo mis à jour"} @api_router.delete("/admin/promos/{code_id}") async def admin_delete_promo(code_id: str, request: Request): await get_admin_user(request) await execute("DELETE FROM promo_codes WHERE code_id=%s", (code_id,)) return {"message": "Code promo supprimé"} @api_router.get("/admin/promos/{code_id}/redemptions") async def admin_get_promo_redemptions(code_id: str, request: Request): await get_admin_user(request) rows = await fetchall(""" SELECT pr.created_at, u.user_id, u.name, u.email, u.subscription_plan FROM promo_redemptions pr LEFT JOIN users u ON u.user_id = pr.user_id WHERE pr.code_id=%s ORDER BY pr.created_at DESC """, (code_id,)) return rows or [] @api_router.post("/admin/promos/{code_id}/toggle") async def admin_toggle_promo(code_id: str, request: Request): await get_admin_user(request) promo = await fetchone("SELECT active FROM promo_codes WHERE code_id=%s", (code_id,)) if not promo: raise HTTPException(status_code=404, detail="Code introuvable") new_active = 0 if promo["active"] else 1 await execute("UPDATE promo_codes SET active=%s WHERE code_id=%s", (new_active, code_id)) return {"active": bool(new_active)} # User redeem promo code @api_router.post("/promo/redeem") async def redeem_promo(request: Request): user = await get_current_user(request) body = await request.json() code = body.get("code", "").upper().strip() if not code: raise HTTPException(status_code=400, detail="Code requis") promo = await fetchone("SELECT * FROM promo_codes WHERE code=%s AND active=1", (code,)) if not promo: raise HTTPException(status_code=404, detail="Code promo invalide ou inactif") now = datetime.now(timezone.utc).isoformat() if promo["start_date"] and now < promo["start_date"]: raise HTTPException(status_code=400, detail="Ce code promo n'est pas encore actif") if promo["end_date"] and now > promo["end_date"]: raise HTTPException(status_code=400, detail="Ce code promo a expiré") if promo["max_uses"] > 0 and promo["current_uses"] >= promo["max_uses"]: raise HTTPException(status_code=400, detail="Ce code promo a atteint sa limite") existing = await fetchone("SELECT id FROM promo_redemptions WHERE code_id=%s AND user_id=%s", (promo["code_id"], user["user_id"])) if existing: raise HTTPException(status_code=400, detail="Vous avez déjà utilisé ce code") await execute("INSERT INTO promo_redemptions (redemption_id,code_id,promo_code,user_id,created_at) VALUES (%s,%s,%s,%s,%s)", (f"pr_{uuid.uuid4().hex[:12]}", promo["code_id"], code, user["user_id"], now)) await execute("UPDATE promo_codes SET current_uses=current_uses+1 WHERE code_id=%s", (promo["code_id"],)) plan = promo.get("plan_upgrade", "duo") months = promo.get("discount_value", 1) trial_end = (datetime.now(timezone.utc) + timedelta(days=30 * months)).isoformat() await execute("UPDATE users SET subscription_plan=%s, subscription_status='promo', trial_end_date=%s WHERE user_id=%s", (plan, trial_end, user["user_id"])) return {"message": f"Code promo activé ! {months} mois de {plan.capitalize()} offert(s)."} @api_router.delete("/admin/users/{user_id}") async def admin_delete_user(user_id: str, request: Request): admin = await get_admin_user(request) if admin["user_id"] == user_id: raise HTTPException(status_code=400, detail="Cannot delete yourself") lists = await fetchall("SELECT list_id FROM shopping_lists WHERE owner_id=%s", (user_id,)) for l in lists: await execute("DELETE FROM list_items WHERE list_id=%s", (l["list_id"],)) await execute("DELETE FROM list_shares WHERE list_id=%s", (l["list_id"],)) await execute("DELETE FROM shopping_lists WHERE owner_id=%s", (user_id,)) await execute("DELETE FROM user_sessions WHERE user_id=%s", (user_id,)) await execute("DELETE FROM notifications WHERE user_id=%s", (user_id,)) await execute("DELETE FROM user_stores WHERE user_id=%s", (user_id,)) await execute("DELETE FROM users WHERE user_id=%s", (user_id,)) return {"message": "Deleted"} @api_router.get("/admin/transactions") async def admin_transactions(request: Request): await get_admin_user(request) txns = await fetchall("SELECT transaction_id,session_id,user_id,email,plan_id,amount,currency,payment_status,created_at FROM payment_transactions ORDER BY created_at DESC LIMIT 200") for t in txns: if t.get("amount"): t["amount"] = float(t["amount"]) return txns # ===== APP SETTINGS (Admin configurable) ===== async def get_setting(key: str, default: str = "") -> str: row = await fetchone("SELECT setting_value FROM app_settings WHERE setting_key=%s", (key,)) return row["setting_value"] if row else default async def set_setting(key: str, value: str): now_iso = datetime.now(timezone.utc).isoformat() existing = await fetchone("SELECT 1 FROM app_settings WHERE setting_key=%s", (key,)) if existing: await execute("UPDATE app_settings SET setting_value=%s, updated_at=%s WHERE setting_key=%s", (value, now_iso, key)) else: await execute("INSERT INTO app_settings (setting_key, setting_value, updated_at) VALUES (%s, %s, %s)", (key, value, now_iso)) @api_router.get("/settings/public") async def public_settings(): """Settings publics que l'app peut lire sans auth admin.""" trial_enabled = await get_setting("trial_enabled", "1") == "1" trial_days = int(await get_setting("trial_duration_days", "7")) return {"trial_enabled": trial_enabled, "trial_duration_days": trial_days} @api_router.get("/admin/settings") async def admin_get_settings(request: Request): await get_admin_user(request) rows = await fetchall("SELECT setting_key, setting_value, updated_at FROM app_settings") return {r["setting_key"]: r["setting_value"] for r in rows} @api_router.put("/admin/settings") async def admin_update_settings(request: Request): await get_admin_user(request) body = await request.json() allowed = {"trial_enabled", "trial_duration_days"} for k, v in body.items(): if k in allowed: await set_setting(k, str(v)) return {"message": "Settings updated", "settings": body} # ===== GOOGLE PLAY BILLING ===== # Service account JSON file path (mounted in container) GOOGLE_PLAY_CREDENTIALS_PATH = os.environ.get("GOOGLE_PLAY_CREDENTIALS_PATH", "/app/google-play-service-account.json") GOOGLE_PLAY_PACKAGE_NAME = os.environ.get("GOOGLE_PLAY_PACKAGE_NAME", "com.smartpanier.app") # Mapping product_id <-> plan_id GOOGLE_PRODUCT_PLAN_MAP = { "smartpanier_duo_monthly": "duo", "smartpanier_family_monthly": "family", } def _google_play_service(): """Crée un client Google Play Developer API à partir du service account.""" from google.oauth2 import service_account from googleapiclient.discovery import build if not os.path.exists(GOOGLE_PLAY_CREDENTIALS_PATH): raise HTTPException(status_code=500, detail=f"Service account JSON introuvable: {GOOGLE_PLAY_CREDENTIALS_PATH}") creds = service_account.Credentials.from_service_account_file( GOOGLE_PLAY_CREDENTIALS_PATH, scopes=["https://www.googleapis.com/auth/androidpublisher"] ) return build("androidpublisher", "v3", credentials=creds, cache_discovery=False) @api_router.post("/subscriptions/google/verify") async def verify_google_purchase(request: Request): """ Vérifie un achat Google Play et active l'abonnement côté serveur. Body: { product_id: str, purchase_token: str, transaction_id?: str } """ user = await get_current_user(request) body = await request.json() product_id = body.get("product_id") purchase_token = body.get("purchase_token") transaction_id = body.get("transaction_id", "") if not product_id or not purchase_token: raise HTTPException(status_code=400, detail="product_id et purchase_token requis") plan_id = GOOGLE_PRODUCT_PLAN_MAP.get(product_id) if not plan_id: raise HTTPException(status_code=400, detail=f"Produit inconnu: {product_id}") # Vérification via Google Play Developer API try: service = _google_play_service() result = service.purchases().subscriptionsv2().get( packageName=GOOGLE_PLAY_PACKAGE_NAME, token=purchase_token ).execute() except Exception as e: logger.error(f"Google Play verification failed: {e}") # Fallback pour tests: si pas de service account, accepter en mode sandbox if os.environ.get("GOOGLE_PLAY_SANDBOX", "0") == "1": logger.warning("Sandbox mode activé, achat accepté sans vérification") result = { "subscriptionState": "SUBSCRIPTION_STATE_ACTIVE", "lineItems": [{"expiryTime": (datetime.now(timezone.utc) + timedelta(days=31)).isoformat()}], "acknowledgementState": "ACKNOWLEDGEMENT_STATE_PENDING", } else: raise HTTPException(status_code=400, detail=f"Vérification Google échouée: {str(e)[:200]}") state = result.get("subscriptionState", "") if state not in ("SUBSCRIPTION_STATE_ACTIVE", "SUBSCRIPTION_STATE_IN_GRACE_PERIOD", "SUBSCRIPTION_STATE_ON_HOLD"): raise HTTPException(status_code=400, detail=f"Abonnement non actif: {state}") # Extraire la date d'expiration line_items = result.get("lineItems", []) expiry_time = None if line_items: expiry_time = line_items[0].get("expiryTime") if not expiry_time: expiry_time = (datetime.now(timezone.utc) + timedelta(days=31)).isoformat() now_iso = datetime.now(timezone.utc).isoformat() sub_id = f"iap_{uuid.uuid4().hex[:12]}" # Enregistre l'abonnement import json as _json await execute( """INSERT INTO iap_subscriptions (subscription_id, user_id, provider, product_id, purchase_token, transaction_id, plan_id, status, auto_renewing, start_time, expiry_time, raw_payload, created_at, updated_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", (sub_id, user["user_id"], "google", product_id, purchase_token, transaction_id, plan_id, "active", 1, now_iso, expiry_time, _json.dumps(result)[:60000], now_iso, now_iso) ) # Active l'abonnement utilisateur await execute( "UPDATE users SET subscription_plan=%s, subscription_status='active', subscription_end_date=%s, payment_provider='google', iap_subscription_id=%s WHERE user_id=%s", (plan_id, expiry_time, sub_id, user["user_id"]) ) # Enregistre aussi dans payment_transactions pour cohérence admin plan_price = SUBSCRIPTION_PLANS.get(plan_id, {}).get("price", 0.0) await execute( "INSERT INTO payment_transactions (transaction_id,session_id,user_id,email,plan_id,amount,currency,payment_status,created_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", (f"txn_{uuid.uuid4().hex[:12]}", f"google_{transaction_id or purchase_token[:20]}", user["user_id"], user["email"], plan_id, plan_price, "eur", "paid", now_iso) ) # Acknowledge (obligatoire sous 3 jours sinon Google rembourse) try: if result.get("acknowledgementState") == "ACKNOWLEDGEMENT_STATE_PENDING": service.purchases().subscriptions().acknowledge( packageName=GOOGLE_PLAY_PACKAGE_NAME, subscriptionId=product_id, token=purchase_token, body={"developerPayload": f"user:{user['user_id']}"} ).execute() except Exception as e: logger.warning(f"Acknowledgement failed (non fatal): {e}") return { "success": True, "plan_id": plan_id, "expiry_time": expiry_time, "subscription_id": sub_id } @api_router.post("/webhooks/google-play") async def google_play_rtdn(request: Request): """ Real-Time Developer Notifications (RTDN) depuis Google Cloud Pub/Sub. Reçoit les événements: renouvellement, annulation, expiration, remboursement. """ import base64 as _b64 import json as _json body = await request.json() # Pub/Sub enveloppe : { message: { data: base64, messageId, publishTime }, subscription: str } message = body.get("message", {}) data_b64 = message.get("data", "") if not data_b64: return {"status": "no data"} try: decoded = _json.loads(_b64.b64decode(data_b64).decode("utf-8")) except Exception as e: logger.error(f"RTDN decode error: {e}") return {"status": "decode_error"} sub_notif = decoded.get("subscriptionNotification") if not sub_notif: return {"status": "ignored"} purchase_token = sub_notif.get("purchaseToken") notification_type = sub_notif.get("notificationType") # Types: 1=recovered, 2=renewed, 3=cancelled, 4=purchased, 5=on_hold, 6=in_grace_period, # 7=restarted, 8=price_changed, 9=deferred, 10=paused, 12=revoked, 13=expired existing = await fetchone("SELECT * FROM iap_subscriptions WHERE purchase_token=%s", (purchase_token,)) if not existing: logger.warning(f"RTDN: token inconnu {purchase_token[:30]}") return {"status": "unknown_token"} user_id = existing["user_id"] now_iso = datetime.now(timezone.utc).isoformat() if notification_type in (3, 12, 13): # cancelled, revoked, expired await execute("UPDATE iap_subscriptions SET status='cancelled', updated_at=%s WHERE subscription_id=%s", (now_iso, existing["subscription_id"])) if notification_type in (12, 13): # Rétrograde l'utilisateur en solo uniquement si expiré/revoked await execute("UPDATE users SET subscription_plan='solo', subscription_status='active', payment_provider='none', iap_subscription_id=NULL WHERE user_id=%s AND iap_subscription_id=%s", (user_id, existing["subscription_id"])) elif notification_type in (2, 1, 7, 4): # renewed, recovered, restarted, purchased # Refresh expiry depuis Google try: service = _google_play_service() result = service.purchases().subscriptionsv2().get( packageName=GOOGLE_PLAY_PACKAGE_NAME, token=purchase_token ).execute() line_items = result.get("lineItems", []) new_expiry = line_items[0].get("expiryTime") if line_items else None if new_expiry: await execute("UPDATE iap_subscriptions SET status='active', expiry_time=%s, updated_at=%s WHERE subscription_id=%s", (new_expiry, now_iso, existing["subscription_id"])) await execute("UPDATE users SET subscription_status='active', subscription_end_date=%s WHERE user_id=%s", (new_expiry, user_id)) except Exception as e: logger.error(f"RTDN refresh failed: {e}") return {"status": "ok", "type": notification_type} @api_router.get("/subscriptions/my") async def get_my_subscription(request: Request): """Retourne l'abonnement actif de l'utilisateur avec son provider.""" user = await get_current_user(request) provider = user.get("payment_provider", "none") iap_sub = None if provider in ("google", "apple") and user.get("iap_subscription_id"): iap_sub = await fetchone("SELECT subscription_id, provider, product_id, plan_id, status, expiry_time, auto_renewing FROM iap_subscriptions WHERE subscription_id=%s", (user["iap_subscription_id"],)) return { "plan": user.get("subscription_plan", "solo"), "status": user.get("subscription_status", "active"), "provider": provider, "end_date": user.get("subscription_end_date"), "iap": iap_sub, } # ===== WEBSOCKET TEMPS RÉEL ===== import asyncio from fastapi import WebSocket, WebSocketDisconnect class ConnectionManager: """Gère les connexions WebSocket par list_id pour broadcaster les updates.""" def __init__(self): # list_id -> set of WebSocket self.connections: Dict[str, set] = {} self._lock = asyncio.Lock() async def connect(self, ws: WebSocket, list_id: str): await ws.accept() async with self._lock: self.connections.setdefault(list_id, set()).add(ws) async def disconnect(self, ws: WebSocket, list_id: str): async with self._lock: if list_id in self.connections: self.connections[list_id].discard(ws) if not self.connections[list_id]: del self.connections[list_id] async def broadcast(self, list_id: str, message: dict, exclude_ws: WebSocket = None): """Diffuse un message à toutes les WS connectées à cette liste.""" if list_id not in self.connections: return dead = [] for ws in list(self.connections.get(list_id, set())): if ws is exclude_ws: continue try: await ws.send_json(message) except Exception: dead.append(ws) # Cleanup connexions mortes if dead: async with self._lock: for ws in dead: self.connections.get(list_id, set()).discard(ws) ws_manager = ConnectionManager() async def notify_list_update(list_id: str, event: str = "update", data: dict = None, actor_user_id: str = None): """Helper appelé après chaque modification d'une liste pour notifier les clients.""" payload = {"event": event, "list_id": list_id, "ts": datetime.now(timezone.utc).isoformat()} if data: payload["data"] = data if actor_user_id: payload["actor_user_id"] = actor_user_id try: await ws_manager.broadcast(list_id, payload) except Exception as e: logger.warning(f"WS broadcast failed: {e}") @api_router.websocket("/ws/lists/{list_id}") async def websocket_list(websocket: WebSocket, list_id: str, token: str = ""): """ WebSocket pour recevoir les updates en temps réel sur une liste. URL: ws://server/api/ws/lists/{list_id}?token=JWT_TOKEN """ if not token: await websocket.close(code=4401, reason="No token") return try: payload = jwt.decode(token, get_jwt_secret(), algorithms=[JWT_ALGORITHM]) user_id = payload.get("sub") if not user_id: await websocket.close(code=4401, reason="Invalid token") return except Exception: await websocket.close(code=4401, reason="Token decode error") return # Vérifier que l'utilisateur a accès à cette liste user = await fetchone("SELECT * FROM users WHERE user_id=%s", (user_id,)) if not user: await websocket.close(code=4401, reason="User not found") return lst = await fetchone("SELECT * FROM shopping_lists WHERE list_id=%s", (list_id,)) if not lst: await websocket.close(code=4404, reason="List not found") return is_shared = await fetchone("SELECT id FROM list_shares WHERE list_id=%s AND shared_email=%s", (list_id, user["email"])) if lst["owner_id"] != user["user_id"] and not is_shared: await websocket.close(code=4403, reason="No access") return await ws_manager.connect(websocket, list_id) try: # Confirme la connexion await websocket.send_json({"event": "connected", "list_id": list_id}) # Garde la connexion ouverte (ping/pong basique) while True: msg = await websocket.receive_text() if msg == "ping": await websocket.send_text("pong") except WebSocketDisconnect: pass except Exception as e: logger.warning(f"WS error: {e}") finally: await ws_manager.disconnect(websocket, list_id) # Include router app.include_router(api_router) app.add_middleware(CORSMiddleware, allow_origin_regex=".*", allow_credentials=True, allow_methods=["*"], allow_headers=["*"]) @app.on_event("startup") async def startup(): await get_db() # Ensure required tables exist (idempotent migrations) migrations = [ # ============== TABLES PRINCIPALES (au cas où la DB est neuve) ============== """CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(50) UNIQUE NOT NULL, email VARCHAR(255) UNIQUE NOT NULL, name VARCHAR(255), picture TEXT, password_hash VARCHAR(255), role VARCHAR(20) DEFAULT 'user', subscription_plan VARCHAR(20) DEFAULT 'solo', subscription_status VARCHAR(30) DEFAULT 'inactive', subscription_end_date VARCHAR(50), payment_provider VARCHAR(20) DEFAULT 'none', iap_subscription_id VARCHAR(50), stripe_customer_id VARCHAR(255), push_token TEXT, created_at VARCHAR(50), INDEX idx_email (email), INDEX idx_user_id (user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS shopping_lists ( id INT AUTO_INCREMENT PRIMARY KEY, list_id VARCHAR(50) UNIQUE NOT NULL, name VARCHAR(255) NOT NULL, owner_id VARCHAR(50) NOT NULL, owner_email VARCHAR(255), status VARCHAR(20) DEFAULT 'active', position INT DEFAULT 0, created_at VARCHAR(50), updated_at VARCHAR(50), completed_at VARCHAR(50), INDEX idx_owner (owner_id), INDEX idx_list_id (list_id), INDEX idx_status (status) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS list_items ( id INT AUTO_INCREMENT PRIMARY KEY, item_id VARCHAR(50) UNIQUE NOT NULL, list_id VARCHAR(50) NOT NULL, name VARCHAR(255) NOT NULL, quantity VARCHAR(50) DEFAULT '1', category VARCHAR(50) DEFAULT 'other', checked TINYINT(1) DEFAULT 0, position INT DEFAULT 0, added_by VARCHAR(50), created_at VARCHAR(50), INDEX idx_list (list_id), INDEX idx_item_id (item_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS list_shares ( id INT AUTO_INCREMENT PRIMARY KEY, list_id VARCHAR(50) NOT NULL, shared_email VARCHAR(255) NOT NULL, INDEX idx_list (list_id), INDEX idx_email (shared_email), UNIQUE KEY uniq_share (list_id, shared_email) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS notifications ( id INT AUTO_INCREMENT PRIMARY KEY, notif_id VARCHAR(50) UNIQUE NOT NULL, user_id VARCHAR(50) NOT NULL, type VARCHAR(50), message TEXT, list_id VARCHAR(50), read_status TINYINT(1) DEFAULT 0, created_at VARCHAR(50), INDEX idx_user (user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS notification_settings ( id INT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(50) UNIQUE NOT NULL, enabled TINYINT(1) DEFAULT 1, list_updates TINYINT(1) DEFAULT 1, INDEX idx_user (user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS push_tokens ( id INT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(50) NOT NULL, push_token TEXT NOT NULL, created_at VARCHAR(50), INDEX idx_user (user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS user_sessions ( id INT AUTO_INCREMENT PRIMARY KEY, session_token VARCHAR(255) UNIQUE NOT NULL, user_id VARCHAR(50) NOT NULL, expires_at DATETIME, created_at VARCHAR(50), INDEX idx_user (user_id), INDEX idx_token (session_token) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS login_attempts ( id INT AUTO_INCREMENT PRIMARY KEY, identifier VARCHAR(255) NOT NULL, count INT DEFAULT 0, locked_until DATETIME NULL, created_at VARCHAR(50), INDEX idx_identifier (identifier) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS payment_transactions ( id INT AUTO_INCREMENT PRIMARY KEY, transaction_id VARCHAR(100) UNIQUE NOT NULL, session_id VARCHAR(255), user_id VARCHAR(50), user_email VARCHAR(255), plan_id VARCHAR(20), amount DECIMAL(10,2), currency VARCHAR(10) DEFAULT 'eur', payment_status VARCHAR(30) DEFAULT 'pending', metadata TEXT, created_at VARCHAR(50), updated_at VARCHAR(50), INDEX idx_user (user_id), INDEX idx_session (session_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS user_stores ( id INT AUTO_INCREMENT PRIMARY KEY, store_id VARCHAR(50) UNIQUE NOT NULL, user_id VARCHAR(50) NOT NULL, name VARCHAR(255), address TEXT, lat DOUBLE, lon DOUBLE, category VARCHAR(50), created_at VARCHAR(50), INDEX idx_user (user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", # ============== TABLES SECONDAIRES ============== """CREATE TABLE IF NOT EXISTS password_reset_codes ( id INT AUTO_INCREMENT PRIMARY KEY, email VARCHAR(255) NOT NULL, code VARCHAR(10) NOT NULL, expires_at DATETIME NOT NULL, used TINYINT(1) DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_email_code (email, code) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS promo_codes ( id INT AUTO_INCREMENT PRIMARY KEY, code_id VARCHAR(50) UNIQUE NOT NULL, code VARCHAR(50) UNIQUE NOT NULL, description TEXT, discount_type VARCHAR(30) DEFAULT 'free_months', discount_value INT DEFAULT 1, plan_upgrade VARCHAR(20) DEFAULT 'duo', start_date VARCHAR(50), end_date VARCHAR(50), max_uses INT DEFAULT 0, current_uses INT DEFAULT 0, active TINYINT(1) DEFAULT 1, created_by VARCHAR(50), created_at VARCHAR(50) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS app_stats ( stat_key VARCHAR(50) PRIMARY KEY, stat_value INT DEFAULT 0 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS referrals ( id INT AUTO_INCREMENT PRIMARY KEY, referral_id VARCHAR(50) UNIQUE, owner_id VARCHAR(50) NOT NULL, owner_email VARCHAR(255), code VARCHAR(50) UNIQUE NOT NULL, uses INT DEFAULT 0, rewarded_months INT DEFAULT 0, created_at VARCHAR(50) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS referral_redemptions ( id INT AUTO_INCREMENT PRIMARY KEY, redemption_id VARCHAR(50) UNIQUE, referral_code VARCHAR(50), referral_owner_id VARCHAR(50), redeemed_by VARCHAR(50), created_at VARCHAR(50), INDEX idx_owner (referral_owner_id), INDEX idx_redeemer (redeemed_by) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """CREATE TABLE IF NOT EXISTS promo_redemptions ( id INT AUTO_INCREMENT PRIMARY KEY, redemption_id VARCHAR(50) UNIQUE, code_id VARCHAR(50), promo_code VARCHAR(50), user_id VARCHAR(50), created_at VARCHAR(50), INDEX idx_code (code_id), INDEX idx_user (user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """INSERT IGNORE INTO app_stats (stat_key, stat_value) VALUES ('total_lists_created', 0)""", """ALTER TABLE users ADD COLUMN subscription_end_date VARCHAR(50) NULL""", """CREATE TABLE IF NOT EXISTS user_groups ( id INT AUTO_INCREMENT PRIMARY KEY, group_id VARCHAR(50) UNIQUE NOT NULL, owner_id VARCHAR(50) NOT NULL, plan_type VARCHAR(20) NOT NULL, created_at VARCHAR(50), INDEX idx_owner (owner_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci""", """CREATE TABLE IF NOT EXISTS group_members ( id INT AUTO_INCREMENT PRIMARY KEY, group_id VARCHAR(50) NOT NULL, email VARCHAR(255) NOT NULL, user_id VARCHAR(50), added_at VARCHAR(50), UNIQUE KEY unique_member (group_id, email), INDEX idx_email (email), INDEX idx_group (group_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci""", """ALTER TABLE group_members CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE user_groups CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", # === In-App Purchases (Google Play / Apple) === """CREATE TABLE IF NOT EXISTS app_settings ( setting_key VARCHAR(50) PRIMARY KEY, setting_value VARCHAR(255) NOT NULL, updated_at VARCHAR(50) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """INSERT IGNORE INTO app_settings (setting_key, setting_value, updated_at) VALUES ('trial_enabled', '1', '2025-01-01T00:00:00+00:00')""", """INSERT IGNORE INTO app_settings (setting_key, setting_value, updated_at) VALUES ('trial_duration_days', '7', '2025-01-01T00:00:00+00:00')""", """CREATE TABLE IF NOT EXISTS iap_subscriptions ( id INT AUTO_INCREMENT PRIMARY KEY, subscription_id VARCHAR(50) UNIQUE NOT NULL, user_id VARCHAR(50) NOT NULL, provider VARCHAR(20) NOT NULL, product_id VARCHAR(100) NOT NULL, purchase_token TEXT, transaction_id VARCHAR(255), plan_id VARCHAR(20) NOT NULL, status VARCHAR(30) DEFAULT 'active', auto_renewing TINYINT(1) DEFAULT 1, start_time VARCHAR(50), expiry_time VARCHAR(50), raw_payload MEDIUMTEXT, created_at VARCHAR(50), updated_at VARCHAR(50), INDEX idx_user (user_id), INDEX idx_provider (provider), INDEX idx_token (purchase_token(255)) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", """ALTER TABLE users ADD COLUMN payment_provider VARCHAR(20) DEFAULT 'none'""", """ALTER TABLE users ADD COLUMN iap_subscription_id VARCHAR(50) NULL""", # Position pour drag & drop """ALTER TABLE shopping_lists ADD COLUMN position INT DEFAULT 0""", """ALTER TABLE list_items ADD COLUMN position INT DEFAULT 0""", # Fix login_attempts si la table a une mauvaise structure (déploiement précédent) """ALTER TABLE login_attempts ADD COLUMN identifier VARCHAR(255)""", """ALTER TABLE login_attempts ADD COLUMN count INT DEFAULT 0""", """ALTER TABLE login_attempts ADD COLUMN locked_until DATETIME NULL""", """ALTER TABLE login_attempts ADD INDEX idx_identifier (identifier)""", # Rendre nullable les anciennes colonnes pour éviter de bloquer les INSERT """ALTER TABLE login_attempts MODIFY email VARCHAR(255) NULL""", """ALTER TABLE login_attempts MODIFY attempted_at DATETIME NULL""", # Standardiser les collations pour éviter "Illegal mix of collations" lors des JOINs """ALTER TABLE users CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE shopping_lists CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE list_items CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE list_shares CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE notifications CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE notification_settings CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE push_tokens CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE user_sessions CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE login_attempts CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE payment_transactions CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", """ALTER TABLE user_stores CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""", ] for sql in migrations: try: await execute(sql) except Exception as e: logger.warning(f"Migration skipped: {e}") # Sync app_stats with existing data (only if count is 0) try: current = await fetchone("SELECT stat_value FROM app_stats WHERE stat_key='total_lists_created'") if current and current["stat_value"] == 0: row = await fetchone("SELECT COUNT(*) AS c FROM shopping_lists") if row and row["c"] > 0: await execute("UPDATE app_stats SET stat_value=%s WHERE stat_key='total_lists_created'", (row["c"],)) except Exception as e: logger.warning(f"Stats sync skipped: {e}") admin_email = os.environ.get("ADMIN_EMAIL", "admin@smartpanier.com") admin_password = os.environ.get("ADMIN_PASSWORD", "Admin123!") existing = await fetchone("SELECT * FROM users WHERE email=%s", (admin_email,)) if not existing: admin_id = f"user_{uuid.uuid4().hex[:12]}" await execute("INSERT INTO users (user_id,email,name,password_hash,role,subscription_plan,subscription_status,created_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", (admin_id, admin_email, "Admin", hash_password(admin_password), "admin", "family", "active", datetime.now(timezone.utc).isoformat())) logger.info(f"Admin seeded: {admin_email}") @app.on_event("shutdown") async def shutdown(): global pool if pool: pool.close() await pool.wait_closed()