Spaces:
Sleeping
Sleeping
File size: 4,790 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from swiggy_scraper import fetch_swiggy_orders
from datetime import datetime
from openai import OpenAI
import os, json
router = APIRouter()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class QueryInput(BaseModel):
query: str
class DateRange(BaseModel):
start_date: str
end_date: str
# @router.post("/parse_query")
# def parse_query_llm(input: QueryInput):
# print("\ninput query:")
# print(input)
# today_str = datetime.today().strftime("%d-%b-%Y")
# system = (
# f"You are a date range extractor.\n"
# f"Today is {today_str}.\n"
# "Extract start_date and end_date in 'DD-MMM-YYYY' format.\n"
# "Respond with:\n"
# "Output ONLY a valid JSON object like:\n"
# '{ "start_date": "17-May-2025", "end_date": "18-May-2025" }\n'
# 'no extra commentry needed.'
# )
# try:
# rsp = client.chat.completions.create(
# model="gpt-4o-mini",
# temperature=0,
# messages=[
# {"role": "system", "content": system},
# {"role": "user", "content": input.query},
# ]
# )
# result = json.loads(rsp.choices[0].message.content.strip())
# if "start_date" not in result or "end_date" not in result:
# raise ValueError("Invalid response format")
# print("results:", result)
# return result
# except Exception as e:
# raise HTTPException(status_code=400, detail=str(e))
def _llm(messages, model="gpt-4o-mini", temperature=0):
rsp = client.chat.completions.create(
model=model,
temperature=temperature,
messages=messages,
)
return rsp.choices[0].message.content.strip()
# ---------- Stage 1: classify + extract dates --------------------------
def _extract_scope(user_query: str):
today_str = datetime.today().strftime("%d-%b-%Y")
sys_prompt = f"""
Today is {today_str}.
You are a SCOPING assistant: decide if the user's text is about Swiggy food orders,
extract ONE date range, and keep the leftover words.
Return ONLY valid JSON like:
{{
"is_swiggy_query": true,
"start_date": "15-May-2025",
"end_date": "20-May-2025",
"remainder": "non veg expense"
}}
Rules:
• Accept natural phrases (“last week”, “since 1 May”).
• If no dates → start_date & end_date = null.
• If not Swiggy related → is_swiggy_query=false and remainder is full original text.
• Do NOT invent a remainder; it is literally whatever words follow the date phrase(s).
"""
raw = _llm(
[
{"role": "system", "content": sys_prompt},
{"role": "user", "content": user_query}
]
)
return json.loads(raw)
# ---------- Stage 2: shrink “remainder” into an intent -----------------
def _extract_intent(remainder: str):
sys_prompt = """
You are an INTENT classifier for Swiggy-order analytics.
Map the sentence into one concise snake_case intent.
Allowed intents (extendable):
• calculate_expense
• list_orders
• list_items
• list_nonveg_items
• list_veg_items
• count_orders
• unknown
Return JSON: { "intent": "calculate_expense" }
If unsure choose "unknown".
"""
raw = _llm(
[
{"role": "system", "content": sys_prompt},
{"role": "user", "content": remainder.strip()}
]
)
return json.loads(raw)["intent"]
# ---------- FastAPI route ----------------------------------------------
@router.post("/parse_query")
def parse_query_llm(input: QueryInput):
try:
scope = _extract_scope(input.query)
print("scope")
print(scope)
# If it is a Swiggy query, classify intent; else, intent = "unrelated"
if scope.get("is_swiggy_query", False):
intent = _extract_intent(scope.get("remainder", ""))
else:
intent = "unrelated"
result = {
"is_swiggy_query": scope["is_swiggy_query"],
"start_date": scope["start_date"],
"end_date": scope["end_date"],
"intent": intent
}
print("result")
print(result)
return result
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.post("/get_orders")
def get_orders(range: DateRange):
try:
orders = fetch_swiggy_orders(range.start_date, range.end_date)
return orders
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
|