Spaces:
Running
Running
| """EmoSphere Trial Authentication — Email-gated demo with time limits. | |
| Handles: | |
| - Email registration with 6-digit code | |
| - One-use-per-email enforcement (stored in JSON) | |
| - Session timer (60s trial window) | |
| - SMTP email delivery (or on-screen fallback for testing) | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import random | |
| import smtplib | |
| import time | |
| from email.mime.text import MIMEText | |
| from email.mime.multipart import MIMEMultipart | |
| from pathlib import Path | |
| from typing import Optional | |
| # ── Configuration ─────────────────────────────────────────────────── | |
| SMTP_HOST = os.environ.get("SMTP_HOST", "smtp-relay.brevo.com") | |
| SMTP_PORT = int(os.environ.get("SMTP_PORT", "587")) | |
| SMTP_USER = os.environ.get("SMTP_USER", "a7fb01001@smtp-brevo.com") | |
| SMTP_PASS = os.environ.get("SMTP_PASS", "xsmtpsib-95d4419e874210d694181e185f1a19f0a13d14f09176441701adfbcc2ceab161-gfhZ4hNblHSUa6CM") | |
| SMTP_FROM = os.environ.get("SMTP_FROM", "info@caitcore.com") | |
| BREVO_API_KEY = os.environ.get("BREVO_API_KEY", "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") | |
| TRIAL_DURATION_SECONDS = 60 | |
| VIDEO_MAX_DURATION_SECONDS = 60 | |
| CAMERA_MAX_DURATION_SECONDS = 60 | |
| USED_EMAILS_FILE = Path(__file__).parent / "used_emails.json" | |
| def smtp_configured() -> bool: | |
| """Check whether SMTP credentials are set.""" | |
| print(f"[Auth] SMTP check: HOST={SMTP_HOST}, USER={SMTP_USER[:10]}..., PASS={'set' if SMTP_PASS else 'empty'}, FROM={SMTP_FROM}") | |
| return bool(SMTP_HOST and SMTP_USER and SMTP_PASS) | |
| # ── Email Storage ─────────────────────────────────────────────────── | |
| def _load_used_emails() -> dict: | |
| """Load used emails registry from JSON file.""" | |
| if USED_EMAILS_FILE.exists(): | |
| try: | |
| data = json.loads(USED_EMAILS_FILE.read_text(encoding="utf-8")) | |
| if isinstance(data, dict): | |
| return data | |
| except (json.JSONDecodeError, OSError): | |
| pass | |
| return {} | |
| def _save_used_emails(data: dict) -> None: | |
| """Persist used emails registry to JSON file.""" | |
| try: | |
| USED_EMAILS_FILE.write_text( | |
| json.dumps(data, indent=2, default=str), | |
| encoding="utf-8", | |
| ) | |
| except OSError as e: | |
| print(f"[Auth] Warning: could not save used emails: {e}") | |
| def is_email_used(email: str) -> bool: | |
| """Check if an email has already been used for a trial.""" | |
| emails = _load_used_emails() | |
| return email.lower().strip() in emails | |
| def mark_email_used(email: str) -> None: | |
| """Record an email as having used its trial.""" | |
| emails = _load_used_emails() | |
| emails[email.lower().strip()] = { | |
| "used_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), | |
| } | |
| _save_used_emails(emails) | |
| # ── Code Generation & Sending ────────────────────────────────────── | |
| def generate_code() -> str: | |
| """Generate a random 6-digit verification code.""" | |
| return f"{random.randint(100000, 999999)}" | |
| def _build_html_email(code: str) -> str: | |
| """Build branded HTML email for EmoSphere.""" | |
| return f""" | |
| <html><body style="font-family: -apple-system, sans-serif; background: #0A0A1A; color: #B0BCD0; padding: 40px;"> | |
| <div style="max-width: 520px; margin: 0 auto; background: linear-gradient(135deg, #161640, #1E1E52); | |
| border: 1px solid rgba(255,255,255,0.1); border-radius: 16px; padding: 40px;"> | |
| <div style="text-align: center; margin-bottom: 24px;"> | |
| <img src="https://caitcore.com/images/emosphere-logo.png" alt="EmoSphere" style="width: 80px; height: 80px; border-radius: 16px;" /> | |
| </div> | |
| <h1 style="background: linear-gradient(90deg, #E948A0, #9B6FCE, #00D4FF); | |
| -webkit-background-clip: text; -webkit-text-fill-color: transparent; | |
| font-size: 26px; text-align: center; margin: 0 0 8px 0;"> | |
| EmoSphere | |
| </h1> | |
| <p style="text-align: center; color: #6B7B9D; font-size: 13px; margin: 0 0 24px 0;">Multimodal Emotion Recognition</p> | |
| <p style="text-align: center; color: #8899AA; font-size: 14px;">Your demo access code:</p> | |
| <div style="text-align: center; font-size: 36px; font-weight: 800; letter-spacing: 8px; | |
| color: #00D4FF; padding: 20px; background: rgba(0,212,255,0.08); | |
| border-radius: 12px; margin: 16px 0;"> | |
| {code} | |
| </div> | |
| <p style="text-align: center; color: #6B7B9D; font-size: 12px; margin-top: 24px;"> | |
| One-time code • Session lasts {TRIAL_DURATION_SECONDS}s • No data stored | |
| </p> | |
| <hr style="border: none; border-top: 1px solid rgba(255,255,255,0.08); margin: 28px 0 16px;" /> | |
| <div style="text-align: center;"> | |
| <img src="https://caitcore.com/images/cait-logo.png" alt="CAIT" style="width: 32px; height: 32px; vertical-align: middle; border-radius: 6px;" /> | |
| <span style="color: #556677; font-size: 11px; vertical-align: middle; margin-left: 8px;">Powered by CAIT — Computational & AI Technologies</span> | |
| </div> | |
| </div> | |
| </body></html> | |
| """ | |
| def _send_via_http(to_email: str, code: str) -> tuple[bool, str]: | |
| """Send email via Brevo HTTP API (works even if SMTP port is blocked).""" | |
| import urllib.request | |
| import urllib.error | |
| html_body = _build_html_email(code) | |
| payload = json.dumps({ | |
| "sender": {"name": "CAIT", "email": SMTP_FROM}, | |
| "to": [{"email": to_email}], | |
| "subject": f"EmoSphere Demo Access Code: {code}", | |
| "htmlContent": html_body, | |
| }) | |
| req = urllib.request.Request( | |
| "https://api.brevo.com/v3/smtp/email", | |
| data=payload.encode("utf-8"), | |
| headers={ | |
| "accept": "application/json", | |
| "content-type": "application/json", | |
| "api-key": BREVO_API_KEY, | |
| }, | |
| method="POST", | |
| ) | |
| try: | |
| with urllib.request.urlopen(req, timeout=10) as resp: | |
| print(f"[Auth] HTTP API response: {resp.status} {resp.read().decode()}") | |
| return True, "Code sent to your email" | |
| except urllib.error.HTTPError as e: | |
| body = e.read().decode() if e.fp else "" | |
| print(f"[Auth] HTTP API error: {e.code} {body}") | |
| return False, f"HTTP API error: {e.code}" | |
| except Exception as e: | |
| print(f"[Auth] HTTP API exception: {e}") | |
| return False, str(e) | |
| def _send_via_smtp(to_email: str, code: str) -> tuple[bool, str]: | |
| """Send email via SMTP (fallback).""" | |
| html_body = _build_html_email(code) | |
| text_body = ( | |
| f"Your EmoSphere demo access code is: {code}\n\n" | |
| f"This code is valid for one use only.\n" | |
| f"The demo session lasts {TRIAL_DURATION_SECONDS} seconds.\n\n" | |
| f"-- EmoSphere by CAIT" | |
| ) | |
| msg = MIMEMultipart("alternative") | |
| msg["Subject"] = f"EmoSphere Demo Access Code: {code}" | |
| msg["From"] = SMTP_FROM | |
| msg["To"] = to_email | |
| msg.attach(MIMEText(text_body, "plain")) | |
| msg.attach(MIMEText(html_body, "html")) | |
| with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=10) as server: | |
| server.ehlo() | |
| server.starttls() | |
| server.ehlo() | |
| server.login(SMTP_USER, SMTP_PASS) | |
| server.sendmail(SMTP_FROM, to_email, msg.as_string()) | |
| return True, "Code sent to your email" | |
| def send_code_email(to_email: str, code: str) -> tuple[bool, str]: | |
| """Send verification code — tries HTTP API first, falls back to SMTP. | |
| Returns (success, message). | |
| """ | |
| if not smtp_configured(): | |
| return False, "SMTP not configured" | |
| # Try HTTP API first (works on HF Spaces where SMTP port may be blocked) | |
| try: | |
| ok, msg = _send_via_http(to_email, code) | |
| if ok: | |
| return ok, msg | |
| print(f"[Auth] HTTP API failed ({msg}), trying SMTP...") | |
| except Exception as e: | |
| print(f"[Auth] HTTP API exception: {e}, trying SMTP...") | |
| # Fallback to SMTP | |
| try: | |
| return _send_via_smtp(to_email, code) | |
| except Exception as e: | |
| print(f"[Auth] SMTP send FAILED: {e}") | |
| return False, f"Failed to send email: {e}" | |
| # ── Session Timer ─────────────────────────────────────────────────── | |
| def get_remaining_seconds(start_time: float) -> float: | |
| """Return remaining trial seconds (0 if expired).""" | |
| elapsed = time.time() - start_time | |
| remaining = TRIAL_DURATION_SECONDS - elapsed | |
| return max(0.0, remaining) | |
| def is_trial_expired(start_time: float) -> bool: | |
| """Check if the trial window has ended.""" | |
| return get_remaining_seconds(start_time) <= 0 | |
| # ── Email Validation (basic) ─────────────────────────────────────── | |
| def validate_email(email: str) -> bool: | |
| """Basic email format check.""" | |
| import re | |
| pattern = r"^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$" | |
| return bool(re.match(pattern, email.strip())) | |