# app.py from fastapi import FastAPI, HTTPException, Response from fastapi.staticfiles import StaticFiles from pydantic import BaseModel import subprocess import os import shutil import logging from datetime import datetime import tempfile from pathlib import Path from dotenv import load_dotenv from urllib.parse import quote # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def env_to_cookies(env_content: str, output_file: str) -> None: """Convert environment variable content to cookie file""" try: # Extract content from env format if '="' not in env_content: raise ValueError("Invalid env content format") content = env_content.split('="', 1)[1].strip('"') # Replace escaped newlines with actual newlines cookie_content = content.replace('\\n', '\n') # Write to cookie file with open(output_file, 'w') as f: f.write(cookie_content) logger.info(f"Successfully created cookie file at {output_file}") except Exception as e: logger.error(f"Error creating cookie file: {str(e)}") raise ValueError(f"Error converting to cookie file: {str(e)}") def get_cookies() -> str: """Get cookies from environment variable""" load_dotenv() cookie_content = os.getenv('COOKIES') if not cookie_content: raise ValueError("COOKIES environment variable not set") return cookie_content def env_to_cookies_from_env(output_file: str) -> None: """Convert environment variable from .env file to cookie file""" try: load_dotenv() env_content = os.getenv('COOKIES') logger.info("Retrieved cookies from environment variable") if not env_content: raise ValueError("COOKIES not found in environment variables") env_to_cookies(f'COOKIES="{env_content}"', output_file) except Exception as e: logger.error(f"Error creating cookie file from env: {str(e)}") raise ValueError(f"Error converting to cookie file: {str(e)}") app = FastAPI( title="GAMDL API", description="API for downloading Google Drive files using gamdl", version="1.0.0" ) # Create downloads directory if it doesn't exist DOWNLOADS_DIR = "downloads" os.makedirs(DOWNLOADS_DIR, exist_ok=True) # Mount the downloads directory with cache control headers class CacheControlStaticFiles(StaticFiles): def __init__(self, directory: str): super().__init__(directory=directory) async def get_response(self, path: str, scope): response = await super().get_response(path, scope) response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" response.headers["Pragma"] = "no-cache" response.headers["Expires"] = "0" return response app.mount("/files", CacheControlStaticFiles(directory=DOWNLOADS_DIR), name="files") class DownloadRequest(BaseModel): url: str class DownloadResponse(BaseModel): success: bool message: str filename: str = None download_url: str = None file_size: int = None # Include file size in the response @app.post("/download", response_model=DownloadResponse) async def download_file(request: DownloadRequest): try: # Create a unique subdirectory for this download timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") download_subdir = os.path.join(DOWNLOADS_DIR, timestamp) os.makedirs(download_subdir, exist_ok=True) # Log the current working directory and download directory logger.info(f"Current working directory: {os.getcwd()}") logger.info(f"Download directory: {download_subdir}") # Create cookies file from environment variable cookie_path = os.path.join(download_subdir, "cookies.txt") logger.info(f"Creating cookies file at: {cookie_path}") env_to_cookies_from_env(cookie_path) # Change to download directory original_dir = os.getcwd() os.chdir(download_subdir) # Log the command being executed cmd = ["votify", "--audio-quality", "vorbis-medium", request.url] logger.info(f"Executing command: {' '.join(cmd)}") # Run gamdl command with more detailed output process = subprocess.run( cmd, capture_output=True, text=True ) # Log the command output logger.info(f"Command stdout: {process.stdout}") logger.info(f"Command stderr: {process.stderr}") # Check if the command was successful process.check_returncode() # Get the downloaded filename files = [f for f in os.listdir() if f != "cookies.txt"] logger.info(f"Files in download directory: {files}") if not files: raise Exception("No files found in download directory after download attempt") downloaded_file = files[0] logger.info(f"Downloaded file: {downloaded_file}") # Get the file size file_size = os.path.getsize(downloaded_file) # Generate the download URL and URL encode the filename space_url = os.getenv("SPACE_URL", "https://tecuts-vob.hf.space") encoded_filename = quote(downloaded_file) # URL encode the filename download_url = f"{space_url}/files/{timestamp}/{encoded_filename}" logger.info(f"Generated download URL: {download_url}") # Move back to original directory return DownloadResponse( success=True, message="File downloaded successfully", filename=downloaded_file, download_url=download_url, file_size=file_size # Include the file size in the response ) except subprocess.CalledProcessError as e: logger.error(f"Download process failed: stdout={e.stdout}, stderr={e.stderr}") raise HTTPException( status_code=400, detail=f"Failed to download: {e.stderr or e.stdout or str(e)}" ) except Exception as e: logger.error(f"Unexpected error: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail=f"Error: {str(e)}" ) finally: # Always try to return to the original directory if 'original_dir' in locals(): os.chdir(original_dir) @app.get("/test") async def test(): """Test endpoint to verify setup""" try: # Test cookie creation temp_cookie = os.path.join(DOWNLOADS_DIR, "test_cookies.txt") env_to_cookies_from_env(temp_cookie) # Test gamdl installation process = subprocess.run(["votify", "--version"], capture_output=True, text=True) return { "gamdl_version": process.stdout.strip(), "cookies_created": os.path.exists(temp_cookie), "cookies_size": os.path.getsize(temp_cookie) if os.path.exists(temp_cookie) else 0, "installed": True, "error": process.stderr if process.stderr else None } except Exception as e: return { "installed": False, "error": str(e) }