Refactor Validation and Create RollCall module

AGENTS.md

@@ -17,6 +17,7 @@
 
 ### Core Modules
 - `__main__.py` - Main entry point with CLI argument parsing and workflow orchestration
+- `roll_call.py` - CSV validation and input handling functions
 - `download.py` - Core image downloading functionality with retry logic
 - `buddy_check.py` - Checksum validation and download verification using BuddyCheck class
 - `utils.py` - Helper functions for CSV processing, logging, and image downsampling

src/cautiousrobot/__main__.py

@@ -8,12 +8,12 @@
 import pandas as pd
 import argparse
 import hashlib
-import os
 import sys
 from sumbuddy import get_checksums
-from cautiousrobot.utils import process_csv, check_existing_images
+from cautiousrobot.utils import process_csv
 from cautiousrobot.buddy_check import BuddyCheck
 from cautiousrobot.download import download_images
+from cautiousrobot.roll_call import RollCall
 from cautiousrobot.__about__ import __version__
 
 def parse_args():
@@ -53,46 +53,6 @@ def parse_args():
     return parser.parse_args()
 
 
-def validate_csv_extension(csv_path):
-    """Validate that the input file has a .csv extension."""
-    if not csv_path.endswith(".csv"):
-        sys.exit("Expected CSV for input file; extension should be '.csv'")
-
-
-def setup_expected_columns(args):
-    """Set up the expected columns dictionary for CSV processing."""
-    subfolders = args.subdir_col
-    expected_cols = {
-        "filename_col": args.img_name_col.lower(),
-        "url_col": args.url_col.lower()
-    }
-    if subfolders:
-        subfolders = subfolders.lower()
-        expected_cols["subfolders"] = subfolders
-    return expected_cols, subfolders
-
-
-def validate_filename_uniqueness(data_df, filename_col):
-    """Validate that the filename column contains unique values."""
-    if data_df.loc[data_df[filename_col].notna()].shape[0] != data_df[filename_col].nunique():
-        sys.exit(f"{filename_col} is not a unique identifier for this dataset, please choose a column with unique values for filenames.")
-
-
-def handle_missing_filenames(data_df, filename_col, url_col):
-    """Handle cases where URLs exist but filenames are missing."""
-    urls_no_name = len(data_df.loc[(data_df[filename_col].isna() & (data_df[url_col].notna()))])
-    if urls_no_name > 0:
-        ignore = input(f"'{filename_col}' is missing values for {urls_no_name} URLs. Proceed with download ignoring these URLs? [y/n]: ")
-        if ignore.lower() != "y":
-            sys.exit("Exited without executing.")
-
-
-def validate_output_directory(img_dir):
-    """Validate that the output directory doesn't already exist."""
-    if os.path.exists(img_dir):
-        sys.exit(f"'{img_dir}' already exists. Exited without executing.")
-
-
 def setup_log_paths(csv_path):
     """Set up the log file paths based on the CSV path."""
     metadata_path = csv_path.split(".")[0]
@@ -147,11 +107,13 @@ def main():
     args = parse_args()
     csv_path = args.input_file
 
+    roll_call = RollCall(csv_path)
+
     # Validate CSV extension
-    validate_csv_extension(csv_path)
+    roll_call.validate_csv_extension(csv_path)
 
     # Set up expected columns and process CSV
-    expected_cols, subfolders = setup_expected_columns(args)
+    expected_cols, subfolders = roll_call.setup_expected_columns(args)
     try:
         data_df = process_csv(csv_path, expected_cols)
     except Exception as missing_cols:
@@ -160,15 +122,35 @@ def main():
     # Validate data and handle missing filenames
     filename_col = expected_cols["filename_col"]
     url_col = expected_cols["url_col"]
-    validate_filename_uniqueness(data_df, filename_col)
-    handle_missing_filenames(data_df, filename_col, url_col)
+    roll_call.validate_filename_uniqueness(data_df, filename_col)
+    roll_call.handle_missing_filenames(data_df, filename_col, url_col)
 
     # Set source DataFrame for only non-null filename values
     source_df = data_df.loc[data_df[filename_col].notna()].copy()
 
+    # If a verifier column is provided, check for duplicate checksums before downloading
+    if args.verifier_col:
+        roll_call.check_duplicate_checksums(
+            data_df=source_df,
+            hash_col=args.verifier_col.lower(),
+            ignore_duplicates=getattr(args, "ignore_duplicates", False)
+        )
+
     # Validate and handle existing output directory
     img_dir = args.output_dir
-    source_df, filtered_df = check_existing_images(csv_path, img_dir, source_df, filename_col, subfolders)
+    source_df, filtered_df = roll_call.check_existing_images(csv_path, img_dir, source_df, filename_col, subfolders)
+
+    # Prepare summary info
+    num_images = filtered_df.shape[0]
+    downsample_dir = img_dir + "_downsized" if isinstance(args.side_length, int) else None
+
+    # Print RollCall summary
+    roll_call.print_download_summary(
+        img_dir=img_dir,
+        downsample_dir=downsample_dir,
+        subfolders=subfolders,
+        num_images=num_images
+    )
 
     # Set up log paths
     log_filepath, error_log_filepath, metadata_path = setup_log_paths(csv_path)

src/cautiousrobot/roll_call.py

@@ -0,0 +1,263 @@
+# CSV validation and input handling functions.
+
+import os
+import sys
+from sumbuddy import gather_file_paths
+from sumbuddy.exceptions import EmptyInputDirectoryError
+
+
+class RollCall:
+    """Pre-download validation and directory verification."""
+
+    def __init__(self, csv_path=None):
+        """
+        Initialize the roll-call validator with an optional CSV path.
+
+        Parameters:
+            csv_path (str): Path to the source CSV file used for reporting and output naming.
+        """
+        self.csv_path = csv_path
+
+    def validate_csv_extension(self, csv_path):
+        """
+        Validate that the supplied input path points to a CSV file.
+
+        Parameters:
+            csv_path (str): Path to the input CSV file.
+
+        Raises:
+            SystemExit: If the input file does not end with the '.csv' extension.
+        """
+        # The CLI expects CSV input, so the validation stops early when the file type is wrong.
+        if not csv_path.lower().endswith(".csv"):
+            sys.exit("Expected CSV for input file; extension should be '.csv'")
+
+    def validate_filename_uniqueness(self, data_df, filename_col):
+        """
+        Validate that the selected filename column contains unique values.
+
+        Parameters:
+            data_df (pd.DataFrame): DataFrame loaded from the CSV file.
+            filename_col (str): Name of the column that should uniquely identify each image.
+
+        Raises:
+            SystemExit: If duplicate filenames are found in the column.
+        """
+        # Count only non-missing values because empty filenames should not be treated as duplicates.
+        if data_df.loc[data_df[filename_col].notna()].shape[0] != data_df[filename_col].nunique():
+            sys.exit(f"{filename_col} is not a unique identifier for this dataset, please choose a column with unique values for filenames.")
+
+    def handle_missing_filenames(self, data_df, filename_col, url_col):
+        """
+        Handle cases where URLs are present but filenames are missing.
+
+        Parameters:
+            data_df (pd.DataFrame): DataFrame loaded from the CSV file.
+            filename_col (str): Name of the column containing image filenames.
+            url_col (str): Name of the column containing image URLs.
+
+        Returns:
+            pd.DataFrame | None: Rows with missing filenames when present; otherwise None.
+        """
+        # Find rows that have a URL but no filename, since those records cannot be downloaded safely.
+        missing = data_df.loc[data_df[filename_col].isna() & data_df[url_col].notna()]
+        if missing.empty:
+            # No missing names were found, so there is nothing to report.
+            return None
+        if len(missing) <= 5:
+            # Show the full set of problematic rows directly in the console for small cases.
+            print("\n Missing filenames detected (showing all):")
+            print(missing)
+        else:
+            # For larger problems, save the affected rows to disk so they can be fixed offline.
+            csv_base = os.path.splitext(self.csv_path)[0]
+            save_path = f"{csv_base}_missing_filenames.csv"
+            missing.to_csv(save_path, index=False)
+            print(
+                f"\n Missing filenames detected for {len(missing)} rows.\n"
+                f"Because there are more than 5, they were saved to:\n  {save_path}\n"
+                f"Please correct the CSV and re-run."
+            )
+        # Return the flagged rows so the calling code can inspect or report them further.
+        return missing
+
+    def setup_expected_columns(self, args):
+        """
+        Build the expected column mapping used by the download workflow.
+
+        Parameters:
+            args (argparse.Namespace): Parsed command-line arguments.
+
+        Returns:
+            tuple[dict, str | None]: Column mapping to use for filename, URL, and optional subfolder handling.
+        """
+        # Normalize the CLI input into the internal column names used throughout the workflow.
+        subfolders = args.subdir_col
+        expected_cols = {
+            "filename_col": args.img_name_col.lower(),
+            "url_col": args.url_col.lower()
+        }
+        if subfolders:
+            # Keep the subfolder column name consistent with the rest of the input handling.
+            subfolders = subfolders.lower()
+            expected_cols["subfolders"] = subfolders
+        return expected_cols, subfolders
+
+    def check_existing_images(self, csv_path, img_dir, source_df, filename_col, subfolders=None):
+        """
+        Checks which files from the CSV already exist in the image directory.
+
+        Adds a new boolean column `in_img_dir` to source_df indicating which images
+        are already in the directory.
+
+        If all images already exist in the directory, the function will exit early
+        by calling `sys.exit()`, and no further processing will occur.
+
+        Parameters:
+            csv_path (str): Path to the CSV file containing image information.
+            img_dir (str): Path to the directory where images are to be stored.
+            source_df (pd.DataFrame): DataFrame loaded from the CSV, containing image metadata.
+            filename_col (str): Name of the column in source_df that contains image filenames.
+            subfolders (str): Name of the column in source_df that contains subfolder names. (optional)
+
+        Returns:
+            updated_df (pd.DataFrame): DataFrame with new column 'in_img_dir' indicating presence in img_dir.
+            filtered_df (pd.DataFrame): DataFrame filtered to only files not present in img_dir.
+        """
+        # Create a copy to avoid modifying the original DataFrame
+        df = source_df.copy()
+
+        if not os.path.exists(img_dir):
+            # Directory doesn't exist, so nothing to check
+            df["in_img_dir"] = False
+            # Return the updated df and the filtered dataframe of items that still need downloading
+            filtered_df = df[~df["in_img_dir"]].copy()
+            return df, filtered_df
+
+        try:
+            existing_files = gather_file_paths(img_dir)
+        except EmptyInputDirectoryError:
+            # If the directory exists but is empty, sumbuddy raises an error.
+            # We catch it and treat it as an empty file list.
+            existing_files = []
+
+        existing_full_paths = {os.path.normpath(os.path.relpath(f, img_dir)) for f in existing_files}
+
+        if subfolders:
+            # We use a generic join here, but the apply(os.path.normpath) below fixes it for the specific OS
+            raw_paths = df[subfolders].astype(str) + os.sep + df[filename_col].astype(str)
+            # This converts '/' to '\' on Windows, or vice versa, ensuring a match
+            df["expected_path"] = raw_paths.apply(os.path.normpath)
+        else:
+            # Normalize even simple filenames just in case they contain pathing characters
+            df["expected_path"] = df[filename_col].astype(str).apply(os.path.normpath)
+
+        # Determine which expected paths physically exist
+        expected_present = df["expected_path"].isin(existing_full_paths)
+        df["in_img_dir"] = expected_present.copy()
+
+        # Clean up the temporary column before returning.
+        df = df.drop(columns=["expected_path"])
+
+        # Create filtered DataFrame
+        filtered_df = df[~df["in_img_dir"]].copy()
+
+        # Exit if all images are already there
+        if filtered_df.empty:
+            sys.exit(f"'{img_dir}' already contains all images. Exited without executing.")
+        else:
+            # Print directory status message - pre-download
+            num_existing = len(existing_files)
+            print(f"There are {num_existing} of the desired files already in {img_dir}. Based on {csv_path}, {filtered_df.shape[0]} images should be downloaded.")
+
+        return df, filtered_df
+
+    def _preview_or_save(self, label, df):
+        """
+        Print a small preview of a DataFrame or save it to disk when the result is larger.
+
+        Parameters:
+            label (str): Descriptive prefix for the output file and console message.
+            df (pd.DataFrame): DataFrame containing the rows to preview or save.
+        """
+        if len(df) <= 5:
+            # Small result sets are easier to inspect directly in the terminal.
+            print(f"\n {label.replace('_', ' ').title()} detected (showing all):")
+            print(df)
+        else:
+            # Larger result sets are written to a CSV so the user can review them without clutter.
+            csv_base = os.path.splitext(self.csv_path)[0]
+            save_path = f"{csv_base}_{label}.csv"
+            df.to_csv(save_path, index=False)
+            print(
+                f"\n {len(df)} {label.replace('_', ' ')} detected.\n"
+                f"Full list saved to:\n  {save_path}\n"
+            )
+
+    def check_duplicate_checksums(self, data_df, hash_col, ignore_duplicates=False):
+        """
+        Detect duplicate checksum values and optionally block execution.
+
+        Parameters:
+            data_df (pd.DataFrame): DataFrame containing the checksum column.
+            hash_col (str): Name of the checksum column to inspect.
+            ignore_duplicates (bool): If True, allow execution to continue after reporting duplicates.
+
+        Raises:
+            SystemExit: If duplicate checksums are found and duplicates are not ignored.
+        """
+        # Only consider non-null hashes when searching for repeated values.
+        dupes = (
+            data_df[data_df[hash_col].notna()]
+            .groupby(hash_col)
+            .filter(lambda x: len(x) > 1)
+        )
+
+        if dupes.empty:
+            # No repeated checksums were found, so the download process can continue.
+            print(" No duplicate checksums detected.")
+            return
+
+        # If duplicates exist, preview or save
+        self._preview_or_save("duplicate_checksums", dupes)
+
+        if ignore_duplicates:
+            # The user explicitly allowed duplicates, so the workflow reports them and proceeds.
+            print(
+                f"\n Duplicate checksums detected ({len(dupes)} rows), "
+                f"but --ignore-duplicates was passed. Continuing.\n"
+            )
+            return
+
+        # Default behavior: block execution
+        sys.exit(
+            " Duplicate checksums detected. "
+            "Use --ignore-duplicates to allow downloading duplicates."
+        )
+
+    def print_download_summary(self, img_dir, downsample_dir, subfolders, num_images):
+        """
+        Print a summary of where images and downsized images will be saved.
+
+        Parameters:
+            img_dir (str): Destination directory for downloaded images.
+            downsample_dir (str | None): Optional directory for resized images.
+            subfolders (str | None): Optional subfolder naming column.
+            num_images (int): Number of images that will be downloaded.
+        """
+        # Print the planned download layout so the user can confirm the output locations.
+        print("\n Download Summary")
+        print("--------------------")
+        print(f"Images will be downloaded to: {img_dir}")
+
+        if downsample_dir:
+            print(f"Downsampled images will be saved to: {downsample_dir}")
+        else:
+            print("Downsampled images: not requested")
+
+        if subfolders:
+            print(f"Subfolders enabled: {subfolders}")
+        else:
+            print("Subfolders: none")
+
+        print(f"Images to download: {num_images}\n")