Add Feature : Deepfake

edenai_apis/api_keys/sightengine_settings_template.json

@@ -0,0 +1,4 @@
+{
+    "api_key": "",
+    "api_user": ""
+}

edenai_apis/apis/__init__.py

@@ -53,6 +53,7 @@
 from .sapling import SaplingApi
 from .senseloaf import SenseloafApi
 from .sentisight import SentiSightApi
+from .sightengine import SightEngineApi
 from .skybiometry import SkybiometryApi
 from .smartclick import SmartClickApi
 from .speechmatics import SpeechmaticsApi

edenai_apis/apis/sightengine/__init__.py

@@ -0,0 +1 @@
+from .sightengine_api import SightEngineApi

edenai_apis/apis/sightengine/info.json

@@ -0,0 +1,13 @@
+{
+    "image": {
+        "deepfake": {
+            "version" : "v1beta"
+        }
+    },
+
+    "video": {
+        "deepfake": {
+            "version" : "v1beta"
+        }
+    }
+}

edenai_apis/apis/sightengine/outputs/image/deepfake_detection_output.json

@@ -0,0 +1,21 @@
+{
+  "original_response": {
+    "status": "success",
+    "request": {
+      "id": "req_h4tygWOFSudQCj175YtZg",
+      "timestamp": 1727355854.923456,
+      "operations": 1
+    },
+    "type": {
+      "deepfake": 0.01
+    },
+    "media": {
+      "id": "med_h4tyD5FNLdWjes2V36427",
+      "uri": "https://cdn.discordapp.com/attachments/1285184350261870694/1288046848560463892/tom-cruise_Deepfake2.jpeg?ex=66f6659b&is=66f5141b&hm=0eeade51129487243fddf2424f17e5cf24f640a4ecade235d87c1dab8c7e442b&"
+    }
+  },
+  "standardized_response": {
+    "ai_score": 0.99,
+    "prediction": "ai-generated"
+  }
+}

edenai_apis/apis/sightengine/outputs/video/deepfake_detection_async_output.json

@@ -0,0 +1,138 @@
+{
+  "status": "succeeded",
+  "provider_job_id": "med_h4tlkcZQ7f5ESX90CuvvZ",
+  "original_response": {
+    "media": {
+      "id": "med_h4tlkcZQ7f5ESX90CuvvZ",
+      "uri": "https://d14uq1pz7dzsdq.cloudfront.net/4f8f3bb1-b322-4ef3-a166-da641d37d746_/tmp/tmp08h_rdxu.upload.mp4?Expires=1727709680&Signature=DmyYhOaki-tGoHz2QT1ekCk7B0iDBuaHQbX6kpwUM9m1WlEZqbxAN0uqDruu0g7f3i8OrUProh1Kgi91H4HciN0s74o80P8QAd7LjQmBM86aLedvAaoeGeBJis3dXQrj2jHtMyohq~yf1mRi57YbCdxcI5jxtcWUWVEdtmHi0vFGuINhWMymaFGnrKcvEXRFNdTrwpWt9Hs6mVqX9xbYuVd0UFUo1wTyFoYwL4FGbSh7tXLljsItAL7BfB5LKjMryMiD4kTe3NGOBIxPtDRfiw2FKXK0ij2ksTPeHyA3vUaB--3~l6GCIYoUS5RXyNc-GLpuCNdLb5baY8wKrumrOQ__&Key-Pair-Id=K1F55BTI9AHGIK"
+    },
+    "request": "req_h4tltii7O4PVL5g3rp7ib",
+    "data": {
+      "status": "finished",
+      "started": 1727355075.403806,
+      "last_update": 1727355077.599314,
+      "operations": 8,
+      "progress": 1,
+      "frames": [
+        {
+          "info": {
+            "id": "med_h4tlkcZQ7f5ESX90CuvvZ_1",
+            "position": 0
+          },
+          "type": {
+            "deepfake": 0.001
+          }
+        },
+        {
+          "info": {
+            "id": "med_h4tlkcZQ7f5ESX90CuvvZ_2",
+            "position": 2000
+          },
+          "type": {
+            "deepfake": 0.001
+          }
+        },
+        {
+          "info": {
+            "id": "med_h4tlkcZQ7f5ESX90CuvvZ_3",
+            "position": 4000
+          },
+          "type": {
+            "deepfake": 0.001
+          }
+        },
+        {
+          "info": {
+            "id": "med_h4tlkcZQ7f5ESX90CuvvZ_4",
+            "position": 6000
+          },
+          "type": {
+            "deepfake": 0.001
+          }
+        },
+        {
+          "info": {
+            "id": "med_h4tlkcZQ7f5ESX90CuvvZ_5",
+            "position": 8000
+          },
+          "type": {
+            "deepfake": 0.001
+          }
+        },
+        {
+          "info": {
+            "id": "med_h4tlkcZQ7f5ESX90CuvvZ_6",
+            "position": 10000
+          },
+          "type": {
+            "deepfake": 0.001
+          }
+        },
+        {
+          "info": {
+            "id": "med_h4tlkcZQ7f5ESX90CuvvZ_7",
+            "position": 12000
+          },
+          "type": {
+            "deepfake": 0.001
+          }
+        },
+        {
+          "info": {
+            "id": "med_h4tlkcZQ7f5ESX90CuvvZ_8",
+            "position": 14000
+          },
+          "type": {
+            "deepfake": 0.001
+          }
+        }
+      ]
+    }
+  },
+  "standardized_response": {
+    "average_score": 0.001,
+    "prediction": "original",
+    "details_per_frame": [
+      {
+        "position": 0,
+        "score": 0.001,
+        "prediction": "original"
+      },
+      {
+        "position": 2000,
+        "score": 0.001,
+        "prediction": "original"
+      },
+      {
+        "position": 4000,
+        "score": 0.001,
+        "prediction": "original"
+      },
+      {
+        "position": 6000,
+        "score": 0.001,
+        "prediction": "original"
+      },
+      {
+        "position": 8000,
+        "score": 0.001,
+        "prediction": "original"
+      },
+      {
+        "position": 10000,
+        "score": 0.001,
+        "prediction": "original"
+      },
+      {
+        "position": 12000,
+        "score": 0.001,
+        "prediction": "original"
+      },
+      {
+        "position": 14000,
+        "score": 0.001,
+        "prediction": "original"
+      }
+    ]
+  }
+}

edenai_apis/apis/sightengine/sightengine_api.py

@@ -0,0 +1,171 @@
+import json
+from typing import Dict, Any, Optional
+import requests
+from edenai_apis.apis.amazon.helpers import check_webhook_result
+from edenai_apis.features import ProviderInterface, ImageInterface
+
+from edenai_apis.features.video.deepfake_detection_async.deepfake_detection_async_dataclass import (
+    DeepfakeDetectionAsyncDataClass as VideoDeepfakeDetectionAsyncDataclass,
+)
+from edenai_apis.features.image.deepfake_detection.deepfake_detection_dataclass import (
+    DeepfakeDetectionDataClass as ImageDeepfakeDetectionDataclass,
+)
+from edenai_apis.loaders.data_loader import ProviderDataEnum
+from edenai_apis.loaders.loaders import load_provider
+from edenai_apis.utils.exception import ProviderException
+from edenai_apis.utils.parsing import extract
+from edenai_apis.utils.types import AsyncBaseResponseType, AsyncLaunchJobResponseType, AsyncPendingResponseType, AsyncResponseType, ResponseType
+from edenai_apis.utils.upload_s3 import upload_file_to_s3
+
+class SightEngineApi(ProviderInterface, ImageInterface):
+    provider_name = "sightengine"
+
+    def __init__(self, api_keys: Optional[Dict[str, Any]] = None):
+        self.api_settings = load_provider(
+            ProviderDataEnum.KEY,
+            provider_name=self.provider_name,
+            api_keys=api_keys or {},
+        )
+        self.api_url = "https://api.sightengine.com/1.0"
+        self.headers = {
+            "Content-Type": "application/json",
+            "Authorization": f'Bearer {self.api_settings["api_key"]}',
+
+        }
+        self.webhook_settings = load_provider(ProviderDataEnum.KEY, "webhooksite")
+        self.webhook_token = self.webhook_settings["webhook_token"]
+        self.webhook_url = f"https://webhook.site/{self.webhook_token}"
+
+    def image__deepfake_detection(
+        self, file: Optional[str] = None, file_url: Optional[str] = None
+    ) -> ResponseType[ImageDeepfakeDetectionDataclass]:
+        if not file_url and not file:
+            raise ProviderException("file or file_url required")
+
+        payload = {"url": file_url or upload_file_to_s3(file, file),
+                              "models":"deepfake",
+                              "api_user": self.api_settings["api_user"],
+                              "api_secret": self.api_settings["api_key"]}
+
+        response = requests.get(
+            f"{self.api_url}/check.json",
+            params=payload,
+        )
+
+        print(response.json())
+
+        if response.status_code != 200:
+            raise ProviderException(response.json(), code=response.status_code)
+
+        original_response = response.json()
+
+        score = 1 - extract(original_response,["type","deepfake"],None)
+        if score is None:
+            raise ProviderException(response.json())
+        prediction = ImageDeepfakeDetectionDataclass.set_label_based_on_score(score)
+
+        standardized_response = ImageDeepfakeDetectionDataclass(
+            ai_score=score,
+            prediction=prediction,
+        )
+
+
+        return ResponseType[ImageDeepfakeDetectionDataclass](
+            original_response=original_response,
+            standardized_response=standardized_response,
+        )
+
+    def video__deepfake_detection_async__launch_job(
+        self, file: Optional[str] = None, file_url: Optional[str] = None
+    ) -> AsyncLaunchJobResponseType:
+        if not file_url and not file:
+            raise ProviderException("file or file_url required")
+
+        payload = {
+            "models": "deepfake",
+            "api_user": self.api_settings["api_user"],
+            "api_secret": self.api_settings["api_key"],
+            "callback_url": self.webhook_url,
+        }
+
+        if file:
+            with open(file, 'rb') as video_file:
+                files = {'media': video_file}
+
+                response = requests.post(
+                    f"{self.api_url}/video/check.json",
+                    files=files,
+                    data=payload,
+                )
+        elif file_url:
+            payload['stream_url'] = file_url
+
+            response = requests.get(
+                f"{self.api_url}/video/check.json",
+                params=payload,
+            )
+
+        if response.status_code != 200:
+            raise ProviderException(response.json(), code=response.status_code)
+
+
+        original_response = response.json()
+
+        media_id = original_response.get("media", {}).get("id")
+
+        requests.post(
+            self.webhook_url,
+            json=media_id,
+            headers={"content-type": "application/json"}
+        )
+
+        return AsyncLaunchJobResponseType(provider_job_id=media_id)
+
+    def video__deepfake_detection_async__get_job_result(
+        self, media_id: str
+    ) -> AsyncBaseResponseType[VideoDeepfakeDetectionAsyncDataclass]:
+        wehbook_result, response_status = check_webhook_result(
+            media_id, self.webhook_settings
+        )
+
+        if response_status != 200:
+            raise ProviderException(wehbook_result, code=response_status)
+
+        try:
+            original_response = json.loads(wehbook_result[0]["content"])
+        except (IndexError, json.JSONDecodeError):
+            return AsyncPendingResponseType[VideoDeepfakeDetectionAsyncDataclass](
+                provider_media_id=media_id
+            )
+
+        if original_response.get("status") == "pending":
+            return AsyncPendingResponseType[VideoDeepfakeDetectionAsyncDataclass](
+                provider_media_id=media_id
+            )
+
+        score = extract(original_response, ["data", "frames", 0, "type", "deepfake"], None)
+        if score is None:
+            raise ProviderException("Deepfake score not found in response.")
+
+        prediction = VideoDeepfakeDetectionAsyncDataclass.set_label_based_on_score(score)
+
+        standardized_response = VideoDeepfakeDetectionAsyncDataclass(
+            average_score=score,
+            prediction=prediction,
+            details_per_frame=[
+                {
+                    "position": frame.get("info", {}).get("position"),
+                    "score": frame.get("type", {}).get("deepfake"),
+                    "prediction": VideoDeepfakeDetectionAsyncDataclass.set_label_based_on_score(
+                        frame.get("type", {}).get("deepfake")
+                    ), 
+                }
+                for frame in extract(original_response, ["data", "frames"], [])
+            ],
+        )
+
+        return AsyncResponseType[VideoDeepfakeDetectionAsyncDataclass](
+            original_response=original_response,
+            standardized_response=standardized_response,
+            provider_job_id=media_id,
+        )

edenai_apis/features/image/__init__.py

@@ -2,6 +2,7 @@
     AnonymizationDataClass,
     anonymization_arguments,
 )
+from .deepfake_detection import deepfake_detection_arguments
 from .ai_detection import ai_detection_arguments
 from .automl_classification import (
     automl_classification_create_project_arguments,

edenai_apis/features/image/deepfake_detection/__init__.py

@@ -0,0 +1,2 @@
+from .deepfake_detection_args import deepfake_detection_arguments
+from .deepfake_detection_dataclass import DeepfakeDetectionDataClass

edenai_apis/features/image/deepfake_detection/deepfake_detection_args.py

@@ -0,0 +1,10 @@
+dnld_t_dpf = ("https://cdn.discordapp.com/attachments/1285184350261870694/1288046846865969163/donald_trump_Deepfake1.jpeg?ex=66f3c29b&is=66f2711b&hm=adda1bc1ae8253fd7722745d4ecdc827873cfa819e8a1abbea65e1cf80545703&")
+t_cruise_not_dpf = ("https://cdn.discordapp.com/attachments/1285184350261870694/1288046848560463892/tom-cruise_Deepfake2.jpeg?ex=66f6659b&is=66f5141b&hm=0eeade51129487243fddf2424f17e5cf24f640a4ecade235d87c1dab8c7e442b&")
+
+
+HUMAN_IMAGE_EXAMPLE = t_cruise_not_dpf
+
+def deepfake_detection_arguments(provider_name: str) -> dict:
+    return {
+        "file_url": HUMAN_IMAGE_EXAMPLE
+    }