Spaces:
Running
Running
| """ | |
| The public Business Footing Framework at /bff: the WHAT, WHY and HOW questions | |
| from onboarding, with no sign-in, no AI and no follow-up questions. | |
| The browser sends the answers and the person's contact details; this route | |
| checks them and stores one document per submission in bff_submissions. | |
| 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 routes.bff import BUSINESS_STAGES | |
| from utils.business_type import describe | |
| from utils.public_forms import RateLimiter, clean_email, clean_phone | |
| router = APIRouter() | |
| COLLECTION = "bff_submissions" | |
| PER_SESSION_LIMIT = (5, 3600) # 5 submissions per hour per browser | |
| GLOBAL_LIMIT = (500, 3600) # 500 per hour across everyone | |
| limiter = RateLimiter() | |
| Short = Field(default="", max_length=200) | |
| Long = Field(default="", max_length=4000) | |
| Choices = Field(default_factory=list, max_length=40) | |
| class What(BaseModel): | |
| businessStage: str = Field(max_length=40) | |
| businessName: str = Field(min_length=1, max_length=120) | |
| sectorId: str = Field(min_length=1, max_length=80) | |
| subsectorId: str = Field(min_length=1, max_length=80) | |
| nicheId: str = Field(min_length=1, max_length=80) | |
| customNiche: str = Short | |
| products: list[str] = Choices | |
| class Why(BaseModel): | |
| whyRaw: str = Field(min_length=1, max_length=4000) | |
| valueProposition: str = Field(min_length=1, max_length=4000) | |
| class Hours(BaseModel): | |
| open: str = Field(max_length=5) | |
| close: str = Field(max_length=5) | |
| off: bool | |
| class How(BaseModel): | |
| locationType: str = Field(default="online", pattern="^(online|physical|hybrid)$") | |
| state: str = Short | |
| lga: str = Short | |
| city: str = Short | |
| address: str = Short | |
| hours: dict[str, Hours] = Field(default_factory=dict, max_length=7) | |
| demographics: dict[str, list[str]] = Field(default_factory=dict, max_length=12) | |
| psychographics: str = Long | |
| channels: dict[str, list[str]] = Field(default_factory=dict, max_length=6) | |
| keyPartners: str = Long | |
| revenueStreams: str = Long | |
| class SubmissionIn(BaseModel): | |
| name: str = Field(min_length=1, max_length=120) | |
| email: str = Field(min_length=3, max_length=254) | |
| phone: str = Field(min_length=1, max_length=30) | |
| consent: bool | |
| what: What | |
| why: Why | |
| how: How | |
| startedAt: int = Field(ge=0) # ms since epoch, from the browser | |
| sessionId: str = Field(min_length=1, max_length=64) | |
| def _clean_list(items: list[str]) -> list[str]: | |
| return [i.strip()[:200] for i in items if i.strip()] | |
| def submit(body: SubmissionIn): | |
| 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 send your answers.") | |
| email = clean_email(body.email) | |
| phone = clean_phone(body.phone) | |
| if body.what.businessStage not in BUSINESS_STAGES: | |
| raise HTTPException(status_code=400, detail="Choose how far along your business is.") | |
| if body.what.nicheId == "other" and not body.what.customNiche.strip(): | |
| raise HTTPException(status_code=400, detail="Tell us your niche.") | |
| what = body.what.model_dump() | |
| what["products"] = _clean_list(body.what.products) | |
| how = body.how.model_dump() | |
| how["demographics"] = {k[:40]: _clean_list(v) for k, v in body.how.demographics.items()} | |
| how["channels"] = {k[:40]: _clean_list(v) for k, v in body.how.channels.items()} | |
| now_ms = int(time.time() * 1000) | |
| doc = { | |
| "name": body.name.strip(), | |
| "email": email, | |
| "phone": phone, | |
| "consent": True, | |
| "what": what, | |
| "why": {k: v.strip() for k, v in body.why.model_dump().items()}, | |
| "how": how, | |
| # Readable labels next to the ids, so the collection can be read without the app. | |
| "labels": {**describe(what), "stage": BUSINESS_STAGES[body.what.businessStage]}, | |
| "startedAt": body.startedAt, | |
| "durationSeconds": max(0, (now_ms - body.startedAt) // 1000) if body.startedAt else None, | |
| "submittedAt": 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 answers.") | |
| return {"success": True, "id": ref.id} | |