perf: Improve Anilife _aldata extraction speed and reliability by adding resource blocking, implementing element polling, and streamlining browser interactions.

lib/camoufox_anilife.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 """
-Camoufox 기반 Anilife 비디오 URL 추출 스크립트 (비동기 버전)
+Camoufox 기반 Anilife 비디오 URL 추출 스크립트 (최적화 비동기 버전)
-강력한 봇 감지 우회 기능이 있는 스텔스 Firefox
 """
 
 import sys
@@ -10,129 +9,101 @@ import asyncio
 import re
 import os
 
-async def _run_browser(browser, detail_url, episode_num, result):
+async def _wait_for_aldata(page, timeout=10):
-    """실제 브라우저 작업을 수행하는 내부 비동기 함수"""
+    """_aldata 변수가 나타날 때까지 폴링 (최대 timeout초)"""
-    page = await browser.new_page()
+    start_time = asyncio.get_event_loop().time()
-    try:
+    while asyncio.get_event_loop().time() - start_time < timeout:
-        # 1. Detail 페이지로 이동
-        print(f"1. Navigating to detail page: {detail_url}", file=sys.stderr)
-        await page.goto(detail_url, wait_until="domcontentloaded", timeout=30000)
-        await asyncio.sleep(2)
-        
-        print(f"   Current URL: {page.url}", file=sys.stderr)
-        
-        # 2. 에피소드 목록으로 스크롤
-        await page.mouse.wheel(0, 800)
-        await asyncio.sleep(1)
-        
-        # 3. 해당 에피소드 찾아서 클릭
-        print(f"2. Looking for episode {episode_num}", file=sys.stderr)
-        
-        episode_clicked = False
         try:
-            # epl-num 클래스의 div에서 에피소드 번호 찾기
+            # 1. JS 변수 확인
-            episode_link = page.locator(f'a:has(.epl-num:text("{episode_num}"))').first
+            aldata = await page.evaluate("typeof _aldata !== 'undefined' ? _aldata : null")
-            if await episode_link.is_visible(timeout=5000):
+            if aldata:
-                href = await episode_link.get_attribute("href")
+                return aldata, "JS"
-                print(f"   Found episode link: {href}", file=sys.stderr)
+            
-                await episode_link.click()
+            # 2. HTML 소스 패턴 확인
-                episode_clicked = True
+            html = await page.content()
-                await asyncio.sleep(3)
+            match = re.search(r'_aldata\s*=\s*["\']([A-Za-z0-9+/=]+)["\']', html)
-        except Exception as e:
+            if match:
-            print(f"   Method 1 failed: {e}", file=sys.stderr)
+                return match.group(1), "HTML"
+        except:
+            pass
+        await asyncio.sleep(0.3)
+    return None, None
+
+async def _run_browser(browser, detail_url, episode_num, result):
+    """최적화된 브라우저 작업 수행"""
+    # 1. 컨텍스트 및 페이지 생성 (이미지/CSS 차단 옵션 적용 가능 시 적용)
+    page = await browser.new_page()
+    
+    # 리소스 차단 (속도 향상의 핵심)
+    async def intercept(route):
+        if route.request.resource_type in ["image", "media", "font", "stylesheet"]:
+            await route.abort()
+        else:
+            await route.continue_()
+    
+    await page.route("**/*", intercept)
+    
+    try:
+        # 1. Detail 페이지 이동
+        print(f"1. Navigating to detail page: {detail_url}", file=sys.stderr)
+        await page.goto(detail_url, wait_until="commit", timeout=20000) # domcontentloaded보다 빠른 commit 대기
         
-        if not episode_clicked:
+        # 2. 에피소드 링크 찾기 (폴링 대기)
+        print(f"2. Searching for episode {episode_num}...", file=sys.stderr)
+        episode_link = None
+        for _ in range(25): # 약 5초간 대기
             try:
-                # provider 링크들 중에서 에피소드 번호가 포함된 것 클릭
+                episode_link = page.locator(f'a:has(.epl-num:text("{episode_num}"))').first
+                if await episode_link.is_visible():
+                    break
+                
+                # 대체 수단: provider 링크 검색
                 links = await page.locator('a[href*="/ani/provider/"]').all()
                 for link in links:
                     text = await link.inner_text()
                     if episode_num in text:
-                        print(f"   Found: {text}", file=sys.stderr)
+                        episode_link = link
-                        await link.click()
-                        episode_clicked = True
-                        await asyncio.sleep(3)
                         break
-            except Exception as e:
+                if episode_link: break
-                print(f"   Method 2 failed: {e}", file=sys.stderr)
+            except: pass
+            await asyncio.sleep(0.2)
         
-        if not episode_clicked:
+        if not episode_link:
             result["error"] = f"Episode {episode_num} not found"
             result["html"] = await page.content()
             return result
+
+        # 3. 에피소드 클릭 및 이동
+        print(f"3. Clicking episode {episode_num}", file=sys.stderr)
+        await episode_link.click()
         
-        # 4. Provider 페이지에서 _aldata 추출
+        # 4. _aldata 추출 (폴링)
-        print(f"3. Provider page URL: {page.url}", file=sys.stderr)
+        print("4. Waiting for _aldata...", file=sys.stderr)
-        result["current_url"] = page.url
+        aldata, source = await _wait_for_aldata(page, timeout=8)
         
-        # 리다이렉트 확인
+        if aldata:
-        if "/ani/provider/" not in page.url:
+            result["aldata"] = aldata
-            result["error"] = f"Redirected to {page.url}"
-            result["html"] = await page.content()
-            return result
-        
-        # _aldata 추출 시도
-        try:
-            aldata_value = await page.evaluate("typeof _aldata !== 'undefined' ? _aldata : null")
-            if aldata_value:
-                result["aldata"] = aldata_value
-                result["success"] = True
-                print(f"   SUCCESS! _aldata found: {aldata_value[:60]}...", file=sys.stderr)
-                return result
-        except Exception as js_err:
-            print(f"   JS error: {js_err}", file=sys.stderr)
-        
-        # HTML에서 _aldata 패턴 추출 시도
-        html_content = await page.content()
-        aldata_match = re.search(r'_aldata\s*=\s*["\']([A-Za-z0-9+/=]+)["\']', html_content)
-        if aldata_match:
-            result["aldata"] = aldata_match.group(1)
             result["success"] = True
-            print(f"   SUCCESS! _aldata from HTML: {result['aldata'][:60]}...", file=sys.stderr)
+            result["current_url"] = page.url
+            print(f"   SUCCESS! Got _aldata from {source}", file=sys.stderr)
             return result
-        
-        # 5. CloudVideo 버튼 클릭 시도
-        print("4. Trying CloudVideo button click...", file=sys.stderr)
-        try:
-            await page.mouse.wheel(0, 500)
-            await asyncio.sleep(1)
             
-            cloudvideo_btn = page.locator('a[onclick*="moveCloudvideo"], a[onclick*="moveJawcloud"]').first
+        # 5. 추출 실패 시 CloudVideo 버튼 강제 클릭 시도
-            if await cloudvideo_btn.is_visible(timeout=3000):
+        print("5. Aldata not found yet. Trying player button...", file=sys.stderr)
-                await cloudvideo_btn.click()
+        await page.mouse.wheel(0, 500)
-                await asyncio.sleep(3)
+        btn = page.locator('a[onclick*="moveCloudvideo"], a[onclick*="moveJawcloud"]').first
-                
+        if await btn.is_visible(timeout=2000):
+            await btn.click()
+            aldata, source = await _wait_for_aldata(page, timeout=5)
+            if aldata:
+                result["aldata"] = aldata
+                result["success"] = True
                 result["current_url"] = page.url
-                print(f"   After click URL: {page.url}", file=sys.stderr)
+                return result
-                
+
-                # 리다이렉트 확인 (구글로 갔는지)
+        result["error"] = "Could not extract aldata"
-                if "google.com" in page.url:
+        result["html"] = await page.content()
-                    result["error"] = "Redirected to Google - bot detected"
+        result["current_url"] = page.url
-                    return result
-                
-                # 플레이어 페이지에서 _aldata 추출
-                try:
-                    aldata_value = await page.evaluate("typeof _aldata !== 'undefined' ? _aldata : null")
-                    if aldata_value:
-                        result["aldata"] = aldata_value
-                        result["success"] = True
-                        print(f"   SUCCESS! _aldata: {aldata_value[:60]}...", file=sys.stderr)
-                        return result
-                except:
-                    pass
-                
-                # HTML에서 추출
-                html_content = await page.content()
-                aldata_match = re.search(r'_aldata\s*=\s*["\']([A-Za-z0-9+/=]+)["\']', html_content)
-                if aldata_match:
-                    result["aldata"] = aldata_match.group(1)
-                    result["success"] = True
-                    return result
-                    
-                result["html"] = html_content
-        except Exception as click_err:
-            print(f"   Click error: {click_err}", file=sys.stderr)
-            result["html"] = await page.content()
             
     finally:
         await page.close()
@@ -140,57 +111,40 @@ async def _run_browser(browser, detail_url, episode_num, result):
     return result
 
 async def extract_aldata(detail_url: str, episode_num: str) -> dict:
-    """AsyncCamoufox로 Detail 페이지에서 _aldata 추출"""
+    """AsyncCamoufox로 최적화된 추출 수행"""
-    
     try:
         from camoufox.async_api import AsyncCamoufox
     except ImportError as e:
         return {"error": f"Camoufox not installed: {e}"}
     
-    result = {
+    result = {"success": False, "aldata": None, "current_url": None, "error": None}
-        "success": False, "aldata": None, "html": None,
-        "current_url": None, "error": None, "vod_url": None
-    }
     
     try:
-        # Docker/서버 환경에서는 DISPLAY가 없으므로 Xvfb 가상 디스플레이 사용 시도
         has_display = os.environ.get('DISPLAY') is not None
-        
+        camou_args = {"headless": False}
         if not has_display:
-            print("   No DISPLAY detected. Using Virtual Display (Xvfb) for better stealth.", file=sys.stderr)
+            camou_args["xvfb"] = True
-            camou_args = {"headless": False, "xvfb": True}
-        else:
-            camou_args = {"headless": False}
         
-        # xvfb 인자 지원 여부에 따른 안전한 실행 (Try-Except Fallback)
+        # 속도 최 최적화를 위한 추가 인자 (필요 시)
         try:
             async with AsyncCamoufox(**camou_args) as browser:
                 return await _run_browser(browser, detail_url, episode_num, result)
-        except TypeError as e:
+        except TypeError:
-            if "xvfb" in str(e):
+            # xvfb 미지원 버전 대비
-                print(f"   Warning: Local Camoufox version too old for 'xvfb'. Falling back to headless.", file=sys.stderr)
+            async with AsyncCamoufox(headless=True) as browser:
-                async with AsyncCamoufox(headless=True) as browser:
+                return await _run_browser(browser, detail_url, episode_num, result)
-                    return await _run_browser(browser, detail_url, episode_num, result)
-            raise e
             
     except Exception as e:
         result["error"] = str(e)
-        import traceback
-        print(traceback.format_exc(), file=sys.stderr)
     
     return result
 
 if __name__ == "__main__":
     if len(sys.argv) < 3:
-        print(json.dumps({"error": "Usage: python camoufox_anilife.py <detail_url> <episode_num>"}))
         sys.exit(1)
     
     detail_url = sys.argv[1]
     episode_num = sys.argv[2]
     
-    # 비동기 실행 루프 시작
+    res = asyncio.run(extract_aldata(detail_url, episode_num))
-    try:
+    print(json.dumps(res, ensure_ascii=False))
-        res = asyncio.run(extract_aldata(detail_url, episode_num))
-        print(json.dumps(res, ensure_ascii=False))
-    except Exception as e:
-        print(json.dumps({"error": str(e), "success": False}))