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
This commit is contained in:
Rahul Ailaboina
committed by
Katon Minhas
Katon Minhas
parent
38ae8c64ba
commit
73f345da86
@@ -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 = (
|
||||
|
||||
@@ -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,7 +893,52 @@ def generate_pre_doczy_report(input_dict, run_timestamp):
|
||||
|
||||
df = pd.DataFrame(master_rows)
|
||||
else:
|
||||
# No master file — just use results as-is
|
||||
# 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 ──
|
||||
@@ -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(
|
||||
|
||||
@@ -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]:
|
||||
|
||||
Reference in New Issue
Block a user