feat(S3): Conditional s3 upload

Only upload to S3 when file checksum is different.

Signed-off-by: [email protected]

Author: Lorenzovagliano

dags/common/scoap3_s3.py

@@ -1,7 +1,9 @@
+import hashlib
 import os
 from uuid import uuid4
 
 import requests
+from botocore.exceptions import ClientError
 from common.repository import IRepository
 from common.s3_service import S3Service
 from structlog import get_logger
@@ -21,6 +23,11 @@ def update_filename_extension(filename, type):
         return f"{filename}{extension}"
 
 
+def get_file_checksum(data):
+    """Calculate MD5 checksum of file data"""
+    return hashlib.md5(data).hexdigest()
+
+
 class Scoap3Repository(IRepository):
     def __init__(self):
         super().__init__()
@@ -30,6 +37,31 @@ def __init__(self):
         self.s3 = S3Service(self.bucket)
         self.client = self.s3.meta.client
 
+    def file_exists_with_same_checksum(self, bucket, key, data=None):
+        """Check if a file exists at the destination and has the same checksum"""
+        try:
+            if data:
+                # Calculate checksum of data
+                data_checksum = get_file_checksum(data)
+
+                # Get destination file if it exists
+                try:
+                    dest_response = self.client.head_object(Bucket=bucket, Key=key)
+                    dest_checksum = dest_response.get("ETag", "").strip('"')
+
+                    # Compare checksums
+                    return data_checksum == dest_checksum
+                except ClientError as e:
+                    if e.response["Error"]["Code"] == "404":
+                        # File doesn't exist at destination
+                        return False
+                    raise
+
+            return False
+        except Exception as e:
+            logger.error("Error checking file existence", error=str(e))
+            return False
+
     def copy_file(self, source_bucket, source_key, prefix=None, type=None):
         if not self.upload_enabled:
             return ""
@@ -129,9 +161,18 @@ def download_and_upload_to_s3(self, url, prefix=None, headers=None, type=None):
             logger.error("Failed to download file", error=str(e), url=url)
             return
 
+        if self.file_exists_with_same_checksum(
+            self.bucket, destination_key, data=response.content
+        ):
+            logger.info(
+                "File already exists with the same checksum, skipping upload",
+                url=url,
+                destination=f"{self.bucket}/{destination_key}",
+            )
+            return f"{self.bucket}/{destination_key}"
+
         try:
-            # Upload the file to S3
-            self.client.put_object(
+            response = self.client.put_object(
                 Body=response.content,
                 Bucket=self.bucket,
                 Key=destination_key,