Spaces:
Sleeping
Sleeping
File size: 4,115 Bytes
7c68554 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
#!/usr/bin/env python3
"""
CLI host for the FastAPI-powered Swiggy server.
Flow
1. Send natural-language query → /parse_query
2. If is_swiggy_query == True and dates are present → /get_orders
3. Pretty-print the result
"""
import requests, sys
from dateutil import parser as dtparse
API_BASE = "http://127.0.0.1:8000"
# ----------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------
def nice_date(dt_str: str | None) -> str:
if not dt_str:
return "??"
return dtparse.parse(dt_str).strftime("%d %b %Y")
def pretty_order(order: dict) -> str:
if "error" in order:
return f" - Email #{order['email_number']}: ❌ {order['error']}"
head = (
f"Restaurant : {order['restaurant_name']}\n"
f"Date : {nice_date(order['order_date'])} {order['order_time']}\n"
f"Total : ₹{order['total_price']:.0f}\n"
"Items:"
)
items = "\n".join(
f" • {it['quantity']} × {it['name']} – ₹{it['price']:.0f}"
for it in order["items"]
)
return head + "\n" + items
# ----------------------------------------------------------------------
# Main REPL
# ----------------------------------------------------------------------
def main() -> None:
# Quick health-check
try:
requests.get(f"{API_BASE}/docs").raise_for_status()
except Exception as e:
print("❌ Cannot reach FastAPI server:", e)
sys.exit(1)
print("✅ Connected. Type a Swiggy question (or Ctrl-C to quit).")
while True:
try:
query = input("\n🗨️ You: ").strip()
except (EOFError, KeyboardInterrupt):
break
if not query:
continue
# ── Stage 1: parse_query ────────────────────────────────────
try:
r = requests.post(f"{API_BASE}/parse_query", json={"query": query})
r.raise_for_status()
meta = r.json() # {is_swiggy_query, start_date, end_date, intent}
except Exception as e:
print("⚠️ Parse error:", e)
print("🔎 Server:", r.text if 'r' in locals() else "no response")
continue
# Handle non-Swiggy or missing dates
if not meta.get("is_swiggy_query"):
print("🤷 That doesn’t look like a Swiggy order query.")
continue
if not meta.get("start_date") or not meta.get("end_date"):
print("⚠️ Couldn’t find a full date range in your question.")
continue
print(f" ↳ Date range: {meta['start_date']} → {meta['end_date']}")
print(f" ↳ Intent : {meta['intent']}")
# ── Stage 2: get_orders ────────────────────────────────────
try:
r2 = requests.post(
f"{API_BASE}/get_orders",
json={
"start_date": meta["start_date"],
"end_date": meta["end_date"],
},
)
r2.raise_for_status()
orders = r2.json()
except Exception as e:
print("⚠️ Failed to fetch orders:", e)
print("🔎 Server:", r2.text if 'r2' in locals() else "no response")
continue
# ── Output ────────────────────────────────────────────────
if not orders:
print("😕 No orders found for that range.")
continue
print("\n📦 Orders:")
for o in orders:
print(pretty_order(o))
print("-" * 40)
print("\n👋 Bye!")
# ----------------------------------------------------------------------
if __name__ == "__main__":
main()
|