from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from webscout import WEBS, transcriber from webscout.websx_search import WEBSX from pydantic import BaseModel, Field from typing import List, Optional from fastapi.encoders import jsonable_encoder app = FastAPI() origins = [ "http://localhost", "http://localhost:8080", "http://127.0.0.1", "http://127.0.0.1:8080", "https://localhost", "https://localhost:8080", "https://127.0.0.1", "https://127.0.0.1:8080" ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class SearchQuery(BaseModel): q: str = Field(..., description="The search query string") max_results: int = Field(10, description="The maximum number of results to return", ge=1, le=100) timelimit: Optional[str] = Field(None, description="The time limit for the search (e.g., 'd' for day, 'w' for week, 'm' for month, 'y' for year)") safesearch: str = Field("moderate", description="The safe search level ('on', 'moderate', or 'off')") region: str = Field("wt-wt", description="The region for the search (e.g., 'us-en', 'uk-en', 'ru-ru')") backend: str = Field("api", description="The backend to use for search ('api', 'html', or 'lite')") class ImageSearchQuery(BaseModel): q: str = Field(..., description="The search query string") max_results: int = Field(10, description="The maximum number of results to return", ge=1, le=100) safesearch: str = Field("moderate", description="The safe search level ('on', 'moderate', or 'off')") region: str = Field("wt-wt", description="The region for the search (e.g., 'us-en', 'uk-en', 'ru-ru')") timelimit: Optional[str] = Field(None, description="The time limit for the search ('Day', 'Week', 'Month', or 'Year')") size: Optional[str] = Field(None, description="The size of the images ('Small', 'Medium', 'Large', or 'Wallpaper')") color: Optional[str] = Field(None, description="The color of the images ('color', 'Monochrome', 'Red', etc.)") type_image: Optional[str] = Field(None, description="The type of images ('photo', 'clipart', 'gif', etc.)") layout: Optional[str] = Field(None, description="The layout of the images ('Square', 'Tall', or 'Wide')") license_image: Optional[str] = Field(None, description="The license of the images ('any', 'Public', 'Share', etc.)") class VideoSearchQuery(BaseModel): q: str = Field(..., description="The search query string") max_results: int = Field(10, description="The maximum number of results to return", ge=1, le=100) safesearch: str = Field("moderate", description="The safe search level ('on', 'moderate', or 'off')") region: str = Field("wt-wt", description="The region for the search (e.g., 'us-en', 'uk-en', 'ru-ru')") timelimit: Optional[str] = Field(None, description="The time limit for the search (e.g., 'd' for day, 'w' for week, 'm' for month)") resolution: Optional[str] = Field(None, description="The resolution of the videos ('high' or 'standard')") duration: Optional[str] = Field(None, description="The duration of the videos ('short', 'medium', or 'long')") license_videos: Optional[str] = Field(None, description="The license of the videos ('creativeCommon' or 'youtube')") class NewsSearchQuery(BaseModel): q: str = Field(..., description="The search query string") max_results: int = Field(10, description="The maximum number of results to return", ge=1, le=100) safesearch: str = Field("moderate", description="The safe search level ('on', 'moderate', or 'off')") region: str = Field("wt-wt", description="The region for the search (e.g., 'us-en', 'uk-en', 'ru-ru')") timelimit: Optional[str] = Field(None, description="The time limit for the search (e.g., 'd' for day, 'w' for week, 'm' for month, 'y' for year)") class AnswersSearchQuery(BaseModel): q: str = Field(..., description="The search query string") class SuggestionsSearchQuery(BaseModel): q: str = Field(..., description="The search query string") region: str = Field("wt-wt", description="The region for suggestions (e.g., 'us-en', 'uk-en', 'ru-ru')") class MapsSearchQuery(BaseModel): q: str = Field(..., description="The search query string") place: Optional[str] = Field(None, description="Simplified search - if set, other location parameters are ignored") street: Optional[str] = Field(None, description="Street address") city: Optional[str] = Field(None, description="City") county: Optional[str] = Field(None, description="County") state: Optional[str] = Field(None, description="State") country: Optional[str] = Field(None, description="Country") postalcode: Optional[str] = Field(None, description="Postal code") latitude: Optional[str] = Field(None, description="Latitude (if used, other location parameters are ignored)") longitude: Optional[str] = Field(None, description="Longitude (if used, other location parameters are ignored)") radius: int = Field(0, description="Expand the search radius in kilometers") max_results: int = Field(10, description="The maximum number of results to return", ge=1, le=100) class TranslateSearchQuery(BaseModel): q: str = Field(..., description="The text to translate") from_: Optional[str] = Field(None, description="The source language (defaults to automatic detection)") to: str = Field("en", description="The target language (defaults to English)") class TranscriptQuery(BaseModel): video_id: str = Field(..., description="The YouTube video ID") languages: str = Field("en", description="Comma-separated list of language codes (e.g., 'en,es')") preserve_formatting: bool = Field(False, description="Whether to preserve text formatting") @app.get("/") async def root(): return {"message": "Welcome to HelpingAI API!"} @app.get("/health") async def health_check(): return {"status": "OK"} @app.get("/api/search") async def search(request: Request, query: SearchQuery): """Perform a text search.""" try: with WEBS() as webs: results = webs.text(keywords=query.q, region=query.region, safesearch=query.safesearch, timelimit=query.timelimit, backend=query.backend, max_results=query.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(request: Request, query: ImageSearchQuery): """Perform an image search.""" try: with WEBS() as webs: results = webs.images(keywords=query.q, region=query.region, safesearch=query.safesearch, timelimit=query.timelimit, size=query.size, color=query.color, type_image=query.type_image, layout=query.layout, license_image=query.license_image, max_results=query.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(request: Request, query: VideoSearchQuery): """Perform a video search.""" try: with WEBS() as webs: results = webs.videos(keywords=query.q, region=query.region, safesearch=query.safesearch, timelimit=query.timelimit, resolution=query.resolution, duration=query.duration, license_videos=query.license_videos, max_results=query.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(request: Request, query: NewsSearchQuery): """Perform a news search.""" try: with WEBS() as webs: results = webs.news(keywords=query.q, region=query.region, safesearch=query.safesearch, timelimit=query.timelimit, max_results=query.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/answers") async def answers(request: Request, query: AnswersSearchQuery): """Get instant answers for a query.""" try: with WEBS() as webs: results = webs.answers(keywords=query.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(request: Request, query: SuggestionsSearchQuery): """Get search suggestions for a query.""" try: with WEBS() as webs: results = webs.suggestions(keywords=query.q, region=query.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/maps") async def maps(request: Request, query: MapsSearchQuery): """Perform a maps search.""" try: with WEBS() as webs: results = webs.maps(keywords=query.q, place=query.place, street=query.street, city=query.city, county=query.county, state=query.state, country=query.country, postalcode=query.postalcode, latitude=query.latitude, longitude=query.longitude, radius=query.radius, max_results=query.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(request: Request, query: TranslateSearchQuery): """Translate text.""" try: with WEBS() as webs: results = webs.translate(keywords=query.q, from_=query.from_, to=query.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(request: Request, query: TranscriptQuery): """Get the transcript of a YouTube video.""" try: languages = query.languages.split(",") transcript = transcriber.get_transcript(query.video_id, languages=languages, preserve_formatting=query.preserve_formatting) return JSONResponse(content=jsonable_encoder(transcript)) except Exception as e: raise HTTPException(status_code=500, detail=f"Error getting YouTube transcript: {e}") # Run the API server if this script is executed if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080)