rpa

src/cluster/rpa.md

@@ -1,4 +1,7 @@
-
+modal setup
+```shell
+python -m modal setup
+```
 
 deploy by env
 ```shell

src/cluster/rpa.py

@@ -11,6 +11,8 @@ from playwright.async_api import async_playwright, Browser, Page
 from pydantic import BaseModel
 from tenacity import retry, stop_after_attempt, wait_fixed
 from ulid import ULID
+from .rpa_comm import LiveStreamProductWatchRequest,LiveStreamResult,ProductSession
+
 
 rpa_image = (modal.Image.debian_slim(python_version="3.11")
              .pip_install_from_pyproject("../pyproject_rpa.toml")
@@ -25,47 +27,6 @@ app = modal.App(name='rpa', image=rpa_image, include_source=False)
 
 ############################ Modal app set############################
 
-# Define Pydantic models
-class ProductSession(BaseModel):
-    start_time: Optional[int] = 0
-    """当前商品直播开始世界时间（毫秒时间戳）"""
-    end_time: Optional[int] = 0
-    """当前商品直播结束世界时间（毫秒时间戳）"""
-    duration: Optional[int] = 0
-    """当前商品直播持续时间（毫秒）"""
-    product_name: Optional[str] = None
-    """商品名称"""
-    product_id: Optional[str] = None
-    """商品id"""
-    product_img: Optional[str] = None
-    """商品图片url"""
-    start_canvas_img: Optional[str] = None
-    """当前商品直播开始截图"""
-    end_canvas_img: Optional[str] = None
-    """当前商品直播结束截图"""
-    ext_data: Optional[Any] = None
-    """扩展字段"""
-
-
-class LiveStreamResult(BaseModel):
-    title: Optional[str] = None
-    """直播间名称"""
-    live_id: Optional[str] = None
-    """直播间id"""
-    start_time: Optional[int] = 0
-    """直播开始世界时间（毫秒时间戳）"""
-    product_sessions: List[ProductSession] = []
-    """直播所有商品记录"""
-    duration: Optional[int] = 0
-    """直播持续时间"""
-    cut_commands: List[str] = []
-    """冗余字段"""
-    ext_data: Optional[Any] = None
-    """扩展字段"""
-    error: Optional[str] = None
-    msg: Optional[str] = None
-
-
 EXPLAINING: str = "讲解中"
 LIVE_END: str = "直播已结束"
 

src/cluster/rpa_comm.py

@@ -0,0 +1,62 @@
+from pydantic import BaseModel
+from typing import List, Optional,Any
+
+class LiveStreamProductWatchRequest(BaseModel):
+    """
+    直播流商品观测
+    """
+
+    live_stream_name: Optional[str] = None
+    """直播间推流名称"""
+    live_stream_start_time: Optional[int] = 0
+    """直播推流开始世界时间（毫秒时间戳）"""
+    live_stream_url: Optional[str] = None
+    """直播间推流链接"""
+    live_id: Optional[str] = None
+    """直播间id"""
+    cookie_string: Optional[str] = None
+    """cookie string ,split('; ') """
+    ext_data: Optional[Any] = None
+    """扩展字段"""
+
+
+# Define Pydantic models
+class ProductSession(BaseModel):
+    start_time: Optional[int] = 0
+    """当前商品直播开始世界时间（毫秒时间戳）"""
+    end_time: Optional[int] = 0
+    """当前商品直播结束世界时间（毫秒时间戳）"""
+    duration: Optional[int] = 0
+    """当前商品直播持续时间（毫秒）"""
+    product_name: Optional[str] = None
+    """商品名称"""
+    product_id: Optional[str] = None
+    """商品id"""
+    product_img: Optional[str] = None
+    """商品图片url"""
+    start_canvas_img: Optional[str] = None
+    """当前商品直播开始截图"""
+    end_canvas_img: Optional[str] = None
+    """当前商品直播结束截图"""
+    ext_data: Optional[Any] = None
+    """扩展字段"""
+
+
+class LiveStreamResult(BaseModel):
+    title: Optional[str] = None
+    """直播间名称"""
+    live_id: Optional[str] = None
+    """直播间id"""
+    start_time: Optional[int] = 0
+    """直播开始世界时间（毫秒时间戳）"""
+    product_sessions: List[ProductSession] = []
+    """直播所有商品记录"""
+    duration: Optional[int] = 0
+    """直播持续时间"""
+    cut_commands: List[str] = []
+    """冗余字段"""
+    ext_data: Optional[Any] = None
+    """扩展字段"""
+    error: Optional[str] = None
+    msg: Optional[str] = None
+

src/cluster/rpa_web_end_point.py

@@ -0,0 +1,42 @@
+# my_job_queue_endpoint.py
+import fastapi
+from typing import List, Optional,Any
+import modal
+from .rpa_comm import LiveStreamProductWatchRequest,LiveStreamResult,ProductSession
+
+
+image = modal.Image.debian_slim().pip_install("fastapi[standard]")
+app = modal.App("fastapi_rpa", image=image)
+
+############################ Modal app set############################
+
+web_app = fastapi.FastAPI()
+
+
+@app.function()
+@modal.asgi_app()
+def fastapi_app():
+    return web_app
+
+
+@web_app.post("/submit")
+async def submit_job_endpoint(data:LiveStreamProductWatchRequest):
+    process_job = modal.Function.from_name("rpa", "rpa_run")
+
+    call = process_job.spawn(data)
+    return {"call_id": call.object_id}
+
+
+@web_app.get("/result/{call_id}")
+async def get_job_result_endpoint(call_id: str):
+    function_call = modal.FunctionCall.from_id(call_id)
+    try:
+        result = function_call.get(timeout=0)
+    except modal.exception.OutputExpiredError:
+        return fastapi.responses.JSONResponse(content="", status_code=404)
+    except TimeoutError:
+        return fastapi.responses.JSONResponse(content="", status_code=202)
+
+    return result
+
+# modal serve my_job_queue_endpoint.py