Spaces:
Sleeping
Sleeping
| # 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 --- | |
| 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 | |
| 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 --- | |
| 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 | |
| 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 | |
| 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 --- | |
| 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 --- | |
| def read_root(): | |
| return RedirectResponse(url="/docs") | |
| # --- Archive.org Provider Routes --- | |
| def search_archive(q: str, has_transcript: bool = False): | |
| result = archive_provider.search(q, has_transcript=has_transcript) | |
| _handle_provider_error(result) | |
| return result | |
| def get_item_details(item_id: str): | |
| result = archive_provider.get_details(item_id) | |
| _handle_provider_error(result) | |
| return result | |
| # --- Dailymotion Provider Routes --- | |
| 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 | |
| def get_dailymotion_details(identifier: str): | |
| result = dailymotion_provider.get_details(identifier=identifier) | |
| _handle_provider_error(result) | |
| return result | |
| 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 --- | |
| 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 | |
| def get_ted_details(talk_slug: str): | |
| result = ted_provider.get_details(talk_slug) | |
| _handle_provider_error(result) | |
| return result | |
| 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. | |
| 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 | |
| def get_youtube_details(video_id: str): | |
| result = youtube_provider.get_details(video_id) | |
| _handle_youtube_error(result) | |
| return result | |
| 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 | |