From 73f345da861d1b36f2dd996cf49a57d821691048 Mon Sep 17 00:00:00 2001 From: Rahul Ailaboina Date: Fri, 17 Apr 2026 16:32:49 +0000 Subject: [PATCH] Merged in feature/Pre_doczy_report (pull request #968) pre_doczy_ru_base_prefix_changes * pre_doczy_ru_base_prefix_changes * Merged dev into feature/Pre_doczy_report * Merged dev into feature/Pre_doczy_report Approved-by: Katon Minhas --- src/config.py | 14 +++ src/document_classification/main.py | 143 ++++++++++++++++++++++++---- src/utils/io_utils.py | 24 +++++ 3 files changed, 165 insertions(+), 16 deletions(-) diff --git a/src/config.py b/src/config.py index d4b55f6..e71e652 100644 --- a/src/config.py +++ b/src/config.py @@ -428,6 +428,20 @@ PDF_S3_PREFIX = get_arg_value("pdf_s3_prefix", "") # S3 prefix for original PDF MASTER_FILE_LIST = get_arg_value( "master_file_list", "" ) # Path to master Excel file for cross-referencing (e.g. Molina_files_Report.xlsx) +RUN_BASE_PREFIX = get_arg_value( + "run_base_prefix", "" +) # Base S3 prefix for the run folder (e.g., utah-contracts/04162026-run/) +# When set, auto-derives all subfolder paths from this base prefix + +# Auto-derive S3_PREFIX, PDF_S3_PREFIX, and PDF_S3_BUCKET from RUN_BASE_PREFIX +if RUN_BASE_PREFIX: + _run_base = RUN_BASE_PREFIX.rstrip("/") + "/" + if not S3_PREFIX: + S3_PREFIX = f"{_run_base}txt-files/" + if not PDF_S3_PREFIX: + PDF_S3_PREFIX = f"{_run_base}pdf-files/" + if not PDF_S3_BUCKET: + PDF_S3_BUCKET = S3_BUCKET ############## POST-DOCZY VALIDATION REPORT SETTINGS ############## PERFORM_POST_DOCZY = ( diff --git a/src/document_classification/main.py b/src/document_classification/main.py index 3fd5f85..08c2852 100644 --- a/src/document_classification/main.py +++ b/src/document_classification/main.py @@ -594,15 +594,42 @@ def _normalize_filename(filename): return filename +_STRIPPABLE_EXTS = ( + ".txt", + ".pdf", + ".docx", + ".doc", + ".xlsx", + ".xls", + ".xlsb", + ".pptx", + ".htm", + ".html", + ".msg", + ".eml", + ".rtf", + ".xml", + ".thmx", + ".tif", + ".tiff", + ".jpg", + ".jpeg", + ".gif", + ".png", +) + + def _strip_ext(fname): """Strip known file extensions for comparison (preserves date-like suffixes). Strips recursively to handle double extensions like .pdf.xlsx. + Covers source formats that land in all-files/ (email/image/office variants) + so a `.htm` source and its `.txt` conversion normalize to the same key. """ changed = True while changed: changed = False - for ext in [".txt", ".pdf", ".docx", ".doc", ".xlsx", ".xls"]: + for ext in _STRIPPABLE_EXTS: if fname.lower().endswith(ext): fname = fname[: -len(ext)] changed = True @@ -728,6 +755,21 @@ def generate_pre_doczy_report(input_dict, run_timestamp): ) txt_base_names = {_normalize_filename(fname) for fname in input_dict} + # ── Step 2b: Get additional S3 folder listings (when run_base_prefix is set) ── + run_base = ( + config.RUN_BASE_PREFIX.rstrip("/") + "/" if config.RUN_BASE_PREFIX else "" + ) + reconciliation_bucket = config.PDF_S3_BUCKET or config.S3_BUCKET + all_files_list = io_utils.list_s3_folder_files( + reconciliation_bucket, f"{run_base}all-files/" if run_base else "" + ) + invalid_files_list = io_utils.list_s3_folder_files( + reconciliation_bucket, f"{run_base}invalid-files/" if run_base else "" + ) + non_pdf_files_list = io_utils.list_s3_folder_files( + reconciliation_bucket, f"{run_base}non-pdf-files/" if run_base else "" + ) + # Build set of all S3 base names for cross-referencing s3_all_base_names = set() if pdf_filenames: @@ -759,13 +801,19 @@ def generate_pre_doczy_report(input_dict, run_timestamp): else: result_data["STATUS"] = "PROCESSED" + invalid_file_set = set(invalid_files_list) | set(non_pdf_files_list) + # ── Step 4: Add PDF-only files (no TXT available) ── + # A PDF in pdf-files/ with no matching TXT = textract failed. Such files + # also land in invalid-files/ per pipeline behavior, so we do NOT exclude + # them here — they belong in TEXTRACT_FAILED, not INVALID. Step 4b's + # `inv_fname not in results` guard prevents double-counting. if pdf_filenames: no_txt_count = 0 for pdf_name in sorted(pdf_filenames): base_name = _strip_ext(pdf_name) if pdf_name not in txt_base_names and base_name not in txt_base_names: - results[pdf_name] = {**na_placeholder, "STATUS": "INVALID"} + results[pdf_name] = {**na_placeholder, "STATUS": "TEXTRACT_FAILED"} no_txt_count += 1 # Add non-PDF files from the bucket (files with other extensions) @@ -776,10 +824,23 @@ def generate_pre_doczy_report(input_dict, run_timestamp): logging.info( f"PDF reconciliation: {len(pdf_filenames)} total PDFs, " f"{len(txt_base_names)} with TXT, " - f"{no_txt_count} without TXT (NO_TXT_AVAILABLE), " + f"{no_txt_count} without TXT (TEXTRACT_FAILED), " f"{len(non_pdf_filenames)} non-PDF files" ) + # ── Step 4b: Add files from invalid-files/ and non-pdf-files/ buckets ── + invalid_added = 0 + for inv_fname in sorted(invalid_file_set): + if inv_fname not in results: + results[inv_fname] = {**na_placeholder, "STATUS": "INVALID"} + invalid_added += 1 + if invalid_file_set: + logging.info( + f"Invalid/non-PDF reconciliation: {len(invalid_files_list)} invalid-files, " + f"{len(non_pdf_files_list)} non-pdf-files, " + f"{invalid_added} new files added with INVALID status" + ) + # ── Step 5: Build normalized lookup from results for master matching ── results_lookup = {} for fname, result_data in results.items(): @@ -832,8 +893,53 @@ def generate_pre_doczy_report(input_dict, run_timestamp): df = pd.DataFrame(master_rows) else: - # No master file — just use results as-is - df = pd.DataFrame([{"FILE_NAME": k, **v} for k, v in results.items()]) + # No master file — use all-files/ S3 folder as synthetic master if available + if all_files_list: + logging.info( + f"No master Excel provided. Using all-files/ folder as source of truth " + f"({len(all_files_list)} files)" + ) + master_rows = [] + matched_count = 0 + not_found_count = 0 + for s3_fname in all_files_list: + norm_key = _strip_ext(s3_fname) + if norm_key in results_lookup: + master_rows.append( + {"FILE_NAME": s3_fname, **results_lookup[norm_key]} + ) + matched_count += 1 + else: + master_rows.append( + { + "FILE_NAME": s3_fname, + **na_placeholder, + "STATUS": "FILE_NOT_FOUND", + } + ) + not_found_count += 1 + + # Add any results not in the all-files list (safety net) + all_files_normalized = set(_strip_ext(f) for f in all_files_list) + extra_results = [ + fname + for fname in results + if _strip_ext(_normalize_filename(fname)) not in all_files_normalized + ] + for fname in extra_results: + master_rows.append({"FILE_NAME": fname, **results[fname]}) + + logging.info( + f"S3-based reconciliation (all-files): " + f"{len(all_files_list)} source files, " + f"{matched_count} matched, " + f"{not_found_count} not found, " + f"{len(extra_results)} extra results not in all-files" + ) + df = pd.DataFrame(master_rows) + else: + # Ultimate fallback: no master Excel and no run_base_prefix configured + df = pd.DataFrame([{"FILE_NAME": k, **v} for k, v in results.items()]) # ── Mark duplicates from two sources ── df["DUPLICATE_STATUS"] = "Non-Duplicate" @@ -868,17 +974,9 @@ def generate_pre_doczy_report(input_dict, run_timestamp): f"{master_dup_count} master filename duplicates" ) - # Add flag columns based on STATUS - df["INVALID_FLAG"] = (df["STATUS"] == "INVALID").map({True: "YES", False: "NO"}) - df["FILE_NOT_FOUND_FLAG"] = (df["STATUS"] == "FILE_NOT_FOUND").map( - {True: "YES", False: "NO"} - ) - col_order = [ "FILE_NAME", "STATUS", - "INVALID_FLAG", - "FILE_NOT_FOUND_FLAG", "DUPLICATE_STATUS", "IS_CONTRACT", "DOCUMENT_TYPE", @@ -905,6 +1003,9 @@ def generate_pre_doczy_report(input_dict, run_timestamp): (df["STATUS"] == "DUPLICATE").sum() if "STATUS" in df.columns else 0 ) invalid_count = (df["STATUS"] == "INVALID").sum() if "STATUS" in df.columns else 0 + textract_failed_count = ( + (df["STATUS"] == "TEXTRACT_FAILED").sum() if "STATUS" in df.columns else 0 + ) file_not_found_count = ( (df["STATUS"] == "FILE_NOT_FOUND").sum() if "STATUS" in df.columns else 0 ) @@ -919,13 +1020,18 @@ def generate_pre_doczy_report(input_dict, run_timestamp): # Verify counts add up status_sum = ( - processed_count + duplicate_count + invalid_count + file_not_found_count + processed_count + + duplicate_count + + invalid_count + + textract_failed_count + + file_not_found_count ) if status_sum != total_files: logging.warning( f"STATUS count mismatch: {status_sum} (sum) != {total_files} (total). " f"Processed={processed_count}, Duplicate={duplicate_count}, " - f"Invalid={invalid_count}, File not found={file_not_found_count}" + f"Invalid={invalid_count}, Textract failed={textract_failed_count}, " + f"File not found={file_not_found_count}" ) # ── Clean reconciliation table ── @@ -941,21 +1047,26 @@ def generate_pre_doczy_report(input_dict, run_timestamp): logging.info(f" Unique master filenames: {unique_master_filenames}") logging.info(f" Duplicate master filenames: {master_filename_dups}") logging.info("-" * 70) + logging.info(f" Source (all-files): {len(all_files_list)}") logging.info(f" S3 PDF objects: {total_pdfs}") logging.info( f" S3 non-PDF objects: {len(non_pdf_filenames) if pdf_filenames else 0}" ) + logging.info(f" Invalid-files bucket: {len(invalid_files_list)}") + logging.info(f" Non-PDF-files bucket: {len(non_pdf_files_list)}") logging.info(f" TXT files loaded: {total_txts}") logging.info("-" * 70) logging.info(f" Final report rows: {total_files}") logging.info(f" Processed: {processed_count}") logging.info(f" Invalid: {invalid_count}") + logging.info(f" Textract failed: {textract_failed_count}") logging.info(f" Duplicate: {duplicate_count}") logging.info(f" - Content-based duplicates: {content_dup_count}") logging.info(f" - Master filename duplicates: {master_dup_count}") logging.info(f" File not found: {file_not_found_count}") logging.info( - f" Check: {processed_count}+{invalid_count}+{duplicate_count}+{file_not_found_count}={status_sum}" + f" Check: {processed_count}+{invalid_count}+{textract_failed_count}" + f"+{duplicate_count}+{file_not_found_count}={status_sum}" ) logging.info("-" * 70) logging.info( diff --git a/src/utils/io_utils.py b/src/utils/io_utils.py index 4f36a76..383dec9 100644 --- a/src/utils/io_utils.py +++ b/src/utils/io_utils.py @@ -180,6 +180,30 @@ def get_pdf_filenames_from_s3(pdf_s3_bucket, pdf_s3_prefix): return pdf_filenames, non_pdf_filenames +def list_s3_folder_files(bucket, prefix): + """List all filenames (basenames) from an S3 folder. + + Generic helper for listing files in any S3 prefix. Used for reconciliation + against all-files/, invalid-files/, non-pdf-files/ folders. + + Args: + bucket: S3 bucket name. + prefix: S3 prefix (folder path). + + Returns: + list[str]: List of basenames found under the prefix. + Returns [] if bucket or prefix is empty. + """ + if not bucket or not prefix: + return [] + + logging.info(f"Listing files from s3://{bucket}/{prefix}") + all_keys = list_s3_files_paginated(prefix=prefix, bucket=bucket) + filenames = [os.path.basename(key) for key in all_keys if os.path.basename(key)] + logging.info(f"Found {len(filenames)} files in s3://{bucket}/{prefix}") + return filenames + + def list_s3_child_prefixes( prefix=config.S3_PREFIX, bucket=config.S3_BUCKET ) -> list[str]: