diff --git a/src/document_classification/main.py b/src/document_classification/main.py index 1924263..e01721e 100644 --- a/src/document_classification/main.py +++ b/src/document_classification/main.py @@ -21,7 +21,7 @@ from src.pipelines.shared.preprocessing.preprocessing_funcs import ( clean_law_symbols, ) from src.prompts.fieldset import Field -from src.utils import io_utils, llm_utils +from src.utils import io_utils, llm_utils, duplicate_detection from src import config # Load prompts from JSON using Field class (with thread-safe caching) @@ -617,6 +617,15 @@ def generate_pre_doczy_report(input_dict, run_timestamp): f"Max pages to check for classification: {config.DTC_MAX_PAGES_TO_CHECK}" ) + # Detect duplicate contracts + duplicate_groups = duplicate_detection.find_duplicate_groups(input_dict) + original_files_to_process, file_status_map = ( + duplicate_detection.get_files_to_process(input_dict, duplicate_groups) + ) + logging.info( + f"Pre-Doczy duplicate detection complete: originals={len(original_files_to_process)}" + ) + # Initialize progress _pre_doczy_progress["total"] = len(input_dict) _pre_doczy_progress["completed"] = 0 @@ -636,6 +645,7 @@ def generate_pre_doczy_report(input_dict, run_timestamp): _process_file_for_pre_doczy_report, fname, ftext, json_folder ): fname for fname, ftext in input_dict.items() + if fname in original_files_to_process } for future in concurrent.futures.as_completed(future_to_file): @@ -655,8 +665,41 @@ def generate_pre_doczy_report(input_dict, run_timestamp): # Build DataFrame with explicit column ordering df = pd.DataFrame([{"FILE_NAME": k, **v} for k, v in results.items()]) + # Add duplicate detection columns + df["DUPLICATE_STATUS"] = "" + df["DUPLICATE_GROUP_ID"] = "" + df["DUPLICATE_FILE_LIST"] = "" + + # Populate duplicate detection columns based on file_status_map + for filename, status_str in file_status_map.items(): + if filename in df["FILE_NAME"].values: + status_parts = status_str.split("____") + status = status_parts[0] + group_id = status_parts[1] if len(status_parts) > 1 else "" + + mask = df["FILE_NAME"] == filename + df.loc[mask, "DUPLICATE_STATUS"] = status + + if status in ["ORIGINAL_IN_GROUP", "DUPLICATE"] and group_id: + df.loc[mask, "DUPLICATE_GROUP_ID"] = group_id + # Get list of all files in this group + files_in_group = [ + f + for f, s in file_status_map.items() + if s.split("____")[1] == group_id + and s.split("____")[0] != "ORIGINAL_FILE" + ] + df.loc[mask, "DUPLICATE_FILE_LIST"] = ",".join(sorted(files_in_group)) + + # Copy results from original files to their duplicates + if file_status_map: + duplicate_detection.duplicate_copy_results(df, file_status_map, "FILE_NAME") + col_order = [ "FILE_NAME", + "DUPLICATE_STATUS", + "DUPLICATE_GROUP_ID", + "DUPLICATE_FILE_LIST", "IS_CONTRACT", "DOCUMENT_TYPE", "TIN_FOUND", diff --git a/src/pipelines/saas/main.py b/src/pipelines/saas/main.py index 9eb9ffb..0ad20e4 100644 --- a/src/pipelines/saas/main.py +++ b/src/pipelines/saas/main.py @@ -20,6 +20,7 @@ import src.utils.llm_utils as llm_utils import src.utils.logging_utils as logging_utils import src.utils.usage_tracking as usage_tracking import src.utils.timing_utils as timing_utils +import src.utils.duplicate_detection as duplicate_detection from src.constants.constants import Constants from src import config from src.parent_child import main as parent_child_main @@ -135,6 +136,16 @@ def main(testing=False, test_params={}): logging.info(f"Total Input Files : {len(input_dict)}") + # Detect and handle duplicate contracts + with timing_utils.timed_block("duplicate_detection", log_level="INFO"): + duplicate_groups = duplicate_detection.find_duplicate_groups(input_dict) + original_files_to_process, file_status_map = ( + duplicate_detection.get_files_to_process(input_dict, duplicate_groups) + ) + logging.debug( + f"Saas main: duplicate_groups={len(duplicate_groups)}, original_files={len(original_files_to_process)}, file_status_map length={len(file_status_map)}" + ) + # Run document type classification if enabled - DTC mode only, then exit if config.PERFORM_DTC: logging.info("=" * 80) @@ -179,9 +190,11 @@ def main(testing=False, test_params={}): # Each submitted task will set its own logging context in file_processing.process_file() # We use an executor to process files concurrently; use submit() instead of map() # so that we can catch exceptions and continue processing other files + # Only process original files; duplicates will have results copied after futures = [ executor.submit(safe_process_file, item, constants, run_timestamp) for item in input_dict.items() + if item[0] in original_files_to_process ] # collect results as they complete @@ -255,6 +268,30 @@ def main(testing=False, test_params={}): else: ERROR_RESULT_DF = pd.DataFrame() + # Copy results from original files to their duplicates + logging.debug( + f"Before copy: file_status_map length={len(file_status_map)}, file_status_map keys={list(file_status_map.keys())[:5]}..." + ) + logging.debug( + f"Before copy: FINAL_RESULT_DF_CC shape={FINAL_RESULT_DF_CC.shape}, rows={len(FINAL_RESULT_DF_CC)}" + ) + + if len(file_status_map) > 0: + with timing_utils.timed_block("copy_duplicate_results", log_level="INFO"): + logging.info( + f"Calling duplicate_copy_results with {len(file_status_map)} file statuses and {len(FINAL_RESULT_DF_CC)} result rows" + ) + duplicate_detection.duplicate_copy_results( + FINAL_RESULT_DF_CC, file_status_map, "FILE_NAME" + ) + logging.debug( + f"After copy: FINAL_RESULT_DF_CC shape={FINAL_RESULT_DF_CC.shape}, rows={len(FINAL_RESULT_DF_CC)}" + ) + if not FINAL_RESULT_DF_DASHBOARD.empty: + duplicate_detection.duplicate_copy_results( + FINAL_RESULT_DF_DASHBOARD, file_status_map, "FILE_NAME" + ) + # Run QC/QA validation by default (preserves automatic behavior for single files and batches) # QC/QA runs on CC version only (dashboard version and error files do not need QC/QA) validated_cc_df = None diff --git a/src/tests/test_duplicate_detection.py b/src/tests/test_duplicate_detection.py new file mode 100644 index 0000000..ebd5571 --- /dev/null +++ b/src/tests/test_duplicate_detection.py @@ -0,0 +1,465 @@ +"""Tests for duplicate contract detection functionality. + +This test suite covers: +- Text normalization with various Textract variations +- Duplicate detection with mock data +- Representative file selection +- Result copying for duplicates +- Integration scenarios +""" + +import pytest +import pandas as pd +from src.utils.duplicate_detection import ( + normalize_text_for_duplicate_detection, + compute_duplicate_hash, + find_duplicate_groups, + select_original_file, + create_duplicate_status_map, + get_files_to_process, + duplicate_copy_results, +) + + +class TestNormalizeText: + """Test text normalization for Textract variations.""" + + def test_normalize_multiple_spaces(self): + """Should collapse multiple spaces to single space.""" + text = "Provider Name Here" + normalized = normalize_text_for_duplicate_detection(text) + assert normalized == "Provider Name Here" + + def test_normalize_multiple_newlines(self): + """Should collapse multiple newlines to single newline.""" + text = "Line 1\n\n\n\nLine 2" + normalized = normalize_text_for_duplicate_detection(text) + assert normalized == "Line 1\nLine 2" + + def test_normalize_tabs_and_form_feeds(self): + """Should replace tabs and form feeds with spaces.""" + text = "Before\tAfter\r\nNext" + normalized = normalize_text_for_duplicate_detection(text) + assert "\t" not in normalized and "\r" not in normalized + + def test_normalize_leading_trailing_whitespace(self): + """Should strip leading and trailing whitespace.""" + text = " Content \n\n" + normalized = normalize_text_for_duplicate_detection(text) + assert normalized == "Content" + + def test_normalize_empty_string(self): + """Should handle empty strings gracefully.""" + assert normalize_text_for_duplicate_detection("") == "" + + def test_normalize_whitespace_only(self): + """Should return empty string for whitespace-only input.""" + assert normalize_text_for_duplicate_detection(" \n\n \t ") == "" + + def test_normalize_complex_textract_output(self): + """Should normalize realistic Textract variations.""" + text1 = "Provider ABC\n123 Main Street\n\n\nNew York, NY 10001" + text2 = "Provider ABC\n\nNew York, NY 10001\n123 Main Street" + + # Both should normalize to a consistent format + norm1 = normalize_text_for_duplicate_detection(text1) + norm2 = normalize_text_for_duplicate_detection(text2) + + # They won't be the same (different order), but should both be normalized + assert "\n\n" not in norm1 and "\n\n" not in norm2 + assert " " not in norm1 and " " not in norm2 + + +class TestComputeTextHash: + """Test SHA256 hash computation for text.""" + + def test_identical_text_same_hash(self): + """Identical text should produce identical hashes.""" + text = "This is a contract" + hash1 = compute_duplicate_hash(text) + hash2 = compute_duplicate_hash(text) + assert hash1 == hash2 + + def test_different_text_different_hash(self): + """Different text should produce different hashes.""" + hash1 = compute_duplicate_hash("Contract A") + hash2 = compute_duplicate_hash("Contract B") + assert hash1 != hash2 + + def test_normalized_variation_same_hash(self): + """Text with normalization variations should have same hash.""" + text1 = "Provider ABC\n\nNew York" + text2 = "Provider ABC\nNew York" + + hash1 = compute_duplicate_hash(text1) + hash2 = compute_duplicate_hash(text2) + assert hash1 == hash2 + + def test_hash_is_hex_string(self): + """Hash should be a 64-character hex string (SHA256).""" + hash_val = compute_duplicate_hash("test") + assert len(hash_val) == 64 + assert all(c in "0123456789abcdef" for c in hash_val) + + def test_empty_text_hash(self): + """Empty text should still produce a valid hash.""" + hash_val = compute_duplicate_hash("") + assert len(hash_val) == 64 + + +class TestFindDuplicates: + """Test duplicate detection across file groups.""" + + def test_no_duplicates(self): + """Should return empty dict when no duplicates exist.""" + input_dict = { + "file1.txt": "Contract A", + "file2.txt": "Contract B", + "file3.txt": "Contract C", + } + duplicates = find_duplicate_groups(input_dict) + assert duplicates == {} + + def test_simple_duplicates(self): + """Should detect exact duplicates.""" + input_dict = { + "file1.txt": "Contract ABC", + "file2.txt": "Contract ABC", # duplicate + "file3.txt": "Contract XYZ", + } + duplicates = find_duplicate_groups(input_dict) + + # Should have 1 group with 2 files + assert len(duplicates) == 1 + group = list(duplicates.values())[0] + assert len(group) == 2 + assert "file1.txt" in group and "file2.txt" in group + + def test_multiple_duplicate_groups(self): + """Should detect multiple separate duplicate groups.""" + input_dict = { + "file1.txt": "Contract A", + "file2.txt": "Contract A", # duplicate group 1 + "file3.txt": "Contract B", + "file4.txt": "Contract B", # duplicate group 2 + "file5.txt": "Contract C", # unique + } + duplicates = find_duplicate_groups(input_dict) + + assert len(duplicates) == 2 + group_sizes = [len(group) for group in duplicates.values()] + assert sorted(group_sizes) == [2, 2] + + def test_three_way_duplicates(self): + """Should detect groups with more than 2 duplicates.""" + input_dict = { + "file1.txt": "Contract X", + "file2.txt": "Contract X", + "file3.txt": "Contract X", + } + duplicates = find_duplicate_groups(input_dict) + + assert len(duplicates) == 1 + group = list(duplicates.values())[0] + assert len(group) == 3 + + def test_duplicates_with_textract_variations(self): + """Should detect duplicates despite Textract variations.""" + input_dict = { + "file1.txt": "Provider ABC\n\nNew York", + "file2.txt": "Provider ABC\nNew York", # extra space, different newline + "file3.txt": "Different Contract", + } + duplicates = find_duplicate_groups(input_dict) + + assert len(duplicates) == 1 + group = list(duplicates.values())[0] + assert len(group) == 2 + assert "file1.txt" in group and "file2.txt" in group + + def test_empty_input(self): + """Should handle empty input gracefully.""" + duplicates = find_duplicate_groups({}) + assert duplicates == {} + + def test_single_file(self): + """Should not consider single file as duplicate.""" + duplicates = find_duplicate_groups({"file1.txt": "Contents"}) + assert duplicates == {} + + +class TestSelectRepresentativeFile: + """Test representative file selection from duplicate groups.""" + + def test_select_shortest_filename(self): + """Should prefer shorter filenames.""" + group = ["very_long_filename.txt", "short.txt", "medium_file.txt"] + original = select_original_file(group) + assert original == "short.txt" + + def test_select_alphabetically_first(self): + """Should use alphabetical order for ties in length.""" + group = ["file_b.txt", "file_a.txt", "file_c.txt"] + original = select_original_file(group) + assert original == "file_a.txt" + + def test_select_deterministic(self): + """Should always select same file for same input.""" + group = ["z.txt", "a.txt", "m.txt"] + rep1 = select_original_file(group) + rep2 = select_original_file(group) + assert rep1 == rep2 == "a.txt" + + def test_empty_group_raises_error(self): + """Should raise error for empty group.""" + with pytest.raises(ValueError): + select_original_file([]) + + def test_single_file_group(self): + """Should return the only file in group.""" + original = select_original_file(["only_file.txt"]) + assert original == "only_file.txt" + + +class TestCreateDuplicateStatusMap: + """Test creation of file status mappings.""" + + def test_simple_mapping(self): + """Should create correct mapping for simple duplicates.""" + duplicate_groups = { + "hash1": ["file1.txt", "file2.txt"], + } + mapping = create_duplicate_status_map(duplicate_groups) + + assert mapping["file1.txt"]["status"] == "ORIGINAL_IN_GROUP" + assert mapping["file2.txt"]["status"] == "DUPLICATE" + assert mapping["file1.txt"]["group_id"] == "hash1" + assert mapping["file2.txt"]["group_id"] == "hash1" + assert mapping["file1.txt"]["original_file"] == "file1.txt" + assert mapping["file2.txt"]["original_file"] == "file1.txt" + + def test_multiple_groups_mapping(self): + """Should handle multiple duplicate groups separately.""" + duplicate_groups = { + "hash1": ["a.txt", "b.txt"], + "hash2": ["x.txt", "y.txt", "z.txt"], + } + mapping = create_duplicate_status_map(duplicate_groups) + + assert len(mapping) == 5 + assert mapping["a.txt"]["group_id"] == "hash1" + assert mapping["b.txt"]["group_id"] == "hash1" + assert mapping["x.txt"]["group_id"] == "hash2" + assert mapping["y.txt"]["group_id"] == "hash2" + assert mapping["z.txt"]["group_id"] == "hash2" + + assert mapping["a.txt"]["status"] == "ORIGINAL_IN_GROUP" + assert mapping["b.txt"]["status"] == "DUPLICATE" + assert mapping["x.txt"]["status"] == "ORIGINAL_IN_GROUP" + assert mapping["y.txt"]["status"] == "DUPLICATE" + assert mapping["z.txt"]["status"] == "DUPLICATE" + + def test_empty_groups(self): + """Should handle empty duplicate groups.""" + mapping = create_duplicate_status_map({}) + assert mapping == {} + + +class TestGetFilesToProcess: + """Test files selection for processing.""" + + def test_representative_only_selected(self): + """Should select only one file per duplicate group.""" + input_dict = { + "file1.txt": "Contract", + "file2.txt": "Contract", + "file3.txt": "Other", + } + duplicate_groups = find_duplicate_groups(input_dict) + representative_files, file_to_rep = get_files_to_process( + input_dict, duplicate_groups + ) + + # Should have fewer files to process if duplicates exist + if duplicate_groups: + assert len(representative_files) < len(input_dict) + + def test_all_unique_files(self): + """When no duplicates, all files should be selected.""" + input_dict = { + "file1.txt": "Contract A", + "file2.txt": "Contract B", + "file3.txt": "Contract C", + } + duplicate_groups = find_duplicate_groups(input_dict) + representative_files, file_to_rep = get_files_to_process( + input_dict, duplicate_groups + ) + + # All files should be representatives since no duplicates + assert len(representative_files) == len(input_dict) + + +class TestDuplicateCopyResults: + """Test copying results from original to duplicate files.""" + + def test_copy_single_row(self): + """Should copy original results to duplicate filename.""" + result_df = pd.DataFrame( + {"FILE_NAME": ["file1.txt"], "RESULT": ["Success"], "VALUE": [42]} + ) + + file_status_map = { + "file1.txt": "ORIGINAL_IN_GROUP____hash1", + "file2.txt": "DUPLICATE____hash1", + } + + duplicate_copy_results(result_df, file_status_map) + + # Should have 2 rows now + assert len(result_df) == 2 + assert all(result_df["RESULT"] == "Success") + assert all(result_df["VALUE"] == 42) + + # Should have entries for both files + assert set(result_df["FILE_NAME"]) == {"file1.txt", "file2.txt"} + + def test_copy_multiple_duplicates(self): + """Should copy to all duplicates in a group.""" + result_df = pd.DataFrame({"FILE_NAME": ["file1.txt"], "RESULT": ["Success"]}) + + file_status_map = { + "file1.txt": "ORIGINAL_IN_GROUP____hash1", + "file2.txt": "DUPLICATE____hash1", + "file3.txt": "DUPLICATE____hash1", + } + + duplicate_copy_results(result_df, file_status_map) + + assert len(result_df) == 3 + assert set(result_df["FILE_NAME"]) == {"file1.txt", "file2.txt", "file3.txt"} + + def test_copy_multiple_groups(self): + """Should handle multiple original groups.""" + result_df = pd.DataFrame( + {"FILE_NAME": ["rep1.txt", "rep2.txt"], "RESULT": ["Success1", "Success2"]} + ) + + file_status_map = { + "rep1.txt": "ORIGINAL_IN_GROUP____hash1", + "dup1a.txt": "DUPLICATE____hash1", + "dup1b.txt": "DUPLICATE____hash1", + "rep2.txt": "ORIGINAL_IN_GROUP____hash2", + "dup2a.txt": "DUPLICATE____hash2", + } + + duplicate_copy_results(result_df, file_status_map) + + # Should have 5 rows total + assert len(result_df) == 5 + assert set(result_df["FILE_NAME"]) == { + "rep1.txt", + "dup1a.txt", + "dup1b.txt", + "rep2.txt", + "dup2a.txt", + } + + def test_empty_result_df(self): + """Should handle empty result dataframe gracefully.""" + result_df = pd.DataFrame() + file_status_map = {"file1.txt": "ORIGINAL_FILE____"} + + # Should not crash + duplicate_copy_results(result_df, file_status_map) + assert result_df.empty + + def test_no_duplicates_to_copy(self): + """Should not modify dataframe if no duplicates to copy.""" + result_df = pd.DataFrame( + {"FILE_NAME": ["file1.txt", "file2.txt"], "VALUE": [1, 2]} + ) + file_status_map = { + "file1.txt": "ORIGINAL_FILE____", + "file2.txt": "ORIGINAL_FILE____", + } + + original_len = len(result_df) + duplicate_copy_results(result_df, file_status_map) + + assert len(result_df) == original_len + + def test_preserve_column_names(self): + """Should preserve all column names when copying.""" + result_df = pd.DataFrame( + {"FILE_NAME": ["rep.txt"], "COL_A": [1], "COL_B": ["B"], "COL_C": [3.14]} + ) + file_status_map = { + "rep.txt": "ORIGINAL_IN_GROUP____hash1", + "dup.txt": "DUPLICATE____hash1", + } + + duplicate_copy_results(result_df, file_status_map) + + assert set(result_df.columns) == {"FILE_NAME", "COL_A", "COL_B", "COL_C"} + assert ( + result_df.loc[result_df["FILE_NAME"] == "dup.txt", "COL_B"].values[0] == "B" + ) + + +class TestIntegrationScenarios: + """Test realistic integration scenarios.""" + + def test_realistic_contract_duplicates(self): + """Test with realistic contract text variations.""" + # Simulate Textract output from same contract but different versions + contract_base = """PROVIDER SERVICE AGREEMENT +Effective Date: January 1, 2024 +Provider TIN: 12-3456789 +This agreement is between...""" + + contract_variations = { + "contract_101.txt": contract_base, + "contract_102.txt": contract_base.replace("\n", "\n\n"), # extra newlines + "contract_201.txt": contract_base.replace(" ", " "), # more spaces + "contract_different.txt": "COMPLETELY DIFFERENT AGREEMENT", + } + + duplicates = find_duplicate_groups(contract_variations) + + # Should find 1 group with 3 duplicates + assert len(duplicates) == 1 + group = list(duplicates.values())[0] + assert len(group) == 3 + + def test_end_to_end_workflow(self): + """Test complete workflow from detection to result copying.""" + input_files = { + "contract_A1.txt": "CONTRACT TYPE A\nTIN: 111", + "contract_A2.txt": "CONTRACT TYPE A\nTIN: 111", # duplicate + "contract_B1.txt": "CONTRACT TYPE B\nTIN: 222", + } + + # Step 1: Detect duplicates + duplicate_groups = find_duplicate_groups(input_files) + representatives, mapping = get_files_to_process(input_files, duplicate_groups) + + # Step 2: Create initial results (only for representatives) + results = [] + for filename in representatives: + results.append( + {"FILE_NAME": filename, "EXTRACTED": "Field1=Value1, Field2=Value2"} + ) + + result_df = pd.DataFrame(results) + + # Step 3: Copy results to duplicates + duplicate_copy_results(result_df, mapping) + + # Verify + assert len(result_df) == 3 # One for each original file + assert all(result_df["EXTRACTED"] == "Field1=Value1, Field2=Value2") + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/src/utils/duplicate_detection.py b/src/utils/duplicate_detection.py new file mode 100644 index 0000000..e47d8e6 --- /dev/null +++ b/src/utils/duplicate_detection.py @@ -0,0 +1,356 @@ +"""Duplicate contract detection utilities. + +This module provides functions to detect duplicate contracts in a batch of files. +It uses text normalization and hashing to identify identical contracts despite +having different filenames or minor Textract formatting variations. + +Functions: + normalize_text_for_duplicate_detection: Normalize text using existing preprocessing + compute_duplicate_hash: Compute SHA256 hash of normalized text + find_duplicate_groups: Find and group duplicate files + select_original_file: Select one file as original for each duplicate group + create_duplicate_status_map: Create a mapping showing original/duplicate status for all files + duplicate_copy_results: Copy results from original to duplicate files +""" + +import hashlib +import logging +import re +from typing import Dict, List, Tuple, Set + +from src.pipelines.shared.preprocessing import preprocessing_funcs +from src.pipelines.shared.postprocessing import postprocessing_funcs + + +def normalize_text_for_duplicate_detection(text: str) -> str: + """Normalize text using existing preprocessing functions to handle Textract variations. + + Reuses existing preprocessing logic and applies additional normalization: + - Uses preprocessing_funcs.remove_page_indicators() to remove "Page X of Y" + - Uses preprocessing_funcs.clean_law_symbols() to clean special symbols + - Uses preprocessing_funcs.clean_newlines() to handle newline formatting + - Collapses multiple consecutive spaces + - Strips leading/trailing whitespace + + Args: + text: Raw text that may contain Textract variations + + Returns: + Normalized text string suitable for hashing and comparison + """ + if not text: + return "" + + # Use existing preprocessing functions + text = preprocessing_funcs.remove_page_indicators(text) + text = preprocessing_funcs.clean_law_symbols(text) + text = preprocessing_funcs.clean_newlines(text) + + # Replace tabs and carriage returns with spaces + text = text.replace("\t", " ").replace("\r", " ") + + # Additional normalization: collapse multiple spaces + text = re.sub(r" +", " ", text) + + # Collapse multiple newlines to single newline (after clean_newlines converted some to spaces) + text = re.sub(r"\n\n+", "\n", text) + + # Strip leading and trailing whitespace + text = text.strip() + + return text + + +def compute_duplicate_hash(text: str) -> str: + """Compute SHA256 hash of normalized text for efficient grouping. + + Uses normalize_text_for_duplicate_detection to normalize before hashing. + + Args: + text: Text to hash + + Returns: + Lowercase SHA256 hex digest of the normalized text + """ + normalized = normalize_text_for_duplicate_detection(text) + return hashlib.sha256(normalized.encode("utf-8")).hexdigest() + + +def find_duplicate_groups(input_dict: Dict[str, str]) -> Dict[str, List[str]]: + """Find groups of duplicate files based on normalized content. + + Groups files by their normalized content hash. Only returns groups with + more than one file (actual duplicates). + + Args: + input_dict: Dictionary mapping filename -> file_text + + Returns: + Dictionary mapping group_id (content hash) -> list of filenames in that group. + Only groups with >1 file are included in the result. + + Example: + >>> input_dict = { + ... "file1.txt": "Contract ABC", + ... "file2.txt": "Contract ABC", # duplicate + ... "file3.txt": "Contract XYZ" + ... } + >>> duplicates = find_duplicate_groups(input_dict) + >>> # Returns: {: ["file1.txt", "file2.txt"]} + """ + # Group files by content hash + hash_to_files: Dict[str, List[str]] = {} + + for filename, file_text in input_dict.items(): + content_hash = compute_duplicate_hash(file_text) + if content_hash not in hash_to_files: + hash_to_files[content_hash] = [] + hash_to_files[content_hash].append(filename) + + # Filter to only keep groups with duplicates (>1 file) + duplicate_groups = { + group_id: files for group_id, files in hash_to_files.items() if len(files) > 1 + } + + return duplicate_groups + + +def select_original_file(duplicate_group: List[str]) -> str: + """Select one original file from a duplicate group to process. + + Uses deterministic selection: shortest filename first, then alphabetically. + This ensures consistent selection across runs. The selected file will be + processed through LLM; other files in the group will reuse its results. + + Args: + duplicate_group: List of duplicate filenames + + Returns: + The original (representative) filename to process + """ + if not duplicate_group: + raise ValueError("Cannot select original file from empty group") + + # Sort by length (shorter first) then alphabetically for deterministic selection + sorted_files = sorted(duplicate_group, key=lambda f: (len(f), f)) + return sorted_files[0] + + +def create_duplicate_status_map( + duplicate_groups: Dict[str, List[str]], +) -> Dict[str, Dict[str, str]]: + """Create a mapping showing duplicate status for each file. + + For each file, shows: + - "ORIGINAL_FILE": Not part of any duplicate group + - "ORIGINAL_IN_GROUP": The one file processed in a duplicate group + - "DUPLICATE": A copy/duplicate of the original + + Args: + duplicate_groups: Output from find_duplicate_groups() + + Returns: + Dictionary mapping filename -> {status, group_id, original_file} + + Example: + >>> groups = {hash1: ["file1.txt", "file2.txt"]} + >>> status = create_duplicate_status_map(groups) + >>> # Returns: { + >>> # "file1.txt": {status: "ORIGINAL_IN_GROUP", group_id: hash1, original_file: "file1.txt"}, + >>> # "file2.txt": {status: "DUPLICATE", group_id: hash1, original_file: "file1.txt"} + >>> # } + """ + file_status_map = {} + + for group_id, files in duplicate_groups.items(): + original = select_original_file(files) + for filename in files: + if filename == original: + file_status_map[filename] = { + "status": "ORIGINAL_IN_GROUP", + "group_id": group_id, + "original_file": original, + } + else: + file_status_map[filename] = { + "status": "DUPLICATE", + "group_id": group_id, + "original_file": original, + } + + return file_status_map + + +def get_files_to_process( + input_dict: Dict[str, str], duplicate_groups: Dict[str, List[str]] +) -> Tuple[Set[str], Dict[str, str]]: + """Determine which files to process as originals and create status map. + + Returns only the original files to process (one per duplicate group), + plus a mapping showing the status of ALL files. + + Args: + input_dict: Original input dictionary + duplicate_groups: Output from find_duplicate_groups() + + Returns: + Tuple of: + - Set of original filenames to process through LLM + - Mapping of all files to their status (ORIGINAL_IN_GROUP, DUPLICATE, or ORIGINAL_FILE) + """ + file_status_map = create_duplicate_status_map(duplicate_groups) + + # Get all files to process: originals in groups + unique single files + # Skip only files marked as DUPLICATE (they reuse results) + original_files = set() + all_files_status = {} + + for filename in input_dict.keys(): + if filename in file_status_map: + status_info = file_status_map[filename] + status = status_info["status"] + group_id = status_info["group_id"] + all_files_status[filename] = f"{status}____{group_id}" + + # Add to processing list if it's an original (not a duplicate) + if status == "ORIGINAL_IN_GROUP": + original_files.add(filename) + else: + # Single file not part of any duplicate group + all_files_status[filename] = "ORIGINAL_FILE____" + # Add single files to processing list + original_files.add(filename) + + logging.info(f"Duplicate Detection Summary:") + logging.info(f" Total files: {len(input_dict)}") + logging.info(f" Duplicate groups found: {len(duplicate_groups)}") + logging.info(f" Original files to process: {len(original_files)}") + logging.info( + f" Duplicate files (reusing results): {len(file_status_map) - len(original_files)}" + ) + + # Log details for each duplicate group + for group_id, files in duplicate_groups.items(): + original = file_status_map[files[0]]["original_file"] + logging.info(f" Duplicate Group: {original} (ORIGINAL_IN_GROUP)") + for filename in files: + if filename != original: + logging.info(f" └─ {filename} (DUPLICATE)") + + return original_files, all_files_status + + +def duplicate_copy_results( + result_df, file_status_map: Dict[str, str], file_name_column: str = "FILE_NAME" +) -> None: + """Copy results from original files to their duplicates in-place. + + This function modifies the input DataFrame by duplicating rows for each + file in a duplicate group, using the original file's results. + + For each duplicate file in a group, creates a new row with the same + extraction results but with the duplicate file's name. + + Args: + result_df: DataFrame with results (modified in-place) + file_status_map: Mapping from get_files_to_process() showing file status + file_name_column: Name of the column containing filenames + + Returns: + None (modifies result_df in-place) + + Example: + >>> # result_df has row for "file1.txt" only + >>> # file_status_map shows file2.txt is duplicate of file1.txt + >>> duplicate_copy_results(result_df, file_status_map) + >>> # Now result_df also has a row for "file2.txt" with same results + """ + if result_df.empty or file_name_column not in result_df.columns: + logging.warning( + "Result DataFrame is empty or FILE_NAME column not found, skipping duplicate copy" + ) + return + + # Normalize FILE_NAME in the result DataFrame (postprocessing may strip .txt). + standardize = postprocessing_funcs.standardize_file_name + originals_in_results = set( + standardize(fn) for fn in result_df[file_name_column].unique() + ) + logging.debug( + f"duplicate_copy_results: originals_in_results={originals_in_results}, total rows in result_df={len(result_df)}" + ) + + # Preserve raw file names when copying outputs, while using standardized names for matching + normalized_file_status_map = { + standardize(filename): (filename, status_str) + for filename, status_str in file_status_map.items() + } + + # Build a mapping of group_id to standardized original file name (1:1) + group_to_original = {} + for filename_std, (_, status_str) in normalized_file_status_map.items(): + status_parts = status_str.split("____") + status = status_parts[0] + group_id = status_parts[1] if len(status_parts) > 1 else "" + + if status == "ORIGINAL_IN_GROUP" and group_id: + group_to_original[group_id] = filename_std + logging.debug( + f"Found original (standardized) for group {group_id}: {filename_std}" + ) + + # For each file in status map that's a duplicate, find its original and copy + rows_to_add = [] + for filename_std, (filename_raw, status_str) in normalized_file_status_map.items(): + status_parts = status_str.split("____") + status = status_parts[0] + group_id = status_parts[1] if len(status_parts) > 1 else "" + + if status == "DUPLICATE" and group_id: + original_std = group_to_original.get(group_id) + logging.debug( + f"Processing duplicate (standardized) {filename_std}, group={group_id}, original={original_std}" + ) + + if original_std and original_std in originals_in_results: + orig_rows = result_df[ + result_df[file_name_column].apply(standardize) == original_std + ] + logging.debug( + f" Found {len(orig_rows)} result rows for original {original_std}" + ) + + if not orig_rows.empty: + for _, row in orig_rows.iterrows(): + new_row = row.copy() + new_row[file_name_column] = filename_raw + rows_to_add.append(new_row) + logging.debug( + f" Added {len(orig_rows)} rows for duplicate {filename_raw}" + ) + else: + logging.debug( + f" Original {original_std} not found in results or not in originals_in_results" + ) + + # Add all new rows at once + if rows_to_add: + import pandas as pd + + logging.debug(f"Adding {len(rows_to_add)} duplicate rows to result_df") + new_rows_df = pd.DataFrame(rows_to_add) + + # Combine original and new rows + combined = pd.concat([result_df, new_rows_df], ignore_index=True) + + # Replace existing dataframe content in-place + result_df.drop(result_df.index, inplace=True) + for i, row in combined.iterrows(): + result_df.loc[i] = row + + result_df.reset_index(drop=True, inplace=True) + + logging.info(f"Copied results for {len(rows_to_add)} duplicate files") + logging.debug(f"Final result_df has {len(result_df)} rows after copying") + else: + logging.debug(f"No rows to add - no duplicates found with matching originals")