Spaces:
Sleeping
Sleeping
File size: 4,069 Bytes
3002e1b 872d043 3002e1b a6a0614 3002e1b 8ce97f0 3002e1b 8ce97f0 3002e1b a6a0614 3002e1b 872d043 a6a0614 |
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 |
import requests
import json
import re
def get_endpoint_info( endpoint:str ,base_url="https://reveltrends.vercel.app/"):
"""
Fetches FastAPI OpenAPI docs for a specific endpoint.
Returns a dictionary with method and parameters.
"""
resp = requests.get(f"{base_url}/openapi.json")
resp.raise_for_status()
spec = resp.json()
if endpoint not in spec["paths"]:
return {"error": "Endpoint not found"}
# Take the first method for the endpoint
for method, details in spec["paths"][endpoint].items():
params = []
# Collect query/path parameters
for param in details.get("parameters", []):
params.append(param.get("name"))
# Collect requestBody 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
# Return only one method and its parameters
return {"method": method.upper(), "parameters": params}
return {"error": "No method found for endpoint"}
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
|