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,
@@ -140,7 +181,10 @@ def download_and_upload_to_s3(self, url, prefix=None, headers=None, type=None):
                 },
                 ACL="public-read",
             )
-            return f"{self.bucket}/{destination_key}"
+
+            version_id = response.get("VersionId")
+            s3_url = f"{self.bucket}/{destination_key}"
+            return {"url": s3_url, "version_id": version_id}
         except Exception as e:
             logger.error(
                 "Failed to upload file",

tests/integration/common/cassettes/download_pdf.yaml

@@ -0,0 +1,42 @@
+interactions:
+  - request:
+      body: null
+      headers:
+        Accept:
+          - "*/*"
+        Accept-Encoding:
+          - gzip, deflate
+        Connection:
+          - keep-alive
+        User-Agent:
+          - python-requests/2.31.0
+      method: HEAD
+      uri: https://scoap3-prod-backend.s3.cern.ch/media/harvested_files/10.1155/2024/3681297/3681297.pdf
+    response:
+      body:
+        string: ""
+      headers:
+        Accept-Ranges:
+          - bytes
+        Bucket:
+          - scoap3-prod-backend
+        Content-Length:
+          - "1013660"
+        Content-Type:
+          - binary/octet-stream
+        Date:
+          - Wed, 19 Mar 2025 14:45:45 GMT
+        Etag:
+          - '"e737b90022b130cf83e58f253ad47a5f"'
+        Last-Modified:
+          - Tue, 25 Feb 2025 00:15:39 GMT
+        X-Amz-Meta-Source-Url:
+          - https://s3.amazonaws.com/downloads.hindawi.com/journals/ahep/2024/3681297.pdf
+        X-Amz-Request-Id:
+          - tx00000176051ff8bcdcbfb-0067dad899-3b453ea1-default
+        X-Rgw-Object-Type:
+          - Normal
+      status:
+        code: 200
+        message: OK
+version: 1