Update app.py

app.py

@@ -1,228 +1,138 @@
-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__":
+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]:
+    try:
+        async with async_playwright() as p:
+            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(viewport={'width': 1280, 'height': 800})
+            page = await context.new_page()
+
+            video_urls = []
+
+            # Intercept requests to capture video URLs
+            async def on_response(response):
+                try:
+                    url = response.url
+                    content_type = response.headers.get("content-type", "")
+                    if ".mp4" in url or "video" in content_type:
+                        video_urls.append(url)
+                except Exception as e:
+                    logger.error(f"Error processing response: {e}")
+
+            page.on("response", on_response)
+
+            await page.goto("https://app.klingai.com/global/community/material", timeout=90000, wait_until="networkidle")
+            await page.wait_for_timeout(5000)
+
+            for _ in range(6):
+                await page.evaluate("window.scrollBy(0, 800)")
+                await page.wait_for_timeout(1500)
+
+            elements = await page.query_selector_all("video, [class*='video'], [class*='card'], [class*='media']")
+            elements = [e for e in elements if await e.is_visible()]
+            random.shuffle(elements)
+
+            for element in elements[:5]:
+                try:
+                    await element.scroll_into_view_if_needed()
+                    await page.wait_for_timeout(1000)
+                    await element.click()
+                    await page.wait_for_timeout(3000)
+
+                    if video_urls:
+                        video_url = video_urls[0]
+                        title = await page.title()
+                        return {
+                            "url": video_url,
+                            "thumbnail": None,
+                            "title": title if title != "Kling AI" else None,
+                            "status": "success"
+                        }
+                except Exception as e:
+                    logger.warning(f"Element interaction failed: {e}")
+
+            raise HTTPException(status_code=404, detail="Could not find a downloadable video")
+    except Exception as e:
+        logger.error(f"Unexpected error: {e}")
+        raise HTTPException(status_code=500, detail=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)