"""
RewardPilot API
===============
FastAPI backend powering the RewardPilot mobile app.
Run locally:
cd backend
pip install -r requirements.txt
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
Endpoints:
GET /health
GET /cards -> full catalogue
POST /recommend -> optimal card for a text/structured query
POST /voice -> audio -> transcript -> recommendation
POST /statements/parse -> upload PDF/CSV -> normalized transactions
POST /transactions/analyze -> were optimal cards used? how much lost?
GET /discover/{card_id} -> apply-for details + extra benefits
"""
import os
import re
import sys
from typing import List, Optional
# allow `from card_catalogue import ...` style imports inside app/
sys.path.insert(0, os.path.dirname(__file__))
# load OPENAI_API_KEY (and others) from backend/.env before importing llm
try:
from dotenv import load_dotenv
load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))
except ImportError:
pass
import anyio
import asyncio
import time as _time
from fastapi import FastAPI, File, Form, UploadFile, Header, HTTPException, Depends, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from card_catalogue import all_cards, get_card
from merchants import resolve_merchant
from scoring_engine import TxnContext, score_transaction, analyse_transaction_history
import llm
import demo_data
# Optional app-key gate. If APP_KEY is set as a Space secret, sensitive endpoints
# (statement upload, recommend, voice) require a matching X-App-Key header. If
# APP_KEY is unset (dev), they stay open. Basic abuse protection for the mobile
# client, not end-user authentication.
APP_KEY = os.getenv("APP_KEY")
def require_app_key(x_app_key: Optional[str] = Header(default=None, alias="X-App-Key")):
if APP_KEY and x_app_key != APP_KEY:
raise HTTPException(status_code=401, detail="invalid_app_key")
# ---------------------------------------------------------------------------
# Overload guards (per worker process).
# - Token-bucket rate limit per client IP on the expensive endpoints, so one
# runaway client can't drain the LLM budget or monopolise parsing.
# - Parse semaphore: statement parsing is CPU-bound; beyond the limit callers
# wait briefly, then get 429 "busy" instead of an opaque timeout.
# ---------------------------------------------------------------------------
_BUCKETS: dict = {}
def _rate_ok(key: str, rate: float, per: float = 60.0) -> bool:
now = _time.time()
tokens, last = _BUCKETS.get(key, (rate, now))
tokens = min(rate, tokens + (now - last) * rate / per)
if tokens < 1.0:
_BUCKETS[key] = (tokens, now)
return False
_BUCKETS[key] = (tokens - 1.0, now)
if len(_BUCKETS) > 10000: # bound memory under address churn
_BUCKETS.clear()
return True
def _client_ip(request: Request) -> str:
fwd = request.headers.get("x-forwarded-for")
return (fwd.split(",")[0].strip() if fwd else None) or (request.client.host if request.client else "unknown")
def _limit(request: Request, path: str, rate: float):
if not _rate_ok(f"{_client_ip(request)}|{path}", rate):
raise HTTPException(status_code=429, detail="rate_limited")
_PARSE_SEM = asyncio.Semaphore(3) # concurrent parses per worker
_PARSE_WAIT_S = 20.0 # max time a caller queues before 429
app = FastAPI(title="RewardPilot API", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# ---------------------------------------------------------------------------
# Models
# ---------------------------------------------------------------------------
class RecommendRequest(BaseModel):
query: Optional[str] = None # free text, e.g. "iPhone at Croma 80000"
held_cards: Optional[List[str]] = None
category: Optional[str] = None
amount: Optional[float] = None
merchant: Optional[str] = None
persona: Optional[List[str]] = None
rail: Optional[str] = None # "card" | "upi"; if None, inferred from query
city: Optional[str] = None # e.g. "Mumbai"; overrides query/GPS detection
lat: Optional[float] = None # GPS fallback for city detection
lng: Optional[float] = None
class AnalyzeRequest(BaseModel):
held_cards: Optional[List[str]] = None
transactions: Optional[List[dict]] = None
def _card_public(card) -> dict:
f = card.fees
return {
"id": card.id,
"name": card.name,
"issuer": card.issuer,
"network": card.network,
"segment": card.segment,
"kind": getattr(card, "kind", "credit"),
"reward_unit": card.reward_unit,
"point_value_inr": card.point_value_inr,
"highlights": card.highlights,
"annual_fee_text": card.annual_fee_text,
"fees": {"joining": f.joining_fee, "annual": f.annual_fee, "waiver_spend": f.fee_waiver_spend} if f else None,
"persona_fit": card.persona_fit,
}
# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------
@app.get("/privacy")
def privacy():
"""Hosted privacy policy (required by Google Play for location + financial data)."""
from fastapi.responses import HTMLResponse
return HTMLResponse("""
RewardPilot Privacy Policy
RewardPilot Privacy Policy
Effective 6 July 2026
RewardPilot recommends the most rewarding credit card and booking platform for each purchase. This policy explains what data the app handles and how. It is written to comply with India's Digital Personal Data Protection Act, 2023.
What we process, and where
Card selections, preferences and imported transactions are stored only on your device. They are not uploaded to our servers and we cannot see them.
Bank statements (PDF) that you choose to import are sent to our server over an encrypted connection, parsed in memory, and returned to your device as transaction summaries. Raw statement files are not written to disk, stored, or retained after parsing. CSV statements never leave your device.
Location, if you grant it, is used to identify a nearby store and suggest the right card. Coordinates are used for the lookup and are not stored.
Voice queries, if you use them, are transcribed on your device.
What we do not do
We do not collect card numbers, account numbers, or credentials. We do not sell or share personal data with advertisers. We do not run third-party advertising or tracking SDKs.
Third-party services
Statement parsing and query narration run on our servers hosted with Hugging Face. Store lookups use Google Places. Card application links open the issuing bank's own website; commissions we may earn do not change our recommendations, which are computed from published reward rates.
Your rights and grievances
You can delete all app data at any time from the Wallet tab (Clear all), or by uninstalling the app; because data lives on your device, deletion is immediate and complete. For questions, corrections, or grievances under the DPDP Act, contact the Grievance Officer: Salim Shaikh, statsguysalim@gmail.com. We respond within 30 days.
Changes
We will update this page when our practices change and revise the effective date above.
""")
@app.get("/health")
def health():
from places import GOOGLE_PLACES_KEY, FSQ_API_KEY
return {
"status": "ok",
"stt": bool(os.getenv("OPENAI_API_KEY")),
"llm": bool(os.getenv("ANTHROPIC_API_KEY") or os.getenv("OPENAI_API_KEY")),
"places": bool(GOOGLE_PLACES_KEY or FSQ_API_KEY),
"cards": len(all_cards()),
}
@app.get("/cards")
def cards():
return {"cards": [_card_public(c) for c in all_cards()]}
@app.get("/offers")
def offers():
"""All currently-active offers (dated). Fed by the ingestion connectors."""
from offers import all_active
items = all_active()
return {"count": len(items), "offers": items}
@app.get("/offers/sources")
def offer_sources():
"""Live view of the ingestion pipeline: which connectors feed the offers
store and how many offers each currently contributes."""
from offers import all_active
by_source: dict = {}
for o in all_active():
by_source[o["source"]] = by_source.get(o["source"], 0) + 1
connectors = [
{"family": "issuer", "sources": ["hdfc_smartbuy", "amex_offers", "sbi", "axis"]},
{"family": "clo", "sources": ["fidel", "kard"]},
{"family": "affiliate", "sources": ["cashkaro", "grabon", "bank_partner"]},
{"family": "platform", "sources": ["dineout", "district"]},
]
return {"connectors": connectors, "active_by_source": by_source, "total_active": sum(by_source.values())}
@app.get("/offers/{merchant_key}")
def offers_for(merchant_key: str):
from offers import active_offers_for
items = active_offers_for(merchant_key)
return {"merchant_key": merchant_key, "count": len(items), "offers": items}
@app.get("/places/nearby")
def places_nearby(lat: float, lng: float, radius: float = 150):
"""Resolve nearby known-merchant candidates from coordinates (Google/Foursquare,
key held server-side). Empty if no provider key is configured."""
from places import nearby
return nearby(lat, lng, radius)
@app.get("/demo")
def demo():
"""Everything the app needs to demo with zero setup."""
return {
"wallet": demo_data.DEMO_WALLET,
"persona": demo_data.DEMO_PERSONA,
"transactions": demo_data.DEMO_TRANSACTIONS,
}
@app.post("/recommend", dependencies=[Depends(require_app_key)])
def recommend(req: RecommendRequest, request: Request = None):
if request is not None:
_limit(request, "rec", rate=90) # 90/min per IP per worker
held = req.held_cards or demo_data.DEMO_WALLET
persona = req.persona or demo_data.DEMO_PERSONA
from merchants import detect_rail
if req.query:
intent = llm.parse_intent(req.query)
category = intent["category"]
amount = intent.get("amount") or req.amount or 5000
merchant = intent.get("merchant")
offers = intent.get("offers") or []
brand_key = intent.get("brand_key")
rail = req.rail or detect_rail(req.query) or "card"
else:
category = req.category or "general"
amount = req.amount or 5000
merchant = req.merchant
resolved = resolve_merchant(merchant or "")
offers = resolved["offers"]
brand_key = resolved["brand_key"]
rail = req.rail or "card"
ctx = TxnContext(category=category, amount=float(amount), merchant=merchant,
brand_key=brand_key, offers=offers, rail=rail)
result = score_transaction(held, ctx, include_discovery=True, persona=persona)
result["narration"] = llm.narrate_recommendation(
result["context"], result["held_ranked"], result["discovery"]
)
# City: explicit request field > city named in the query > nearest metro from GPS.
from availability import normalize_city, extract_city, city_from_latlng
city = normalize_city(req.city)
if not city and req.query:
city = extract_city(req.query)
if not city and req.lat is not None and req.lng is not None:
city = city_from_latlng(req.lat, req.lng)
result["intent"] = {"category": category, "amount": amount, "merchant": merchant, "rail": rail, "city": city}
# Dine-in / movies / travel: same purchase, several booking apps. Rank the
# channels too (best card PER channel), so the answer is "pay via District
# with your Axis card", not just "use your Axis card".
from channels import compare_channels
result["channel_comparison"] = compare_channels(
held, category, float(amount), brand_key=brand_key, merchant=merchant,
rail=rail, persona=persona, city=city,
)
return result
@app.post("/voice", dependencies=[Depends(require_app_key)])
async def voice(audio: UploadFile = File(...), held_cards: Optional[str] = Form(None)):
raw = await audio.read()
transcript = llm.transcribe(raw, filename=audio.filename or "audio.m4a")
held = held_cards.split(",") if held_cards else demo_data.DEMO_WALLET
rec = recommend(RecommendRequest(query=transcript, held_cards=held))
rec["transcript"] = transcript
return rec
@app.post("/statements/parse", dependencies=[Depends(require_app_key)])
async def parse_statements(
request: Request,
files: List[UploadFile] = File(...),
password: Optional[str] = Form(None),
):
_limit(request, "parse", rate=12) # 12/min per IP per worker
from statement_parser import parse_statement, detect_cards, detect_issuers, find_pdf_password
# Smart unlock: the password field may carry MULTIPLE candidates (newline or
# comma separated), derived app-side from the user's details the way every
# Indian issuer's statement password is (name4+DDMM and friends).
pw_candidates = [p.strip() for p in re.split(r"[\n,]", password or "") if p.strip()][:32]
# queue politely for a slot; overload gets a clear 429 instead of a timeout
try:
await asyncio.wait_for(_PARSE_SEM.acquire(), timeout=_PARSE_WAIT_S)
except asyncio.TimeoutError:
raise HTTPException(status_code=429, detail="busy_try_again")
try:
all_txns = []
all_reversals = []
errors = []
detected: List[str] = []
issuers: List[str] = []
for f in files:
content = await f.read()
# Resolve THE working password for this file up front (cheap pypdf
# probe), so a candidate list costs one parse, not N parses.
pw: Optional[str] = pw_candidates[0] if len(pw_candidates) == 1 else None
if f.filename.lower().endswith(".pdf") and pw_candidates:
found = await anyio.to_thread.run_sync(
lambda c=content: find_pdf_password(c, pw_candidates))
pw = found if found else pw
try:
# run the CPU-bound parse in a thread so this worker's event loop
# keeps serving /health and /offers while parsing
revs: list = []
txns = await anyio.to_thread.run_sync(
lambda c=content, n=f.filename, p=pw, r=revs: parse_statement(n, c, password=p, reversals=r)
)
for t in txns:
t["source_file"] = f.filename
for r in revs:
r["source_file"] = f.filename
all_txns.extend(txns)
all_reversals.extend(revs)
for cid in await anyio.to_thread.run_sync(
lambda c=content, n=f.filename, p=pw: detect_cards(n, c, password=p)
):
if cid not in detected:
detected.append(cid)
# issuer fallback: even when no exact card matched, knowing the
# bank lets the app ask "which of your cards?"
for iss in await anyio.to_thread.run_sync(
lambda c=content, n=f.filename, p=pw: detect_issuers(n, c, password=p)
):
if iss not in issuers:
issuers.append(iss)
except Exception as e:
errors.append({"file": f.filename, "error": str(e)})
# A locked/broken file can still identify its card from the
# FILENAME ("Millennia_Statement_Jun.pdf") - never lose that.
try:
for cid in detect_cards(f.filename, b""):
if cid not in detected:
detected.append(cid)
for iss in detect_issuers(f.filename, b""):
if iss not in issuers:
issuers.append(iss)
except Exception:
pass
return {"transactions": all_txns, "count": len(all_txns), "errors": errors,
"detected_cards": detected, "detected_issuers": issuers, "reversals": all_reversals}
finally:
_PARSE_SEM.release()
# ---------------------------------------------------------------------------
# Email import (phase 1: forwarding, no inbox OAuth). The user forwards bank
# statement emails to their private address rp-@; the
# email provider (Mailgun-style route) POSTs the message here; attachments are
# parsed IN MEMORY like every other statement and held briefly until the app
# collects them with the same token. Nothing is written to disk.
# ---------------------------------------------------------------------------
_INBOX: dict = {}
_INBOX_MAX_ITEMS = 20
_INBOX_MAX_TOKENS = 500
def _inbox_token(addr: str) -> Optional[str]:
local = (addr or "").split("@")[0].strip().lower()
if local.startswith("rp-"):
local = local[3:]
return local if 8 <= len(local) <= 64 and local.replace("-", "").isalnum() else None
@app.post("/email/inbound")
async def email_inbound(request: Request):
"""Inbound-email webhook (provider-agnostic: reads recipient/to plus any file
fields from the multipart form)."""
_limit(request, "inbound", rate=30)
from statement_parser import parse_statement, detect_cards
form = await request.form()
token = _inbox_token(str(form.get("recipient") or form.get("to") or ""))
if not token:
raise HTTPException(status_code=400, detail="unknown_recipient")
results = []
for key in form:
val = form[key]
if not hasattr(val, "filename") or not getattr(val, "filename", None):
continue
try:
content = await val.read()
revs: list = []
txns = await anyio.to_thread.run_sync(
lambda c=content, n=val.filename, r=revs: parse_statement(n, c, reversals=r))
cards = await anyio.to_thread.run_sync(
lambda c=content, n=val.filename: detect_cards(n, c))
for t in txns:
t["source_file"] = val.filename
results.append({"file": val.filename, "count": len(txns),
"transactions": txns, "detected_cards": cards, "reversals": revs})
except Exception as e:
msg = str(e)
short = "PASSWORD_REQUIRED" if "PASSWORD_REQUIRED" in msg else msg[:120]
results.append({"file": val.filename, "count": 0, "transactions": [],
"detected_cards": [], "error": short})
if results:
if len(_INBOX) >= _INBOX_MAX_TOKENS and token not in _INBOX:
_INBOX.clear() # bound memory; queue is best-effort by design
q = _INBOX.setdefault(token, [])
q.extend(results)
del q[:-_INBOX_MAX_ITEMS]
return {"accepted": len(results)}
@app.get("/email/pending")
def email_pending(request: Request, token: str):
"""The app collects (and clears) whatever its forwarding address received."""
_limit(request, "pending", rate=30)
tok = _inbox_token(token)
items = _INBOX.pop(tok, []) if tok else []
return {"items": items, "count": sum(i.get("count", 0) for i in items)}
@app.post("/transactions/analyze")
def analyze(req: AnalyzeRequest):
held = req.held_cards or demo_data.DEMO_WALLET
txns = req.transactions or demo_data.DEMO_TRANSACTIONS
return analyse_transaction_history(held, txns)
@app.get("/discover/{card_id}")
def discover(card_id: str):
card = get_card(card_id)
if not card:
return {"error": "card_not_found"}
from scoring_engine import apply_url_for
return {
"card": _card_public(card),
"apply_url": apply_url_for(card_id), # official issuer page (affiliate deep link later)
"extra_benefits": card.highlights,
"why": (
f"Based on your spending pattern, {card.name} would beat your current "
f"best card in its strong categories ({', '.join(list(card.category_rates.keys())[:3]).replace('_',' ')})."
),
}