app.py

from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from webscout import WEBS, YTTranscriber, LLM, fastai
from stream import fastai_stream
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
import threading
import json
import os
import time
from huggingface_hub import HfApi
from huggingface_hub import InferenceClient
from PIL import Image
import io

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",
    proxy: Optional[str] = None  # Add proxy parameter here
):
    """Perform a text search."""
    try:
        with WEBS(proxy=proxy) as webs:  # Pass proxy to WEBS instance
            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,
    proxy: Optional[str] = None # Add proxy parameter here
):
    """Perform an image search."""
    try:
        with WEBS(proxy=proxy) as webs:  # Pass proxy to WEBS instance
            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,
    proxy: Optional[str] = None # Add proxy parameter here
):
    """Perform a video search."""
    try:
        with WEBS(proxy=proxy) as webs:  # Pass proxy to WEBS instance
            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,
    proxy: Optional[str] = None  # Add proxy parameter here
):
    """Perform a news search."""
    try:
        with WEBS(proxy=proxy) as webs:  # Pass proxy to WEBS instance
            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/fastAI")
async def fast_ai(user: str, model: str = "llama3-70b", system: str = "Answer as concisely as possible."):
    """Get a response from the Snova AI service."""
    try:
        response = await asyncio.to_thread(fastai, user, model, system)
        return JSONResponse(content={"response": response})
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during Snova AI request: {e}")

@app.get("/api/streaming-fastAI")
async def fast_ai(user: str, model: str = "llama3-8b", system: str = "Answer as concisely as possible."):
    """Get a streaming response from the Snova AI service."""
    try:
        return StreamingResponse(fastai_stream(user, model, system), media_type="text/event-stream")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during Snova AI request: {e}")

@app.get("/api/answers")
async def answers(q: str, proxy: Optional[str] = None):
    """Get instant answers for a query."""
    try:
        with WEBS(proxy=proxy) 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/chat")
async def chat(
    q: str,
    model: str = "gpt-4o-mini",
    proxy: Optional[str] = None
):
    """Perform a text search."""
    try:
        with WEBS(proxy=proxy) 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, proxy: Optional[str] = None):
    """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"}, proxy=proxy) 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
    proxy: Optional[str] = None
):
    """Extracts text from a given URL."""
    try:
        result = await fetch_and_extract(url, max_chars, proxy)
        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 = "html",
    max_chars: int = 6000,
    extract_only: bool = True,
    proxy: Optional[str] = None
):
    """
    Searches using WEBS, extracts text from the top results, and returns both.
    """
    try:
        with WEBS(proxy=proxy) 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, proxy) 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}")

def extract_text_from_webpage2(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

def fetch_and_extract2(url, max_chars, proxy: Optional[str] = None):
    """Fetches a URL and extracts text using threading."""
    proxies = {'http': proxy, 'https': proxy} if proxy else None
    try:
        response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, proxies=proxies)
        response.raise_for_status()
        html_content = response.text
        visible_text = extract_text_from_webpage2(html_content)
        if len(visible_text) > max_chars:
            visible_text = visible_text[:max_chars] + "..."
        return {"link": url, "text": visible_text}
    except (requests.exceptions.RequestException) as e:
        print(f"Error fetching or processing {url}: {e}")
        return {"link": url, "text": None}

@app.get("/api/websearch-and-extract-threading")
def web_search_and_extract_threading(
    q: str,
    max_results: int = 3,
    timelimit: Optional[str] = None,
    safesearch: str = "moderate",
    region: str = "wt-wt",
    backend: str = "html",
    max_chars: int = 6000,
    extract_only: bool = True,
    proxy: Optional[str] = None
):
    """
    Searches using WEBS, extracts text from the top results using threading, and returns both.
    """
    try:
        with WEBS(proxy=proxy) 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 using threading
            extracted_results = []
            threads = []
            for result in search_results:
                if 'href' in result:
                    thread = threading.Thread(target=lambda: extracted_results.append(fetch_and_extract2(result['href'], max_chars, proxy)))
                    threads.append(thread)
                    thread.start()

            # Wait for all threads to finish
            for thread in threads:
                thread.join()

            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 = "llama3-8b",
    max_results: int = 3,
    timelimit: Optional[str] = None,
    safesearch: str = "moderate",
    region: str = "wt-wt",
    backend: str = "html",
    max_chars: int = 6000,
    system_prompt: str = "You are an advanced AI chatbot. Provide the best answer to the user based on Google search results.",
    proxy: Optional[str] = None
):
    """
    Combines web search, web extraction, and FastAI chat for advanced search.
    """
    try:
        with WEBS(proxy=proxy) 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, proxy) 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 FastAI
        ai_prompt = (
            f"User Query: {q}\n\n"
            f"Please provide a detailed and accurate answer to the user's query. Include relevant information extracted from the search results below. Ensure to cite sources by providing links to the original content where applicable. Format your response as follows:\n\n"
            f"1. **Answer:** Provide a clear and comprehensive answer to the user's query.\n"
            f"2. **Details:** Include any additional relevant details or explanations.\n"
            f"3. **Sources:** List the sources of the information with clickable links for further reading.\n\n"
            f"Search Results:\n{extracted_text}"
        )

        # 4. Get the FastAI's response using FastAI service
        try:
            response = await asyncio.to_thread(fastai, ai_prompt, model=model, system=system_prompt)
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error during FastAI request: {e}")

        # 5. Return the results
        return JSONResponse(content={"response": 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, proxy: Optional[str] = None):
    """Summarizes the content of a given URL using a chat model."""
    try:
        # Extract text from the given URL
        proxies = {'http': proxy, 'https': proxy} if proxy else None
        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)
        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(proxy=proxy) as webs:
            summary_prompt = f"Summarize this in detail in Paragraph: {visible_text}"
            summary_result = webs.chat(keywords=summary_prompt, model="gpt-4o-mini")

        # 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", proxy: Optional[str] = None):
    """
    Asks a question about the content of a given website.
    """
    try:
        # Extract text from the given URL
        proxies = {'http': proxy, 'https': proxy} if proxy else None
        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)
        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(proxy=proxy) 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}")

client_sd3 = InferenceClient("stabilityai/stable-diffusion-3-medium-diffusers")

@app.get("/api/sd3")
def sd3(prompt :str = "",
        steps: int = 20,
        width: int = 1000,
        height: int = 1000
       ):
    try:
        image = client_sd3.text_to_image(prompt = f"{prompt}, hd, high quality, 4k, masterpiece",
                num_inference_steps = steps,
                width = width, height = height )
        image = Image.open(io.BytesIO(image))
        return image
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during image generation: {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,
    proxy: Optional[str] = None
):
    """Perform a maps search."""
    try:
        with WEBS(proxy=proxy) 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",
    proxy: Optional[str] = None
):
    """Translate text."""
    try:
        with WEBS(proxy=proxy) 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}")

from easygoogletranslate import EasyGoogleTranslate

@app.get("/api/google_translate")
def google_translate(q: str, from_: Optional[str] = 'auto', to: str = "en"):
    try:
        translator = EasyGoogleTranslate(
    source_language=from_,
    target_language=to,
    timeout=10
)
        result = translator.translate(q)
        return JSONResponse(content=jsonable_encoder({"detected_language": from_ , "original": q , "translated": result}))
    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_url: str,
    preserve_formatting: bool = False,
    proxy: Optional[str] = None  # Add proxy parameter
):
    """Get the transcript of a YouTube video."""
    try:
        proxies = {"http": proxy, "https": proxy} if proxy else None
        transcript = YTTranscriber.get_transcript(video_url, languages=None, preserve_formatting=preserve_formatting, proxies=proxies)
        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)

# def main():
#     # Retrieve the space ID and token from environment variables
#     space_id = os.getenv("SPACE_ID")
#     token = os.getenv("HF_TOKEN")

#     # Initialize the HfApi with the retrieved token
#     api = HfApi(token=token)

#     while True:
#         try:
#             # Restart the space
#             api.restart_space(space_id, factory_reboot=False)
#             print(f"Successfully restarted the space: {space_id}")
#         except Exception as e:
#             print(f"Error restarting the space: {e}")

#         # Wait for 10 minutes before restarting again
#         time.sleep(600)  # Sleep for 600 seconds (10 minutes)

# if __name__ == "__main__":
#     main()