""" FastAPI app — Twilio WhatsApp webhook + health endpoint. Endpoints: GET /health — liveness check, returns engine + bot status POST /whatsapp — Twilio webhook (form-encoded); returns TwiML XML POST /chat — debug JSON endpoint (no Twilio); useful for curl Run locally: uvicorn whatsapp_bot.server:app --reload --port 8000 """ from __future__ import annotations import sys import os from contextlib import asynccontextmanager from typing import Any, Dict # Make project-root imports work when running `uvicorn whatsapp_bot.server:app` _HERE = os.path.dirname(os.path.abspath(__file__)) _ROOT = os.path.dirname(_HERE) if _ROOT not in sys.path: sys.path.insert(0, _ROOT) try: from fastapi import FastAPI, Form, Header, HTTPException, Query, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, Response except ImportError as e: raise RuntimeError( "fastapi not installed. Run: pip install -r requirements.txt" ) from e from matching_engine import LogisticsContext, run_matching from sample_data.loader import load_all_sample_data as _load_csv from sample_data.loader import load_real_data as _load_real # Precomputed data paths (resolved relative to project root so they work # both locally and inside the Docker container) _HERE_SERVER = os.path.dirname(os.path.abspath(__file__)) _PROJECT_ROOT = os.path.dirname(_HERE_SERVER) _ANOMALIES_PATH = os.path.join(_PROJECT_ROOT, "sample_data", "anomalies", "anomalies_all.json") _FORECASTS_PATH = os.path.join(_PROJECT_ROOT, "sample_data", "forecasts", "forecast_all.json") def _load_data_backend() -> dict: """ Select the data backend via the DATA_BACKEND env var. DATA_BACKEND=csv (default) — REAL BPS Jawa Timur 2022 data from sample_data/surplus_deficit_real.csv. Offline-safe. DATA_BACKEND=demo — the synthetic 19-commodity fixture. Test data only; never serve it to users. DATA_BACKEND=postgres — load from Supabase/Postgres via db.db_loader. WHY THE DEFAULT IS REAL DATA ---------------------------- This used to call load_all_sample_data(), whose default file is the synthetic surplus_deficit.csv. Every served response — dashboard map, WhatsApp reply, API — was therefore built on invented numbers while the real BPS-derived file sat unused beside it. That also caused a visible failure. The synthetic file prices rice demand in consumer cities at Rp16,400-17,000/kg against a 2022 farmgate-derived threshold whose 3-sigma ceiling is Rp15,100, so the D3 gate excluded EVERY rice deficit node and both beras commodities returned zero matches. On real data there are no anomaly exclusions at all, and matches go from 23 to 84. The synthetic fixture is still what 13 test files load directly, which is fine: it exercises engine logic across more commodities than the real data covers. It just must not be what users see. The Postgres path raises RuntimeError if SUPABASE_DB_URL is not set, so misconfiguration is loud rather than silent. """ import os backend = os.environ.get("DATA_BACKEND", "csv").strip().lower() if backend == "postgres": from db.db_loader import load_all as _load_pg return _load_pg() if backend == "demo": return _load_csv() # Default: real BPS data (offline-safe) return _load_real() from . import billing from .auth import AuthUser, GatedUser, RequireUser, auth_configured, require_auth_enabled from .config import settings from .gemini_client import GeminiClient from .handlers import ( MISSING_SLOT_PREFIX, OUT_OF_COVERAGE_PREFIX, EngineData, dispatch, ) from .intent import ( INTENT_ANOMALI, INTENT_CARI_PEMBELI, INTENT_CARI_PENJUAL, INTENT_FORECAST, INTENT_HARGA_LOOKUP, classify, ) from .subscription import SubscriptionService, hash_phone from .twilio_client import make_twiml_response, validate_signature # Intents that consume free-tier quota. The fallback intent is excluded on # purpose: a Gemini chit-chat answer is not the product, and charging for it # would let a vague question burn the user's daily allowance. METERED_INTENTS = frozenset({ INTENT_HARGA_LOOKUP, INTENT_CARI_PEMBELI, INTENT_CARI_PENJUAL, INTENT_FORECAST, INTENT_ANOMALI, }) # ============================================================================= # APP STATE — loaded once at startup # ============================================================================= class AppState: data: EngineData | None = None gemini: GeminiClient | None = None subs: SubscriptionService | None = None state = AppState() @asynccontextmanager async def lifespan(app: FastAPI): state.data = EngineData(_load_data_backend()) state.gemini = GeminiClient() state.subs = SubscriptionService() yield # No teardown needed app = FastAPI( title="AgriFlow WhatsApp Bot", version="0.1.0", description="Twilio webhook + Gemini RAG over the AgriFlow matching engine.", lifespan=lifespan, ) # CORS so the Next.js dashboard can hit /api/v1/* from dev (localhost) # and from any *.vercel.app preview / production URL. Regex covers branch # previews like agriflow-git-feature-x.vercel.app without re-deploys. # Also allows *.hf.space (Hugging Face Spaces) for direct curl/browser testing # against the API itself when it's hosted there. app.add_middleware( CORSMiddleware, allow_origins=[ "http://localhost:3000", "http://127.0.0.1:3000", ], allow_origin_regex=r"https://.*\.(vercel\.app|hf\.space)", allow_methods=["GET", "POST", "OPTIONS"], allow_headers=["*"], ) # ============================================================================= # CORE — pure function (no Twilio coupling), reused by /whatsapp and /chat # ============================================================================= def _ensure_state() -> None: """Eager init for non-FastAPI callers (CLI, tests not using TestClient).""" if state.data is None: state.data = EngineData(_load_data_backend()) if state.gemini is None: state.gemini = GeminiClient() if state.subs is None: state.subs = SubscriptionService() def handle_message(message: str, sender: str | None = None) -> str: """ Pure pipeline: text in → text out. Easy to unit-test. `sender` is the raw WhatsApp identifier ('whatsapp:+62...'). When it is absent the message is treated as anonymous and no quota is applied — that is the debug path (/chat, CLI). The Twilio webhook always passes a sender, so real users are always metered. Order of operations matters here: 1. Billing/help commands run first and are never metered, so a user at their limit can still reach STATUS and UPGRADE. 2. The quota check runs before dispatch, so an over-limit user gets the upgrade offer instead of an answer. 3. Quota is consumed only *after* a metered intent produced a real answer, so incomplete questions cost nothing. """ _ensure_state() assert state.data is not None and state.gemini is not None and state.subs is not None phone_hash = hash_phone(sender) if sender else "" # 1. Commands — free, and available even at zero remaining quota. # Skipped entirely when the paywall is off: with no quota there is no # billing surface, and a STATUS reply quoting a limit nobody enforces # would be a lie. if settings.quota_enabled and phone_hash: command = billing.parse_command(message) if command is not None: return billing.handle_command(command, phone_hash, state.subs) intent = classify( message, state.gemini, state.data.kabupaten, state.data.komoditas, ) metered = ( settings.quota_enabled and bool(phone_hash) and intent.name in METERED_INTENTS ) # 2. Paywall. if metered: decision = state.subs.check(phone_hash) if not decision.allowed: order = state.subs.start_upgrade(phone_hash) return billing.quota_exceeded(decision, order) reply = dispatch(intent, state.data, state.gemini) # 3. Bill only a query we actually answered. Asking the user to rephrase # is free, and so is telling them a commodity is outside our data — # neither delivered the thing they asked for. if metered and not reply.startswith((MISSING_SLOT_PREFIX, OUT_OF_COVERAGE_PREFIX)): state.subs.consume(phone_hash) return reply # ============================================================================= # ROUTES # ============================================================================= @app.get("/health") async def health() -> Dict[str, Any]: data_loaded = state.data is not None return { "status": "ok", "version": "0.1.0", "mock_mode": settings.mock_mode, "data_loaded": data_loaded, "kabupaten_count": len(state.data.kabupaten) if data_loaded else 0, "komoditas_count": len(state.data.komoditas) if data_loaded else 0, "gemini_mock": state.gemini.mock if state.gemini else None, "auth_configured": auth_configured(), "require_auth": require_auth_enabled(), "quota_enabled": settings.quota_enabled, "free_daily_quota": settings.free_daily_quota, "quota_backend": settings.quota_backend, "billing_mock": settings.billing_mock, # Surfaced so a deployment check can catch an unsalted hash without # exposing the salt itself. "phone_hash_salted": bool(settings.phone_hash_salt), } @app.post("/whatsapp") async def whatsapp_webhook( request: Request, Body: str = Form(...), From: str = Form(...), x_twilio_signature: str | None = Header(default=None), ) -> Response: """ Twilio WhatsApp webhook entrypoint. Body — message text from user From — 'whatsapp:+62xxx' sender Returns TwiML XML that Twilio will send back to the user. """ # Optional signature validation — enable once webhook is reachable from Twilio if settings.twilio_validate_signature and not settings.mock_mode: form = await request.form() url = str(request.url) if not validate_signature( settings.twilio_auth_token, x_twilio_signature or "", url, form, ): raise HTTPException(status_code=403, detail="Invalid Twilio signature") reply = handle_message(Body, sender=From) return Response(content=make_twiml_response(reply), media_type="application/xml") @app.post("/chat") async def chat_debug(payload: Dict[str, str]) -> JSONResponse: """ Debug endpoint — bypasses Twilio. Useful for local curl testing: curl -X POST localhost:8000/chat -H 'Content-Type: application/json' \\ -d '{"message": "Harga cabai di Malang"}' Pass an optional "from" field to exercise the quota flow end to end: -d '{"message": "Harga cabai di Malang", "from": "whatsapp:+628123"}' WITHOUT "from" this endpoint is unmetered, so it bypasses the paywall by design. Set DEBUG_CHAT_ENABLED=false in any deployment where that matters — the Twilio webhook is the metered path, this one is a development tool. """ if not settings.debug_chat_enabled: raise HTTPException(status_code=404, detail="Not found") message = payload.get("message", "").strip() if not message: raise HTTPException(status_code=400, detail="message field required") reply = handle_message(message, sender=payload.get("from")) return JSONResponse({"reply": reply}) # ============================================================================= # BILLING — upgrade flow # # The payment page and confirm endpoint form the gateway seam. In mock mode # they are self-contained; to go live, point PUBLIC_BASE_URL's payment link at # Midtrans/Xendit and have their webhook POST /billing/confirm with the order id # after verifying the provider signature. # ============================================================================= def _ensure_subs() -> SubscriptionService: if state.subs is None: state.subs = SubscriptionService() return state.subs @app.get("/billing/pay/{order_id}") async def billing_pay_page(order_id: str) -> Response: """ Minimal payment page the WhatsApp link opens. In mock mode this renders a confirm button that settles the order. With a real gateway this route would instead redirect to the provider's hosted checkout for this order. """ subs = _ensure_subs() order = subs.store.get_order(order_id) if order is None: return Response( content="

Pesanan tidak ditemukan

" "

Silakan balas UPGRADE di WhatsApp untuk membuat pesanan baru.

", media_type="text/html", status_code=404, ) amount = f"Rp {order.amount_idr:,.0f}".replace(",", ".") if order.status == "PAID": body = "

Pesanan ini sudah dibayar. Akun Anda sudah PRO.

" elif billing.billing_mock_enabled(): body = ( f"
" f"" f"
" f"

Mode demo — tidak ada transaksi sungguhan.

" ) else: body = "

Menunggu pengalihan ke penyedia pembayaran.

" return Response( content=( "" "" "AgriFlow PRO" "" f"

AgriFlow PRO

Pesanan {order.order_id}
" f"Jumlah {amount} untuk 30 hari

{body}" ), media_type="text/html", ) @app.post("/billing/confirm") async def billing_confirm(request: Request) -> Response: """ Settle an order and grant PRO — the gateway webhook seam. Accepts either form-encoded (the mock page) or JSON (a webhook). A real integration MUST verify the provider's signature here before trusting the order id; right now anyone who knows an order id can settle it, which is acceptable only because mock mode charges nothing. """ subs = _ensure_subs() ctype = request.headers.get("content-type", "") if "application/json" in ctype: payload = await request.json() order_id = str(payload.get("order_id", "")) else: form = await request.form() order_id = str(form.get("order_id", "")) if not order_id: raise HTTPException(status_code=400, detail="order_id required") if not billing.billing_mock_enabled(): raise HTTPException( status_code=501, detail="Live payment confirmation is not wired yet. " "Implement provider signature verification before enabling.", ) account = subs.confirm_payment(order_id) if account is None: raise HTTPException(status_code=404, detail=f"unknown order: {order_id}") if "application/json" in ctype: return JSONResponse({ "status": "ok", "plan": account.plan, "expires_at": account.expires_at.isoformat() if account.expires_at else None, }) return Response( content="" "" "

Pembayaran berhasil

" "

Akun WhatsApp Anda sekarang PRO selama 30 hari. " "Silakan kembali ke WhatsApp dan lanjutkan bertanya.

", media_type="text/html", ) @app.get("/billing/status") async def billing_status( phone: str = Query(..., description="WhatsApp number, e.g. +628123456789"), user: AuthUser = RequireUser, ) -> JSONResponse: """ Plan + remaining quota for one number. Powers the dashboard account panel. The number is hashed before lookup and never stored by this call. """ subs = _ensure_subs() phone_hash = hash_phone(phone) if not phone_hash: raise HTTPException(status_code=400, detail="invalid phone number") decision = subs.check(phone_hash) return JSONResponse({ "plan": decision.account.plan, "is_pro": decision.account.is_pro, "expires_at": ( decision.account.expires_at.isoformat() if decision.account.expires_at else None ), "used_today": decision.used_today, "limit": decision.limit, "remaining": decision.remaining, }) # ============================================================================= # DASHBOARD API — /api/v1/* (consumed by Next.js dashboard) # ============================================================================= def _ensure_engine() -> EngineData: if state.data is None: state.data = EngineData(_load_data_backend()) return state.data # Cached full engine run. # # run_matching() is a pure function of the data loaded at startup, so re-running # it per request was burning ~1.5 ms of CPU to recompute a byte-identical # answer — about 60x the cost of everything else in the request and the binding # constraint on how many concurrent users one worker can serve. # # The cache is keyed on the EngineData object itself (not id(), which a garbage # collector can recycle onto a different object). Reloading data rebinds # state.data to a new instance, which misses the cache and recomputes. _matching_cache: Dict[str, Any] = {"data": None, "report": None} def _cached_report(): data = _ensure_engine() if _matching_cache["data"] is not data: _matching_cache["report"] = run_matching( surplus_nodes=data.surplus, deficit_nodes=data.deficit, logistics=LogisticsContext(), weather_forecasts=data.weather, historical_prices=data.historical, ) _matching_cache["data"] = data return _matching_cache["report"] @app.get("/api/v1/commodities") async def api_commodities() -> JSONResponse: data = _ensure_engine() out = [ {"code": c.code, "nama": c.nama} for c in sorted(data.komoditas.values(), key=lambda c: c.nama) ] return JSONResponse(out) @app.get("/api/v1/kabupaten") async def api_kabupaten() -> JSONResponse: data = _ensure_engine() out = [ { "id": k.id, "nama": k.nama, "lat": k.latitude, "lng": k.longitude, "tier": k.tier.value, "ipm": k.ipm, "population": k.population, } for k in sorted(data.kabupaten.values(), key=lambda k: k.nama) ] return JSONResponse(out) @app.get("/api/v1/surplus-deficit") async def api_surplus_deficit( commodity: str = Query(..., description="Commodity code, e.g. cabai_merah"), ) -> JSONResponse: """Per-kab surplus/deficit volume for one commodity — powers the map bubbles.""" data = _ensure_engine() if commodity not in data.komoditas: raise HTTPException(status_code=404, detail=f"unknown commodity: {commodity}") commo = data.komoditas[commodity] rows = [] for s in data.surplus: if s.commodity.code != commodity: continue rows.append({ "kab_id": s.kabupaten.id, "kab_nama": s.kabupaten.nama, "lat": s.kabupaten.latitude, "lng": s.kabupaten.longitude, "tier": s.kabupaten.tier.value, "role": "surplus", "volume_tons": s.volume_tons, "price_per_kg": s.price_per_kg, }) for d in data.deficit: if d.commodity.code != commodity: continue rows.append({ "kab_id": d.kabupaten.id, "kab_nama": d.kabupaten.nama, "lat": d.kabupaten.latitude, "lng": d.kabupaten.longitude, "tier": d.kabupaten.tier.value, "role": "deficit", "volume_tons": d.volume_tons, "price_per_kg": d.price_per_kg, }) total_surplus = sum(r["volume_tons"] for r in rows if r["role"] == "surplus") total_deficit = sum(r["volume_tons"] for r in rows if r["role"] == "deficit") return JSONResponse({ "commodity": {"code": commo.code, "nama": commo.nama}, "rows": rows, "totals": { "surplus_tons": total_surplus, "deficit_tons": total_deficit, "balance_tons": total_surplus - total_deficit, }, }) def _serialize_match(m) -> Dict[str, Any]: return { "surplus": { "kab_id": m.surplus.kabupaten.id, "kab_nama": m.surplus.kabupaten.nama, "lat": m.surplus.kabupaten.latitude, "lng": m.surplus.kabupaten.longitude, "price_per_kg": m.surplus.price_per_kg, }, "deficit": { "kab_id": m.deficit.kabupaten.id, "kab_nama": m.deficit.kabupaten.nama, "lat": m.deficit.kabupaten.latitude, "lng": m.deficit.kabupaten.longitude, "price_per_kg": m.deficit.price_per_kg, }, "commodity_code": m.surplus.commodity.code, "commodity_nama": m.surplus.commodity.nama, "matched_volume_tons": m.matched_volume_tons, "distance_km": m.distance_km, "final_score": m.final_score, "confidence": m.confidence.value, "flags": list(m.flags), } @app.get("/api/v1/matches") async def api_matches( user: AuthUser | None = GatedUser, commodity: str | None = Query(None, description="Filter by commodity code"), kab_id: str | None = Query(None, description="Filter where this kab is surplus OR deficit side"), limit: int = Query(50, ge=1, le=500), ) -> JSONResponse: """Serve scored matches for map flow lines + side panel, from a cached engine run.""" report = _cached_report() # Copy before sorting. `report.matches` is the shared cached list, and an # unfiltered request would otherwise sort it in place under every other # concurrent caller. matches = list(report.matches) if commodity: matches = [m for m in matches if m.surplus.commodity.code == commodity] if kab_id: matches = [ m for m in matches if m.surplus.kabupaten.id == kab_id or m.deficit.kabupaten.id == kab_id ] matches.sort(key=lambda m: m.final_score, reverse=True) matches = matches[:limit] return JSONResponse({ "count": len(matches), "matches": [_serialize_match(m) for m in matches], }) # ============================================================================= # FORECAST + ANOMALY API -- /api/v1/forecast and /api/v1/anomalies # # Both endpoints serve precomputed JSON files that were generated offline by: # python analysis/precompute_anomalies.py # python analysis/forecast_timesfm.py # # The server NEVER imports timesfm at runtime (HF Space OOM guard). # ============================================================================= import json as _json import functools @functools.lru_cache(maxsize=1) def _load_forecasts() -> list: """Load forecast_all.json once and cache in-process.""" if not os.path.exists(_FORECASTS_PATH): return [] with open(_FORECASTS_PATH, encoding="utf-8") as fh: return _json.load(fh) @functools.lru_cache(maxsize=1) def _load_anomalies() -> list: """Load anomalies_all.json once and cache in-process.""" if not os.path.exists(_ANOMALIES_PATH): return [] with open(_ANOMALIES_PATH, encoding="utf-8") as fh: return _json.load(fh) @app.get("/api/v1/forecast") async def api_forecast( user: AuthUser | None = GatedUser, commodity: str = Query(..., description="Commodity code, e.g. cabai_rawit"), city: str = Query(..., description="IHK city_id, e.g. 3578 (Surabaya)"), ) -> JSONResponse: """ 30-day price forecast (point + P10/P90) for one commodity × city pair. Data is precomputed offline (seasonal-naive baseline unless TimesFM was available at precompute time). The 'method' field in the response tells you which model was used. Query params: commodity AgriFlow commodity code (e.g. cabai_rawit, bawang_merah) city IHK city_id (e.g. 3578 for Kota Surabaya) Response schema: commodity_code str city_id str city_name str method str ("timesfm_2.0" | "seasonal_naive_baseline") generated_at str ISO 8601 horizon_days int history_end_date str ISO 8601 forecasts list of {date, point, p10, p90} """ records = _load_forecasts() if not records: raise HTTPException( status_code=503, detail=( "Forecast data not yet precomputed. " "Run: python analysis/forecast_timesfm.py" ), ) match = next( (r for r in records if r["commodity_code"] == commodity and r["city_id"] == city), None, ) if match is None: # List available (commodity, city) pairs so caller can self-correct available = sorted({(r["commodity_code"], r["city_id"]) for r in records}) raise HTTPException( status_code=404, detail={ "error": f"No forecast for commodity={commodity!r} city={city!r}", "available_pairs": [{"commodity": c, "city": ci} for c, ci in available[:20]], }, ) return JSONResponse(match) @app.get("/api/v1/anomalies") async def api_anomalies( user: AuthUser | None = GatedUser, commodity: str | None = Query(None, description="Filter by commodity code"), city: str | None = Query(None, description="Filter by IHK city_id"), limit: int = Query(50, ge=1, le=500, description="Max records returned (sorted by score desc)"), since: str | None = Query(None, description="ISO date lower-bound, e.g. 2024-01-01"), ) -> JSONResponse: """ Detected price anomalies from the S-H-ESD scanner (precomputed offline). All filters are optional. Without filters returns top-N anomalies by score. Query params: commodity optional commodity code filter city optional IHK city_id filter limit max records (default 50, max 500) since ISO date — only return anomalies on or after this date Response schema: count int method str ("shesd_v2") anomalies list of { date str ISO 8601 price float IDR/kg rolling_median float deviation_pct float (positive = spike, negative = drop) type str SPIKE | DROP score float (higher = more anomalous) commodity_code str city_id str city_name str persistent bool } """ records = _load_anomalies() if not records: raise HTTPException( status_code=503, detail=( "Anomaly data not yet precomputed. " "Run: python analysis/precompute_anomalies.py" ), ) filtered = records if commodity: filtered = [r for r in filtered if r["commodity_code"] == commodity] if city: filtered = [r for r in filtered if r["city_id"] == city] if since: filtered = [r for r in filtered if r["date"] >= since] # Already sorted by score desc in the precomputed file; slice to limit filtered = filtered[:limit] return JSONResponse({ "count": len(filtered), "method": "shesd_v2", "anomalies": filtered, }) # ============================================================================= # CLI helper: python -m whatsapp_bot.server "Harga cabai di Malang" # ============================================================================= def _cli_main() -> None: # Force UTF-8 stdout on Windows so emoji in replies don't crash cp1252 consoles if sys.platform == "win32": try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass if len(sys.argv) < 2: print("Usage: python -m whatsapp_bot.server \"\" [--from +628123]") sys.exit(1) argv = sys.argv[1:] sender = None if "--from" in argv: i = argv.index("--from") sender = argv[i + 1] if i + 1 < len(argv) else None argv = argv[:i] + argv[i + 2:] msg = " ".join(argv) print(handle_message(msg, sender=sender)) if __name__ == "__main__": _cli_main()