Upload app.py

app.py

@@ -0,0 +1,228 @@
+from fastapi import FastAPI, HTTPException, BackgroundTasks
+from fastapi.responses import JSONResponse
+import asyncio
+import random
+import logging
+from typing import List, Dict, Optional, Any
+from playwright.async_api import async_playwright
+import uvicorn
+from pydantic import BaseModel
+
+# Set up logging
+logging.basicConfig(level=logging.INFO, 
+                   format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+app = FastAPI(title="Kling Video Fetcher", 
+             description="API to fetch random videos from Kling AI gallery")
+
+# Model for the response
+class VideoResponse(BaseModel):
+    url: str
+    thumbnail: Optional[str] = None
+    title: Optional[str] = None
+    status: str = "success"
+
+# Cache to store recent video URLs
+video_cache: List[Dict[str, Any]] = []
+last_fetch_time = 0
+
+async def fetch_random_video_from_kling() -> Dict[str, Any]:
+    """
+    Fetches a random video from Kling AI gallery and returns its URL.
+    Returns a dictionary with video information.
+    """
+    try:
+        async with async_playwright() as p:
+            # Configure browser with a realistic user agent and longer timeout
+            browser_context_options = {
+                'viewport': {'width': 1280, 'height': 800},
+                'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36',
+            }
+            
+            # Launch the browser in headless mode with anti-detection args
+            browser = await p.chromium.launch(
+                headless=True,
+                args=[
+                    '--no-sandbox',
+                    '--disable-setuid-sandbox',
+                    '--disable-dev-shm-usage',
+                    '--disable-accelerated-2d-canvas',
+                    '--no-first-run',
+                    '--no-zygote',
+                    '--disable-gpu',
+                    '--disable-web-security',
+                ]
+            )
+            context = await browser.new_context(**browser_context_options)
+            page = await context.new_page()
+            
+            try:
+                logger.info("Navigating to Kling AI gallery...")
+                # Use a longer timeout and wait until networkidle
+                await page.goto(
+                    "https://app.klingai.com/global/community/material", 
+                    timeout=90000,
+                    wait_until="networkidle"
+                )
+                logger.info("Successfully loaded gallery page")
+            except Exception as e:
+                logger.error(f"Error navigating to gallery: {e}")
+                raise HTTPException(status_code=504, detail=f"Unable to access Kling AI gallery: {str(e)}")
+            
+            # Wait for dynamic content to load
+            logger.info("Waiting for dynamic content to load...")
+            await page.wait_for_timeout(5000)
+            
+            # Scroll to load content
+            logger.info("Scrolling to load gallery content...")
+            scroll_attempts = 6
+            for i in range(scroll_attempts):
+                logger.info(f"Scroll {i+1}/{scroll_attempts}")
+                await page.evaluate("window.scrollBy(0, 800)")
+                await page.wait_for_timeout(1500)
+            
+            # Find potential video elements
+            logger.info("Searching for video elements...")
+            all_potential_elements = []
+            
+            # Direct video elements
+            videos = await page.query_selector_all("video")
+            if videos:
+                logger.info(f"Found {len(videos)} direct video elements")
+                all_potential_elements.extend(videos)
+            
+            # Look for video containers
+            video_containers = await page.query_selector_all(
+                '[class*="video"], [class*="player"], [class*="media"], ' +
+                '[class*="thumbnail"], [class*="card"], [class*="item"]'
+            )
+            if video_containers:
+                logger.info(f"Found {len(video_containers)} potential video containers")
+                all_potential_elements.extend(video_containers)
+            
+            # Filter to keep only visible elements
+            visible_elements = []
+            for element in all_potential_elements:
+                try:
+                    is_visible = await element.is_visible()
+                    if is_visible:
+                        visible_elements.append(element)
+                except Exception:
+                    continue
+            
+            if visible_elements:
+                logger.info(f"Found {len(visible_elements)} visible elements")
+                all_potential_elements = visible_elements
+            
+            if not all_potential_elements:
+                logger.error("No potential video elements found")
+                raise HTTPException(status_code=404, detail="No video elements found on the page")
+            
+            # Try multiple elements until we find a video
+            # Shuffling the elements to increase randomness
+            random.shuffle(all_potential_elements)
+            
+            # We'll try up to 5 random elements
+            max_attempts = min(5, len(all_potential_elements))
+            
+            for attempt in range(max_attempts):
+                selected_element = all_potential_elements[attempt]
+                
+                try:
+                    # Scroll the element into view
+                    await selected_element.scroll_into_view_if_needed()
+                    await page.wait_for_timeout(1000)
+                    
+                    # Click on the selected element
+                    logger.info(f"Clicking on element (attempt {attempt+1}/{max_attempts})...")
+                    await selected_element.click()
+                    await page.wait_for_timeout(3000)
+                    
+                    # Look for video elements that appeared after clicking
+                    post_click_videos = await page.query_selector_all("video")
+                    
+                    if post_click_videos:
+                        logger.info(f"Found {len(post_click_videos)} video elements after clicking!")
+                        
+                        # Find a video with a source
+                        for video in post_click_videos:
+                            try:
+                                video_src = await video.evaluate("el => el.src || ''")
+                                poster_src = await video.evaluate("el => el.poster || ''")
+                                
+                                if video_src:
+                                    # Try to get title from page
+                                    title = await page.title()
+                                    # Check if we're on a different page now
+                                    current_url = page.url
+                                    
+                                    result = {
+                                        "url": video_src,
+                                        "thumbnail": poster_src if poster_src else None,
+                                        "title": title if title != "Kling AI" else None,
+                                        "page_url": current_url
+                                    }
+                                    
+                                    await browser.close()
+                                    return result
+                            except Exception as e:
+                                logger.error(f"Error extracting video details: {e}")
+                                continue
+                
+                except Exception as e:
+                    logger.error(f"Error with element attempt {attempt+1}: {e}")
+                    continue
+            
+            # If we get here, we couldn't find a video after multiple attempts
+            await browser.close()
+            raise HTTPException(status_code=404, detail="Could not find any video with a valid source")
+            
+    except Exception as e:
+        logger.error(f"Unexpected error: {e}")
+        raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}")
+
+@app.get("/api/random-video", response_model=VideoResponse)
+async def get_random_video():
+    """
+    Endpoint to get a random video from Kling AI gallery.
+    Returns the video URL and metadata.
+    """
+    try:
+        # Fetch a random video
+        video_info = await fetch_random_video_from_kling()
+        
+        # Return the video information
+        return {
+            "url": video_info["url"],
+            "thumbnail": video_info.get("thumbnail"),
+            "title": video_info.get("title"),
+            "status": "success"
+        }
+    except HTTPException as e:
+        # Re-raise HTTP exceptions
+        raise e
+    except Exception as e:
+        # Log and convert other exceptions to HTTPException
+        logger.error(f"Error in endpoint: {e}")
+        raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}")
+
+@app.get("/")
+async def root():
+    """Root endpoint with API information"""
+    return {
+        "name": "Kling Video Fetcher API",
+        "version": "1.0",
+        "endpoints": {
+            "/api/random-video": "Get a random video URL from Kling AI gallery"
+        }
+    }
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    return {"status": "healthy"}
+
+# Run the server
+if __name__ == "__main__":
+    uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)