from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, Form, File, UploadFile, Query, status from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.middleware.cors import CORSMiddleware from fastapi import Request import json import asyncio import os import configparser import logging # Load environment variables from .env file if available try: from dotenv import load_dotenv load_dotenv() except ImportError: # python-dotenv not installed, use os.environ directly pass from typing import List, Dict, Any, Optional from datetime import datetime from urllib.parse import urlencode # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Import authentication module from .auth import init_oauth_manager, get_oauth_manager # Import scheduled tasks modules from .database import init_database from .scheduler import start_scheduler, stop_scheduler from .scheduled_tasks import router as scheduled_tasks_router # Import BDFR API layer import sys import os # Add the parent directory (BDFR root) to Python path bdfr_root = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), 'bdfr') sys.path.insert(0, bdfr_root) # Also add the current bdfr directory to path sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), '')) try: from bdfr.api import BDFRManager, ProgressEvent, ProgressCallback, get_bdfr_manager BDFR_AVAILABLE = True except ImportError as e: logger.warning(f"Failed to import BDFR API: {e}. BDFR modules may not be available.") BDFR_AVAILABLE = False # Create a mock BDFRManager for when BDFR is not available class MockBDFRManager: def __init__(self, *args, **kwargs): pass def download_subreddit(self, *args, **kwargs): raise NotImplementedError("BDFR not available") def download_user(self, *args, **kwargs): raise NotImplementedError("BDFR not available") def get_download_status(self, *args, **kwargs): return None def cancel_download(self, *args, **kwargs): return False class MockProgressCallback: pass class MockProgressEvent: def __init__(self, *args, **kwargs): pass BDFRManager = MockBDFRManager ProgressEvent = MockProgressEvent ProgressCallback = MockProgressCallback def get_bdfr_manager(*args, **kwargs): return MockBDFRManager() app = FastAPI(title="BDFR Web Interface", version="1.0.0") # Application lifecycle events @app.on_event("startup") async def startup_event(): """Initialize services on application startup""" try: logger.info("Starting up BDFR Web Interface...") # Initialize database init_database() logger.info("Database initialized") # Ensure database schema is up to date before starting scheduler from .database import upgrade_database_schema upgrade_database_schema() logger.info("Database schema check complete") # Start scheduler start_scheduler() logger.info("Scheduler started") logger.info("Startup complete!") except Exception as e: logger.error(f"Startup error: {e}", exc_info=True) raise @app.on_event("shutdown") async def shutdown_event(): """Cleanup on application shutdown""" try: logger.info("Shutting down BDFR Web Interface...") # Stop scheduler stop_scheduler() logger.info("Scheduler stopped") # Stop task queue from .task_queue import task_queue await task_queue.stop() logger.info("Task queue stopped") logger.info("Shutdown complete!") except Exception as e: logger.error(f"Shutdown error: {e}", exc_info=True) # Initialize OAuth2 manager def init_oauth(): """Initialize OAuth2 manager with credentials from environment or BDFR config""" try: # Try to get credentials from environment variables first client_id = os.getenv("BDFR_CLIENT_ID") client_secret = os.getenv("BDFR_CLIENT_SECRET") if client_id and client_secret: # Use environment credentials logger.info("Using OAuth credentials from environment variables") scopes = ["identity", "history", "read", "save", "mysubreddits"] else: # Fall back to BDFR config file logger.info("Using OAuth credentials from BDFR config file") config = configparser.ConfigParser() config.read("../bdfr/default_config.cfg") client_id = config.get("DEFAULT", "client_id") client_secret = config.get("DEFAULT", "client_secret") scopes_str = config.get("DEFAULT", "scopes", fallback="identity,read") # Parse scopes scopes = [scope.strip() for scope in scopes_str.split(",")] init_oauth_manager(client_id, client_secret, scopes) logger.info("OAuth2 manager initialized successfully") except Exception as e: logger.error(f"Failed to initialize OAuth2 manager: {e}") # Use default credentials if config fails init_oauth_manager("U-6gk4ZCh3IeNQ", "7CZHY6AmKweZME5s50SfDGylaPg") # Initialize OAuth2 on startup # Use a configurable redirect URI - this should match your Reddit OAuth app settings redirect_uri = os.getenv("BDFR_REDIRECT_URI", "http://localhost:8000/auth/callback") init_oauth() # Enable CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Mount static files import os current_dir = os.path.dirname(os.path.abspath(__file__)) static_dir = os.path.join(current_dir, "..", "static") template_dir = os.path.join(current_dir, "..", "templates") # Ensure directories exist os.makedirs(static_dir, exist_ok=True) os.makedirs(template_dir, exist_ok=True) app.mount("/static", StaticFiles(directory=static_dir), name="static") templates = Jinja2Templates(directory=template_dir) # Include scheduled tasks router app.include_router(scheduled_tasks_router) # WebSocket connection manager class ConnectionManager: def __init__(self): self.active_connections: List[WebSocket] = [] async def connect(self, websocket: WebSocket): logger.info(f"[WEBSOCKET] WebSocket connection attempt from {websocket.client}") await websocket.accept() self.active_connections.append(websocket) logger.info(f"[WEBSOCKET] New connection established from {websocket.client}. Total connections: {len(self.active_connections)}") def disconnect(self, websocket: WebSocket): if websocket in self.active_connections: self.active_connections.remove(websocket) logger.info(f"[WEBSOCKET] Connection disconnected from {websocket.client}. Total connections: {len(self.active_connections)}") async def send_personal_message(self, message: str, websocket: WebSocket): try: await websocket.send_text(message) except Exception as e: logger.warning(f"Failed to send personal message: {e}") self.disconnect(websocket) async def broadcast(self, message: str): """Broadcast message to all active connections - simplified version""" logger.debug(f"[WEBSOCKET] Broadcasting message to {len(self.active_connections)} connections") if not self.active_connections: logger.debug("[WEBSOCKET] No active connections to broadcast to!") return # Simple approach: try to send to all connections, remove failed ones immediately alive_connections = [] for connection in self.active_connections: try: await connection.send_text(message) alive_connections.append(connection) logger.debug(f"[WEBSOCKET] Message sent successfully to connection {id(connection)}") except Exception as e: logger.warning(f"[WEBSOCKET] Removing failed connection {id(connection)}: {e}") # Connection failed, don't add it to alive_connections # Update active connections to only include successful ones self.active_connections = alive_connections logger.debug(f"[WEBSOCKET] Broadcast complete. Active connections: {len(self.active_connections)}") async def is_connection_alive(self, websocket: WebSocket) -> bool: """Check if a WebSocket connection is still alive""" try: # Try to send a ping frame (this is a WebSocket protocol ping) await websocket.ping() return True except Exception: return False async def cleanup_dead_connections(self): """Simplified cleanup - just log the current state""" logger.info(f"[WEBSOCKET] Cleanup check: {len(self.active_connections)} active connections") manager = ConnectionManager() # Initialize BDFR Manager bdfr_manager = get_bdfr_manager("./downloads") # Store auth token for BDFR manager if available _bdfr_auth_token = None # Active downloads tracking (now managed by BDFRManager, but kept for WebSocket compatibility) active_downloads = {} # WebSocket-compatible progress callback class WebSocketProgressCallback(ProgressCallback if BDFR_AVAILABLE else MockProgressCallback): """Progress callback that sends updates to WebSocket clients""" def __init__(self, download_id: str, connection_manager: ConnectionManager): self.download_id = download_id self.connection_manager = connection_manager async def on_progress(self, event: ProgressEvent): """Send progress update to WebSocket clients""" logger.info(f"[WEBSOCKET-PROGRESS] Received progress event for download {event.download_id}: {event.message} ({event.progress}%)") logger.info(f"[WEBSOCKET-PROGRESS] Active connections before broadcast: {len(self.connection_manager.active_connections)}") try: # Convert BDFR API event to web interface format progress_data = { "type": "progress", "id": event.download_id, "download_id": event.download_id, "status": "running", "message": event.message, "progress": event.progress or 0, "data": event.data, "timestamp": event.timestamp.isoformat() } # Update active_downloads for WebSocket compatibility # Need to find the web interface download ID that corresponds to this BDFR download ID web_download_id = None for download_id, download_info in active_downloads.items(): if download_info.get("bdfr_download_id") == event.download_id: web_download_id = download_id break if web_download_id: # Keep server-side state in sync active_downloads[web_download_id]["status"] = "running" active_downloads[web_download_id]["progress"] = event.progress or 0 if "items_processed" in event.data: active_downloads[web_download_id]["items_processed"] = event.data["items_processed"] if "items_found" in event.data: active_downloads[web_download_id]["items_found"] = event.data["items_found"] if "current_item" in event.data: active_downloads[web_download_id]["current_item"] = event.data["current_item"] if "phase" in event.data: active_downloads[web_download_id]["phase"] = event.data["phase"] # Unify IDs for frontend to prevent duplicate cards progress_data["id"] = web_download_id progress_data["download_id"] = web_download_id progress_data["bdfr_download_id"] = event.download_id progress_data["web_download_id"] = web_download_id progress_data["subreddit"] = active_downloads[web_download_id].get("subreddit") progress_data["username"] = active_downloads[web_download_id].get("username") progress_data["limit"] = active_downloads[web_download_id].get("limit") else: # No mapping yet; include bdfr id for debugging progress_data["bdfr_download_id"] = event.download_id progress_data["limit"] = (event.data or {}).get("limit") logger.info(f"[WEBSOCKET-PROGRESS] Broadcasting progress data: {progress_data}") logger.info(f"[WEBSOCKET-PROGRESS] About to broadcast progress message for download {event.download_id}") await self.connection_manager.broadcast(json.dumps(progress_data)) logger.info(f"[WEBSOCKET-PROGRESS] Progress message broadcast completed for download {event.download_id}") except Exception as e: logger.warning(f"Failed to send progress update for {event.download_id}: {e}") async def on_error(self, event: ProgressEvent): """Send error update to WebSocket clients""" logger.info(f"[WEBSOCKET-ERROR] Received error event for download {event.download_id}: {event.message}") logger.info(f"[WEBSOCKET-ERROR] Event data: {event.data}") try: error_data = { "type": "error", "id": event.download_id, "download_id": event.download_id, "status": "failed", "message": event.message, "data": event.data, "timestamp": event.timestamp.isoformat() } # Update active_downloads - need to find the web interface download ID # that corresponds to this BDFR download ID web_download_id = None for download_id, download_info in active_downloads.items(): if download_info.get("bdfr_download_id") == event.download_id: web_download_id = download_id break if web_download_id: active_downloads[web_download_id]["status"] = "failed" active_downloads[web_download_id]["error"] = event.message active_downloads[web_download_id]["end_time"] = datetime.now().isoformat() # Persist failure phase for status_update broadcasting (e.g., 'rate_limited') try: if isinstance(event.data, dict) and event.data.get("phase"): active_downloads[web_download_id]["phase"] = event.data.get("phase") except Exception: pass logger.info(f"Updated download {web_download_id} to failed status") # Unify IDs for frontend and include mapping error_data["id"] = web_download_id error_data["download_id"] = web_download_id error_data["bdfr_download_id"] = event.download_id error_data["web_download_id"] = web_download_id error_data["subreddit"] = active_downloads[web_download_id].get("subreddit") error_data["username"] = active_downloads[web_download_id].get("username") # Also surface phase at top-level for clients that check data.phase or phase try: if isinstance(event.data, dict) and event.data.get("phase"): error_data["phase"] = event.data.get("phase") except Exception: pass else: error_data["bdfr_download_id"] = event.download_id await self.connection_manager.broadcast(json.dumps(error_data)) except Exception as e: logger.warning(f"Failed to send error update for {event.download_id}: {e}") logger.warning(f"Error data type: {type(event.data.get('exception'))}") async def on_completed(self, event: ProgressEvent): """Send completion update to WebSocket clients""" logger.info(f"[WEBSOCKET-COMPLETED] Received completion event for download {event.download_id}: {event.message}") try: completed_data = { "type": "completed", "id": event.download_id, "download_id": event.download_id, "status": "completed", "message": event.message, "progress": 100.0, "data": event.data, "timestamp": event.timestamp.isoformat() } # Update active_downloads - need to find the web interface download ID # that corresponds to this BDFR download ID web_download_id = None for download_id, download_info in active_downloads.items(): if download_info.get("bdfr_download_id") == event.download_id: web_download_id = download_id break if web_download_id: active_downloads[web_download_id]["status"] = "completed" active_downloads[web_download_id]["progress"] = 100.0 active_downloads[web_download_id]["end_time"] = datetime.now().isoformat() logger.info(f"Updated download {web_download_id} to completed status") # Unify IDs and include mapping data completed_data["id"] = web_download_id completed_data["download_id"] = web_download_id completed_data["bdfr_download_id"] = event.download_id completed_data["web_download_id"] = web_download_id completed_data["subreddit"] = active_downloads[web_download_id].get("subreddit") completed_data["username"] = active_downloads[web_download_id].get("username") else: completed_data["bdfr_download_id"] = event.download_id await self.connection_manager.broadcast(json.dumps(completed_data)) except Exception as e: logger.warning(f"Failed to send completion update for {event.download_id}: {e}") # Helper function to create download with BDFR API async def create_download_with_bdfr_api(download_type: str, name: str, **kwargs): """Create a download using the new BDFR API layer""" # Check if BDFR is available if not hasattr(bdfr_manager, 'download_subreddit'): # BDFR not available, create a mock failed download download_id = f"{download_type}_{name.replace('/', '_').replace(' ', '_')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}" active_downloads[download_id] = { "id": download_id, "type": download_type, "status": "failed", "progress": 0, "start_time": datetime.now().isoformat(), "authenticated": False, "items_processed": 0, "items_found": 0, "current_item": None, "current_item_type": None, "phase": "failed", "error": "BDFR API not available" } # Add type-specific fields if download_type in ["subreddit", "archive", "clone"]: active_downloads[download_id]["subreddit"] = name elif download_type == "user": active_downloads[download_id]["username"] = name return download_id # Get global auth token if available global _bdfr_auth_token auth_token = _bdfr_auth_token logger.info(f"[DEBUG] Global auth token available: {auth_token is not None}") logger.info(f"[DEBUG] Auth token for BDFR: {auth_token[:10]}..." if auth_token else "None") # Create unique download ID download_id = f"{download_type}_{name.replace('/', '_').replace(' ', '_')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}" # Get auth token if provided auth_token = None logger.info(f"[DEBUG] Checking auth_state: {kwargs.get('auth_state')}") if kwargs.get('auth_state'): try: oauth_manager = get_oauth_manager() logger.info(f"[DEBUG] OAuth manager exists: {oauth_manager is not None}") auth_token = oauth_manager.get_valid_token(kwargs['auth_state']) logger.info(f"[DEBUG] Auth token retrieved for state {kwargs['auth_state']}: {auth_token[:10]}..." if auth_token else "None") except Exception as e: logger.warning(f"Failed to get auth token: {e}") # Initialize download tracking for WebSocket compatibility active_downloads[download_id] = { "id": download_id, "type": download_type, "status": "queued", "progress": 0, "start_time": datetime.now().isoformat(), "limit": kwargs.get('limit'), "authenticated": auth_token is not None, "items_processed": 0, "items_found": 0, "current_item": None, "current_item_type": None, "phase": "queued" } # Add type-specific fields if download_type in ["subreddit", "archive", "clone"]: active_downloads[download_id]["subreddit"] = name elif download_type == "user": active_downloads[download_id]["username"] = name # Create progress callback callback = WebSocketProgressCallback(download_id, manager) # Start download based on type if download_type == "subreddit": bdfr_download_id = bdfr_manager.download_subreddit( name, limit=kwargs.get('limit'), sort=kwargs.get('sort', 'hot'), time_filter=kwargs.get('time_filter', 'all'), no_dupes=kwargs.get('no_dupes', False), progress_callbacks=[callback] ) elif download_type == "archive": # Archive mode - metadata only bdfr_download_id = bdfr_manager.archive_subreddit( name, format_type=kwargs.get('format', 'json'), limit=kwargs.get('limit'), progress_callbacks=[callback] ) elif download_type == "clone": # Clone mode - both download and archive bdfr_download_id = bdfr_manager.clone_subreddit( name, limit=kwargs.get('limit'), format_type=kwargs.get('format', 'json'), no_dupes=kwargs.get('no_dupes', False), progress_callbacks=[callback] ) elif download_type == "user": # Set auth token in the BDFR manager if provided if auth_token: bdfr_manager.auth_token = auth_token logger.info(f"[DEBUG] Set auth token in BDFR manager for user download") bdfr_download_id = bdfr_manager.download_user( name, limit=kwargs.get('limit'), submitted=kwargs.get('submitted', True), upvoted=kwargs.get('upvoted', False), saved=kwargs.get('saved', False), no_dupes=kwargs.get('no_dupes', False), progress_callbacks=[callback] ) else: raise ValueError(f"Unsupported download type: {download_type}") # Store BDFR download ID for tracking active_downloads[download_id]["bdfr_download_id"] = bdfr_download_id return download_id @app.get("/", response_class=HTMLResponse) async def home(request: Request): """Serve the main interface""" return templates.TemplateResponse("index.html", {"request": request}) @app.get("/health") async def health_check(): """Health check endpoint""" return {"status": "healthy", "timestamp": datetime.now().isoformat()} # OAuth2 Authentication Endpoints @app.get("/auth/login") async def oauth_login(redirect_uri: str = None): """Initiate OAuth2 login flow""" try: # Use provided redirect_uri or fall back to configured default if redirect_uri is None: redirect_uri = os.getenv("BDFR_REDIRECT_URI", "http://localhost:8000/auth/callback") oauth_manager = get_oauth_manager() auth_data = oauth_manager.get_authorization_url(redirect_uri) return { "authorization_url": auth_data["authorization_url"], "state": auth_data["state"], "redirect_uri": redirect_uri, "message": "Redirect user to the authorization URL" } except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to initiate OAuth2 login: {str(e)}" ) @app.get("/auth/callback") async def oauth_callback( request: Request, code: str = Query(..., description="Authorization code from Reddit"), state: str = Query(..., description="State parameter for security"), error: Optional[str] = Query(None, description="Error from OAuth2 provider") ): """Handle OAuth2 callback""" try: if error: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"OAuth2 error: {error}" ) if not code or not state: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Missing code or state parameter" ) oauth_manager = get_oauth_manager() # Use the same redirect URI that was used for authorization redirect_uri = os.getenv("BDFR_REDIRECT_URI", "http://localhost:8000/auth/callback") token_data = await oauth_manager.exchange_code_for_token(code, state, redirect_uri) # Get user info user_info = await oauth_manager.get_user_info(token_data["access_token"]) # Return success page instead of JSON return templates.TemplateResponse("auth_success.html", {"request": request}) except HTTPException: raise except Exception as e: logger.error(f"OAuth2 callback error: {str(e)}") # Return error page for unexpected errors return templates.TemplateResponse("auth_error.html", { "request": request, "error": "Authentication failed", "details": str(e) }) except HTTPException: raise except Exception as e: logger.error(f"OAuth2 callback error: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Authentication failed" ) @app.get("/auth/status") async def auth_status(state: Optional[str] = Query(None)): """Get current authentication status""" try: oauth_manager = get_oauth_manager() status_data = oauth_manager.get_auth_status(state) return { "authenticated": status_data["authenticated"], "message": status_data.get("message", "Authenticated" if status_data["authenticated"] else "Not authenticated"), "expires_at": status_data.get("expires_at"), "scopes": status_data.get("scopes"), "username": status_data.get("username") } except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get auth status: {str(e)}" ) @app.get("/api/auth/status") async def api_auth_status(state: Optional[str] = Query(None)): """Get current authentication status (JSON API)""" try: oauth_manager = get_oauth_manager() # If no state provided, check if there are any active sessions if not state and oauth_manager: # Check if there are any stored access tokens if oauth_manager.access_tokens: # Use the first available state for checking state = next(iter(oauth_manager.access_tokens.keys())) logger.info(f"No state provided, using first available: {state}") status_data = oauth_manager.get_auth_status(state) if oauth_manager else {"authenticated": False, "message": "No OAuth manager"} # Debug logging logger.info(f"Auth status check - State: {state}, Authenticated: {status_data['authenticated']}") # Store auth token for BDFR if authenticated global _bdfr_auth_token if status_data['authenticated'] and state: _bdfr_auth_token = oauth_manager.get_valid_token(state) logger.info(f"Stored auth token for BDFR manager: {_bdfr_auth_token is not None}") elif not status_data['authenticated']: _bdfr_auth_token = None logger.info("Cleared auth token for BDFR manager") return { "authenticated": status_data["authenticated"], "message": status_data.get("message", "Authenticated" if status_data["authenticated"] else "Not authenticated"), "expires_at": status_data.get("expires_at", 0), "scopes": status_data.get("scopes", []), "username": status_data.get("username"), "debug_info": { "state_provided": state is not None, "state_used": state, "oauth_manager_exists": oauth_manager is not None, "available_states": len(oauth_manager.oauth_states) if oauth_manager else 0, "available_tokens": len(oauth_manager.access_tokens) if oauth_manager else 0, "bdfr_auth_token_set": _bdfr_auth_token is not None } } except Exception as e: logger.error(f"Auth status error: {str(e)}") return { "authenticated": False, "message": f"Error checking auth status: {str(e)}", "error": True, "debug_info": { "error_details": str(e) } } @app.post("/auth/logout") async def auth_logout(state: str = Form(...)): """Logout and revoke OAuth2 session""" try: oauth_manager = get_oauth_manager() oauth_manager.revoke_session(state) return { "message": "Successfully logged out", "state": state } except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Logout failed: {str(e)}" ) @app.post("/api/download/subreddit") async def download_subreddit( subreddit: str = Form(...), limit: int = Form(10), sort: str = Form("hot"), time_filter: str = Form(""), min_score: str = Form(""), no_dupes: bool = Form(False), simple_check: bool = Form(False), make_hard_links: bool = Form(False), download_mode: str = Form("download"), auth_state: str = Form(None) ): """Download from subreddit using BDFR API""" # Map download_mode to appropriate BDFR operation if download_mode == "archive": # Use archive mode download_id = await create_download_with_bdfr_api( "archive", subreddit, limit=limit, sort=sort, time_filter=time_filter or "all", format="json", simple_check=simple_check, auth_state=auth_state ) message = f"Starting archive for r/{subreddit}" elif download_mode == "clone": # Use clone mode (download + archive) download_id = await create_download_with_bdfr_api( "clone", subreddit, limit=limit, sort=sort, time_filter=time_filter or "all", no_dupes=no_dupes, simple_check=simple_check, format="json", auth_state=auth_state ) message = f"Starting clone for r/{subreddit}" else: # Default: download mode download_id = await create_download_with_bdfr_api( "subreddit", subreddit, limit=limit, sort=sort, time_filter=time_filter or "all", no_dupes=no_dupes, simple_check=simple_check, auth_state=auth_state ) message = f"Starting download for r/{subreddit}" return { "download_id": download_id, "message": message, "mode": download_mode, "estimated_time": "2-3 minutes" } @app.post("/api/download/user") async def download_user( username: str = Form(...), limit: int = Form(10), submitted: bool = Form(True), sort: str = Form("hot"), time_filter: str = Form(""), no_dupes: bool = Form(False), simple_check: bool = Form(False), make_hard_links: bool = Form(False), download_mode: str = Form("download"), auth_state: str = Form(None) ): """Download from user using BDFR API""" # Map download_mode to appropriate BDFR operation if download_mode == "archive": # Use archive mode download_id = await create_download_with_bdfr_api( "archive", username, limit=limit, sort=sort, time_filter=time_filter or "all", format="json", submitted=submitted, simple_check=simple_check, auth_state=auth_state ) message = f"Starting archive for u/{username}" elif download_mode == "clone": # Use clone mode (download + archive) download_id = await create_download_with_bdfr_api( "clone", username, limit=limit, sort=sort, time_filter=time_filter or "all", no_dupes=no_dupes, simple_check=simple_check, format="json", submitted=submitted, auth_state=auth_state ) message = f"Starting clone for u/{username}" else: # Default: download mode download_id = await create_download_with_bdfr_api( "user", username, limit=limit, sort=sort, submitted=submitted, no_dupes=no_dupes, simple_check=simple_check, auth_state=auth_state ) message = f"Starting download for u/{username}" return { "download_id": download_id, "message": message, "mode": download_mode, "estimated_time": "1-2 minutes" } @app.get("/api/downloads") async def get_downloads(): """Get all active downloads""" return {"downloads": active_downloads} @app.get("/api/downloads/{download_id}") async def get_download_status(download_id: str): """Get specific download status""" if download_id not in active_downloads: raise HTTPException(status_code=404, detail="Download not found") return {"download": active_downloads[download_id]} @app.delete("/api/downloads/{download_id}") async def cancel_download(download_id: str): """Cancel a download""" if download_id not in active_downloads: raise HTTPException(status_code=404, detail="Download not found") # Cancel in BDFR manager if we have the BDFR download ID if "bdfr_download_id" in active_downloads[download_id]: bdfr_manager.cancel_download(active_downloads[download_id]["bdfr_download_id"]) active_downloads[download_id]["status"] = "cancelled" return {"message": "Download cancelled"} @app.websocket("/ws/progress") async def websocket_endpoint(websocket: WebSocket): """WebSocket endpoint for real-time progress updates""" logger.info(f"[WEBSOCKET-ENDPOINT] New WebSocket connection attempt from {websocket.client}") await manager.connect(websocket) logger.info(f"[WEBSOCKET-ENDPOINT] WebSocket connection established successfully") logger.info(f"[WEBSOCKET-ENDPOINT] Connection count: {len(manager.active_connections)}") # Track last broadcast state to avoid redundant messages last_broadcast_state = {} cleanup_counter = 0 keepalive_counter = 0 try: while True: # Clean up dead connections periodically (every 10 iterations = 20 seconds) cleanup_counter += 1 if cleanup_counter >= 10: await manager.cleanup_dead_connections() cleanup_counter = 0 # Send keepalive ping every 15 iterations (30 seconds) to prevent timeout keepalive_counter += 1 if keepalive_counter >= 15: try: await websocket.send_json({"type": "keepalive", "timestamp": datetime.now().isoformat()}) logger.debug(f"[WEBSOCKET-ENDPOINT] Sent keepalive ping") except Exception as e: logger.warning(f"[WEBSOCKET-ENDPOINT] Failed to send keepalive: {e}") break keepalive_counter = 0 # Check for changes in download states has_changes = False for download_id, download in list(active_downloads.items()): # Only check active downloads if download.get("status") in ["running", "queued", "failed"]: # Create a state snapshot for comparison current_state = { "status": download.get("status"), "progress": download.get("progress", 0), "phase": download.get("phase", "unknown"), "items_processed": download.get("items_processed", 0), "items_found": download.get("items_found", 0), "current_item": download.get("current_item"), } # Compare with last broadcast state if download_id not in last_broadcast_state or last_broadcast_state[download_id] != current_state: has_changes = True last_broadcast_state[download_id] = current_state logger.debug(f"[WEBSOCKET-ENDPOINT] State change detected for {download_id}: {download.get('status')}") status_message = { "type": "status_update", "id": download_id, "download_id": download_id, "status": download["status"], "progress": download.get("progress", 0), "phase": download.get("phase", "unknown"), "items_processed": download.get("items_processed", 0), "items_found": download.get("items_found", 0), "limit": download.get("limit", 0), "current_item": download.get("current_item"), "current_item_type": download.get("current_item_type"), "current_subreddit": download.get("current_subreddit"), "username": download.get("username"), "subreddit": download.get("subreddit"), "message": f"{download.get('subreddit', download.get('username', 'content'))} - {download['status']}" } try: await manager.broadcast(json.dumps(status_message)) except Exception as e: logger.warning(f"Error broadcasting status update: {e}") if not has_changes: logger.debug(f"[WEBSOCKET-ENDPOINT] No state changes detected, skipping broadcast") await asyncio.sleep(2) # Check every 2 seconds except WebSocketDisconnect: logger.info(f"[WEBSOCKET-ENDPOINT] WebSocket disconnected normally from {websocket.client}") manager.disconnect(websocket) except Exception as e: logger.error(f"[WEBSOCKET-ENDPOINT] WebSocket error from {websocket.client}: {e}") manager.disconnect(websocket) @app.get("/api/bdfr/status") async def bdfr_status(): """Get BDFR status and available options""" return { "bdfr_available": BDFR_AVAILABLE, "version": "2.0.0", "supported_operations": [ "subreddit_download", "user_download", "custom_filter" ], "output_formats": ["json", "csv", "xml"] } @app.get("/api/websocket/status") async def websocket_status(): """Get WebSocket connection status for debugging""" return { "active_connections": len(manager.active_connections), "connection_failures": manager.connection_failures, "active_downloads": len(active_downloads), "websocket_endpoint": "/ws/progress", "server_host": "0.0.0.0", "server_port": 8000 } @app.get("/api/oauth/debug") async def oauth_debug(): """Debug endpoint to check OAuth state""" try: oauth_manager = get_oauth_manager() return { "oauth_manager_initialized": oauth_manager is not None, "oauth_states_count": len(oauth_manager.oauth_states) if oauth_manager else 0, "access_tokens_count": len(oauth_manager.access_tokens) if oauth_manager else 0, "refresh_tokens_count": len(oauth_manager.refresh_tokens) if oauth_manager else 0, "client_id_configured": oauth_manager.client_id != "U-6gk4ZCh3IeNQ" if oauth_manager else False, "redirect_uri": os.getenv("BDFR_REDIRECT_URI", "http://localhost:8000/auth/callback") } except Exception as e: return { "error": str(e), "oauth_manager_initialized": False } if __name__ == "__main__": import uvicorn uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)