#!/usr/bin/env python3 """ Trigger Auth Emails — Diagnostic Script ========================================= Szimulálja az éles API végpontok működését a teljes Auth logikán keresztül, hogy kiderüljön, hol akad el a folyamat. Használat: docker compose exec sf_api python3 /app/trigger_auth_emails.py Környezeti változók: - TEST_EMAIL: Cél e-mail cím (alapértelmezett: accipe.t@aplast.hu) - EMAIL_PROVIDER: smtp | brevo_api | brevo_smtp """ import os import sys import asyncio import logging # Ensure the backend package is importable sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # ── Load .env ── from dotenv import load_dotenv env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.env') load_dotenv(env_path, override=False) # ── Logging setup ── logging.basicConfig( level=logging.DEBUG, format='%(asctime)s [%(name)s] %(levelname)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S', ) logger = logging.getLogger("TriggerAuthEmails") # Increase log level for noisy libs logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) async def main(): print("=" * 70) print(" SERVICE FINDER — TRIGGER AUTH EMAILS (DIAGNOSTIC)") print("=" * 70) # ── 1. Target email ── target_email = os.getenv("TEST_EMAIL", "accipe.t@aplast.hu") print(f"\n🎯 Target email: {target_email}") # ── 2. Show email config ── provider = os.getenv("EMAIL_PROVIDER", "smtp") print(f"\n📧 Current EMAIL_PROVIDER: {provider}") print(f" SMTP_HOST: {os.getenv('SMTP_HOST', '(not set)')}") print(f" SMTP_PORT: {os.getenv('SMTP_PORT', '(not set)')}") print(f" MAIL_FROM: {os.getenv('MAIL_FROM', '(not set)')}") print(f" MAIL_FROM_NAME: {os.getenv('MAIL_FROM_NAME', '(not set)')}") print(f" BREVO_API_KEY: {'✅ Set' if os.getenv('BREVO_API_KEY') else '❌ NOT SET'}") print(f" BREVO_SMTP_USER: {'✅ Set' if os.getenv('BREVO_SMTP_USER') else '❌ NOT SET'}") print(f" BREVO_SMTP_PASSWORD: {'✅ Set' if os.getenv('BREVO_SMTP_PASSWORD') else '❌ NOT SET'}") # ── 3. Import and execute ── try: from app.db.session import AsyncSessionLocal from app.services.auth_service import AuthService from sqlalchemy import text # ── 3a. Check if user exists ── print(f"\n🔍 Checking user: {target_email}...") async with AsyncSessionLocal() as db: result = await db.execute( text("SELECT id, email, is_active, preferred_language, region_code FROM identity.users WHERE email = :email AND is_deleted = false"), {"email": target_email} ) user = result.fetchone() if not user: print(f"❌ User not found with email: {target_email}") print(" Check if the user exists in the database.") return user_id, email, is_active, lang, region = user print(f" ✅ User found:") print(f" ID: {user_id}") print(f" Email: {email}") print(f" Active: {is_active}") print(f" Lang: {lang}") print(f" Region: {region}") # ── 3b. Check existing verification tokens ── result = await db.execute( text(""" SELECT id, token, token_type, is_used, expires_at, created_at FROM identity.verification_tokens WHERE user_id = :uid ORDER BY created_at DESC LIMIT 5 """), {"uid": user_id} ) tokens = result.fetchall() print(f"\n🔑 Existing verification tokens for user #{user_id}:") if tokens: for t in tokens: print(f" ID={t[0]} type={t[2]} used={t[3]} expires={t[4]} created={t[5]}") else: print(" (none)") # ── 3c. Test 1: Resend Verification ── print(f"\n{'=' * 70}") print(f" 📬 TEST 1: Resend Verification Email") print(f"{'=' * 70}") print(f" Calling AuthService.resend_verification(email='{target_email}')...\n") async with AsyncSessionLocal() as db: try: result = await AuthService.resend_verification(db=db, email=target_email) print(f"\n✅ Result: {result}") except Exception as e: print(f"\n❌ Exception: {type(e).__name__}: {e}") import traceback traceback.print_exc() # ── 3d. Test 2: Initiate Password Reset ── print(f"\n{'=' * 70}") print(f" 📬 TEST 2: Initiate Password Reset Email") print(f"{'=' * 70}") print(f" Calling AuthService.initiate_password_reset(email='{target_email}')...\n") async with AsyncSessionLocal() as db: try: result = await AuthService.initiate_password_reset(db=db, email=target_email) print(f"\n✅ Result: {result}") except Exception as e: print(f"\n❌ Exception: {type(e).__name__}: {e}") import traceback traceback.print_exc() # ── 3e. Test 3: Direct EmailManager call ── print(f"\n{'=' * 70}") print(f" 📬 TEST 3: Direct EmailManager.send_email (template_key='reg')") print(f"{'=' * 70}") print(f" Calling EmailManager.send_email() directly...\n") from app.services.email_manager import EmailManager async with AsyncSessionLocal() as db: try: result = await EmailManager.send_email( recipient=target_email, template_key="reg", variables={ "first_name": "Teszt", "link": "https://dev.servicefinder.hu/verify?token=test-token-123", }, lang="hu", db=db, ) print(f"\n✅ Result: {result}") except Exception as e: print(f"\n❌ Exception: {type(e).__name__}: {e}") import traceback traceback.print_exc() # ── 3f. Test 4: Direct EmailManager call with pwd_reset template ── print(f"\n{'=' * 70}") print(f" 📬 TEST 4: Direct EmailManager.send_email (template_key='pwd_reset')") print(f"{'=' * 70}") print(f" Calling EmailManager.send_email() directly...\n") async with AsyncSessionLocal() as db: try: result = await EmailManager.send_email( recipient=target_email, template_key="pwd_reset", variables={ "link": "https://dev.servicefinder.hu/reset-password?token=test-token-456", }, lang="hu", db=db, ) print(f"\n✅ Result: {result}") except Exception as e: print(f"\n❌ Exception: {type(e).__name__}: {e}") import traceback traceback.print_exc() except ImportError as e: print(f"\n❌ Import Error: {e}") print(" Make sure you run this inside the sf_api container!") print(" Command: docker compose exec sf_api python3 /app/trigger_auth_emails.py") sys.exit(1) print(f"\n{'=' * 70}") print(" ✅ DIAGNOSTIC COMPLETE") print(f"{'=' * 70}") if __name__ == "__main__": asyncio.run(main())