Spaces:
Sleeping
Sleeping
| 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 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 | |