Add batch video deepfake detection support

README.md

@@ -147,6 +147,28 @@ output = client.deepfakes.get_result(pid=response.pid)
 
 See more in our [API documentation](https://behavioralsignals.readme.io/v5.4.0/docs/generator-detection#/).
 
+#### 🎬 Video Deepfake Detection (Batch Only)
+
+In addition to audio, the Deepfakes API can detect deepfakes in video files. You upload a video the same way you upload audio, and the API analyzes both the audio track and the video frames.
+
+```python
+from behavioralsignals import Client
+
+client = Client(YOUR_CID, YOUR_API_KEY)
+
+response = client.deepfakes.upload_video(file_path="video.mp4")
+output = client.deepfakes.get_video_result(pid=response.pid)
+```
+
+Unlike `get_result`, the video result response returns two separate lists — `audio_results` (deepfake detection on the audio track) and `video_results` (deepfake detection on the video frames):
+
+```python
+for item in output.video_results:
+    print(item.task, item.finalLabel)
+```
+
+You can also submit a video via an S3 presigned URL with `client.deepfakes.upload_s3_presigned_video_url(url=...)`, and list/inspect video processes with `client.deepfakes.list_video_processes()` and `client.deepfakes.get_video_process(pid=...)`. The `embeddings` and `enable_generator_detection` options are supported and apply to the audio-track results. Video deepfake detection is currently available in batch mode only.
+
 ### Deepfakes API Streaming Mode
 
 A similar streaming example for the Deepfakes API allows you to send audio data in real-time for speech deepfake detection:

examples/batch/video_deepfake_polling.py

@@ -0,0 +1,83 @@
+"""Example script demonstrating video deepfake detection via the batch API.
+
+The batch API works as follows:
+    1. Submit your video and retrieve a process ID (pid).
+    2. Poll the process until it completes.
+    3. Retrieve the results using this pid. The video result response contains two
+       separate lists: `audio_results` (deepfake detection on the audio track) and
+       `video_results` (deepfake detection on the video frames).
+
+Video deepfake detection is currently available in batch mode only.
+"""
+
+import os
+import json
+import time
+import argparse
+
+from dotenv import load_dotenv
+
+from behavioralsignals import Client
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="Video Deepfake Detection Example")
+    parser.add_argument(
+        "--file_path", type=str, required=True, help="Path to the video file to send"
+    )
+    parser.add_argument(
+        "--output", type=str, default="video_output.json", help="Path to save the output JSON file"
+    )
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    args = parse_args()
+    file_path, output = args.file_path, args.output
+
+    # Step 1. Initialize the client with your client ID and API key.
+    load_dotenv()
+    client = Client(cid=os.getenv("CID"), api_key=os.getenv("API_KEY")).deepfakes
+
+    # Step 2. Send the video file for processing
+    upload_response = client.upload_video(file_path=file_path)
+    pid = upload_response.pid
+    print(f"Sent video for processing! Process ID (pid): {pid}")
+
+    # Step 3. Poll the API to check the status of the process
+    last_status = None
+    while True:
+        process = client.get_video_process(pid=pid)
+        status = process.statusmsg
+
+        if process.is_completed:
+            if last_status != process.statusmsg:
+                print("Processing complete!")
+            break
+        elif process.is_processing:
+            if last_status != process.statusmsg:
+                print("Processing video...")
+        elif process.is_pending:
+            if last_status != process.statusmsg:
+                print("API is busy, waiting...")
+        else:
+            if last_status != process.statusmsg:
+                print(f"Unexpected status: {process.statusmsg}")
+            break
+
+        last_status = status
+        # Wait before polling again
+        time.sleep(1.0)
+
+    # Step 4. Retrieve the results if processing is complete and save to output file
+    if process.is_completed:
+        result = client.get_video_result(pid=pid)
+        result_dict = result.model_dump()
+
+        n_audio = len(result.audio_results or [])
+        n_video = len(result.video_results or [])
+        print(f"Got {n_audio} audio result(s) and {n_video} video result(s).")
+
+        with open(output, "w") as f:
+            json.dump(result_dict, f, indent=4)
+        print(f"Results saved to {output}")

src/behavioralsignals/__init__.py

@@ -1,7 +1,7 @@
 from .client import Client
-from .models import StreamingOptions
 from .deepfakes import Deepfakes
 from .behavioral import Behavioral
+from .models import StreamingOptions, VideoResultResponse
 
 
-__all__ = ["Client", "Behavioral", "Deepfakes", "StreamingOptions"]
+__all__ = ["Client", "Behavioral", "Deepfakes", "StreamingOptions", "VideoResultResponse"]

src/behavioralsignals/deepfakes.py

@@ -10,6 +10,7 @@
     StreamingOptions,
     ProcessListParams,
     ProcessListResponse,
+    VideoResultResponse,
     StreamingResultResponse,
     DeepfakeAudioUploadParams,
     DeepfakeS3UrlUploadParams,
@@ -185,6 +186,175 @@ def get_result(self, pid: int) -> ResultResponse:
         )
         return ResultResponse(**data)
 
+    def upload_video(
+        self,
+        file_path: str,
+        name: Optional[str] = None,
+        embeddings: bool = False,
+        enable_generator_detection: bool = False,
+        meta: Optional[str] = None,
+    ) -> ProcessItem:
+        """Uploads a video file for deepfake detection and returns the process item.
+
+        Args:
+            file_path (str): Path to the video file to upload.
+            name (str, optional): Optional name for the job request. Defaults to filename.
+            embeddings (bool): Whether to include speaker and deepfake embeddings in the audio results. Defaults to False.
+            enable_generator_detection (bool): Whether to include prediction for the source of the deepfake (generator model) in the audio results. Defaults to False.
+            meta (str, optional): Metadata json containing any extra user-defined metadata.
+        Returns:
+            ProcessItem: The process item containing details about the submitted process.
+        """
+        # Create and validate parameters
+        params = DeepfakeAudioUploadParams(
+            file_path=file_path,
+            name=name,
+            embeddings=embeddings,
+            meta=meta,
+            enable_generator_detection=enable_generator_detection,
+        )
+
+        # Use provided name or default to filename
+        job_name = params.name or Path(params.file_path).name
+
+        with open(params.file_path, "rb") as video_file:
+            files = {"file": video_file}
+            data = {
+                "name": job_name,
+                "embeddings": params.embeddings,
+                "enable_generator_detection": params.enable_generator_detection,
+            }
+
+            if params.meta:
+                data["meta"] = params.meta
+
+            data = self._send_request(
+                path=f"detection/clients/{self.config.cid}/processes/video",
+                method="POST",
+                files=files,
+                data=data,
+            )
+
+        return ProcessItem(**data)
+
+    def upload_s3_presigned_video_url(
+        self,
+        url: str,
+        name: Optional[str] = None,
+        embeddings: bool = False,
+        enable_generator_detection: bool = False,
+        meta: Optional[str] = None,
+    ) -> ProcessItem:
+        """Uploads an S3 presigned url pointing to a video file and returns the process item.
+
+        Args:
+            url (str): The S3 presigned url.
+            name (str, optional): Optional name for the job request.
+            embeddings (bool): Whether to include speaker and deepfake embeddings in the audio results. Defaults to False.
+            enable_generator_detection (bool): Whether to include prediction for the source of the deepfake (generator model) in the audio results. Defaults to False.
+            meta (str, optional): Metadata json containing any extra user-defined metadata.
+        Returns:
+            ProcessItem: The process item containing details about the submitted process.
+        """
+        # Create and validate parameters
+        params = DeepfakeS3UrlUploadParams(
+            url=url,
+            name=name,
+            embeddings=embeddings,
+            meta=meta,
+            enable_generator_detection=enable_generator_detection,
+        )
+
+        # Use provided name or default to filename
+        job_name = params.name
+
+        payload = {
+            "url": params.url,
+            "name": job_name,
+            "embeddings": params.embeddings,
+            "enable_generator_detection": params.enable_generator_detection,
+        }
+
+        if params.meta:
+            payload["meta"] = params.meta
+
+        headers = {"content-type": "application/json"}
+
+        response = self._send_request(
+            path=f"detection/clients/{self.config.cid}/processes/s3-presigned-video-url",
+            method="POST",
+            json=payload,
+            headers=headers,
+        )
+
+        return ProcessItem(**response)
+
+    def list_video_processes(
+        self,
+        page: int = 0,
+        page_size: int = 1000,
+        sort: Literal["asc", "desc"] = "asc",
+        start_date: Optional[str] = None,
+        end_date: Optional[str] = None,
+    ) -> ProcessListResponse:
+        """Lists all video deepfake detection processes for the authenticated user.
+
+        Args:
+            page (int): Page number for pagination (default is 0).
+            page_size (int): Number of processes per page (default is 1000).
+            sort (str): Sort order for the processes, should be "asc" or "desc". Defaults to "asc".
+            start_date (str, optional: Filter processes created on or after this date (YYYY-MM-DD).
+            end_date (str, optional): Filter processes created on or before this date (YYYY-MM-DD).
+        Returns:
+            ProcessListResponse: A list of video processes associated with the user.
+        """
+
+        query_params = ProcessListParams(
+            page=page, page_size=page_size, sort=sort, start_date=start_date, end_date=end_date
+        )
+        query_params = query_params.model_dump(by_alias=True, exclude_none=True)
+
+        data = self._send_request(
+            path=f"detection/clients/{self.config.cid}/processes/video",
+            method="GET",
+            data=query_params,
+        )
+
+        return ProcessListResponse(processes=data)
+
+    def get_video_process(self, pid: int) -> ProcessItem:
+        """Retrieves details of a specific video process by its ID.
+
+        Args:
+            pid (int): The process ID to retrieve.
+        Returns:
+            ProcessItem: The process item containing details about the specified process.
+        """
+
+        data = self._send_request(
+            path=f"detection/clients/{self.config.cid}/processes/video/{pid}",
+            method="GET",
+        )
+
+        return ProcessItem(**data)
+
+    def get_video_result(self, pid: int) -> VideoResultResponse:
+        """Retrieves the result of a completed video process by its ID.
+
+        The response contains separate result lists for the audio track
+        (``audio_results``) and the video frames (``video_results``).
+
+        Args:
+            pid (int): The process ID for which to retrieve the result
+        Returns:
+            VideoResultResponse: The result response containing audio and video results.
+        """
+        data = self._send_request(
+            path=f"detection/clients/{self.config.cid}/processes/video/{pid}/results",
+            method="GET",
+        )
+        return VideoResultResponse(**data)
+
     def stream_audio(
         self, audio_stream: Iterator[bytes], options: StreamingOptions
     ) -> Iterator[ResultResponse]:

src/behavioralsignals/models.py

@@ -100,19 +100,21 @@ def validate_meta_json(cls, v):
                 raise ValueError("meta must be valid JSON string")
         return v
 
+
 class DeepfakeAudioUploadParams(AudioUploadParams):
     enable_generator_detection: bool = Field(
-        False, description="Whether to include prediction for the source of the deepfake (generator model)"
+        False,
+        description="Whether to include prediction for the source of the deepfake (generator model)",
     )
 
 
 class DeepfakeS3UrlUploadParams(S3UrlUploadParams):
     enable_generator_detection: bool = Field(
-        False, description="Whether to include prediction for the source of the deepfake (generator model)"
+        False,
+        description="Whether to include prediction for the source of the deepfake (generator model)",
     )
 
 
-
 class ProcessItem(BaseModel):
     """Individual process in the list"""
 
@@ -208,7 +210,7 @@ class ResultItem(BaseModel):
     )
     task: Optional[str] = Field(
         None,
-        description="The behavioral attribute. Can be one of diarization, deepfake, asr, gender, age, language, features, emotion, strength, positivity, speaking_rate, hesitation, politeness. "
+        description="The behavioral attribute. Can be one of diarization, deepfake, visual_deepfake, asr, gender, age, language, features, emotion, strength, positivity, speaking_rate, hesitation, politeness. "
         "Consider visiting the guides in behavioralsignals.readme.io for the latest examples.",
         example="emotion",
     )
@@ -246,6 +248,26 @@ class ResultResponse(BaseModel):
     results: Optional[List[ResultItem]] = None
 
 
+class VideoResultResponse(BaseModel):
+    """Result of a video deepfake detection process.
+
+    Unlike the audio result response, a video process returns two separate result
+    lists: one for the deepfake detection performed on the audio track and one for
+    the deepfake detection performed on the video frames.
+    """
+
+    pid: Optional[int] = Field(None, description="Unique ID for the processing job")
+    cid: Optional[int] = Field(None, description="Client ID that requested the processing")
+    code: Optional[int] = Field(None, description="Code indicating status")
+    message: Optional[str] = Field(None, description="Description of status")
+    audio_results: Optional[List[ResultItem]] = Field(
+        None, description="Audio deepfake detection results"
+    )
+    video_results: Optional[List[ResultItem]] = Field(
+        None, description="Video deepfake detection results"
+    )
+
+
 class StreamingResultResponse(BaseModel):
     pid: Optional[int] = Field(None, description="Unique ID for the processing job")
     cid: Optional[int] = Field(None, description="Client ID that requested the processing")