x1

backend/readme.md

@@ -1,10 +1,9 @@
 ```bash
-pip install fastapi
-pip install uvicorn[standard]
+pip install fastapi[all]
 ```
 
 ```bash
-uvicorn main:app --reload
+python main.py
 ```
 
 ```markdown

backend/routers/challenge.py

@@ -3,22 +3,14 @@
 # Author     : QIN2DIM
 # GitHub     : https://github.com/QIN2DIM
 # Description:
-from io import BytesIO
-from typing import List
 
-from PIL import Image
 from fastapi import APIRouter
 from loguru import logger
-from pydantic import BaseModel, Field, Base64Bytes
 
 import hcaptcha_challenger as solver
-from hcaptcha_challenger import (
-    handle,
-    ModelHub,
-    DataLake,
-    register_pipline,
-    ZeroShotImageClassifier,
-)
+from hcaptcha_challenger import ModelHub, register_pipline
+from hcaptcha_challenger.models import SelfSupervisedResponse, SelfSupervisedPayload
+from hcaptcha_challenger.tools.zero_shot_image_classifier import invoke_clip_tool
 
 router = APIRouter()
 
@@ -34,48 +26,7 @@
 )
 
 
-class SelfSupervisedPayload(BaseModel):
-    """hCaptcha payload of the image_label_binary challenge"""
-
-    prompt: str = Field(..., description="challenge prompt")
-    challenge_images: List[Base64Bytes] = Field(default_factory=list)
-    positive_labels: List[str] | None = Field(default_factory=list)
-    negative_labels: List[str] | None = Field(default_factory=list)
-
-
-class SelfSupervisedResponse(BaseModel):
-    """The binary classification result of the image, in the same order as the challenge_images."""
-
-    results: List[bool] = Field(default_factory=list)
-
-
-def invoke_clip_tool(payload: SelfSupervisedPayload) -> List[bool]:
-    label = handle(payload.prompt)
-
-    if any(payload.positive_labels) and any(payload.negative_labels):
-        serialized = {
-            "positive_labels": payload.positive_labels,
-            "negative_labels": payload.negative_labels,
-        }
-        modelhub.datalake[label] = DataLake.from_serialized(serialized)
-
-    if not (dl := modelhub.datalake.get(label)):
-        dl = DataLake.from_challenge_prompt(label)
-    tool = ZeroShotImageClassifier.from_datalake(dl)
-
-    # Default to `RESNET.OPENAI` perf_counter 1.794s
-    model = clip_model or register_pipline(modelhub)
-
-    response: List[bool] = []
-    for image_data in payload.challenge_images:
-        results = tool(model, image=Image.open(BytesIO(image_data)))
-        trusted = results[0]["label"] in tool.positive_labels
-        response.append(trusted)
-
-    return response
-
-
 @router.post("/image_label_binary", response_model=SelfSupervisedResponse)
 async def challenge_image_label_binary(payload: SelfSupervisedPayload):
-    results = invoke_clip_tool(payload)
+    results = invoke_clip_tool(modelhub, payload, clip_model)
     return SelfSupervisedResponse(results=results)

examples/faker_client.py

@@ -12,7 +12,6 @@
 
 from hcaptcha_challenger.agents import AgentV
 from hcaptcha_challenger.agents import Malenia
-from hcaptcha_challenger.onnx.modelhub import ModelHub
 from hcaptcha_challenger.utils import SiteKey
 
 
@@ -33,17 +32,7 @@ async def main(headless: bool = False):
 async def mime(context: BrowserContext):
     page = await context.new_page()
 
-    modelhub = ModelHub.from_github_repo()
-    modelhub.parse_objects()
-
-    agent = AgentV.into_solver(
-        # page, the control handle of the Playwright Page
-        page=page,
-        # modelhub, Register modelhub externally, and the agent can patch custom configurations
-        modelhub=modelhub,
-        # tmp_dir, Mount the cache directory to the current working folder
-        tmp_dir=Path("tmp_dir"),
-    )
+    agent = AgentV.into_solver(page=page, tmp_dir=Path("tmp_dir"))
 
     sitekey = SiteKey.user_easy
 
@@ -57,7 +46,7 @@ async def mime(context: BrowserContext):
 
 
 if __name__ == "__main__":
-    EXECUTION = "collect"
-    # EXECUTION = "challenge"
+    # EXECUTION = "collect"
+    EXECUTION = "challenge"
 
     encrypted_resp = asyncio.run(main(headless=False))