Spaces:
Running
Running
| """ | |
| Shared checks for the public, no-sign-in forms (/bsa and /bff): a simple rate | |
| limiter and the contact-detail validators. Rate limits are in-memory, per | |
| process (same trade-off as routes/help.py). | |
| """ | |
| import re | |
| import threading | |
| import time | |
| from collections import defaultdict | |
| from fastapi import HTTPException | |
| EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") | |
| # Digits with an optional leading +, allowing the spaces, dashes, dots and brackets people type. | |
| PHONE_RE = re.compile(r"^\+?[\d\s().-]+$") | |
| class RateLimiter: | |
| """Allows `count` hits per `period` seconds for each key.""" | |
| def __init__(self): | |
| self._lock = threading.Lock() | |
| self._hits: dict[str, list[float]] = defaultdict(list) | |
| def allow(self, key: str, count: int, period: int) -> bool: | |
| now = time.time() | |
| with self._lock: | |
| recent = [t for t in self._hits[key] if now - t < period] | |
| allowed = len(recent) < count | |
| if allowed: | |
| recent.append(now) | |
| self._hits[key] = recent | |
| return allowed | |
| def reset(self): | |
| with self._lock: | |
| self._hits.clear() | |
| def clean_email(raw: str) -> str: | |
| email = raw.strip().lower() | |
| if not EMAIL_RE.match(email): | |
| raise HTTPException(status_code=400, detail="Enter a valid email address.") | |
| return email | |
| def clean_phone(raw: str) -> str: | |
| """Trimmed phone number, kept as typed apart from spacing. 7 to 15 digits.""" | |
| phone = " ".join(raw.split()) | |
| digits = re.sub(r"\D", "", phone) | |
| if not PHONE_RE.match(phone) or not 7 <= len(digits) <= 15: | |
| raise HTTPException(status_code=400, detail="Enter a valid phone number.") | |
| return phone | |