|
import os |
|
from fastapi import FastAPI, HTTPException, UploadFile, Depends, Query |
|
from fastapi.security import APIKeyHeader |
|
import aiohttp |
|
import asyncio |
|
import json |
|
import tempfile |
|
from typing import List, Dict |
|
import logging |
|
import textract |
|
import boto3 |
|
from botocore.exceptions import NoCredentialsError |
|
from duckduckgo_search import DDGS |
|
from bs4 import BeautifulSoup |
|
|
|
app = FastAPI() |
|
|
|
|
|
logging.basicConfig(level=logging.INFO) |
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
API_KEY_NAME = "X-API-Key" |
|
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False) |
|
|
|
|
|
INVIDIOUS_INSTANCES = [ |
|
"https://invidious.privacydev.net", |
|
"https://invidious.reallyaweso.me", |
|
"https://invidious.adminforge.de" |
|
] |
|
API_KEY = os.environ.get("API_KEY") |
|
|
|
|
|
S3_ACCESS_KEY_ID = os.environ.get("S3_ACCESS_KEY_ID") |
|
S3_SECRET_ACCESS_KEY = os.environ.get("S3_SECRET_ACCESS_KEY") |
|
S3_BUCKET = os.environ.get("S3_BUCKET") |
|
S3_REGION = os.environ.get("S3_REGION") |
|
|
|
if not all([API_KEY, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_BUCKET, S3_REGION]): |
|
raise ValueError("Missing required environment variables") |
|
|
|
|
|
async def search_and_get_videos(query: str, num_videos: int = 2) -> List[Dict]: |
|
for instance in INVIDIOUS_INSTANCES: |
|
url = f"{instance}/api/v1/search?q={query}&type=video" |
|
try: |
|
async with aiohttp.ClientSession() as session: |
|
async with session.get(url) as response: |
|
response.raise_for_status() |
|
search_results = await response.json() |
|
videos = [ |
|
{ |
|
"id": video.get("videoId"), |
|
"title": video.get("title"), |
|
"thumbnail": video["videoThumbnails"][0]["url"] |
|
if video.get("videoThumbnails") |
|
else "", |
|
} |
|
for video in search_results |
|
][:num_videos] |
|
return videos |
|
except aiohttp.ClientError as e: |
|
logger.error(f"Error performing video search on {instance}: {e}") |
|
logger.error("All Invidious instances failed") |
|
return [] |
|
|
|
async def get_youtube_audio(video_id: str, max_retries: int = 3) -> Dict: |
|
for instance in INVIDIOUS_INSTANCES: |
|
for attempt in range(max_retries): |
|
try: |
|
url = f"{instance}/api/v1/videos/{video_id}" |
|
|
|
async with aiohttp.ClientSession() as session: |
|
async with session.get(url) as response: |
|
response.raise_for_status() |
|
video_data = await response.json() |
|
|
|
audio_format = next((format for format in video_data.get('adaptiveFormats', []) |
|
if format.get('type', '').startswith('audio/mp4')), None) |
|
|
|
if audio_format: |
|
audio_url = audio_format.get('url') |
|
if audio_url: |
|
try: |
|
async with session.get(audio_url) as audio_response: |
|
audio_content = await audio_response.read() |
|
|
|
with tempfile.NamedTemporaryFile(delete=False, suffix='.m4a') as temp_file: |
|
temp_file.write(audio_content) |
|
temp_file_path = temp_file.name |
|
|
|
return {'success': True, 'temp_file_path': temp_file_path} |
|
except aiohttp.ServerDisconnectedError: |
|
if attempt == max_retries - 1: |
|
logger.error(f"Max retries reached for video ID {video_id} on {instance}") |
|
break |
|
await asyncio.sleep(1 * (attempt + 1)) |
|
continue |
|
|
|
logger.warning(f"No suitable audio format found for video ID {video_id} on {instance}") |
|
break |
|
except aiohttp.ClientError as e: |
|
logger.error(f"Network error fetching YouTube audio for video ID {video_id} on {instance}: {e}") |
|
except json.JSONDecodeError: |
|
logger.error(f"Error decoding JSON response for video ID {video_id} on {instance}") |
|
except Exception as e: |
|
logger.error(f"Unexpected error fetching YouTube audio for video ID {video_id} on {instance}: {e}") |
|
if attempt == max_retries - 1: |
|
break |
|
await asyncio.sleep(1 * (attempt + 1)) |
|
|
|
return {'success': False, 'error': "Failed to fetch audio after multiple attempts on all instances"} |
|
|
|
def extract_text_from_document(file: UploadFile) -> dict: |
|
try: |
|
with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(file.filename)[1]) as temp_file: |
|
content = file.file.read() |
|
temp_file.write(content) |
|
temp_file_path = temp_file.name |
|
|
|
text = textract.process(temp_file_path).decode('utf-8') |
|
|
|
os.unlink(temp_file_path) |
|
|
|
return { |
|
'success': True, |
|
'extracted_text': text |
|
} |
|
except Exception as e: |
|
return { |
|
'success': False, |
|
'error': f"Error extracting text from document: {str(e)}" |
|
} |
|
|
|
def upload_to_s3(local_file, s3_file): |
|
s3_client = boto3.client( |
|
"s3", |
|
aws_access_key_id=S3_ACCESS_KEY_ID, |
|
aws_secret_access_key=S3_SECRET_ACCESS_KEY, |
|
region_name=S3_REGION, |
|
) |
|
|
|
try: |
|
s3_client.upload_file(local_file, S3_BUCKET, s3_file) |
|
s3_url = f"https://{S3_BUCKET}.s3.{S3_REGION}.amazonaws.com/{s3_file}" |
|
return s3_url |
|
except NoCredentialsError: |
|
logger.error("Credentials not available") |
|
return None |
|
|
|
def image_search(query: str, num_results: int = 5) -> dict: |
|
try: |
|
with DDGS( |
|
headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'} |
|
) as ddgs: |
|
results = list(ddgs.images(query, max_results=num_results)) |
|
formatted_results = [ |
|
{ |
|
'title': result['title'], |
|
'image_url': result['image'], |
|
'thumbnail_url': result['thumbnail'], |
|
'source_url': result['url'], |
|
'width': result['width'], |
|
'height': result['height'] |
|
} |
|
for result in results |
|
] |
|
return { |
|
'success': True, |
|
'results': formatted_results |
|
} |
|
except Exception as e: |
|
logger.error(f"Error performing image search: {e}") |
|
return { |
|
'success': False, |
|
'error': f"Error performing image search: {str(e)}" |
|
} |
|
|
|
async def verify_api_key(api_key: str = Depends(api_key_header)): |
|
if api_key != API_KEY: |
|
raise HTTPException(status_code=401, detail="Invalid API Key") |
|
return api_key |
|
|
|
@app.get("/search-videos/") |
|
async def search_videos( |
|
query: str, |
|
num_videos: int = Query(default=2, ge=1, le=10), |
|
api_key: str = Depends(verify_api_key) |
|
): |
|
videos = await search_and_get_videos(query, num_videos) |
|
if not videos: |
|
raise HTTPException(status_code=404, detail="No videos found or an error occurred during the search.") |
|
return {"videos": videos} |
|
|
|
@app.get("/get-audio/{video_id}") |
|
async def get_audio(video_id: str, api_key: str = Depends(verify_api_key)): |
|
result = await get_youtube_audio(video_id) |
|
if not result['success']: |
|
raise HTTPException(status_code=404, detail=result['error']) |
|
|
|
s3_file_name = f"{video_id}.m4a" |
|
s3_url = upload_to_s3(result['temp_file_path'], s3_file_name) |
|
|
|
if s3_url: |
|
os.unlink(result['temp_file_path']) |
|
return {"audio_url": s3_url} |
|
else: |
|
raise HTTPException(status_code=500, detail="Failed to upload audio to S3") |
|
|
|
@app.post("/extract-text/") |
|
async def extract_text(file: UploadFile, api_key: str = Depends(verify_api_key)): |
|
result = extract_text_from_document(file) |
|
if not result['success']: |
|
raise HTTPException(status_code=500, detail=result['error']) |
|
return {"extracted_text": result['extracted_text']} |
|
|
|
@app.get("/image-search/") |
|
async def image_search_endpoint(query: str, num_results: int = 5, api_key: str = Depends(verify_api_key)): |
|
result = image_search(query, num_results) |
|
if not result['success']: |
|
raise HTTPException(status_code=500, detail=result['error']) |
|
return result |
|
|
|
class DuckDuckGoSearch: |
|
async def search(self, query: str, num_results: int = 5) -> list: |
|
url = f"https://html.duckduckgo.com/html/?q={query}" |
|
headers = { |
|
"User-Agent": "Mozilla/5.0", |
|
"Referer": "https://google.com/", |
|
"Cookie": "kl=wt-wt", |
|
} |
|
|
|
async with aiohttp.ClientSession() as session: |
|
async with session.get(url, headers=headers) as response: |
|
if response.status != 200: |
|
raise Exception("Failed to fetch data from DuckDuckGo") |
|
|
|
html = await response.text() |
|
soup = BeautifulSoup(html, "html.parser") |
|
results = [] |
|
|
|
for result in soup.select(".result"): |
|
title = result.select_one(".result__title .result__a") |
|
url = result.select_one(".result__url") |
|
desc = result.select_one(".result__snippet") |
|
|
|
if title and url and desc: |
|
results.append({ |
|
"title": title.get_text(strip=True), |
|
"body": desc.get_text(strip=True), |
|
"href": f"https://{url.get_text(strip=True)}", |
|
}) |
|
|
|
if len(results) >= num_results: |
|
break |
|
|
|
return results |
|
|
|
async def web_search(query: str, num_results: int = 5) -> dict: |
|
try: |
|
results = await DuckDuckGoSearch().search(query, num_results) |
|
return { |
|
'success': True, |
|
'results': results |
|
} |
|
except Exception as e: |
|
return { |
|
'success': False, |
|
'error': str(e) |
|
} |
|
|
|
@app.get("/web-search/") |
|
async def web_search_endpoint(query: str, num_results: int = 5, api_key: str = Depends(verify_api_key)): |
|
result = await web_search(query, num_results) |
|
if not result['success']: |
|
raise HTTPException(status_code=500, detail=result['error']) |
|
return result |
