app/main.py

"""
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("""<!doctype html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>RewardPilot Privacy Policy</title>
<style>body{font-family:-apple-system,Segoe UI,Roboto,sans-serif;max-width:720px;margin:40px auto;padding:0 20px;color:#1a1f27;line-height:1.6}h1{font-size:26px}h2{font-size:18px;margin-top:28px}small{color:#666}</style>
</head><body>
<h1>RewardPilot Privacy Policy</h1>
<small>Effective 6 July 2026</small>
<p>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.</p>
<h2>What we process, and where</h2>
<p><b>Card selections, preferences and imported transactions</b> are stored only on your device. They are not uploaded to our servers and we cannot see them.</p>
<p><b>Bank statements (PDF)</b> 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.</p>
<p><b>Location</b>, 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.</p>
<p><b>Voice queries</b>, if you use them, are transcribed on your device.</p>
<h2>What we do not do</h2>
<p>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.</p>
<h2>Third-party services</h2>
<p>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.</p>
<h2>Your rights and grievances</h2>
<p>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.</p>
<h2>Changes</h2>
<p>We will update this page when our practices change and revise the effective date above.</p>
</body></html>""")


@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 <issuer> 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-<token>@<inbound domain>; 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('_',' ')})."
        ),
    }