add function to export mismatches

Author: PascalEgn

digitization/cli.py

@@ -1,5 +1,5 @@
 import click
-from .file_import.file_import import create_import_xml_files
+from .file_import.file_import import create_import_xml_files, get_matching_errors
 from .xml_collect.xml_collect import records_collection
 from .xml_collect.utils import (
     download_files_from_ftp,
@@ -8,6 +8,7 @@
     records_collection_creation,
 )
 import os
+import logging
 
 @click.group()
 def digitization():
@@ -83,6 +84,35 @@ def create_import_xml(data_path, output_path):
     click.echo("✅ XML files created successfully.")
 
 
+@digitization.command("get-s3-matching-errors")
+@click.option("-d", "--data-path", type=str, required=True, help="Path to the boite data files folder.")
+@click.option("-o", "--log-path", type=str, required=True, help="Path to save the log file.")
+def get_s3_matching_errors(data_path, log_path):
+    """Log missing files in S3 and Excel."""
+    log_file = os.path.join(log_path, "matching_errors.log")
+    logging.basicConfig(
+        filename=log_file,
+        level=logging.INFO,
+        format="%(asctime)s - %(levelname)s - %(message)s",
+    )
+    for box_file in os.listdir(data_path):
+        matching_errors = get_matching_errors(data_path, box_file, False)
+        for filetype, missing_files in matching_errors["missing_in_excel"].items():
+            if missing_files:
+                logging.warning(
+                    f"[{box_file}] Missing in Excel ({filetype}): {', '.join(missing_files)}"
+                )
+            else:
+                logging.info(f"[{box_file}] No missing files in Excel for {filetype}.")
+        for filetype, missing_files in matching_errors["missing_in_s3"].items():
+            if missing_files:
+                logging.warning(
+                    f"[{box_file}] Missing in S3 ({filetype}): {', '.join(missing_files)}"
+                )
+            else:
+                logging.info(f"[{box_file}] No missing files in S3 for {filetype}.")
+    click.echo(f"✅ Log file created: {log_file}")
+
 
 if __name__ == "__main__":
     digitization()

digitization/file_import/file_import.py

@@ -2,7 +2,7 @@
 import logging
 import pandas as pd
 from tqdm import tqdm
-from .utils import generate_s3_url, get_s3_client, get_s3_file_path, list_s3_files, create_custom_xml
+from .utils import generate_s3_url, get_s3_client, get_s3_file_path, list_s3_files_and_folders, create_custom_xml, transform_box_file_name
 
 
 
@@ -19,7 +19,7 @@ def process_row(row, box_file, s3_client):
 
     for filetype in ['PDF', 'PDF_LATEX', 'TIFF']:
         s3_prefix = get_s3_file_path(filetype=filetype, box_file=box_file, filename=record_name)
-        files = list_s3_files('cern-archives', s3_prefix, s3_client)
+        files = list_s3_files_and_folders('cern-archives', s3_prefix, s3_client)['files']
 
         if not files:
             logging.info(f"[MISSING] {filetype} for record {record_name} (ID: {record_id}) in {s3_prefix}")
@@ -52,3 +52,41 @@ def create_import_xml_files(data_path, output_path):
         xml_path = os.path.join(xml_output_path, xml_filename)
         create_custom_xml(records_data, xml_path)
         print(f"✅ XML written: {xml_path}")
+
+
+def get_matching_errors(boite_data_path, box_file, corrections_folder=False):
+    """
+    This function reads the Excel file and returns a dict:
+    {
+      'missing_in_excel': {filetype: [...], ...},
+      'missing_in_s3': {filetype: [...], ...}
+    }
+    """
+    s3_client = get_s3_client()
+    boite_data = pd.read_excel(os.path.join(boite_data_path, box_file), header=None)
+    box_file_s3 = transform_box_file_name(box_file)
+    filetypes = ['PDF', 'PDF_LATEX', 'TIFF']
+    boite_values = boite_data[boite_data.columns[1]].tolist()
+
+    missing_in_excel_dict = {}
+    missing_in_s3_dict = {}
+
+    for ft in filetypes:
+        prefix = f'raw/CORRECTIONS/{ft}/{box_file_s3}/' if corrections_folder else f'raw/{ft}/{box_file_s3}/'
+        files_for_type = list_s3_files_and_folders('cern-archives', prefix, s3_client)
+        if ft == 'PDF_LATEX':
+            s3_names = [f.split('/')[-1].split('.')[0] for f in files_for_type['files']]
+        else:
+            s3_names = [f.split('/')[-2] for f in files_for_type['folders']]
+
+        try:
+            s3_names.remove(box_file_s3)
+        except ValueError:
+            pass
+        missing_in_excel_dict[ft] = list(set(s3_names) - set(boite_values))
+        missing_in_s3_dict[ft] = list(set(boite_values) - set(s3_names))
+
+    return {
+        'missing_in_excel': missing_in_excel_dict,
+        'missing_in_s3': missing_in_s3_dict
+    }

digitization/file_import/utils.py

@@ -6,8 +6,11 @@
 load_dotenv()
 
 
+def transform_box_file_name(box_file):
+    return box_file.split('.')[0].upper().replace('-', '_')
+
 def get_s3_file_path(filetype='', box_file='', filename=''):
-    box_file = box_file.split('.')[0].upper().replace('-', '_')
+    box_file = transform_box_file_name(box_file)
     if filetype == 'PDF' or filetype == 'TIFF':
         return f"raw/{filetype}/{box_file}/{filename}/"
     elif filetype == 'PDF_LATEX':
@@ -22,20 +25,32 @@ def get_s3_client():
         endpoint_url='https://s3.cern.ch',
     )
 
-def list_s3_files(bucket_name, prefix, s3_client=None):
+def list_s3_files_and_folders(bucket_name, prefix, s3_client=None):
+    if s3_client is None:
+        s3_client = boto3.client('s3')
+
     try:
         response = s3_client.list_objects_v2(
             Bucket=bucket_name,
             Prefix=prefix,
             Delimiter='/'
         )
+
+        files = []
+        folders = []
+
+        if 'CommonPrefixes' in response:
+            folders = [cp['Prefix'] for cp in response['CommonPrefixes']]
+
         if 'Contents' in response:
-            return [obj['Key'] for obj in response['Contents'] if
-                    not obj['Key'].endswith('/')]
-        else:
-            return []
-    except Exception:
-        return []
+            files = [obj['Key'] for obj in response['Contents'] if not obj['Key'].endswith('/')]
+
+        return {'files': files, 'folders': folders}
+
+    except Exception as e:
+        print(f"Error listing S3 path: {e}")
+        return {'files': [], 'folders': []}
+
 
 def generate_s3_url(bucket_name, file_key, expiration=31556952, s3_client=None):
     return f"{bucket_name}/{file_key}/{expiration}"