Spaces:
Running
Running
| """ | |
| Business Strength Archetype results: the public assessment at /bsa. | |
| No sign-in. The browser sends the person's details and the strength behind each | |
| answer; this route re-scores the answers itself (bsa/scoring.py), then stores one | |
| document per submission in bsa_results. firestore.rules' default deny keeps the | |
| collection backend-only. | |
| """ | |
| import time | |
| from fastapi import APIRouter, HTTPException | |
| from firebase_admin import firestore | |
| from pydantic import BaseModel, Field | |
| from bsa import scoring | |
| from utils.public_forms import RateLimiter, clean_email, clean_phone | |
| router = APIRouter() | |
| COLLECTION = "bsa_results" | |
| PER_SESSION_LIMIT = (5, 3600) # 5 submissions per hour per browser | |
| GLOBAL_LIMIT = (500, 3600) # 500 per hour across everyone | |
| limiter = RateLimiter() | |
| class Answer(BaseModel): | |
| q: int | |
| dim: str = Field(min_length=2, max_length=2) | |
| class ResultIn(BaseModel): | |
| firstName: str = Field(min_length=1, max_length=60) | |
| email: str = Field(min_length=3, max_length=254) | |
| phone: str = Field(min_length=1, max_length=30) | |
| ageRange: str = Field(max_length=40) | |
| stage: str = Field(max_length=120) | |
| field: str = Field(default="", max_length=120) | |
| location: str = Field(default="", max_length=120) | |
| consent: bool | |
| answers: list[Answer] = Field(min_length=1, max_length=100) | |
| startedAt: int = Field(ge=0) # ms since epoch, from the browser | |
| sessionId: str = Field(min_length=1, max_length=64) | |
| def submit_result(body: ResultIn): | |
| if not limiter.allow(f"s:{body.sessionId}", *PER_SESSION_LIMIT) or not limiter.allow("global", *GLOBAL_LIMIT): | |
| raise HTTPException(status_code=429, detail="Too many submissions. Try again later.") | |
| if not body.consent: | |
| raise HTTPException(status_code=400, detail="Consent is required to save your result.") | |
| email = clean_email(body.email) | |
| phone = clean_phone(body.phone) | |
| if body.ageRange not in scoring.AGES or body.stage not in scoring.STAGES: | |
| raise HTTPException(status_code=400, detail="Choose your age range and business stage.") | |
| answers = {a.q: a.dim for a in body.answers} | |
| if len(answers) != len(body.answers): | |
| raise HTTPException(status_code=400, detail="Each question can only be answered once.") | |
| try: | |
| result = scoring.score(answers) | |
| except ValueError as e: | |
| raise HTTPException(status_code=400, detail=str(e)) | |
| now_ms = int(time.time() * 1000) | |
| doc = { | |
| "firstName": body.firstName.strip(), | |
| "email": email, | |
| "phone": phone, | |
| "ageRange": body.ageRange, | |
| "stage": body.stage, | |
| "field": body.field.strip(), | |
| "location": body.location.strip(), | |
| "consent": True, | |
| "answers": [{"q": q, "dim": d} for q, d in sorted(answers.items())], | |
| **result, | |
| "startedAt": body.startedAt, | |
| "durationSeconds": max(0, (now_ms - body.startedAt) // 1000) if body.startedAt else None, | |
| "finishedAt": firestore.SERVER_TIMESTAMP, | |
| "version": 1, | |
| } | |
| try: | |
| _, ref = firestore.client().collection(COLLECTION).add(doc) | |
| except Exception: | |
| raise HTTPException(status_code=500, detail="Could not save your result.") | |
| return {"success": True, "id": ref.id, **result} | |