diff --git a/AGENTS.md b/AGENTS.md index 12bc364..169edb7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -17,6 +17,7 @@ ### Core Modules - `__main__.py` - Main entry point with CLI argument parsing and workflow orchestration +- `roll_call.py` - Pre-download checks and processing. - `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 diff --git a/src/cautiousrobot/__main__.py b/src/cautiousrobot/__main__.py index 464847a..d67636e 100644 --- a/src/cautiousrobot/__main__.py +++ b/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(): @@ -49,50 +49,11 @@ def parse_args(): help = f"checksum algorithm to use on images (default: md5, available: {available_algorithms})" ) opt_args.add_argument("-v", "--verifier-col", required = False, help = "name of column in source CSV with checksums (same hash as -a) to verify download", nargs = "?") + opt_args.add_argument("--ignore-duplicates", action="store_true", help="Continue download when duplicate checksum values are detected in the source CSV. Duplicates are still printed or saved.") 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 +108,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,22 +123,41 @@ 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) # Download images with or without downsampling - if isinstance(args.side_length, int): - downsample_dest_path = img_dir + "_downsized" + if downsample_dir: # Download images from urls & save downsample copy download_images(filtered_df, img_dir=img_dir, @@ -183,12 +165,12 @@ def main(): error_log_filepath=error_log_filepath, filename=filename_col, subfolders=subfolders, - downsample_path=downsample_dest_path, + downsample_path=downsample_dir, downsample=args.side_length, file_url=url_col, wait=args.wait_time, retry=args.max_retries) - print(f"Images downloaded from {csv_path} to {img_dir}, with downsampled images in {downsample_dest_path}.") + print(f"Images downloaded from {csv_path} to {img_dir}, with downsampled images in {downsample_dir}.") else: # Download images from urls without downsample copy download_images(filtered_df, diff --git a/src/cautiousrobot/roll_call.py b/src/cautiousrobot/roll_call.py new file mode 100644 index 0000000..52dda65 --- /dev/null +++ b/src/cautiousrobot/roll_call.py @@ -0,0 +1,258 @@ +# 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: + missing (pd.DataFrame) | None: DataFrame of entries with URLs but missing filenames or 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: + (expected_cols, subfolders) 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. + """ + # 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") \ No newline at end of file diff --git a/src/cautiousrobot/utils.py b/src/cautiousrobot/utils.py index ac4f4d9..22900e3 100644 --- a/src/cautiousrobot/utils.py +++ b/src/cautiousrobot/utils.py @@ -1,12 +1,9 @@ # Helper functions for download import json -import sys import pandas as pd import os from PIL import Image -from sumbuddy import gather_file_paths -from sumbuddy.exceptions import EmptyInputDirectoryError def log_response(log_data, index, image, file_path, response_code): @@ -82,75 +79,4 @@ def downsample_and_save_image(image_dir_path, image_name, downsample_dir_path, d file_path=file_path, response_code=str(e) ) - update_log(log=log_errors, index=image_index, filepath=error_log_filepath) - -def check_existing_images(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 + update_log(log=log_errors, index=image_index, filepath=error_log_filepath) \ No newline at end of file diff --git a/tests/test_download_images.py b/tests/test_download_images.py index c0c4af1..f9cbbdb 100644 --- a/tests/test_download_images.py +++ b/tests/test_download_images.py @@ -401,26 +401,33 @@ def test_main_non_unique_filenames(self, mock_process_csv, mock_parse_args): @patch('cautiousrobot.__main__.parse_args') @patch('cautiousrobot.__main__.process_csv') - @patch('builtins.input', return_value='n') - def test_main_missing_filenames(self, mock_input, mock_process_csv, mock_parse_args): + def test_main_missing_filenames_noninteractive(self, mock_process_csv, mock_parse_args): mock_args = MagicMock() mock_args.input_file = 'test.csv' mock_args.img_name_col = 'filename_col' mock_args.url_col = 'url_col' mock_args.subdir_col = None + mock_args.output_dir = 'output_dir' + mock_args.side_length = None + mock_args.wait_time = 0 + mock_args.max_retries = 3 + mock_args.checksum_algorithm = 'md5' + mock_args.verifier_col = None + mock_parse_args.return_value = mock_args mock_data = pd.DataFrame({ 'filename_col': [None, None, 'file2', 'file3'], 'url_col': ['url1', 'url2', 'url3', 'url4'] }) - + mock_process_csv.return_value = mock_data - with self.assertRaises(SystemExit) as cm: + # Should NOT exit now + try: main() - - self.assertEqual(cm.exception.code, "Exited without executing.") + except SystemExit as e: + self.fail(f"main() raised SystemExit unexpectedly: {e}") if __name__ == '__main__': unittest.main() diff --git a/tests/test_existing_images.py b/tests/test_existing_images.py index eb2b405..579e40f 100644 --- a/tests/test_existing_images.py +++ b/tests/test_existing_images.py @@ -1,7 +1,7 @@ import unittest import pandas as pd from unittest.mock import patch -from cautiousrobot.utils import check_existing_images +from cautiousrobot.roll_call import RollCall class TestCheckExistingImages(unittest.TestCase): @@ -12,11 +12,12 @@ def setUp(self): self.sample_df = pd.DataFrame({ self.filename_col: ["a.jpg", "b.jpg", "c.jpg"] }) + self.rollcall = RollCall() - @patch("cautiousrobot.utils.os.path.exists", return_value=False) + @patch("cautiousrobot.roll_call.os.path.exists", return_value=False) def test_directory_does_not_exist(self, mock_exists): """If image directory doesn't exist, all images marked as not in directory.""" - updated_df, filtered_df = check_existing_images( + updated_df, filtered_df = self.rollcall.check_existing_images( self.csv_path, self.img_dir, self.sample_df, self.filename_col ) @@ -24,12 +25,12 @@ def test_directory_does_not_exist(self, mock_exists): self.assertEqual(len(filtered_df), len(self.sample_df)) mock_exists.assert_called_once_with(self.img_dir) - @patch("cautiousrobot.utils.os.path.exists", return_value=True) - @patch("cautiousrobot.utils.gather_file_paths", return_value=["test_images/a.jpg"]) - @patch("cautiousrobot.utils.print") + @patch("cautiousrobot.roll_call.os.path.exists", return_value=True) + @patch("cautiousrobot.roll_call.gather_file_paths", return_value=["test_images/a.jpg"]) + @patch("cautiousrobot.roll_call.print") def test_some_files_exist(self, mock_print, mock_gather, mock_exists): """Should mark existing files correctly and print status.""" - updated_df, filtered_df = check_existing_images( + updated_df, filtered_df = self.rollcall.check_existing_images( self.csv_path, self.img_dir, self.sample_df, self.filename_col ) @@ -40,24 +41,24 @@ def test_some_files_exist(self, mock_print, mock_gather, mock_exists): mock_print.assert_called_once() self.assertIn("There are 1 of the desired files", mock_print.call_args[0][0]) - @patch("cautiousrobot.utils.os.path.exists", return_value=True) - @patch("cautiousrobot.utils.gather_file_paths", return_value=["test_images/a.jpg", "test_images/b.jpg", "test_images/c.jpg"]) + @patch("cautiousrobot.roll_call.os.path.exists", return_value=True) + @patch("cautiousrobot.roll_call.gather_file_paths", return_value=["test_images/a.jpg", "test_images/b.jpg", "test_images/c.jpg"]) def test_all_files_exist_exits(self, mock_gather, mock_exists): """If all images exist, should exit early with proper message.""" with self.assertRaises(SystemExit) as cm: - check_existing_images( + self.rollcall.check_existing_images( self.csv_path, self.img_dir, self.sample_df, self.filename_col ) self.assertIn("already contains all images", cm.exception.code) mock_exists.assert_called_once_with(self.img_dir) - @patch("cautiousrobot.utils.os.path.exists", return_value=True) - @patch("cautiousrobot.utils.gather_file_paths", return_value=[]) - @patch("cautiousrobot.utils.print") + @patch("cautiousrobot.roll_call.os.path.exists", return_value=True) + @patch("cautiousrobot.roll_call.gather_file_paths", return_value=[]) + @patch("cautiousrobot.roll_call.print") def test_no_files_exist(self, mock_print, mock_gather, mock_exists): """If no files exist, should mark all as not in directory and print message.""" - updated_df, filtered_df = check_existing_images( + updated_df, filtered_df = self.rollcall.check_existing_images( self.csv_path, self.img_dir, self.sample_df, self.filename_col ) @@ -66,8 +67,8 @@ def test_no_files_exist(self, mock_print, mock_gather, mock_exists): mock_print.assert_called_once() self.assertIn("There are 0 of the desired files", mock_print.call_args[0][0]) - @patch("cautiousrobot.utils.os.path.exists", return_value=True) - @patch("cautiousrobot.utils.gather_file_paths", return_value=["test_images/species1/a.jpg", "test_images/shouldnotcount/b.jpg"]) + @patch("cautiousrobot.roll_call.os.path.exists", return_value=True) + @patch("cautiousrobot.roll_call.gather_file_paths", return_value=["test_images/species1/a.jpg", "test_images/shouldnotcount/b.jpg"]) def test_subfolders_handling(self, mock_gather, mock_exists): """When `subfolders` is provided, expected paths should be constructed and matched.""" sub_df = pd.DataFrame({ @@ -75,7 +76,7 @@ def test_subfolders_handling(self, mock_gather, mock_exists): self.filename_col: ["a.jpg", "b.jpg"] }) - updated_df, filtered_df = check_existing_images( + updated_df, filtered_df = self.rollcall.check_existing_images( self.csv_path, self.img_dir, sub_df, self.filename_col, subfolders="subfolder" ) diff --git a/tests/test_roll_call.py b/tests/test_roll_call.py new file mode 100644 index 0000000..40d5cc0 --- /dev/null +++ b/tests/test_roll_call.py @@ -0,0 +1,523 @@ +import unittest +from unittest.mock import patch, MagicMock +import pandas as pd + +from cautiousrobot.roll_call import RollCall + + +class TestValidateCsvExtension(unittest.TestCase): + """Test CSV file extension validation.""" + + def setUp(self): + self.rollcall = RollCall() + + def test_valid_csv_extension(self): + """Should not raise exception for valid .csv file.""" + # Should not raise any exception + try: + self.rollcall.validate_csv_extension("data.csv") + except SystemExit: + self.fail("validate_csv_extension raised SystemExit unexpectedly") + + def test_csv_extension_case_sensitive(self): + """Should accept .csv regardless of path content.""" + try: + self.rollcall.validate_csv_extension("MyData.csv") + self.rollcall.validate_csv_extension("path/to/data.csv") + except SystemExit: + self.fail("validate_csv_extension raised SystemExit unexpectedly") + + def test_invalid_txt_extension(self): + """Should exit if file has .txt extension.""" + with self.assertRaises(SystemExit) as cm: + self.rollcall.validate_csv_extension("data.txt") + self.assertIn("csv", cm.exception.code.lower()) + + def test_invalid_xlsx_extension(self): + """Should exit if file has .xlsx extension.""" + with self.assertRaises(SystemExit) as cm: + self.rollcall.validate_csv_extension("data.xlsx") + self.assertIn("csv", cm.exception.code.lower()) + + def test_invalid_no_extension(self): + """Should exit if file has no extension.""" + with self.assertRaises(SystemExit) as cm: + self.rollcall.validate_csv_extension("data") + self.assertIn("csv", cm.exception.code.lower()) + + def test_invalid_csv_as_substring(self): + """Should require .csv as actual extension, not substring.""" + with self.assertRaises(SystemExit) as cm: + self.rollcall.validate_csv_extension("data.csv.backup") + self.assertIn("csv", cm.exception.code.lower()) + + +class TestValidateFilenameUniqueness(unittest.TestCase): + """Test filename uniqueness validation.""" + + def setUp(self): + self.rollcall = RollCall() + + def test_unique_filenames(self): + """Should not raise exception for unique filenames.""" + df = pd.DataFrame({ + "filename": ["image1.jpg", "image2.jpg", "image3.jpg"] + }) + try: + self.rollcall.validate_filename_uniqueness(df, "filename") + except SystemExit: + self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") + + def test_duplicate_filenames(self): + """Should exit if filenames are not unique.""" + df = pd.DataFrame({ + "filename": ["image1.jpg", "image1.jpg", "image2.jpg"] + }) + with self.assertRaises(SystemExit) as cm: + self.rollcall.validate_filename_uniqueness(df, "filename") + self.assertIn("unique identifier", cm.exception.code.lower()) + + def test_filenames_with_missing_values(self): + """Should only count non-null filenames for uniqueness.""" + df = pd.DataFrame({ + "filename": ["image1.jpg", None, "image2.jpg", None] + }) + try: + self.rollcall.validate_filename_uniqueness(df, "filename") + except SystemExit: + self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") + + def test_all_missing_filenames(self): + """Should not raise exception if all filenames are missing.""" + df = pd.DataFrame({ + "filename": [None, None, None] + }) + try: + self.rollcall.validate_filename_uniqueness(df, "filename") + except SystemExit: + self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") + + def test_single_filename(self): + """Should not raise exception for single unique filename.""" + df = pd.DataFrame({ + "filename": ["image1.jpg"] + }) + try: + self.rollcall.validate_filename_uniqueness(df, "filename") + except SystemExit: + self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") + + def test_empty_dataframe(self): + """Should not raise exception for empty dataframe.""" + df = pd.DataFrame({ + "filename": [] + }) + try: + self.rollcall.validate_filename_uniqueness(df, "filename") + except SystemExit: + self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") + + def test_custom_column_name(self): + """Should work with custom column names.""" + df = pd.DataFrame({ + "image_name": ["photo1.jpg", "photo2.jpg", "photo3.jpg"] + }) + try: + self.rollcall.validate_filename_uniqueness(df, "image_name") + except SystemExit: + self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") + + def test_duplicate_with_custom_column(self): + """Should detect duplicates in custom column names.""" + df = pd.DataFrame({ + "image_id": ["img_001", "img_001", "img_002"] + }) + with self.assertRaises(SystemExit) as cm: + self.rollcall.validate_filename_uniqueness(df, "image_id") + self.assertIn("unique identifier", cm.exception.code.lower()) + + +class TestHandleMissingFilenames(unittest.TestCase): + """Test handling of missing filename values under new non-interactive behavior.""" + + def setUp(self): + self.filename_col = "filename" + self.url_col = "file_url" + # RollCall now requires csv_path for saving missing CSVs + self.rollcall = RollCall(csv_path="testdata.csv") + + @patch("builtins.print") + def test_no_missing_filenames(self, mock_print): + """Should not print anything when no filenames are missing.""" + df = pd.DataFrame({ + self.filename_col: ["image1.jpg", "image2.jpg"], + self.url_col: ["http://url1.com", "http://url2.com"] + }) + result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) + self.assertIsNone(result) + mock_print.assert_not_called() + + @patch("builtins.print") + def test_missing_filenames_prints_when_five_or_fewer(self, mock_print): + """Should print missing rows when count <= 5.""" + df = pd.DataFrame({ + self.filename_col: [None, None, "img3.jpg"], + self.url_col: ["url1", "url2", "url3"] + }) + result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) + + self.assertEqual(len(result), 2) + mock_print.assert_called() # printed the missing rows + + @patch("pandas.DataFrame.to_csv") + @patch("builtins.print") + def test_missing_filenames_saved_when_more_than_five(self, mock_print, mock_to_csv): + """Should save missing rows to CSV when count > 5.""" + df = pd.DataFrame({ + self.filename_col: [None] * 6, + self.url_col: ["url"] * 6 + }) + + result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) + + self.assertEqual(len(result), 6) + mock_to_csv.assert_called_once() + args, kwargs = mock_to_csv.call_args + self.assertIn("testdata_missing_filenames.csv", args[0]) + + @patch("builtins.print") + def test_missing_filenames_no_urls(self, mock_print): + """Should not print anything when URLs are missing (no actionable missing filenames).""" + df = pd.DataFrame({ + self.filename_col: ["img1.jpg", None, "img3.jpg"], + self.url_col: [None, None, None] + }) + result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) + + self.assertIsNone(result) + mock_print.assert_not_called() + + @patch("builtins.print") + def test_missing_filenames_empty_dataframe(self, mock_print): + """Should not print anything for empty dataframe.""" + df = pd.DataFrame({ + self.filename_col: [], + self.url_col: [] + }) + result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) + + self.assertIsNone(result) + mock_print.assert_not_called() + + @patch("builtins.print") + def test_missing_filenames_returns_correct_subset(self, mock_print): + """Should return only rows with missing filenames and valid URLs.""" + df = pd.DataFrame({ + self.filename_col: [None, "img2.jpg", None, "img4.jpg"], + self.url_col: ["url1", None, "url3", "url4"] + }) + + result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) + + # Only rows 0 and 2 qualify + self.assertEqual(len(result), 2) + self.assertListEqual(result.index.tolist(), [0, 2]) + + +class TestSetupExpectedColumns(unittest.TestCase): + """Test expected columns setup.""" + + def setUp(self): + self.rollcall = RollCall() + + def test_basic_columns_setup(self): + """Should return expected columns with basic arguments.""" + args = MagicMock() + args.img_name_col = "filename" + args.url_col = "file_url" + args.subdir_col = None + + expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + + self.assertEqual(expected_cols["filename_col"], "filename") + self.assertEqual(expected_cols["url_col"], "file_url") + self.assertNotIn("subfolders", expected_cols) + self.assertIsNone(subfolders) + + def test_columns_lowercase_conversion(self): + """Should convert column names to lowercase.""" + args = MagicMock() + args.img_name_col = "FileName" + args.url_col = "FileURL" + args.subdir_col = None + + expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + + self.assertEqual(expected_cols["filename_col"], "filename") + self.assertEqual(expected_cols["url_col"], "fileurl") + + def test_subdir_column_setup(self): + """Should include subdir column when provided.""" + args = MagicMock() + args.img_name_col = "filename" + args.url_col = "file_url" + args.subdir_col = "species" + + expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + + self.assertEqual(expected_cols["subfolders"], "species") + self.assertEqual(subfolders, "species") + + def test_subdir_column_lowercase(self): + """Should convert subdir column name to lowercase.""" + args = MagicMock() + args.img_name_col = "filename" + args.url_col = "file_url" + args.subdir_col = "Species" + + expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + + self.assertEqual(expected_cols["subfolders"], "species") + self.assertEqual(subfolders, "species") + + def test_custom_column_names_with_subdir(self): + """Should handle custom column names with subdir.""" + args = MagicMock() + args.img_name_col = "ImageID" + args.url_col = "DownloadURL" + args.subdir_col = "Category" + + expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + + self.assertEqual(expected_cols["filename_col"], "imageid") + self.assertEqual(expected_cols["url_col"], "downloadurl") + self.assertEqual(expected_cols["subfolders"], "category") + self.assertEqual(subfolders, "category") + + def test_returns_subfolders_none_when_not_provided(self): + """Should return None for subfolders when not provided in args.""" + args = MagicMock() + args.img_name_col = "filename" + args.url_col = "file_url" + args.subdir_col = None + + expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + + self.assertIsNone(subfolders) + self.assertNotIn("subfolders", expected_cols) + + def test_returns_subfolders_value_when_provided(self): + """Should return subfolders value when provided.""" + args = MagicMock() + args.img_name_col = "filename" + args.url_col = "file_url" + args.subdir_col = "region" + + expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + + self.assertEqual(subfolders, "region") + self.assertIn("subfolders", expected_cols) + + def test_dictionary_structure(self): + """Should return correctly structured dictionary.""" + args = MagicMock() + args.img_name_col = "fname" + args.url_col = "url" + args.subdir_col = "folder" + + expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + + # Check dictionary has correct keys + self.assertIn("filename_col", expected_cols) + self.assertIn("url_col", expected_cols) + self.assertIn("subfolders", expected_cols) + + # Check values are correct + self.assertEqual(expected_cols["filename_col"], "fname") + self.assertEqual(expected_cols["url_col"], "url") + self.assertEqual(expected_cols["subfolders"], "folder") + +class TestCheckDuplicateChecksums(unittest.TestCase): + """Tests for RollCall.check_duplicate_checksums.""" + + def setUp(self): + self.rollcall = RollCall(csv_path="testdata.csv") + + @patch("builtins.print") + def test_no_duplicates(self, mock_print): + """Should not exit when no duplicate checksums exist.""" + df = pd.DataFrame({"hash": ["a", "b", "c"]}) + + try: + self.rollcall.check_duplicate_checksums(df, "hash") + except SystemExit: + self.fail("Unexpected SystemExit for no duplicates") + + mock_print.assert_called_once() # "✔ No duplicate checksums detected." + + @patch("builtins.print") + def test_duplicates_without_ignore(self, mock_print): + """Should exit when duplicates exist and ignore flag is False.""" + df = pd.DataFrame({"hash": ["a", "a", "b"]}) + + with self.assertRaises(SystemExit): + self.rollcall.check_duplicate_checksums(df, "hash", ignore_duplicates=False) + + # Should have printed preview/save message + mock_print.assert_called() + + @patch("builtins.print") + def test_duplicates_with_ignore(self, mock_print): + """Should not exit when duplicates exist and ignore flag is True.""" + df = pd.DataFrame({"hash": ["a", "a", "b"]}) + + try: + self.rollcall.check_duplicate_checksums(df, "hash", ignore_duplicates=True) + except SystemExit: + self.fail("Unexpected SystemExit when ignore_duplicates=True") + + # Should print warning + self.assertTrue(any("ignore-duplicates" in call.args[0] for call in mock_print.call_args_list)) + + @patch("builtins.print") + def test_preview_mode_for_small_duplicate_set(self, mock_print): + """Should print preview when <=5 duplicate rows.""" + df = pd.DataFrame({"hash": ["x", "x", "x"]}) + + with self.assertRaises(SystemExit): + self.rollcall.check_duplicate_checksums(df, "hash") + + # Should print the DataFrame preview + mock_print.assert_called() + + @patch("pandas.DataFrame.to_csv") + @patch("builtins.print") + def test_save_mode_for_large_duplicate_set(self, mock_print, mock_to_csv): + """Should save CSV when >5 duplicate rows.""" + df = pd.DataFrame({"hash": ["x"] * 10}) + + with self.assertRaises(SystemExit): + self.rollcall.check_duplicate_checksums(df, "hash") + + mock_to_csv.assert_called_once() + args, kwargs = mock_to_csv.call_args + self.assertIn("testdata_duplicate_checksums.csv", args[0]) + +class TestPreviewOrSave(unittest.TestCase): + """Tests for RollCall._preview_or_save.""" + + def setUp(self): + self.rollcall = RollCall(csv_path="testdata.csv") + + @patch("builtins.print") + @patch("pandas.DataFrame.to_csv") + def test_preview_small_dataframe(self, mock_to_csv, mock_print): + """Should print when <=5 rows.""" + df = pd.DataFrame({"x": [1, 2, 3]}) + + self.rollcall._preview_or_save("test_label", df) + + mock_to_csv.assert_called_once() + args, kwargs = mock_to_csv.call_args + self.assertIn("testdata_test_label.csv", args[0]) + mock_print.assert_called() + + @patch("builtins.print") + @patch("pandas.DataFrame.to_csv") + def test_save_large_dataframe(self, mock_to_csv, mock_print): + """Should save CSV when >5 rows.""" + df = pd.DataFrame({"x": list(range(10))}) + + self.rollcall._preview_or_save("test_label", df) + + mock_to_csv.assert_called_once() + args, kwargs = mock_to_csv.call_args + self.assertIn("testdata_test_label.csv", args[0]) + mock_print.assert_called() + +class TestPrintDownloadSummary(unittest.TestCase): + """Tests for RollCall.print_download_summary.""" + + def setUp(self): + self.rollcall = RollCall(csv_path="testdata.csv") + + @patch("builtins.print") + def test_summary_with_downsampling_and_subfolders(self, mock_print): + """Should print full summary including downsized path and subfolders.""" + self.rollcall.print_download_summary( + img_dir="/images", + downsample_dir="/images_downsized", + subfolders="species", + num_images=42 + ) + + printed = " ".join(call.args[0] for call in mock_print.call_args_list) + + self.assertIn("Images will be downloaded to: /images", printed) + self.assertIn("Downsampled images will be saved to: /images_downsized", printed) + self.assertIn("Subfolders enabled: species", printed) + self.assertIn("Images to download: 42", printed) + + @patch("builtins.print") + def test_summary_without_downsampling(self, mock_print): + """Should indicate that downsampling is not requested.""" + self.rollcall.print_download_summary( + img_dir="/images", + downsample_dir=None, + subfolders=None, + num_images=10 + ) + + printed = " ".join(call.args[0] for call in mock_print.call_args_list) + + self.assertIn("Images will be downloaded to: /images", printed) + self.assertIn("Downsampled images: not requested", printed) + self.assertIn("Subfolders: none", printed) + self.assertIn("Images to download: 10", printed) + + @patch("builtins.print") + def test_summary_with_subfolders_only(self, mock_print): + """Should print subfolder info even without downsampling.""" + self.rollcall.print_download_summary( + img_dir="/images", + downsample_dir=None, + subfolders="category", + num_images=5 + ) + + printed = " ".join(call.args[0] for call in mock_print.call_args_list) + + self.assertIn("Subfolders enabled: category", printed) + self.assertIn("Images to download: 5", printed) + + @patch("builtins.print") + def test_summary_zero_images(self, mock_print): + """Should correctly print zero image count.""" + self.rollcall.print_download_summary( + img_dir="/images", + downsample_dir=None, + subfolders=None, + num_images=0 + ) + + printed = " ".join(call.args[0] for call in mock_print.call_args_list) + self.assertIn("Images to download: 0", printed) + + @patch("builtins.print") + def test_summary_formatting_header(self, mock_print): + """Should print the summary header and separator.""" + self.rollcall.print_download_summary( + img_dir="/images", + downsample_dir=None, + subfolders=None, + num_images=1 + ) + + printed_lines = [call.args[0] for call in mock_print.call_args_list] + + self.assertIn(" Download Summary", printed_lines[0]) + self.assertIn("--------------------", printed_lines[1]) + + +if __name__ == "__main__": + unittest.main()