From 9989860cc01c15d4481f7ae86607be997556b2ff Mon Sep 17 00:00:00 2001 From: AndreasPatakis Date: Thu, 2 Jul 2026 11:50:38 +0300 Subject: [PATCH] Add batch video deepfake detection support --- README.md | 22 +++ examples/batch/video_deepfake_polling.py | 83 +++++++++++ src/behavioralsignals/__init__.py | 4 +- src/behavioralsignals/deepfakes.py | 170 +++++++++++++++++++++++ src/behavioralsignals/models.py | 30 +++- 5 files changed, 303 insertions(+), 6 deletions(-) create mode 100644 examples/batch/video_deepfake_polling.py diff --git a/README.md b/README.md index dcaa212..54d38b9 100644 --- a/README.md +++ b/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: diff --git a/examples/batch/video_deepfake_polling.py b/examples/batch/video_deepfake_polling.py new file mode 100644 index 0000000..de8fbe3 --- /dev/null +++ b/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}") diff --git a/src/behavioralsignals/__init__.py b/src/behavioralsignals/__init__.py index 2bbd594..f9391e0 100644 --- a/src/behavioralsignals/__init__.py +++ b/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"] diff --git a/src/behavioralsignals/deepfakes.py b/src/behavioralsignals/deepfakes.py index 74949a3..4caeec3 100644 --- a/src/behavioralsignals/deepfakes.py +++ b/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]: diff --git a/src/behavioralsignals/models.py b/src/behavioralsignals/models.py index 723c54b..422719d 100644 --- a/src/behavioralsignals/models.py +++ b/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")