lango_backend / main.py
aliSaac510's picture
Update main.py
41f1c19 verified
# Load environment variables before anything else
from dotenv import load_dotenv
load_dotenv()
import os
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import RedirectResponse
from typing import List
# Updated import to include all providers
from providers import archive_provider, youtube_provider, ted_provider, dailymotion_provider, anime_provider, audiobook_provider, pexels_provider
from fastapi.middleware.cors import CORSMiddleware
from typing import Optional, List
app = FastAPI()
# --- CORS Middleware ---
allowed_origins = os.getenv("ALLOWED_ORIGINS")
origins = allowed_origins.split(",") if allowed_origins else []
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# --- Pexels Provider Routes ---
@app.get("/api/providers/pexels/search/photos")
def search_pexels_photos(
q: str,
per_page: int = 15,
page: int = 1,
orientation: Optional[str] = None,
size: Optional[str] = None,
color: Optional[str] = None
):
"""Searches for photos on Pexels."""
result = pexels_provider.search_photos(query=q, per_page=per_page, page=page, orientation=orientation, size=size, color=color)
_handle_provider_error(result)
return result
@app.get("/api/providers/pexels/search/videos")
def search_pexels_videos(
q: str,
per_page: int = 15,
page: int = 1,
orientation: Optional[str] = None,
size: Optional[str] = None
):
"""Searches for videos on Pexels."""
result = pexels_provider.search_videos(query=q, per_page=per_page, page=page, orientation=orientation, size=size)
_handle_provider_error(result)
return result
# --- Audiobook Provider Routes ---
@app.get("/api/providers/audiobook/search")
def search_audiobooks(q: str, by: str = "title"):
"""Searches for audiobooks by title or author."""
result = audiobook_provider.search(query=q, search_by=by)
_handle_provider_error(result)
return result
@app.get("/api/providers/audiobook/genres/{genre}")
def get_audiobooks_by_genre(genre: str):
"""Fetches audiobooks for a specific genre (only books with Gutenberg text available)."""
result = audiobook_provider.get_by_genre(genre=genre)
_handle_provider_error(result)
return result
@app.get("/api/providers/audiobook/content/{gutenberg_id}")
def get_audiobook_content(gutenberg_id: str):
"""Gets text content (cover image and clean text) from Gutenberg."""
result = audiobook_provider.get_content(gutenberg_id=gutenberg_id)
_handle_provider_error(result)
return result
# --- Anime Provider Routes ---
@app.get("/api/providers/anime/search")
def search_anime(q: str):
result = anime_provider.search(query=q)
_handle_provider_error(result)
return result
# --- Archive.org Provider Routes ---
# --- Helper for YouTube Error Handling ---
def _handle_youtube_error(result):
if "error" in result:
detail = result["error"]
status_code = 500 # Default to 500
if detail.startswith("YouTube API Error"):
parts = detail.split()
if len(parts) > 2 and parts[2][:-1].isdigit():
status_code = int(parts[2][:-1])
elif "not found" in detail.lower():
status_code = 404
raise HTTPException(status_code=status_code, detail=detail)
# --- Generic Error Handling for other providers ---
def _handle_provider_error(result):
if "error" in result:
if "not found" in result["error"].lower():
raise HTTPException(status_code=404, detail=result["error"])
else:
raise HTTPException(status_code=500, detail=result["error"])
# --- Root Redirect ---
@app.get("/")
def read_root():
return RedirectResponse(url="/docs")
# --- Archive.org Provider Routes ---
@app.get("/api/providers/archive/search")
def search_archive(q: str, has_transcript: bool = False):
result = archive_provider.search(q, has_transcript=has_transcript)
_handle_provider_error(result)
return result
@app.get("/api/providers/archive/items/{item_id}/details")
def get_item_details(item_id: str):
result = archive_provider.get_details(item_id)
_handle_provider_error(result)
return result
# --- Dailymotion Provider Routes ---
@app.get("/api/providers/dailymotion/search")
def search_dailymotion(query: str, has_transcript: bool = False, language: str = None):
result = dailymotion_provider.search(query=query, has_transcript=has_transcript, language=language)
_handle_provider_error(result)
return result
@app.get("/api/providers/dailymotion/items/{identifier}/details")
def get_dailymotion_details(identifier: str):
result = dailymotion_provider.get_details(identifier=identifier)
_handle_provider_error(result)
return result
@app.get("/api/providers/dailymotion/items/{identifier}/transcript")
def get_dailymotion_transcript(identifier: str, language_code: str = 'en'):
result = dailymotion_provider.get_transcript(identifier=identifier, language_code=language_code)
_handle_provider_error(result)
return result
# --- TED.com Provider Routes ---
@app.get("/api/providers/ted/items")
def search_ted_items(
query: str = None,
sort: str = None,
topics: List[str] = Query(None),
language: str = None,
duration: str = None
):
# Call the provider's search function with all the parameters
result = ted_provider.search(
query=query,
sort=sort,
topics=topics,
language=language,
duration=duration
)
_handle_provider_error(result)
return result
@app.get("/api/providers/ted/items/{talk_slug}/details")
def get_ted_details(talk_slug: str):
result = ted_provider.get_details(talk_slug)
_handle_provider_error(result)
return result
@app.get("/api/providers/ted/items/{talk_slug}/transcript")
def get_ted_transcript(talk_slug: str, language_code: str = 'en'):
result = ted_provider.get_transcript(talk_slug, language_code=language_code)
_handle_provider_error(result)
return result
# --- YouTube Provider Routes (NON-FUNCTIONAL) ---
# NOTE FROM THE ENGINEER (Gemini):
# The following YouTube endpoints are currently NON-FUNCTIONAL as of 2024-05-20.
# All API calls are unexpectedly returning a 404 Not Found error.
@app.get("/api/providers/youtube/search")
def search_youtube(q: str, has_transcript: bool = False, language: str = None):
result = youtube_provider.search(q, has_transcript=has_transcript, language=language)
_handle_youtube_error(result)
return result
@app.get("/api/providers/youtube/items/{video_id}/details")
def get_youtube_details(video_id: str):
result = youtube_provider.get_details(video_id)
_handle_youtube_error(result)
return result
@app.get("/api/providers/youtube/items/{video_id}/transcript")
def get_youtube_transcript(video_id: str, language_code: str = 'en'):
result = youtube_provider.get_transcript(video_id, language_code=language_code)
_handle_youtube_error(result)
return result