TypeGPT/Webscout-API
0
1from fastapi import FastAPI, HTTPException, Query2from fastapi.responses import JSONResponse3from webscout import WEBS, transcriber, LLM4from typing import Optional, List, Dict, Union5from fastapi.encoders import jsonable_encoder6from bs4 import BeautifulSoup7import requests8import urllib.parse9import asyncio10import aiohttp11import threading12 13app = FastAPI()14 15@app.get("/")16async def root():17 return {"message": "API documentation can be found at /docs"}18 19@app.get("/health")20async def health_check():21 return {"status": "OK"}22 23@app.get("/api/search")24async def search(25 q: str,26 max_results: int = 10,27 timelimit: Optional[str] = None,28 safesearch: str = "moderate",29 region: str = "wt-wt",30 backend: str = "api",31 proxy: Optional[str] = None # Add proxy parameter here32):33 """Perform a text search."""34 try:35 with WEBS(proxy=proxy) as webs: # Pass proxy to WEBS instance36 results = webs.text(37 keywords=q,38 region=region,39 safesearch=safesearch,40 timelimit=timelimit,41 backend=backend,42 max_results=max_results,43 )44 return JSONResponse(content=jsonable_encoder(results))45 except Exception as e:46 raise HTTPException(status_code=500, detail=f"Error during search: {e}")47 48@app.get("/api/images")49async def images(50 q: str,51 max_results: int = 10,52 safesearch: str = "moderate",53 region: str = "wt-wt",54 timelimit: Optional[str] = None,55 size: Optional[str] = None,56 color: Optional[str] = None,57 type_image: Optional[str] = None,58 layout: Optional[str] = None,59 license_image: Optional[str] = None,60 proxy: Optional[str] = None # Add proxy parameter here61):62 """Perform an image search."""63 try:64 with WEBS(proxy=proxy) as webs: # Pass proxy to WEBS instance65 results = webs.images(66 keywords=q,67 region=region,68 safesearch=safesearch,69 timelimit=timelimit,70 size=size,71 color=color,72 type_image=type_image,73 layout=layout,74 license_image=license_image,75 max_results=max_results,76 )77 return JSONResponse(content=jsonable_encoder(results))78 except Exception as e:79 raise HTTPException(status_code=500, detail=f"Error during image search: {e}")80 81@app.get("/api/videos")82async def videos(83 q: str,84 max_results: int = 10,85 safesearch: str = "moderate",86 region: str = "wt-wt",87 timelimit: Optional[str] = None,88 resolution: Optional[str] = None,89 duration: Optional[str] = None,90 license_videos: Optional[str] = None,91 proxy: Optional[str] = None # Add proxy parameter here92):93 """Perform a video search."""94 try:95 with WEBS(proxy=proxy) as webs: # Pass proxy to WEBS instance96 results = webs.videos(97 keywords=q,98 region=region,99 safesearch=safesearch,100 timelimit=timelimit,101 resolution=resolution,102 duration=duration,103 license_videos=license_videos,104 max_results=max_results,105 )106 return JSONResponse(content=jsonable_encoder(results))107 except Exception as e:108 raise HTTPException(status_code=500, detail=f"Error during video search: {e}")109 110 111@app.get("/api/news")112async def news(113 q: str,114 max_results: int = 10,115 safesearch: str = "moderate",116 region: str = "wt-wt",117 timelimit: Optional[str] = None,118 proxy: Optional[str] = None # Add proxy parameter here119):120 """Perform a news search."""121 try:122 with WEBS(proxy=proxy) as webs: # Pass proxy to WEBS instance123 results = webs.news(124 keywords=q,125 region=region,126 safesearch=safesearch,127 timelimit=timelimit,128 max_results=max_results129 )130 return JSONResponse(content=jsonable_encoder(results))131 except Exception as e:132 raise HTTPException(status_code=500, detail=f"Error during news search: {e}")133 134 135@app.get("/api/llm")136async def llm_chat(137 model: str,138 message: str,139 system_prompt: str = Query(None, description="Optional custom system prompt")140):141 """Interact with a specified large language model with an optional system prompt."""142 try:143 messages = [{"role": "user", "content": message}]144 if system_prompt:145 messages.insert(0, {"role": "system", "content": system_prompt}) # Add system message at the beginning146 147 llm = LLM(model=model) 148 response = llm.chat(messages=messages)149 return JSONResponse(content={"response": response})150 except Exception as e:151 raise HTTPException(status_code=500, detail=f"Error during LLM chat: {e}")152 153 154@app.get("/api/answers")155async def answers(q: str, proxy: Optional[str] = None):156 """Get instant answers for a query."""157 try:158 with WEBS(proxy=proxy) as webs:159 results = webs.answers(keywords=q)160 return JSONResponse(content=jsonable_encoder(results))161 except Exception as e:162 raise HTTPException(status_code=500, detail=f"Error getting instant answers: {e}")163 164@app.get("/api/suggestions")165async def suggestions(q: str, region: str = "wt-wt", proxy: Optional[str] = None):166 """Get search suggestions for a query."""167 try:168 with WEBS(proxy=proxy) as webs:169 results = webs.suggestions(keywords=q, region=region)170 return JSONResponse(content=jsonable_encoder(results))171 except Exception as e:172 raise HTTPException(status_code=500, detail=f"Error getting search suggestions: {e}")173 174@app.get("/api/chat")175async def chat(176 q: str,177 model: str = "gpt-3.5",178 proxy: Optional[str] = None179):180 """Perform a text search."""181 try:182 with WEBS(proxy=proxy) as webs:183 results = webs.chat(keywords=q, model=model)184 return JSONResponse(content=jsonable_encoder(results))185 except Exception as e:186 raise HTTPException(status_code=500, detail=f"Error getting chat results: {e}")187 188def extract_text_from_webpage(html_content):189 """Extracts visible text from HTML content using BeautifulSoup."""190 soup = BeautifulSoup(html_content, "html.parser")191 # Remove unwanted tags192 for tag in soup(["script", "style", "header", "footer", "nav"]):193 tag.extract()194 # Get the remaining visible text195 visible_text = soup.get_text(strip=True)196 return visible_text197 198async def fetch_and_extract(url, max_chars, proxy: Optional[str] = None):199 """Fetches a URL and extracts text asynchronously."""200 201 async with aiohttp.ClientSession() as session:202 try:203 async with session.get(url, headers={"User-Agent": "Mozilla/5.0"}, proxy=proxy) as response:204 response.raise_for_status()205 html_content = await response.text()206 visible_text = extract_text_from_webpage(html_content)207 if len(visible_text) > max_chars:208 visible_text = visible_text[:max_chars] + "..."209 return {"link": url, "text": visible_text}210 except (aiohttp.ClientError, requests.exceptions.RequestException) as e:211 print(f"Error fetching or processing {url}: {e}")212 return {"link": url, "text": None}213 214@app.get("/api/web_extract")215async def web_extract(216 url: str,217 max_chars: int = 12000, # Adjust based on token limit218 proxy: Optional[str] = None219):220 """Extracts text from a given URL."""221 try:222 result = await fetch_and_extract(url, max_chars, proxy)223 return {"url": url, "text": result["text"]}224 except requests.exceptions.RequestException as e:225 raise HTTPException(status_code=500, detail=f"Error fetching or processing URL: {e}")226 227@app.get("/api/search-and-extract")228async def web_search_and_extract(229 q: str,230 max_results: int = 3,231 timelimit: Optional[str] = None,232 safesearch: str = "moderate",233 region: str = "wt-wt",234 backend: str = "html",235 max_chars: int = 6000,236 extract_only: bool = True,237 proxy: Optional[str] = None238):239 """240 Searches using WEBS, extracts text from the top results, and returns both.241 """242 try:243 with WEBS(proxy=proxy) as webs:244 # Perform WEBS search245 search_results = webs.text(keywords=q, region=region, safesearch=safesearch,246 timelimit=timelimit, backend=backend, max_results=max_results)247 248 # Extract text from each result's link asynchronously249 tasks = [fetch_and_extract(result['href'], max_chars, proxy) for result in search_results if 'href' in result]250 extracted_results = await asyncio.gather(*tasks)251 252 if extract_only:253 return JSONResponse(content=jsonable_encoder(extracted_results))254 else:255 return JSONResponse(content=jsonable_encoder({"search_results": search_results, "extracted_results": extracted_results}))256 except Exception as e:257 raise HTTPException(status_code=500, detail=f"Error during search and extraction: {e}")258 259def extract_text_from_webpage2(html_content):260 """Extracts visible text from HTML content using BeautifulSoup."""261 soup = BeautifulSoup(html_content, "html.parser")262 # Remove unwanted tags263 for tag in soup(["script", "style", "header", "footer", "nav"]):264 tag.extract()265 # Get the remaining visible text266 visible_text = soup.get_text(strip=True)267 return visible_text268 269def fetch_and_extract2(url, max_chars, proxy: Optional[str] = None):270 """Fetches a URL and extracts text using threading."""271 proxies = {'http': proxy, 'https': proxy} if proxy else None272 try:273 response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, proxies=proxies)274 response.raise_for_status()275 html_content = response.text276 visible_text = extract_text_from_webpage2(html_content)277 if len(visible_text) > max_chars:278 visible_text = visible_text[:max_chars] + "..."279 return {"link": url, "text": visible_text}280 except (requests.exceptions.RequestException) as e:281 print(f"Error fetching or processing {url}: {e}")282 return {"link": url, "text": None}283 284@app.get("/api/websearch-and-extract-threading")285def web_search_and_extract_threading(286 q: str,287 max_results: int = 3,288 timelimit: Optional[str] = None,289 safesearch: str = "moderate",290 region: str = "wt-wt",291 backend: str = "html",292 max_chars: int = 6000,293 extract_only: bool = True,294 proxy: Optional[str] = None295):296 """297 Searches using WEBS, extracts text from the top results using threading, and returns both.298 """299 try:300 with WEBS(proxy=proxy) as webs:301 # Perform WEBS search302 search_results = webs.text(keywords=q, region=region, safesearch=safesearch,303 timelimit=timelimit, backend=backend, max_results=max_results)304 305 # Extract text from each result's link using threading306 extracted_results = []307 threads = []308 for result in search_results:309 if 'href' in result:310 thread = threading.Thread(target=lambda: extracted_results.append(fetch_and_extract2(result['href'], max_chars, proxy)))311 threads.append(thread)312 thread.start()313 314 # Wait for all threads to finish315 for thread in threads:316 thread.join()317 318 if extract_only:319 return JSONResponse(content=jsonable_encoder(extracted_results))320 else:321 return JSONResponse(content=jsonable_encoder({"search_results": search_results, "extracted_results": extracted_results}))322 except Exception as e:323 raise HTTPException(status_code=500, detail=f"Error during search and extraction: {e}")324 325 326@app.get("/api/adv_web_search")327async def adv_web_search(328 q: str,329 model: str = "gpt-3.5",330 max_results: int = 3, 331 timelimit: Optional[str] = None,332 safesearch: str = "moderate",333 region: str = "wt-wt",334 backend: str = "html",335 max_chars: int = 6000, 336 system_prompt: str = "You are Most Advanced and Powerful Ai chatbot, User ask you questions and you have to answer that, You are also provided with Google Search Results, To increase your accuracy and providing real time data. Your task is to answer in best way to user.",337 proxy: Optional[str] = None338):339 """340 Combines web search, web extraction, and LLM chat for advanced search.341 """342 try:343 with WEBS(proxy=proxy) as webs:344 # 1. Perform the web search345 search_results = webs.text(keywords=q, region=region, 346 safesearch=safesearch,347 timelimit=timelimit, backend=backend, 348 max_results=max_results)349 350 # 2. Extract text from top search result URLs asynchronously 351 extracted_text = ""352 tasks = [fetch_and_extract(result['href'], max_chars, proxy) for result in search_results if 'href' in result]353 extracted_results = await asyncio.gather(*tasks)354 for result in extracted_results:355 if result['text']:356 extracted_text += f"## Content from: {result['link']}\n\n{result['text']}\n\n"357 358 # 3. Construct the prompt for the LLM359 llm_prompt = f"Query by user: {q} , Answer the query asked by user in detail. Now, You are provided with Google Search Results, To increase your accuracy and providing real time data. SEarch Result: {extracted_text}"360 361 # 4. Get the LLM's response using LLM class (similar to /api/llm)362 messages = [{"role": "user", "content": llm_prompt}]363 if system_prompt:364 messages.insert(0, {"role": "system", "content": system_prompt})365 366 llm = LLM(model=model)367 llm_response = llm.chat(messages=messages)368 369 # 5. Return the results370 return JSONResponse(content=jsonable_encoder({ "llm_response": llm_response }))371 372 except Exception as e:373 raise HTTPException(status_code=500, detail=f"Error during advanced search: {e}")374 375 376@app.get("/api/website_summarizer")377async def website_summarizer(url: str, proxy: Optional[str] = None):378 """Summarizes the content of a given URL using a chat model."""379 try:380 # Extract text from the given URL381 proxies = {'http': proxy, 'https': proxy} if proxy else None382 response = requests.get(url, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0"}, proxies=proxies)383 response.raise_for_status()384 visible_text = extract_text_from_webpage(response.text)385 if len(visible_text) > 7500: # Adjust max_chars based on your needs386 visible_text = visible_text[:7500] + "..."387 388 # Use chat model to summarize the extracted text389 with WEBS(proxy=proxy) as webs:390 summary_prompt = f"Summarize this in detail in Paragraph: {visible_text}"391 summary_result = webs.chat(keywords=summary_prompt, model="gpt-3.5")392 393 # Return the summary result394 return JSONResponse(content=jsonable_encoder({summary_result}))395 396 except requests.exceptions.RequestException as e:397 raise HTTPException(status_code=500, detail=f"Error fetching or processing URL: {e}")398 except Exception as e:399 raise HTTPException(status_code=500, detail=f"Error during summarization: {e}")400 401@app.get("/api/ask_website")402async def ask_website(url: str, question: str, model: str = "llama-3-70b", proxy: Optional[str] = None):403 """404 Asks a question about the content of a given website.405 """406 try:407 # Extract text from the given URL408 proxies = {'http': proxy, 'https': proxy} if proxy else None409 response = requests.get(url, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0"}, proxies=proxies)410 response.raise_for_status()411 visible_text = extract_text_from_webpage(response.text)412 if len(visible_text) > 7500: # Adjust max_chars based on your needs413 visible_text = visible_text[:7500] + "..."414 415 # Construct a prompt for the chat model416 prompt = f"Based on the following text, answer this question in Paragraph: [QUESTION] {question} [TEXT] {visible_text}"417 418 # Use chat model to get the answer419 with WEBS(proxy=proxy) as webs:420 answer_result = webs.chat(keywords=prompt, model=model)421 422 # Return the answer result423 return JSONResponse(content=jsonable_encoder({answer_result}))424 425 except requests.exceptions.RequestException as e:426 raise HTTPException(status_code=500, detail=f"Error fetching or processing URL: {e}")427 except Exception as e:428 raise HTTPException(status_code=500, detail=f"Error during question answering: {e}")429 430@app.get("/api/maps")431async def maps(432 q: str,433 place: Optional[str] = None,434 street: Optional[str] = None,435 city: Optional[str] = None,436 county: Optional[str] = None,437 state: Optional[str] = None,438 country: Optional[str] = None,439 postalcode: Optional[str] = None,440 latitude: Optional[str] = None,441 longitude: Optional[str] = None,442 radius: int = 0,443 max_results: int = 10,444 proxy: Optional[str] = None445):446 """Perform a maps search."""447 try:448 with WEBS(proxy=proxy) as webs:449 results = webs.maps(keywords=q, place=place, street=street, city=city, county=county, state=state, country=country, postalcode=postalcode, latitude=latitude, longitude=longitude, radius=radius, max_results=max_results)450 return JSONResponse(content=jsonable_encoder(results))451 except Exception as e:452 raise HTTPException(status_code=500, detail=f"Error during maps search: {e}")453 454@app.get("/api/translate")455async def translate(456 q: str,457 from_: Optional[str] = None,458 to: str = "en",459 proxy: Optional[str] = None460):461 """Translate text."""462 try:463 with WEBS(proxy=proxy) as webs:464 results = webs.translate(keywords=q, from_=from_, to=to)465 return JSONResponse(content=jsonable_encoder(results))466 except Exception as e:467 raise HTTPException(status_code=500, detail=f"Error during translation: {e}")468 469from easygoogletranslate import EasyGoogleTranslate470 471@app.get("/api/google_translate")472def google_translate(q: str, from_: Optional[str] = 'auto', to: str = "en"):473 try:474 translator = EasyGoogleTranslate(475 source_language=from_,476 target_language=to,477 timeout=10478)479 result = translator.translate(q)480 return JSONResponse(content=jsonable_encoder({"detected_language": from_ , "original": q , "translated": result}))481 except Exception as e:482 raise HTTPException(status_code=500, detail=f"Error during translation: {e}")483 484 485@app.get("/api/youtube/transcript")486async def youtube_transcript(487 video_id: str,488 languages: str = "en",489 preserve_formatting: bool = False,490 proxy: Optional[str] = None # Add proxy parameter491):492 """Get the transcript of a YouTube video."""493 try:494 languages_list = languages.split(",")495 transcript = transcriber.get_transcript(video_id, languages=languages_list, preserve_formatting=preserve_formatting, proxies=proxy)496 return JSONResponse(content=jsonable_encoder(transcript))497 except Exception as e:498 raise HTTPException(status_code=500, detail=f"Error getting YouTube transcript: {e}")499 500import requests501@app.get("/weather/json/{location}")502def get_weather_json(location: str):503 url = f"https://wttr.in/{location}?format=j1"504 response = requests.get(url)505 if response.status_code == 200:506 return response.json()507 else:508 return {"error": f"Unable to fetch weather data. Status code: {response.status_code}"}509 510@app.get("/weather/ascii/{location}")511def get_ascii_weather(location: str):512 url = f"https://wttr.in/{location}"513 response = requests.get(url, headers={'User-Agent': 'curl'})514 if response.status_code == 200:515 return response.text516 else:517 return {"error": f"Unable to fetch weather data. Status code: {response.status_code}"}518 519# Run the API server if this script is executed520if __name__ == "__main__":521 import uvicorn522 uvicorn.run(app, host="0.0.0.0", port=8083)