subashpoudel's picture
Implemented the analytics too in orchestrator
8ce97f0
import requests
import json
import re
def generate_api_knowledge(base_url: str):
"""
Fetches FastAPI OpenAPI docs and summarizes endpoints, methods, parameters, and descriptions.
Returns a JSON-friendly Python list of dicts, excluding the root endpoint.
"""
resp = requests.get(f"{base_url}/openapi.json")
resp.raise_for_status()
spec = resp.json()
api_knowledge = []
for path, methods in spec["paths"].items():
if path == "/": # skip root endpoint
continue
for method, details in methods.items():
endpoint = path
description = details.get("description", "No description provided").strip()
# collect parameters
params = []
for param in details.get("parameters", []):
params.append(param.get("name"))
# requestBody can also have parameters
if "requestBody" in details:
try:
body_schema = (
details["requestBody"]["content"]["application/json"]["schema"]
)
if "properties" in body_schema:
params.extend(body_schema["properties"].keys())
except Exception:
pass
api_knowledge.append({
"endpoint": endpoint,
"method": method.upper(),
"parameters": params,
"description": description
})
return api_knowledge
def influencers_name():
return {
"information_of_influencers_usernames": "From the names_dictionary, the key represents their original names and the value represents their usernames (influencer_username).",
"names_dictionary": {
"divya dhakal": "divyadhakal_",
"istu karki": "ishtukarkee",
"kano mama": "kano_mama",
"muna chiya": "munachiya",
"nepal food": "nepal_food",
"ggkaam": "ggkaam610",
"adarsh": "adars_fpv",
"ajay tamang": "ajaytm43",
"anisha kafle": "anishakafle",
"diwash gurung": "diwasg",
"eat grub food": "eatgrubfood",
"surakshya kc": "imsurakshyakc",
"jholey": "jholeyism",
"mrb vlogs": "mrbvlog2"
}
}
def process_query(user_query: str) -> str:
# load mapping from json file
with open("src/genai/analytics_chatbot/utils/name_variations.json", "r") as f:
variations = json.load(f)
processed = user_query
# Build reverse mapping: variation → username
expanded = {}
for username, names in variations.items():
for n in names:
expanded[n.lower()] = username
# Replace names in query
for name in sorted(expanded.keys(), key=len, reverse=True):
pattern = r"\b" + re.escape(name) + r"\b"
processed = re.sub(pattern, expanded[name], processed, flags=re.IGNORECASE)
return processed