Spaces:
Running
Running
| """ | |
| Niches people couldn't find in data/sectors.ts and described in their own words. | |
| One Firestore doc per distinct niche (keyed by its normalised text), so repeat | |
| requests add up instead of piling up as duplicates. Admins read the list in the | |
| console to decide which niches to add. firestore.rules' default deny keeps the | |
| collection backend-only. | |
| """ | |
| import re | |
| from firebase_admin import firestore | |
| COLLECTION = "niche_requests" | |
| STATUSES = {"open", "added", "dismissed"} | |
| MAX_USERS_TRACKED = 1000 # beyond this the count stops rising; plenty to rank by | |
| def niche_key(text: str) -> str: | |
| """'School Stationery!' and 'school stationery' are the same request.""" | |
| return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")[:100] | |
| def record(uid: str, text: str, sector_id: str = "", subsector_id: str = "") -> str: | |
| key = niche_key(text) | |
| if not key: | |
| raise ValueError("Niche text has no letters or numbers.") | |
| db = firestore.client() | |
| ref = db.collection(COLLECTION).document(key) | |
| def _upsert(tx): | |
| snap = ref.get(transaction=tx) | |
| d = snap.to_dict() if snap.exists else {} | |
| uids = list(d.get("uids", [])) | |
| if uid not in uids and len(uids) < MAX_USERS_TRACKED: | |
| uids.append(uid) | |
| sectors = list(d.get("sectors", [])) | |
| where = "/".join(p for p in (sector_id, subsector_id) if p) | |
| if where and where not in sectors: | |
| sectors.append(where) | |
| update = { | |
| "text": " ".join(text.split()), | |
| "uids": uids, | |
| "userCount": len(uids), | |
| "sectors": sectors[-20:], | |
| "status": d.get("status", "open"), | |
| "lastAt": firestore.SERVER_TIMESTAMP, | |
| } | |
| if not snap.exists: | |
| update["firstAt"] = firestore.SERVER_TIMESTAMP | |
| tx.set(ref, update, merge=True) | |
| _upsert(db.transaction()) | |
| return key | |
| def list_requests(status: str | None = None) -> list[dict]: | |
| """Most-requested first. Never returns who asked, only how many.""" | |
| out = [] | |
| for doc in firestore.client().collection(COLLECTION).stream(): | |
| d = doc.to_dict() | |
| if status and d.get("status", "open") != status: | |
| continue | |
| out.append({ | |
| "id": doc.id, | |
| "text": d.get("text", ""), | |
| "userCount": d.get("userCount", 0), | |
| "sectors": d.get("sectors", []), | |
| "status": d.get("status", "open"), | |
| "firstAt": str(d.get("firstAt", "")), | |
| "lastAt": str(d.get("lastAt", "")), | |
| }) | |
| return sorted(out, key=lambda r: (-r["userCount"], r["text"])) | |
| def set_status(key: str, status: str) -> str: | |
| """Returns the niche text, for the audit trail.""" | |
| if status not in STATUSES: | |
| raise ValueError(f"status must be one of {sorted(STATUSES)}") | |
| ref = firestore.client().collection(COLLECTION).document(key) | |
| snap = ref.get() | |
| if not snap.exists: | |
| raise KeyError(key) | |
| ref.update({"status": status}) | |
| return snap.to_dict().get("text", key) | |