app.py

from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse
from webscout import WEBS, transcriber, LLM
from typing import Optional, List, Dict, Union
from fastapi.encoders import jsonable_encoder
from bs4 import BeautifulSoup
import requests
import urllib.parse
import asyncio
import aiohttp
from typing import List

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "API documentation can be found at /docs"}

@app.get("/health")
async def health_check():
    return {"status": "OK"}

@app.get("/api/search")
async def search(
    q: str,
    max_results: int = 10,
    timelimit: Optional[str] = None,
    safesearch: str = "moderate",
    region: str = "wt-wt",
    backend: str = "api"
):
    """Perform a text search."""
    try:
        with WEBS() as webs:
            results = webs.text(keywords=q, region=region, safesearch=safesearch, timelimit=timelimit, backend=backend, max_results=max_results)
            return JSONResponse(content=jsonable_encoder(results))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during search: {e}")

@app.get("/api/images")
async def images(
    q: str,
    max_results: int = 10,
    safesearch: str = "moderate",
    region: str = "wt-wt",
    timelimit: Optional[str] = None,
    size: Optional[str] = None,
    color: Optional[str] = None,
    type_image: Optional[str] = None,
    layout: Optional[str] = None,
    license_image: Optional[str] = None
):
    """Perform an image search."""
    try:
        with WEBS() as webs:
            results = webs.images(keywords=q, region=region, safesearch=safesearch, timelimit=timelimit, size=size, color=color, type_image=type_image, layout=layout, license_image=license_image, max_results=max_results)
            return JSONResponse(content=jsonable_encoder(results))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during image search: {e}")

@app.get("/api/videos")
async def videos(
    q: str,
    max_results: int = 10,
    safesearch: str = "moderate",
    region: str = "wt-wt",
    timelimit: Optional[str] = None,
    resolution: Optional[str] = None,
    duration: Optional[str] = None,
    license_videos: Optional[str] = None
):
    """Perform a video search."""
    try:
        with WEBS() as webs:
            results = webs.videos(keywords=q, region=region, safesearch=safesearch, timelimit=timelimit, resolution=resolution, duration=duration, license_videos=license_videos, max_results=max_results)
            return JSONResponse(content=jsonable_encoder(results))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during video search: {e}")

@app.get("/api/news")
async def news(
    q: str,
    max_results: int = 10,
    safesearch: str = "moderate",
    region: str = "wt-wt",
    timelimit: Optional[str] = None
):
    """Perform a news search."""
    try:
        with WEBS() as webs:
            results = webs.news(keywords=q, region=region, safesearch=safesearch, timelimit=timelimit, max_results=max_results)
            return JSONResponse(content=jsonable_encoder(results))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during news search: {e}")

@app.get("/api/llm")
async def llm_chat(
    model: str,
    message: str,
    system_prompt: str = Query(None, description="Optional custom system prompt")
):
    """Interact with a specified large language model with an optional system prompt."""
    try:
        messages = [{"role": "user", "content": message}]
        if system_prompt:
            messages.insert(0, {"role": "system", "content": system_prompt})  # Add system message at the beginning

        llm = LLM(model=model) 
        response = llm.chat(messages=messages)
        return JSONResponse(content={"response": response})
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during LLM chat: {e}")


@app.get("/api/answers")
async def answers(q: str):
    """Get instant answers for a query."""
    try:
        with WEBS() as webs:
            results = webs.answers(keywords=q)
            return JSONResponse(content=jsonable_encoder(results))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error getting instant answers: {e}")

@app.get("/api/suggestions")
async def suggestions(q: str, region: str = "wt-wt"):
    """Get search suggestions for a query."""
    try:
        with WEBS() as webs:
            results = webs.suggestions(keywords=q, region=region)
            return JSONResponse(content=jsonable_encoder(results))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error getting search suggestions: {e}")

@app.get("/api/chat")
async def chat(
    q: str,
    model: str = "gpt-3.5"
):
    """Perform a text search."""
    try:
        with WEBS() as webs:
            results = webs.chat(keywords=q, model=model)
            return JSONResponse(content=jsonable_encoder(results))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error getting chat results: {e}")

def extract_text_from_webpage(html_content):
    """Extracts visible text from HTML content using BeautifulSoup."""
    soup = BeautifulSoup(html_content, "html.parser")
    # Remove unwanted tags
    for tag in soup(["script", "style", "header", "footer", "nav"]):
        tag.extract()
    # Get the remaining visible text
    visible_text = soup.get_text(strip=True)
    return visible_text

async def fetch_and_extract(url, max_chars):
    """Fetches a URL and extracts text asynchronously."""
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(url, headers={"User-Agent": "Mozilla/5.0"}) as response:
                response.raise_for_status()
                html_content = await response.text()
                visible_text = extract_text_from_webpage(html_content)
                if len(visible_text) > max_chars:
                    visible_text = visible_text[:max_chars] + "..."
                return {"link": url, "text": visible_text}
        except (aiohttp.ClientError, requests.exceptions.RequestException) as e:
            print(f"Error fetching or processing {url}: {e}")
            return {"link": url, "text": None}

@app.get("/api/web_extract")
async def web_extract(
    url: str,
    max_chars: int = 12000,  # Adjust based on token limit
):
    """Extracts text from a given URL."""
    try:
        result = await fetch_and_extract(url, max_chars)
        return {"url": url, "text": result["text"]}
    except requests.exceptions.RequestException as e:
        raise HTTPException(status_code=500, detail=f"Error fetching or processing URL: {e}")

@app.get("/api/search-and-extract")
async def web_search_and_extract(
    q: str,
    max_results: int = 3,
    timelimit: Optional[str] = None,
    safesearch: str = "moderate",
    region: str = "wt-wt",
    backend: str = "api",
    max_chars: int = 6000,
    extract_only: bool = False
):
    """
    Searches using WEBS, extracts text from the top results, and returns both.
    """
    try:
        with WEBS() as webs:
            # Perform WEBS search
            search_results = webs.text(keywords=q, region=region, safesearch=safesearch,
                                     timelimit=timelimit, backend=backend, max_results=max_results)

            # Extract text from each result's link asynchronously
            tasks = [fetch_and_extract(result['href'], max_chars) for result in search_results if 'href' in result]
            extracted_results = await asyncio.gather(*tasks)

            if extract_only:
                return JSONResponse(content=jsonable_encoder({extracted_results}))
            else:
                return JSONResponse(content=jsonable_encoder({"search_results": search_results, "extracted_results": extracted_results}))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during search and extraction: {e}")

@app.get("/api/adv_web_search")
async def adv_web_search(
    q: str,
    model: str = "gpt-3.5",
    max_results: int = 3,  
    timelimit: Optional[str] = None,
    safesearch: str = "moderate",
    region: str = "wt-wt",
    backend: str = "api",
    max_chars: int = 6000,  
    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."
):
    """
    Combines web search, web extraction, and LLM chat for advanced search.
    """
    try:
        with WEBS() as webs:
            # 1. Perform the web search
            search_results = webs.text(keywords=q, region=region, 
                                     safesearch=safesearch,
                                     timelimit=timelimit, backend=backend, 
                                     max_results=max_results)

            # 2. Extract text from top search result URLs asynchronously 
            extracted_text = ""
            tasks = [fetch_and_extract(result['href'], max_chars) for result in search_results if 'href' in result]
            extracted_results = await asyncio.gather(*tasks)
            for result in extracted_results:
                if result['text']:
                    extracted_text += f"## Content from: {result['link']}\n\n{result['text']}\n\n"

        # 3. Construct the prompt for the LLM
        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}"

        # 4. Get the LLM's response using LLM class (similar to /api/llm)
        messages = [{"role": "user", "content": llm_prompt}]
        if system_prompt:
            messages.insert(0, {"role": "system", "content": system_prompt})

        llm = LLM(model=model)
        llm_response = llm.chat(messages=messages)

        # 5. Return the results
        return JSONResponse(content=jsonable_encoder({ "llm_response": llm_response }))

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during advanced search: {e}")

        
@app.get("/api/website_summarizer")
async def website_summarizer(url: str):
    """Summarizes the content of a given URL using a chat model."""
    try:
        # Extract text from the given URL
        response = requests.get(url, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0"})
        response.raise_for_status()
        visible_text = extract_text_from_webpage(response.text)
        if len(visible_text) > 7500:  # Adjust max_chars based on your needs
            visible_text = visible_text[:7500] + "..."

        # Use chat model to summarize the extracted text
        with WEBS() as webs:
            summary_prompt = f"Summarize this in detail in Paragraph: {visible_text}"
            summary_result = webs.chat(keywords=summary_prompt, model="gpt-3.5")

        # Return the summary result
        return JSONResponse(content=jsonable_encoder({summary_result}))

    except requests.exceptions.RequestException as e:
        raise HTTPException(status_code=500, detail=f"Error fetching or processing URL: {e}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during summarization: {e}")

@app.get("/api/ask_website")
async def ask_website(url: str, question: str, model: str = "llama-3-70b"):
    """
    Asks a question about the content of a given website.
    """
    try:
        # Extract text from the given URL
        response = requests.get(url, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0"})
        response.raise_for_status()
        visible_text = extract_text_from_webpage(response.text)
        if len(visible_text) > 7500:  # Adjust max_chars based on your needs
            visible_text = visible_text[:7500] + "..."

        # Construct a prompt for the chat model
        prompt = f"Based on the following text, answer this question in Paragraph: [QUESTION] {question} [TEXT] {visible_text}"

        # Use chat model to get the answer
        with WEBS() as webs:
            answer_result = webs.chat(keywords=prompt, model=model)

        # Return the answer result
        return JSONResponse(content=jsonable_encoder({answer_result}))

    except requests.exceptions.RequestException as e:
        raise HTTPException(status_code=500, detail=f"Error fetching or processing URL: {e}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during question answering: {e}")
        
@app.get("/api/maps")
async def maps(
    q: str,
    place: Optional[str] = None,
    street: Optional[str] = None,
    city: Optional[str] = None,
    county: Optional[str] = None,
    state: Optional[str] = None,
    country: Optional[str] = None,
    postalcode: Optional[str] = None,
    latitude: Optional[str] = None,
    longitude: Optional[str] = None,
    radius: int = 0,
    max_results: int = 10
):
    """Perform a maps search."""
    try:
        with WEBS() as webs:
            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)
            return JSONResponse(content=jsonable_encoder(results))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during maps search: {e}")

@app.get("/api/translate")
async def translate(
    q: str,
    from_: Optional[str] = None,
    to: str = "en"
):
    """Translate text."""
    try:
        with WEBS() as webs:
            results = webs.translate(keywords=q, from_=from_, to=to)
            return JSONResponse(content=jsonable_encoder(results))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during translation: {e}")

@app.get("/api/youtube/transcript")
async def youtube_transcript(
    video_id: str,
    languages: str = "en",
    preserve_formatting: bool = False
):
    """Get the transcript of a YouTube video."""
    try:
        languages_list = languages.split(",")
        transcript = transcriber.get_transcript(video_id, languages=languages_list, preserve_formatting=preserve_formatting)
        return JSONResponse(content=jsonable_encoder(transcript))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error getting YouTube transcript: {e}")
        
import requests
@app.get("/weather/json/{location}")
def get_weather_json(location: str):
    url = f"https://wttr.in/{location}?format=j1"
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        return {"error": f"Unable to fetch weather data. Status code: {response.status_code}"}

@app.get("/weather/ascii/{location}")
def get_ascii_weather(location: str):
    url = f"https://wttr.in/{location}"
    response = requests.get(url, headers={'User-Agent': 'curl'})
    if response.status_code == 200:
        return response.text
    else:
        return {"error": f"Unable to fetch weather data. Status code: {response.status_code}"}

# Run the API server if this script is executed
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8083)