Create app.py

app.py

@@ -0,0 +1,200 @@
+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)