Create APP.PY

APP.PY

@@ -0,0 +1,274 @@
+import os, io, base64, json, time, random
+from typing import Optional, Dict, Any, List, Tuple
+from urllib.parse import quote_plus
+
+from fastapi import FastAPI, Request, BackgroundTasks
+from fastapi.responses import PlainTextResponse
+import httpx
+from bs4 import BeautifulSoup
+from PIL import Image
+
+from twilio.rest import Client as TwilioClient
+
+# OpenAI
+from openai import OpenAI
+oai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+
+# LangChain minimal (structured output)
+from langchain_openai import ChatOpenAI
+from langchain_core.pydantic_v1 import BaseModel, Field
+from langchain_core.prompts import ChatPromptTemplate
+
+# Search tools
+from duckduckgo_search import DDGS
+try:
+    from tavily import TavilyClient
+    _HAS_TAVILY = True
+except Exception:
+    _HAS_TAVILY = False
+
+app = FastAPI(title="SAVE SMS Webhook (Async Reply)")
+
+# ---------- Twilio client ----------
+TW_SID = os.getenv("TWILIO_ACCOUNT_SID", "")
+TW_TOKEN = os.getenv("TWILIO_AUTH_TOKEN", "")
+TW_FROM = os.getenv("TWILIO_FROM", "")  # e.g., +12175898085
+_twilio_ok = bool(TW_SID and TW_TOKEN and TW_FROM)
+twilio_client = TwilioClient(TW_SID, TW_TOKEN) if _twilio_ok else None
+
+# ---------- shared helpers ----------
+lc_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
+
+MERCHANT_DOMAINS = (
+    "walmart.ca","realcanadiansuperstore.ca","amazon.ca","metro.ca",
+    "nofrills.ca","freshco.com","well.ca","costco.ca","iga.net","londondrugs.com"
+)
+
+class Offer(BaseModel):
+    merchant: str = Field(...)
+    title: str = Field(...)
+    price: float = Field(...)
+    url: str = Field(...)
+
+extract_prompt = ChatPromptTemplate.from_messages([
+    ("system","Extract one best CAD offer for the queried item from the page text. "
+              "Return JSON: merchant,title,price(float),url. If none, return empty with price 0."),
+    ("human","Query: {query}\nURL: {url}\n--- PAGE TEXT ---\n{text}\n--- END ---")
+])
+chain_extract = extract_prompt | lc_llm.with_structured_output(Offer)
+
+def img_or_pdf_to_image_bytes(data: bytes, filename: str) -> bytes:
+    name = (filename or "").lower()
+    if name.endswith((".jpg",".jpeg",".png",".webp")):
+        img = Image.open(io.BytesIO(data)).convert("RGB")
+        buf = io.BytesIO(); img.save(buf, format="JPEG", quality=90); return buf.getvalue()
+    if name.endswith(".pdf"):
+        try:
+            img = Image.open(io.BytesIO(data)).convert("RGB")
+            buf = io.BytesIO(); img.save(buf, format="JPEG", quality=90); return buf.getvalue()
+        except Exception:
+            return data
+    return data
+
+def b64_data_uri(data: bytes, mime: str) -> str:
+    return f"data:{mime};base64," + base64.b64encode(data).decode("utf-8")
+
+def call_openai_vision_for_receipt(image_bytes: bytes) -> Dict[str, Any]:
+    is_pdf = image_bytes[0:4] == b"%PDF"
+    mime = "application/pdf" if is_pdf else "image/jpeg"
+    system = ("You are a strict, no-chitchat receipt parser for Canadian grocery receipts. "
+              "Return ONLY JSON; prices in CAD.")
+    user_prompt = """
+{ "store":{"name":"string","address":"string|null","date":"YYYY-MM-DD|null"},
+  "items":[{"name":"string","size":"string|null","qty":1,"unit_price":0.00,"line_total":0.00}],
+  "subtotal":0.00,"tax":0.00,"total":0.00 }
+Rules: shopper-friendly names; qty>=1; unit_price before tax; line_total=qty*unit_price; use null if missing.
+Return ONLY JSON.
+"""
+    resp = oai_client.chat.completions.create(
+        model="gpt-4o-mini", temperature=0,
+        messages=[{"role":"system","content":system},
+                  {"role":"user","content":[
+                      {"type":"text","text":user_prompt},
+                      {"type":"image_url","image_url":{"url":b64_data_uri(image_bytes,mime)}}
+                  ]}]
+    )
+    s = resp.choices[0].message.content.strip()
+    if s.startswith("```"):
+        s = s.split("```",2)[1]
+        if s.lower().startswith("json"): s = s.split("\n",1)[1]
+    return json.loads(s)
+
+def _fallback_store_search_urls(q: str, k: int = 5) -> List[str]:
+    qenc = quote_plus(q)
+    urls = [
+        f"https://www.walmart.ca/search?q={qenc}",
+        f"https://www.realcanadiansuperstore.ca/search?search-bar={qenc}",
+        f"https://www.amazon.ca/s?k={qenc}",
+        f"https://www.metro.ca/en/online-grocery/search?filter.query={qenc}",
+        f"https://www.nofrills.ca/search?search-bar={qenc}",
+        f"https://www.freshco.com/en/search?search-bar={qenc}",
+        f"https://well.ca/searchresult.html?keyword={qenc}",
+        f"https://www.costco.ca/CatalogSearch?dept=All&keyword={qenc}",
+        f"https://www.iga.net/en/search?search={qenc}",
+        f"https://www.londondrugs.com/search?searchTerm={qenc}",
+    ]
+    return urls[:k]
+
+def _search_web(query: str, k: int = 5) -> List[str]:
+    urls: List[str] = []
+    if _HAS_TAVILY and os.getenv("TAVILY_API_KEY"):
+        try:
+            tv = TavilyClient(os.getenv("TAVILY_API_KEY"))
+            res = tv.search(query=f"{query} price", search_depth="basic", max_results=k,
+                            include_domains=list(MERCHANT_DOMAINS))
+            for r in res.get("results", []):
+                if r.get("url"): urls.append(r["url"])
+        except Exception: pass
+    if not urls:
+        try:
+            with DDGS() as ddgs:
+                q = f"{query} price"
+                for attempt in range(3):
+                    for r in ddgs.text(q, region="ca-en", max_results=k):
+                        u = r.get("href") or r.get("url")
+                        if u: urls.append(u)
+                    if urls: break
+                    time.sleep(0.6*(2**attempt)+random.random()*0.3)
+        except Exception: urls=[]
+    urls = [u for u in urls if any(dom in u for dom in MERCHANT_DOMAINS)]
+    if not urls: urls = _fallback_store_search_urls(query, k=k)
+    seen, out = set(), []
+    for u in urls:
+        if u not in seen: out.append(u); seen.add(u)
+    return out[:k]
+
+def _fetch_text(url: str, timeout=15) -> str:
+    try:
+        headers = {"User-Agent":"Mozilla/5.0 (compatible; PriceAgent/1.0)"}
+        with httpx.Client(follow_redirects=True, timeout=timeout) as client:
+            r = client.get(url, headers=headers)
+        soup = BeautifulSoup(r.text,"html.parser")
+        for t in soup(["script","style","noscript"]): t.decompose()
+        return " ".join(soup.get_text(separator=" ").split())[:12000]
+    except Exception: return ""
+
+def langchain_price_lookup(item_name: str) -> Optional[Dict[str, Any]]:
+    urls = _search_web(item_name, k=5)
+    best: Optional[Offer] = None
+    for u in urls:
+        text = _fetch_text(u)
+        if not text: continue
+        try:
+            offer: Offer = chain_extract.invoke({"query": item_name, "url": u, "text": text})
+        except Exception:
+            continue
+        if not offer or not offer.price or offer.price <= 0: continue
+        if best is None or offer.price < best.price: best = offer
+    if not best: return None
+    return {"title": best.title or item_name, "price": float(best.price),
+            "source": best.merchant or "Other store", "link": best.url or urls[0]}
+
+def normalize_query(item: Dict[str, Any]) -> str:
+    base = item.get("name") or ""; size = item.get("size") or ""
+    q = f"{base} {size}".strip()
+    return " ".join([t for t in q.split() if len(t) > 1])
+
+def research_prices(items: List[Dict[str, Any]], max_items=6) -> List[Dict[str, Any]]:
+    out=[]
+    for it in items[:max_items]:
+        name = normalize_query(it)
+        if not name: continue
+        unit = it.get("unit_price")
+        offer = langchain_price_lookup(name)
+        if not offer: continue
+        cheaper = isinstance(unit,(int,float)) and offer["price"] < float(unit)-0.005
+        out.append({"item_name":it.get("name"),"receipt_unit_price":unit,
+                    "found_price":offer["price"],"found_store":offer["source"],
+                    "found_title":offer["title"],"found_link":offer["link"],
+                    "is_cheaper":cheaper})
+        time.sleep(0.25)
+    return out
+
+def compute_savings(receipt: Dict[str, Any], found: List[Dict[str, Any]]) -> Tuple[float,List[Dict[str, Any]]]:
+    cheaper=[f for f in found if f.get("is_cheaper")]
+    s=0.0
+    for f in cheaper:
+        try: s += float(f["receipt_unit_price"]) - float(f["found_price"])
+        except Exception: pass
+    return round(s,2), cheaper
+
+def format_five_lines(receipt: Dict[str, Any], savings: float, cheaper_list: List[Dict[str, Any]]) -> str:
+    store = (receipt.get("store") or {}).get("name") or "your store"
+    total_val = receipt.get("total") or receipt.get("subtotal")
+    total_txt = "N/A"
+    try: total_txt = f"${float(str(total_val).replace('$','').strip()):.2f}"
+    except Exception: pass
+    lines = [
+        f"Receipt read: {store}, total {total_txt}.",
+        f"I found potential savings of ${savings:.2f} by checking other stores.",
+    ]
+    if cheaper_list:
+        bullets=[]
+        for f in cheaper_list[:3]:
+            item=f.get("item_name") or "Item"; shop=f.get("found_store") or "other store"
+            price=float(f.get("found_price") or 0.0)
+            bullets.append(f"{item} @ {shop} for ${price:.2f}")
+        lines.append("Cheaper picks: " + "; ".join(bullets) + ".")
+    else:
+        lines.append("No clearly cheaper matches found right now for your items.")
+    lines.append("Reply 'DEALS' anytime to get weekly picks tailored to your receipts.")
+    return "\n".join(lines[:5])
+
+# ---------- background job ----------
+def process_and_reply(media_url: str, to_number: str):
+    if not _twilio_ok:
+        return
+    try:
+        with httpx.Client(follow_redirects=True, timeout=20) as client:
+            r = client.get(media_url); r.raise_for_status()
+            content = r.content
+        img_bytes = img_or_pdf_to_image_bytes(content, "mms.jpg")
+        receipt = call_openai_vision_for_receipt(img_bytes)
+        items = receipt.get("items") or []
+        if not items:
+            msg = "I couldn't read items. Send a clearer photo."
+        else:
+            found = research_prices(items)
+            savings, cheaper = compute_savings(receipt, found)
+            msg = format_five_lines(receipt, savings, cheaper)
+            msg = "\n".join(msg.split("\n")[:5])[:1400]
+    except Exception as e:
+        msg = f"Processing error: {e}"
+
+    # outbound SMS
+    try:
+        twilio_client.messages.create(to=to_number, from_=TW_FROM, body=msg)
+    except Exception:
+        pass
+
+# ---------- HTTP routes ----------
+@app.get("/sms")
+async def sms_health():
+    return PlainTextResponse("SMS webhook is up (POST only).", media_type="text/plain")
+
+@app.post("/sms")
+async def sms_webhook(request: Request, background_tasks: BackgroundTasks):
+    form = dict(await request.form())
+    from_number = form.get("From", "")
+    num_media = int(form.get("NumMedia","0") or "0")
+    media_url = form.get("MediaUrl0") if num_media > 0 else None
+
+    # quick acknowledgement to beat 15s timeout
+    if not media_url:
+        ack = "<Response><Message>Please MMS a clear photo of your grocery receipt to analyze savings.</Message></Response>"
+        return PlainTextResponse(ack, media_type="application/xml")
+
+    # run heavy work in background, then send outbound SMS
+    if _twilio_ok and from_number:
+        background_tasks.add_task(process_and_reply, media_url, from_number)
+        reply = "<Response><Message>Got it—processing your receipt now. You’ll get a follow-up text shortly.</Message></Response>"
+    else:
+        reply = "<Response><Message>Got the image—backend SMS sending is misconfigured. Set TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_FROM.</Message></Response>"
+
+    return PlainTextResponse(reply, media_type="application/xml")