app.py

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()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# API key security
API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)

# Constants
INVIDIOUS_INSTANCES = [
    "https://invidious.privacydev.net",
    "https://invidious.reallyaweso.me",
    "https://invidious.adminforge.de"
]
API_KEY = os.environ.get("API_KEY")

# S3 Configuration
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'])  # Remove the temporary file
        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