rpa

src/cluster/rpa.py

@@ -4,13 +4,13 @@ import json
 import os
 import time
 from pathlib import Path
-from typing import List, Optional
+from typing import List, Optional, Any
 
 import modal
 from loguru import logger
 from playwright.async_api import async_playwright, Browser, Page
 from tenacity import retry, stop_after_attempt, wait_fixed
-from ulid import ULID
+from functools import partial
 
 from .rpa_comm import LiveStreamProductWatchRequest, LiveStreamResult, ProductSession
 from .rpa_web_end_point import app as rpa_web_end_point_app
@@ -102,10 +102,14 @@ async def check_live_status(webcast_id: str, page: Optional[Page] = None) -> boo
         #  TODO api
         return True
 
- # 监听网络响应
-async def handle_response(response):
+
+# 监听网络响应
+
+async def handle_response(custom_param, response):
     if "live.douyin.com/live/promotions/page" in response.url:
-        logger.debug(f"URL: {response.url}")
+        print(f"Custom param: {custom_param}")
+        print(f"Response URL: {response.url}")
+
         logger.debug(f"Status: {response.status}")
         # 获取响应内容
         try:
@@ -118,8 +122,6 @@ async def handle_response(response):
             logger.debug("Non-JSON response")
 
 
-
-
 @retry(stop=stop_after_attempt(3), wait=wait_fixed(5))
 async def get_promotion_list_text(page: Page, webcast_id: str, max_duration: int = 8 * 60 * 60) -> List[ProductSession]:
     """
@@ -156,7 +158,8 @@ async def get_promotion_list_text(page: Page, webcast_id: str, max_duration: int
                 break
             last_status_check = time.time()
 
-        page.on("response", handle_response)
+        # page.on("response", handle_response)
+        page.on("response", partial(handle_response, webcast_id))
 
         # Refresh the page to ensure updated innerText
         await reload_page(page)
@@ -367,8 +370,6 @@ with rpa_image.imports():
 
                 logger.info(f"Generated {len(result.cut_commands)} FFmpeg cut commands")
 
-
-
         # Save result to JSON file with ULID naming,TODO 后续s3上传
         await save_result_to_json(result, data)
 
@@ -403,13 +404,13 @@ with rpa_image.imports():
         live_status = None
         logger.debug(f"Getting live status for webcast_id: {webcast_id}")
         try:
-            await page.wait_for_selector("[class='pip-anchor']", timeout=60*000)
+            await page.wait_for_selector("[class='pip-anchor']", timeout=60 * 000)
             await page.wait_for_function(
                 """() => {
                     const element = document.querySelector('[class="pip-anchor"]');
                     return element && element.innerText !== "";
                 }""",
-                timeout=60*000
+                timeout=60 * 000
             )
             live_status_js = """document.querySelector('[class="pip-anchor"]')?.innerText || ''"""
             live_status_text: str = await page.evaluate(live_status_js)