| """ |
| utility.py — RentNow marketplace core. |
| |
| Covers: profiles, listings, search/match, spoof payments, rental pass, |
| viewings, wallet ledger, payouts, leases, Gemini helpers, PDFs. |
| |
| Money is stored as integer cents. Payments are SPOOFED for MVP. |
| Wire real EcoCash later behind create_payment()/confirm_payment(). |
| """ |
|
|
| import os |
| import re |
| import json |
| import time |
| import uuid |
| import logging |
| import io |
| from datetime import datetime, timezone, timedelta |
| from typing import Any, Dict, List, Optional, Tuple |
|
|
| logger = logging.getLogger(__name__) |
|
|
| |
| try: |
| from google import genai |
| from google.genai import types as genai_types |
| except ImportError: |
| genai = None |
| genai_types = None |
|
|
| GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY", "") |
| GEMINI_MODEL = os.environ.get("GEMINI_MODEL", "gemini-3.1-flash-lite") |
| _gemini = None |
| if genai and GOOGLE_API_KEY: |
| try: |
| _gemini = genai.Client(api_key=GOOGLE_API_KEY) |
| except Exception as e: |
| logger.error(f"Gemini init failed: {e}") |
|
|
| |
| db: Optional[Any] = None |
| try: |
| import firebase_admin |
| from firebase_admin import credentials, firestore, storage as fb_storage |
|
|
| def _init_db(): |
| global db |
| try: |
| if firebase_admin._apps: |
| db = firestore.client(database_id=os.environ.get("FIRESTORE_DB_NAME", "(default)")) |
| return db |
| env_json = os.environ.get("FIREBASE", "") |
| if not env_json: |
| logger.warning("FIREBASE not set — running without persistence (seed data only).") |
| return None |
| cred = credentials.Certificate(json.loads(env_json)) |
| opts = {} |
| bucket = os.environ.get("FIREBASE_STORAGE_BUCKET", "") |
| if bucket: |
| opts["storageBucket"] = bucket |
| firebase_admin.initialize_app(cred, opts) |
| db = firestore.client(database_id=os.environ.get("FIRESTORE_DB_NAME", "(default)")) |
| logger.info("Firebase ready.") |
| return db |
| except Exception as e: |
| logger.error(f"Firebase init failed: {e}") |
| return None |
|
|
| _init_db() |
| except ImportError: |
| logger.warning("firebase-admin not installed — persistence disabled.") |
|
|
| |
| APP_NAME = "RentNow" |
| RENTAL_PASS_CENTS = 1000 |
| RENTAL_PASS_CREDITS = 3 |
| RENTAL_PASS_DAYS = 14 |
| VIEWING_FEE_CENTS = 500 |
| LANDLORD_SHARE_CENTS = 200 |
| PLATFORM_SHARE_CENTS = 300 |
|
|
| PROPERTY_TYPES = ["single room", "cottage", "1 bedroom", "2 bedroom", "house"] |
| AMENITIES = ["solar", "borehole", "tiled floors", "fitted kitchen", "parking", |
| "security", "water", "electricity", "furnished", "wifi"] |
| CITIES = ["Harare", "Bulawayo", "Mutare", "Gweru", "Victoria Falls"] |
|
|
| def _pics(seed: str, n: int = 3) -> List[str]: |
| return [f"https://picsum.photos/seed/{seed}-{i}/600/400" for i in range(1, n + 1)] |
|
|
|
|
| SEED_LISTINGS: List[Dict] = [ |
| {"id": "seed-01", "provider_phone": "+263781234001", "provider_name": "T. Moyo", |
| "type": "2 bedroom", "title": "2 Bedroom House, Westgate", |
| "rent_cents": 38000, "deposit_cents": 38000, |
| "city": "Harare", "suburb": "Westgate", "available_from": "immediately", |
| "occupancy_limit": 4, "description": "Modern 2-bed with fitted kitchen, walled yard, borehole. Quiet street near shops and schools.", |
| "amenities": ["solar", "parking", "borehole", "walled", "tiled", "fitted kitchen"], |
| "media": _pics("rent1"), "status": "active", "verification": "verified", "seed": True}, |
| {"id": "seed-02", "provider_phone": "+263781234002", "provider_name": "R. Chikafu", |
| "type": "2 bedroom", "title": "2 Bedroom Cottage, Westgate", |
| "rent_cents": 35000, "deposit_cents": 35000, |
| "city": "Harare", "suburb": "Westgate", "available_from": "immediately", |
| "occupancy_limit": 3, "description": "Neat cottage, own entrance and meter. Tiled throughout, secure yard.", |
| "amenities": ["solar", "parking", "tiled", "secure"], |
| "media": _pics("rent2"), "status": "active", "verification": "verified", "seed": True}, |
| {"id": "seed-03", "provider_phone": "+263781234003", "provider_name": "S. Dube", |
| "type": "1 bedroom", "title": "1 Bedroom Flat, Avondale", |
| "rent_cents": 28000, "deposit_cents": 28000, |
| "city": "Harare", "suburb": "Avondale", "available_from": "immediately", |
| "occupancy_limit": 2, "description": "Sunny 1-bed in secure complex. Fitted kitchen, parking, backup water.", |
| "amenities": ["solar", "borehole", "parking", "security", "fitted kitchen"], |
| "media": _pics("rent3"), "status": "active", "verification": "verified", "seed": True}, |
| {"id": "seed-04", "provider_phone": "+263781234004", "provider_name": "K. Mutasa", |
| "type": "cottage", "title": "Cottage, Glen View", |
| "rent_cents": 15000, "deposit_cents": 15000, |
| "city": "Harare", "suburb": "Glen View", "available_from": "immediately", |
| "occupancy_limit": 3, "description": "Quiet yard cottage near transport. Own meter, water available.", |
| "amenities": ["parking", "water", "electricity", "security"], |
| "media": _pics("rent4"), "status": "active", "verification": "verified", "seed": True}, |
| {"id": "seed-05", "provider_phone": "+263781234005", "provider_name": "B. Nyathi", |
| "type": "single room", "title": "Single Room, Mbare", |
| "rent_cents": 6000, "deposit_cents": 6000, |
| "city": "Harare", "suburb": "Mbare", "available_from": "immediately", |
| "occupancy_limit": 2, "description": "Budget room minutes from bus terminus. Shared ablutions, water daily.", |
| "amenities": ["water", "electricity"], |
| "media": _pics("rent5"), "status": "active", "verification": "verified", "seed": True}, |
| {"id": "seed-06", "provider_phone": "+263781234006", "provider_name": "A. Sibanda", |
| "type": "2 bedroom", "title": "2 Bedroom House, Kuwadzana", |
| "rent_cents": 25000, "deposit_cents": 25000, |
| "city": "Harare", "suburb": "Kuwadzana", "available_from": "flexible", |
| "occupancy_limit": 4, "description": "Family house with borehole and parking. Schools nearby.", |
| "amenities": ["borehole", "parking", "security", "fitted kitchen"], |
| "media": _pics("rent6"), "status": "active", "verification": "verified", "seed": True}, |
| {"id": "seed-07", "provider_phone": "+263781234007", "provider_name": "M. Hove", |
| "type": "cottage", "title": "Cottage, Hatcliffe", |
| "rent_cents": 18000, "deposit_cents": 18000, |
| "city": "Harare", "suburb": "Hatcliffe", "available_from": "immediately", |
| "occupancy_limit": 3, "description": "Spacious cottage, walled stand, pet friendly. Solar lighting.", |
| "amenities": ["solar", "walled", "pet friendly", "parking"], |
| "media": _pics("rent7"), "status": "active", "verification": "verified", "seed": True}, |
| {"id": "seed-08", "provider_phone": "+263781234008", "provider_name": "P. Zulu", |
| "type": "single room", "title": "Single Room, Epworth", |
| "rent_cents": 7000, "deposit_cents": 7000, |
| "city": "Harare", "suburb": "Epworth", "available_from": "immediately", |
| "occupancy_limit": 2, "description": "Clean room, outside toilet, ZESA available. Deposit negotiable.", |
| "amenities": ["water", "electricity"], |
| "media": _pics("rent8"), "status": "active", "verification": "verified", "seed": True}, |
| ] |
|
|
| _mem: Dict[str, Dict] = {} |
| _sess_mem: Dict[str, Dict] = {} |
| SESS_TTL = 1800 |
|
|
|
|
| def _now() -> str: |
| return datetime.now(timezone.utc).isoformat() |
|
|
|
|
| def sess_get(phone: str) -> Optional[Dict]: |
| """Server-side session. Firestore-backed so restarts never drop context.""" |
| s = _sess_mem.get(phone) |
| if s and time.time() - s.get("ts", 0) < SESS_TTL: |
| return s |
| if db: |
| try: |
| doc = db.collection("rentnow_sessions").document(phone).get() |
| if doc.exists: |
| s = doc.to_dict() |
| if time.time() - s.get("ts", 0) < SESS_TTL: |
| _sess_mem[phone] = s |
| return s |
| except Exception as e: |
| logger.error(f"sess_get: {e}") |
| return None |
|
|
|
|
| def sess_set(phone: str, state: Dict) -> None: |
| state = {**state, "ts": time.time(), "updated_at": _now()} |
| _sess_mem[phone] = state |
| if db: |
| try: |
| db.collection("rentnow_sessions").document(phone).set(state) |
| except Exception as e: |
| logger.error(f"sess_set: {e}") |
|
|
|
|
| def sess_clear(phone: str) -> None: |
| _sess_mem.pop(phone, None) |
| if db: |
| try: |
| db.collection("rentnow_sessions").document(phone).delete() |
| except Exception: |
| pass |
|
|
|
|
| def cents_fmt(c: int) -> str: |
| return f"${c/100:.0f}" if c % 100 == 0 else f"${c/100:.2f}" |
|
|
|
|
| |
| def get_profile(phone: str) -> Dict: |
| if db: |
| try: |
| doc = db.collection("rentnow_profiles").document(phone).get() |
| if doc.exists: |
| return doc.to_dict() |
| except Exception as e: |
| logger.error(f"get_profile: {e}") |
| return _mem.get(f"profile:{phone}", {"phone": phone, "roles": []}) |
|
|
|
|
| def save_profile(phone: str, patch: Dict) -> Dict: |
| cur = get_profile(phone) |
| cur.update(patch) |
| cur["phone"] = phone |
| if db: |
| try: |
| db.collection("rentnow_profiles").document(phone).set(cur, merge=True) |
| except Exception as e: |
| logger.error(f"save_profile: {e}") |
| _mem[f"profile:{phone}"] = cur |
| return cur |
|
|
|
|
| def get_chat_history(phone: str, limit: int = 6) -> str: |
| if not db: |
| return "" |
| try: |
| docs = (db.collection("rentnow_profiles").document(phone).collection("chat_logs") |
| .order_by("ts", direction="DESCENDING").limit(limit).stream()) |
| msgs = [f"User: {d.to_dict().get('message','')}\nBot: {d.to_dict().get('response','')}" for d in docs] |
| return "\n".join(reversed(msgs)) |
| except Exception: |
| return "" |
|
|
|
|
| def save_chat_log(phone: str, message: str, response: str, intent: Dict) -> None: |
| if not db: |
| return |
| try: |
| db.collection("rentnow_profiles").document(phone).collection("chat_logs").add( |
| {"message": message, "response": response, "intent": intent, "ts": _now()}) |
| except Exception as e: |
| logger.error(f"chat log: {e}") |
|
|
|
|
| |
| def get_listings(active_only: bool = True) -> List[Dict]: |
| out: List[Dict] = [] |
| if db: |
| try: |
| q = db.collection("rentnow_listings") |
| if active_only: |
| q = q.where("status", "==", "active") |
| for d in q.stream(): |
| item = d.to_dict() |
| item["id"] = d.id |
| out.append(item) |
| except Exception as e: |
| logger.error(f"get_listings: {e}") |
| if not out: |
| |
| mem_extra = [v for k, v in _mem.items() if k.startswith("listing:")] |
| out = list(SEED_LISTINGS) + [v for v in mem_extra] |
| if active_only: |
| out = [x for x in out if x.get("status") == "active"] |
| return out |
|
|
|
|
| def get_listing(listing_id: str) -> Optional[Dict]: |
| for x in get_listings(active_only=False): |
| if x.get("id") == listing_id: |
| return x |
| if db: |
| try: |
| doc = db.collection("rentnow_listings").document(listing_id).get() |
| if doc.exists: |
| item = doc.to_dict() |
| item["id"] = doc.id |
| return item |
| except Exception: |
| pass |
| return None |
|
|
|
|
| def create_listing(data: Dict) -> Dict: |
| lid = f"L-{uuid.uuid4().hex[:6].upper()}" |
| data = {**data, "id": lid, "status": "pending_review", |
| "verification": "pending", "created_at": _now(), "updated_at": _now()} |
| if db: |
| try: |
| db.collection("rentnow_listings").document(lid).set(data) |
| except Exception as e: |
| logger.error(f"create_listing: {e}") |
| _mem[f"listing:{lid}"] = data |
| return data |
|
|
|
|
| def seed_upsert() -> int: |
| """Write the full seed set to Firestore (upsert). Returns count written.""" |
| n = 0 |
| for s in SEED_LISTINGS: |
| item = {**s, "updated_at": _now()} |
| item.setdefault("created_at", _now()) |
| if db: |
| try: |
| db.collection("rentnow_listings").document(s["id"]).set(item) |
| n += 1 |
| except Exception as e: |
| logger.error(f"seed_upsert: {e}") |
| _mem[f"listing:{s['id']}"] = item |
| n += 0 if db else 1 |
| return n |
|
|
|
|
| def delete_listing(listing_id: str) -> bool: |
| """Remove a listing (seed cleanup, admin). Returns True if one existed.""" |
| gone = False |
| if db: |
| try: |
| ref = db.collection("rentnow_listings").document(listing_id) |
| if ref.get().exists: |
| ref.delete() |
| gone = True |
| except Exception as e: |
| logger.error(f"delete_listing: {e}") |
| if _mem.pop(f"listing:{listing_id}", None) is not None: |
| gone = True |
| if any(s.get("id") == listing_id for s in SEED_LISTINGS): |
| gone = True |
| return gone |
|
|
|
|
| def admin_create_listing(data: Dict, reviewer: str = "admin") -> Dict: |
| """Admin lists a home on behalf of an owner. Goes live immediately.""" |
| lid = f"L-{uuid.uuid4().hex[:6].upper()}" |
| try: |
| rent_c = int(float(data.get("rent", 0)) * 100) |
| except Exception: |
| rent_c = 0 |
| try: |
| dep_c = int(float(data.get("deposit", 0)) * 100) |
| except Exception: |
| dep_c = 0 |
| item = { |
| "id": lid, "provider_phone": data.get("provider_phone", ""), |
| "provider_name": data.get("provider_name", ""), |
| "type": (data.get("type") or "").lower(), "title": data.get("title", ""), |
| "rent_cents": rent_c, "deposit_cents": dep_c, |
| "city": data.get("city", "Harare"), "suburb": data.get("suburb", ""), |
| "available_from": data.get("available_from", "immediately"), |
| "occupancy_limit": data.get("occupancy_limit", ""), |
| "description": data.get("description", ""), |
| "amenities": data.get("amenities", []) if isinstance(data.get("amenities"), list) |
| else [a.strip() for a in (data.get("amenities") or "").split(",") if a.strip()], |
| "media": data.get("media", []) if isinstance(data.get("media"), list) |
| else [u.strip() for u in (data.get("media") or "").split(",") if u.strip()], |
| "video_url": data.get("video_url", ""), |
| "rental_requirements": data.get("rental_requirements", ""), |
| "status": "active", "verification": "verified", |
| "created_by": reviewer, "created_at": _now(), "updated_at": _now(), |
| } |
| if db: |
| try: |
| db.collection("rentnow_listings").document(lid).set(item) |
| db.collection("rentnow_audit").add( |
| {"actor": reviewer, "action": "listing.admin_created", |
| "entity": "listing", "entity_id": lid, "ts": _now()}) |
| except Exception as e: |
| logger.error(f"admin_create_listing: {e}") |
| _mem[f"listing:{lid}"] = item |
| return item |
|
|
|
|
| def moderate_listing(listing_id: str, approve: bool, reviewer: str = "admin") -> Optional[Dict]: |
| item = get_listing(listing_id) |
| if not item: |
| return None |
| item["status"] = "active" if approve else "rejected" |
| item["verification"] = "verified" if approve else "rejected" |
| item["updated_at"] = _now() |
| if db: |
| try: |
| db.collection("rentnow_listings").document(listing_id).set(item, merge=True) |
| db.collection("rentnow_verifications").add( |
| {"target": listing_id, "type": "listing", "status": item["status"], |
| "reviewer": reviewer, "ts": _now()}) |
| except Exception as e: |
| logger.error(f"moderate: {e}") |
| _mem[f"listing:{listing_id}"] = item |
| return item |
|
|
|
|
| |
| def match_score(item: Dict, crit: Dict) -> Tuple[int, List[str]]: |
| score, reasons = 0, [] |
| want_type = (crit.get("type") or "").lower() |
| if want_type and want_type in (item.get("type") or "").lower(): |
| score += 30 |
| reasons.append("correct type") |
| suburb = (crit.get("suburb") or "").lower().replace("mt pleasant", "mount pleasant") |
| if suburb and suburb in (item.get("suburb") or "").lower(): |
| score += 25 |
| reasons.append("correct suburb") |
| try: |
| max_b = float(crit.get("max_budget") or 0) |
| except Exception: |
| max_b = 0 |
| rent = (item.get("rent_cents") or 0) / 100 |
| if max_b and rent <= max_b: |
| score += 20 |
| reasons.append("within budget") |
| wants = [a.lower() for a in (crit.get("amenities") or [])] |
| haves = [a.lower() for a in (item.get("amenities") or [])] |
| hit = [a for a in wants if any(a in h for h in haves)] |
| if wants: |
| score += min(20, int(20 * len(hit) / max(1, len(wants)))) |
| reasons += hit |
| if (item.get("available_from") or "") == "immediately": |
| score += 5 |
| reasons.append("available immediately") |
| return min(100, score), reasons |
|
|
|
|
| def search_listings(crit: Dict, limit: int = 5) -> List[Dict]: |
| items = get_listings(active_only=True) |
| want_type = (crit.get("type") or "").lower() |
| suburb = (crit.get("suburb") or "").lower().replace("mt pleasant", "mount pleasant") |
| try: |
| max_b = float(crit.get("max_budget") or 0) |
| except Exception: |
| max_b = 0 |
| min_b = float(crit.get("min_budget") or 0) |
| ranked = [] |
| for it in items: |
| if want_type and want_type not in (it.get("type") or "").lower(): |
| continue |
| if suburb and suburb not in (it.get("suburb") or "").lower().replace("mt pleasant", "mount pleasant"): |
| continue |
| if max_b and (it.get("rent_cents") or 0) / 100 > max_b: |
| continue |
| if min_b and (it.get("rent_cents") or 0) / 100 < min_b: |
| continue |
| wants = [a.lower() for a in crit.get("amenities", [])] |
| haves = [a.lower() for a in it.get("amenities", [])] |
| if any(not any(a in h for h in haves) for a in wants): |
| continue |
| s, reasons = match_score(it, crit) |
| ranked.append({**it, "_score": s, "_reasons": reasons}) |
| ranked.sort(key=lambda x: (-x["_score"], x.get("rent_cents", 0))) |
| return ranked[:limit] |
|
|
|
|
| def format_preview(item: Dict, idx: int = 0, score: Optional[int] = None, |
| reasons: Optional[List[str]] = None) -> str: |
| tag = f"*{idx}. {item.get('title')}* [{item.get('id')}]" if idx else f"*{item.get('title')}* [{item.get('id')}]" |
| lines = [tag, |
| f" Type: {item.get('type')} | {item.get('suburb')}, {item.get('city')}", |
| f" Rent: *{cents_fmt(item.get('rent_cents', 0))}/mo* | Deposit: {cents_fmt(item.get('deposit_cents', 0))}", |
| f" {', '.join(item.get('amenities', [])[:5])}"] |
| if score is not None: |
| lines.append(f" _{score}% match — {', '.join((reasons or [])[:4])}_") |
| lines.append(" Property: " + ("Verified" if item.get("verification") == "verified" else "Verification pending")) |
| provider = get_profile(item.get("provider_phone", "")) |
| lines.append(" Provider: " + ("Verified" if (provider.get("landlord") or {}).get("verification") == "verified" else "Verification pending")) |
| lines.append(" _Free preview. Full details available with a $10 Rental Pass._") |
| return "\n".join(lines) |
|
|
|
|
| def format_full(item: Dict) -> str: |
| lines = [f"*{item.get('title')}* [{item.get('id')}]", |
| f"Type: {item.get('type')} | {item.get('suburb')}, {item.get('city')}", |
| f"Rent: *{cents_fmt(item.get('rent_cents', 0))}/mo* | Deposit: {cents_fmt(item.get('deposit_cents', 0))}", |
| f"Available: {item.get('available_from')} | Sleeps: {item.get('occupancy_limit')}", |
| f"{item.get('description')}", |
| f"Amenities: {', '.join(item.get('amenities', []))}", |
| f"Verified: {'yes' if item.get('verification') == 'verified' else 'pending'}", |
| f"Photos: {len(item.get('media', []))} unlocked", |
| f"Rental requirements: {item.get('rental_requirements') or item.get('requirements') or 'Ask RentNow for the provider requirements.'}", |
| f"Provider: {item.get('provider_name') or 'RentNow coordinates with the provider'}", |
| "Provider contact details are shared after the $5 viewing payment."] |
| return "\n".join(lines) |
|
|
|
|
| |
| def create_payment(phone: str, kind: str, amount_cents: int, |
| meta: Optional[Dict] = None) -> Dict: |
| """Create a spoof EcoCash payment intent. Wire real provider later.""" |
| pid = f"PAY-{uuid.uuid4().hex[:8].upper()}" |
| ref = f"RN-{uuid.uuid4().hex[:6].upper()}" |
| pay = {"id": pid, "user_phone": phone, "kind": kind, "amount_cents": amount_cents, |
| "provider": "ecocash_spoof", "provider_ref": ref, "status": "pending", |
| "spoof": True, "meta": meta or {}, "created_at": _now()} |
| if db: |
| try: |
| db.collection("rentnow_payments").document(pid).set(pay) |
| except Exception as e: |
| logger.error(f"create_payment: {e}") |
| _mem[f"pay:{pid}"] = pay |
| return pay |
|
|
|
|
| def confirm_payment_spoof(payment_id: str, ref_input: str = "") -> Optional[Dict]: |
| """MVP spoof: user taps 'I have paid' and we confirm. Replace with webhook.""" |
| pay = _mem.get(f"pay:{payment_id}") |
| if not pay and db: |
| try: |
| doc = db.collection("rentnow_payments").document(payment_id).get() |
| pay = doc.to_dict() if doc.exists else None |
| except Exception: |
| pay = None |
| if not pay or pay.get("status") == "confirmed": |
| return pay |
| pay["status"] = "confirmed" |
| pay["confirmed_at"] = _now() |
| pay["ref_input"] = ref_input |
| if db: |
| try: |
| db.collection("rentnow_payments").document(payment_id).set(pay, merge=True) |
| db.collection("rentnow_audit").add( |
| {"actor": pay["user_phone"], "action": "payment.confirmed", |
| "entity": "payment", "entity_id": payment_id, "ts": _now()}) |
| except Exception as e: |
| logger.error(f"confirm_payment: {e}") |
| _mem[f"pay:{payment_id}"] = pay |
| return pay |
|
|
|
|
| def ecocash_instructions(pay: Dict) -> str: |
| return ( |
| f"EcoCash (spoofed MVP):\n" |
| f"Amount: *{cents_fmt(pay['amount_cents'])}*\n" |
| f"Reference: *{pay['provider_ref']}*\n" |
| "Demo only. No money is charged. Reply 1 to simulate a successful EcoCash payment." |
| ) |
|
|
|
|
| |
| def active_pass(phone: str) -> Optional[Dict]: |
| best = None |
| cands: List[Dict] = [] |
| if db: |
| try: |
| for d in db.collection("rentnow_passes").where("tenant_phone", "==", phone).stream(): |
| cands.append({**d.to_dict(), "id": d.id}) |
| except Exception as e: |
| logger.error(f"active_pass: {e}") |
| cands += [v for k, v in _mem.items() if k.startswith("pass:") and v.get("tenant_phone") == phone] |
| now = datetime.now(timezone.utc) |
| for p in cands: |
| try: |
| exp = datetime.fromisoformat(p["expires_at"]) |
| if p.get("status") == "active" and exp > now and p.get("credits_used", 0) < p.get("credits_total", 3): |
| if not best or p["expires_at"] > best["expires_at"]: |
| best = p |
| except Exception: |
| continue |
| return best |
|
|
|
|
| def issue_pass(phone: str, payment_id: str) -> Dict: |
| pid = f"PASS-{uuid.uuid4().hex[:6].upper()}" |
| exp = (datetime.now(timezone.utc) + timedelta(days=RENTAL_PASS_DAYS)).isoformat() |
| p = {"id": pid, "tenant_phone": phone, "amount_cents": RENTAL_PASS_CENTS, |
| "credits_total": RENTAL_PASS_CREDITS, "credits_used": 0, "status": "active", |
| "payment_id": payment_id, "created_at": _now(), "expires_at": exp} |
| if db: |
| try: |
| db.collection("rentnow_passes").document(pid).set(p) |
| except Exception as e: |
| logger.error(f"issue_pass: {e}") |
| _mem[f"pass:{pid}"] = p |
| return p |
|
|
|
|
| def is_unlocked(phone: str, listing_id: str) -> bool: |
| if db: |
| try: |
| q = (db.collection("rentnow_unlocks") |
| .where("tenant_phone", "==", phone).where("listing_id", "==", listing_id).limit(1).stream()) |
| if any(True for _ in q): |
| return True |
| except Exception: |
| pass |
| return any(v.get("tenant_phone") == phone and v.get("listing_id") == listing_id |
| for k, v in _mem.items() if k.startswith("unlock:")) |
|
|
|
|
| def consume_unlock(phone: str, listing_id: str) -> Tuple[bool, str]: |
| if is_unlocked(phone, listing_id): |
| return True, "already unlocked" |
| p = active_pass(phone) |
| if not p: |
| return False, "no active pass" |
| p["credits_used"] = p.get("credits_used", 0) + 1 |
| if p["credits_used"] >= p.get("credits_total", 3): |
| p["status"] = "expired" |
| uid = f"UNL-{uuid.uuid4().hex[:6].upper()}" |
| rec = {"id": uid, "pass_id": p["id"], "listing_id": listing_id, |
| "tenant_phone": phone, "unlocked_at": _now()} |
| if db: |
| try: |
| db.collection("rentnow_passes").document(p["id"]).set(p, merge=True) |
| db.collection("rentnow_unlocks").document(uid).set(rec) |
| except Exception as e: |
| logger.error(f"consume_unlock: {e}") |
| _mem[f"pass:{p['id']}"] = p |
| _mem[f"unlock:{uid}"] = rec |
| left = p.get("credits_total", 3) - p.get("credits_used", 0) |
| return True, f"{left} credit(s) left" |
|
|
|
|
| |
| def create_viewing(tenant: str, listing_id: str, payment_id: str) -> Dict: |
| vid = f"V-{uuid.uuid4().hex[:6].upper()}" |
| lst = get_listing(listing_id) or {} |
| v = {"id": vid, "listing_id": listing_id, "tenant_phone": tenant, |
| "landlord_phone": lst.get("provider_phone", ""), |
| "fee_payment_id": payment_id, "status": "paid_appointment_pending", |
| "appointment_at": "", "note": "", "created_at": _now()} |
| if db: |
| try: |
| db.collection("rentnow_viewings").document(vid).set(v) |
| except Exception as e: |
| logger.error(f"create_viewing: {e}") |
| _mem[f"view:{vid}"] = v |
| return v |
|
|
|
|
| def get_viewing(vid: str) -> Optional[Dict]: |
| if f"view:{vid}" in _mem: |
| return _mem[f"view:{vid}"] |
| if db: |
| try: |
| doc = db.collection("rentnow_viewings").document(vid).get() |
| if doc.exists: |
| return {**doc.to_dict(), "id": doc.id} |
| except Exception: |
| pass |
| return None |
|
|
|
|
| def set_appointment(vid: str, when: str, note: str = "") -> Optional[Dict]: |
| v = get_viewing(vid) |
| if not v: |
| return None |
| v.update(status="appointment_set", appointment_at=when, note=note) |
| if db: |
| try: |
| db.collection("rentnow_viewings").document(vid).set(v, merge=True) |
| except Exception as e: |
| logger.error(f"set_appointment: {e}") |
| _mem[f"view:{vid}"] = v |
| return v |
|
|
|
|
| def complete_viewing(vid: str, outcome: str) -> Optional[Dict]: |
| v = get_viewing(vid) |
| if not v: |
| return None |
| v["status"] = outcome |
| if db: |
| try: |
| db.collection("rentnow_viewings").document(vid).set(v, merge=True) |
| except Exception as e: |
| logger.error(f"complete_viewing: {e}") |
| _mem[f"view:{vid}"] = v |
| if outcome == "completed": |
| credit_landlord(v["landlord_phone"], LANDLORD_SHARE_CENTS, vid) |
| return v |
|
|
|
|
| |
| def credit_landlord(phone: str, cents: int, viewing_id: str) -> Dict: |
| eid = f"LED-{uuid.uuid4().hex[:6].upper()}" |
| e = {"id": eid, "landlord_phone": phone, "kind": "earning", "amount_cents": cents, |
| "viewing_id": viewing_id, "status": "earned", "ref": viewing_id, "ts": _now()} |
| if db: |
| try: |
| db.collection("rentnow_wallet_ledger").document(eid).set(e) |
| except Exception as ex: |
| logger.error(f"credit_landlord: {ex}") |
| _mem[f"led:{eid}"] = e |
| return e |
|
|
|
|
| def wallet_balance(phone: str) -> int: |
| total = 0 |
| rows: List[Dict] = [] |
| if db: |
| try: |
| for d in db.collection("rentnow_wallet_ledger").where("landlord_phone", "==", phone).stream(): |
| rows.append(d.to_dict()) |
| except Exception: |
| pass |
| rows += [v for k, v in _mem.items() if k.startswith("led:") and v.get("landlord_phone") == phone] |
| for r in rows: |
| if r.get("kind") == "earning": |
| total += r.get("amount_cents", 0) |
| elif r.get("kind") in ("payout", "adjustment"): |
| total -= abs(r.get("amount_cents", 0)) |
| return total |
|
|
|
|
| def payout_run(period: str) -> Dict: |
| bid = f"BATCH-{uuid.uuid4().hex[:6].upper()}" |
| batch = {"id": bid, "period": period, "status": "processed", |
| "approved_at": _now(), "processed_at": _now(), "payouts": []} |
| |
| phones = set() |
| if db: |
| try: |
| for d in db.collection("rentnow_wallet_ledger").stream(): |
| phones.add(d.to_dict().get("landlord_phone", "")) |
| except Exception: |
| pass |
| phones |= {v.get("landlord_phone", "") for k, v in _mem.items() if k.startswith("led:")} |
| for ph in [p for p in phones if p]: |
| bal = wallet_balance(ph) |
| if bal > 0: |
| pid = f"PO-{uuid.uuid4().hex[:6].upper()}" |
| batch["payouts"].append({"id": pid, "provider_phone": ph, "amount_cents": bal}) |
| adj = {"id": f"LED-{uuid.uuid4().hex[:6].upper()}", "landlord_phone": ph, |
| "kind": "payout", "amount_cents": bal, "payout_id": pid, |
| "status": "paid", "ref": bid, "ts": _now()} |
| if db: |
| try: |
| db.collection("rentnow_wallet_ledger").document(adj["id"]).set(adj) |
| except Exception: |
| pass |
| _mem[f"led:{adj['id']}"] = adj |
| if db: |
| try: |
| db.collection("rentnow_payout_batches").document(bid).set(batch) |
| except Exception: |
| pass |
| return batch |
|
|
|
|
| |
| def create_lease(listing_id: str, tenant: str, term_months: int = 12) -> Dict: |
| lst = get_listing(listing_id) or {} |
| lid = f"LEASE-{uuid.uuid4().hex[:6].upper()}" |
| lease = {"id": lid, "listing_id": listing_id, "tenant_phone": tenant, |
| "landlord_phone": lst.get("provider_phone", ""), |
| "rent_cents": lst.get("rent_cents", 0), "deposit_cents": lst.get("deposit_cents", 0), |
| "term_months": term_months, "status": "draft", "created_at": _now()} |
| if db: |
| try: |
| db.collection("rentnow_leases").document(lid).set(lease) |
| except Exception as e: |
| logger.error(f"create_lease: {e}") |
| _mem[f"lease:{lid}"] = lease |
| return lease |
|
|
|
|
| def lease_status(lid: str, status: str) -> Optional[Dict]: |
| lease = _mem.get(f"lease:{lid}") |
| if not lease and db: |
| try: |
| doc = db.collection("rentnow_leases").document(lid).get() |
| lease = {**doc.to_dict(), "id": doc.id} if doc.exists else None |
| except Exception: |
| lease = None |
| if not lease: |
| return None |
| lease["status"] = status |
| if db: |
| try: |
| db.collection("rentnow_leases").document(lid).set(lease, merge=True) |
| except Exception: |
| pass |
| _mem[f"lease:{lid}"] = lease |
| return lease |
|
|
|
|
| def generate_lease_pdf(lease: Dict, listing: Dict) -> Optional[str]: |
| try: |
| from reportlab.lib.pagesizes import A4 |
| from reportlab.lib.styles import getSampleStyleSheet |
| from reportlab.lib.units import cm |
| from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer |
| outdir = os.path.join(os.getcwd(), "leases") |
| os.makedirs(outdir, exist_ok=True) |
| path = os.path.join(outdir, f"{lease['id']}.pdf") |
| doc = SimpleDocTemplate(path, pagesize=A4, leftMargin=1.5 * cm, rightMargin=1.5 * cm) |
| styles = getSampleStyleSheet() |
| story = [Paragraph("RENTNOW — Residential Lease (DRAFT)", styles["Title"]), |
| Spacer(1, 0.3 * cm), |
| Paragraph("_Template pending Zimbabwean counsel approval. Do not sign as final._", |
| styles["Normal"]), |
| Spacer(1, 0.3 * cm)] |
| rows = [ |
| f"Lease ID: {lease['id']}", |
| f"Property: {listing.get('title')} ({listing.get('suburb')}, {listing.get('city')})", |
| f"Type: {listing.get('type')}", |
| f"Landlord: {lease.get('landlord_phone')}", |
| f"Tenant: {lease.get('tenant_phone')}", |
| f"Rent: {cents_fmt(lease.get('rent_cents', 0))} per month", |
| f"Deposit: {cents_fmt(lease.get('deposit_cents', 0))}", |
| f"Term: {lease.get('term_months')} months", |
| f"Status: {lease.get('status')}", |
| f"Date: {lease.get('created_at')}", |
| ] |
| for r in rows: |
| story.append(Paragraph(r, styles["Normal"])) |
| story.append(Spacer(1, 0.4 * cm)) |
| story.append(Paragraph("Tenant signature: ___________________ Date: _______", styles["Normal"])) |
| story.append(Paragraph("Landlord signature: _________________ Date: _______", styles["Normal"])) |
| doc.build(story) |
| return path |
| except Exception as e: |
| logger.error(f"lease pdf: {e}") |
| return None |
|
|
|
|
| |
| def _safe_json(s: str, fallback: Any) -> Any: |
| try: |
| c = s |
| if "```json" in c: |
| c = c.split("```json")[1].split("```")[0] |
| elif "```" in c: |
| c = c.split("```")[1] |
| return json.loads(c.strip()) |
| except Exception: |
| return fallback |
|
|
|
|
| def gemini_detect_intent(text: str) -> Dict[str, Any]: |
| if not _gemini: |
| return {"intent": "CASUAL", "criteria": {}} |
| prompt = ( |
| "You route WhatsApp messages for RentNow, a Zimbabwe rental finder. " |
| "Return STRICT JSON only. " |
| '{"intent": "SEARCH|REGISTER_TENANT|REGISTER_LANDLORD|CREATE_LISTING|GET_PASS|MY_ACCOUNT|HELP|SUPPORT|LEASE|CASUAL", ' |
| '"criteria": {"type": "", "suburb": "", "max_budget": 0, "amenities": []}} ' |
| "Types, exact strings: single room, cottage, 1 bedroom, 2 bedroom. " |
| "Map 2 bed, two bedroom, 2bed to 2 bedroom. Map 1 bed to 1 bedroom. Map room, bedsitter to single room. " |
| "Extract suburb as written. Extract max monthly USD from under/below/budget/up to/$, else 0. " |
| "Amenities from: solar, borehole, parking, furnished, walled, tiled, security, water, electricity, wifi, fitted kitchen, pet friendly. " |
| "SEARCH covers any home request, even vague ones like looking, need a place, anything available. " |
| "SUPPORT covers help with a person, complaint, problem, agent. " |
| f"Message: {text}" |
| ) |
| try: |
| resp = _gemini.models.generate_content( |
| model=GEMINI_MODEL, contents=prompt, |
| config=genai_types.GenerateContentConfig(response_mime_type="application/json")) |
| return _safe_json(resp.text, {"intent": "CASUAL", "criteria": {}}) |
| except Exception as e: |
| logger.error(f"intent: {e}") |
| return {"intent": "CASUAL", "criteria": {}} |
|
|
|
|
| def gemini_chat(text: str, context: str = "") -> str: |
| """Warm short rental assistant, Pricelyst style. Always ends with a next step.""" |
| if not _gemini: |
| return "" |
| try: |
| resp = _gemini.models.generate_content( |
| model=GEMINI_MODEL, |
| contents=( |
| "You are RentNow, a friendly Zimbabwe rental assistant on WhatsApp. " |
| "Voice: warm, plain, confident. Two short paragraphs max. " |
| "Start by acknowledging what the user wants in one line. " |
| "Then give one concrete next step: search by type plus suburb plus budget, " |
| "unlock with the $10 pass, or book a $5 viewing. " |
| "Never invent listings, prices, or suburbs. No URLs except rentnow.co.zw. " |
| f"Context: {context}\nUser: {text}") |
| ) |
| return (resp.text or "").strip() |
| except Exception: |
| return "" |
|
|
|
|
| def gemini_ack_search(crit: Dict, count: int) -> str: |
| """One-line acknowledgment before results. Gemini when present, template fallback.""" |
| bits = [] |
| if crit.get("type"): |
| bits.append(crit["type"] + "s") |
| if crit.get("suburb"): |
| bits.append("in " + crit["suburb"]) |
| if crit.get("max_budget"): |
| bits.append("up to $" + str(int(crit["max_budget"])) + "/month") |
| want = " ".join(bits) or "homes" |
| if _gemini: |
| try: |
| resp = _gemini.models.generate_content( |
| model=GEMINI_MODEL, |
| contents=(f"Write one warm WhatsApp line. User wants {want}. " |
| f"We found {count} matches. No prices, no listings, one line.")) |
| line = (resp.text or "").strip() |
| if line: |
| return line |
| except Exception: |
| pass |
| return f"Perfect. Searching RentNow for {want}. Found {count} for you." |
|
|
|
|
| def parse_budget(text: str) -> Tuple[Optional[float], Optional[float]]: |
| nums = [float(n.replace(",", "")) for n in re.findall(r"\$?\s*(\d[\d,]*)", text or "")] |
| if not nums: |
| return None, None |
| if len(nums) == 1: |
| return None, nums[0] |
| return min(nums[:2]), max(nums[:2]) |
|
|
|
|
| SUBURBS = ["westgate", "avondale", "borrowdale", "mbare", "kuwadzana", "glen view", |
| "hatcliffe", "epworth", "ruwa", "chitungwiza", "bulawayo", "cbd", "town", |
| "greendale", "mount pleasant", "mt pleasant", "chisipite", "highlands", "southerton"] |
|
|
| TYPE_KEYS = [ |
| ("2 bedroom", ["2 bedroom", "2 bedrooms", "2 bed", "two bedroom", "2bed"]), |
| ("1 bedroom", ["1 bedroom", "1 bedrooms", "1 bed", "one bedroom", "1bed"]), |
| ("cottage", ["cottage", "cottages"]), |
| ("house", ["house", "townhouse"]), |
| ("cottage", ["garden flat"]), |
| ("single room", ["single room", "single rooms", "room", "rooms", "bedsitter"]), |
| ] |
|
|
| AMEN_KEYS = ["solar", "borehole", "parking", "furnished", "walled", "tiled", |
| "secure", "security", "water", "electricity", "wifi", "fitted kitchen", |
| "pet friendly", "pets", "ensuite", "gated community"] |
|
|
|
|
| def parse_search(text: str) -> Dict: |
| """Deterministic natural-language search. No AI needed. |
| |
| Understands: '2 bedroom in Westgate under $400 with solar and parking'. |
| Returns criteria dict; empty values mean 'no constraint'. |
| """ |
| t = (text or "").lower() |
| crit: Dict = {"type": "", "suburb": "", "max_budget": 0, "amenities": []} |
|
|
| t = t.replace("mt pleasant", "mount pleasant") |
| for canon, keys in sorted(TYPE_KEYS, key=lambda pair: pair[0] != "cottage"): |
| if any(k in t for k in keys): |
| crit["type"] = canon |
| break |
|
|
| near = re.search(r"near\s+([a-z ]+?)(?:\s+with|\s+under|\s+below|\s+budget|\s+around|$)", t) |
| if near: |
| cand = near.group(1).strip() |
| hit = next((s for s in SUBURBS if s in cand or cand in s), "") |
| if hit: |
| crit["suburb"] = hit.title() if hit != "cbd" else "CBD" |
| if not crit["suburb"]: |
| hit = next((s for s in SUBURBS if s in t), "") |
| if hit: |
| crit["suburb"] = hit.title() if hit != "cbd" else "CBD" |
|
|
| m = re.search(r"(under|below|up to|upto|max|maximum|around|about|budget)\s*\$?\s*(\d[\d,]*)", t) |
| if m: |
| crit["max_budget"] = float(m.group(2).replace(",", "")) |
| else: |
| rng = re.search(r"(\d[\d,]*)\s*[-\u2013]\s*(\d[\d,]*)", t) |
| if rng: |
| crit["min_budget"] = float(rng.group(1).replace(",", "")) |
| crit["max_budget"] = float(rng.group(2).replace(",", "")) |
| else: |
| nums = re.findall(r"\$\s*(\d[\d,]*)", t) |
| if nums: |
| crit["max_budget"] = float(nums[-1].replace(",", "")) |
| else: |
| _, hi = parse_budget(t) |
| |
| if hi and any(w in t for w in ["budget", "rent", "month", "$", "usd"]): |
| crit["max_budget"] = hi |
|
|
| crit["amenities"] = [a for a in AMEN_KEYS if a in t] |
| return crit |
|
|
|
|
| def looks_like_search(text: str) -> bool: |
| """True if free text carries any search signal at all.""" |
| c = parse_search(text) |
| t = (text or "").lower() |
| return bool(c["type"] or c["suburb"] or c["max_budget"] or c["amenities"] |
| or any(w in t for w in ["bedroom", "cottage", "room", "rent", "house", "flat", |
| "looking", "need", "want", "find", "search", "place"])) |
|
|
