From c9cd7ddb43a65794dae14edf85d474471fa90634 Mon Sep 17 00:00:00 2001 From: Rahul Ailaboina Date: Fri, 13 Mar 2026 21:06:44 +0000 Subject: [PATCH] Merged in feature/doczy_reports (pull request #911) Feature/doczy reports * Added post_doczy_reports * black format fix * Merged DEV into feature/doczy_reports * fixed code review Approved-by: Katon Minhas --- src/config.py | 19 + src/constants/regex_patterns.py | 17 + src/document_classification/main.py | 934 ++++++++++++++++++++++++++-- src/utils/io_utils.py | 106 +++- 4 files changed, 1006 insertions(+), 70 deletions(-) diff --git a/src/config.py b/src/config.py index 235b268..7dcf202 100644 --- a/src/config.py +++ b/src/config.py @@ -404,6 +404,25 @@ DTC_OUTPUT_FILE = ( ) DTC_JSON_OUTPUT_FOLDER = "dtc_json_results" # Folder for individual JSON results +############## PRE-DOCZY REPORT SETTINGS ############## +PERFORM_PRE_DOCZY = ( + get_arg_value("perform_pre_doczy", "False") == "True" +) # Enable Pre-Doczy classification + keyword report + +############## POST-DOCZY VALIDATION REPORT SETTINGS ############## +PERFORM_POST_DOCZY = ( + get_arg_value("perform_post_doczy", "False") == "True" +) # Enable post-Doczy validation report +PERFORM_POST_DOCZY_EXCEL = ( + get_arg_value("perform_post_doczy_excel", "False") == "True" +) # Enable Excel-only post-Doczy report (no raw text files needed) +POST_DOCZY_JSON_OUTPUT_FOLDER = ( + "post_doczy_json_results" # Folder for individual JSON results +) +DOCZY_RESULTS_PATH = get_arg_value( + "doczy_results_path", "" +) # Path to Doczy extraction output CSV/Excel (local path or S3 URI) + ############## DASHBOARD POSTPROCESSING ############## # Run dashboard postprocessing (comma-separated list format). Default: CC only. # Set to True via run_dashboard arg when dashboard output is needed. diff --git a/src/constants/regex_patterns.py b/src/constants/regex_patterns.py index e82d3e5..18b6285 100644 --- a/src/constants/regex_patterns.py +++ b/src/constants/regex_patterns.py @@ -109,6 +109,23 @@ EFFECTIVE_DATE_FORMAT_PATTERNS = [ r"\b(?:19|20)\d{2}[/\-](?:0?[1-9]|1[0-2])[/\-](?:0?[1-9]|[12]\d|3[01])\b", ] +# ── Signature Patterns (for Post-Doczy Report) ── +# Keyword patterns that indicate signature presence in contract documents. +SIGNATURE_KEYWORD_PATTERNS = [ + r"\bsignature\b", + r"\bsigned\s+by\b", + r"\bauthorized\s+signature\b", + r"\bexecuted\s+by\b", + r"\bsignatory\b", + r"\bIN\s+WITNESS\s+WHEREOF\b", + r"\bsigning\s+authority\b", + r"\bduly\s+authorized\b", + r"\bby\s*:\s*_+", + r"\bsign\s+here\b", + r"\bsigned\s+and\s+sealed\b", + r"\bexecuted\s+this\b", +] + # dba patterns DBA_PATTERNS = [r"\bD/B/A\b", r"\bDBA\b", r"\bDOING BUSINESS AS\b", r"\bD B A\b"] diff --git a/src/document_classification/main.py b/src/document_classification/main.py index 33b3456..53fb5ec 100644 --- a/src/document_classification/main.py +++ b/src/document_classification/main.py @@ -10,8 +10,10 @@ import pandas as pd from src.constants.regex_patterns import ( TIN_PATTERN, + FILE_NAME_TIN_PATTERN, EFFECTIVE_DATE_KEYWORD_PATTERNS, EFFECTIVE_DATE_FORMAT_PATTERNS, + SIGNATURE_KEYWORD_PATTERNS, ) from src.pipelines.shared.preprocessing.preprocessing_funcs import ( clean_newlines, @@ -47,6 +49,11 @@ _EFFECTIVE_DATE_COMPILED = [ for p in EFFECTIVE_DATE_KEYWORD_PATTERNS + EFFECTIVE_DATE_FORMAT_PATTERNS ] +# ── Pre-compiled patterns for Post-Doczy Report ────────────────────────────── +_SIGNATURE_COMPILED = [re.compile(p, re.IGNORECASE) for p in SIGNATURE_KEYWORD_PATTERNS] +_FILE_NAME_TIN_COMPILED = re.compile(FILE_NAME_TIN_PATTERN) +_DATE_FORMAT_COMPILED = [re.compile(p) for p in EFFECTIVE_DATE_FORMAT_PATTERNS] + # ── LOB keywords loaded from JSON mappings (single source of truth) ───────── _MAPPINGS_DIR = os.path.join(os.path.dirname(__file__), "..", "constants", "mappings") @@ -117,9 +124,37 @@ _LOB_COMPILED = [ for kw, normalized_lob in _LOB_KEYWORD_TO_NORMALIZED.items() ] -# Separate progress tracker for the DTC report (avoids conflicts with main()) -_report_progress_lock = Lock() -_report_progress = {"completed": 0, "total": 0} +# ── Error result templates (used in per-file handler + as_completed fallback) ─ +_PRE_DOCZY_ERROR_RESULT = { + "IS_CONTRACT": "ERROR", + "DOCUMENT_TYPE": "ERROR", + "TIN_FOUND": "ERROR", + "TIN_PAGES": "ERROR", + "EFFECTIVE_DATE_FOUND": "ERROR", + "EFFECTIVE_DATE_PAGES": "ERROR", + "LOB_FOUND": "ERROR", + "LOB_PAGES": "ERROR", + "LOB_KEYWORDS_MATCHED": "ERROR", + "LOB_DERIVED": "ERROR", +} + +_POST_DOCZY_ERROR_RESULT = { + "TIN_STATUS": "ERROR", + "TIN_DOCZY_VALUE": "ERROR", + "LOB_STATUS": "ERROR", + "LOB_DOCZY_VALUE": "ERROR", + "LOB_DETECTED_VALUE": "ERROR", + "EFF_DATE_STATUS": "ERROR", + "EFF_DATE_DOCZY_VALUE": "ERROR", + "EFF_DATE_DETECTED_VALUE": "ERROR", + "SIGNATURE_STATUS": "ERROR", + "SIGNATURE_DOCZY_VALUE": "ERROR", + "PRICING_READY": "ERROR", +} + +# Separate progress tracker for the Pre-Doczy report (avoids conflicts with main()) +_pre_doczy_progress_lock = Lock() +_pre_doczy_progress = {"completed": 0, "total": 0} def call_llm(filename, context, question): @@ -225,7 +260,7 @@ def process_single_file(filename, contract_text, json_folder): progress_counter["completed"] += 1 completed = progress_counter["completed"] total = progress_counter["total"] - if ( + if total > 0 and ( completed % 5 == 0 or completed == total ): # Log every 5 files or at completion logging.info( @@ -317,7 +352,8 @@ def main(input_dict, run_timestamp): duration = (end_time - start_time).total_seconds() logging.info(f"DTC classification complete in {duration:.2f} seconds") - logging.info(f"Average time per file: {duration/len(input_dict):.2f} seconds") + if len(input_dict) > 0: + logging.info(f"Average time per file: {duration/len(input_dict):.2f} seconds") # Create DataFrame and save results df = pd.DataFrame([{"FILE_NAME": k, **v} for k, v in answer_dict.items()]) @@ -397,7 +433,7 @@ def main(input_dict, run_timestamp): return answer_dict -# ── DTC Report: Keyword Search Helpers ────────────────────────────────────── +# ── Pre-Doczy Report: Keyword Search Helpers ──────────────────────────────── def _search_tin_in_page(page_text): @@ -440,11 +476,11 @@ def _search_lob_in_page(page_text): ] -# ── DTC Report: Per-file Processing ──────────────────────────────────────── +# ── Pre-Doczy Report: Per-file Processing ─────────────────────────────────── -def _process_file_for_report(filename, file_text, json_folder): - """Process a single file for the DTC report: classify + keyword search. +def _process_file_for_pre_doczy_report(filename, file_text, json_folder): + """Process a single file for the Pre-Doczy report: classify + keyword search. Combines LLM-based document type classification with regex-based keyword detection for TIN, effective date, and line of business across all pages. @@ -529,41 +565,32 @@ def _process_file_for_report(filename, file_text, json_folder): io_utils.save_result_to_json(filename, result, json_folder) # Update progress - with _report_progress_lock: - _report_progress["completed"] += 1 - done = _report_progress["completed"] - total = _report_progress["total"] - if done % 5 == 0 or done == total: + with _pre_doczy_progress_lock: + _pre_doczy_progress["completed"] += 1 + done = _pre_doczy_progress["completed"] + total = _pre_doczy_progress["total"] + if total > 0 and (done % 5 == 0 or done == total): logging.info( - f"DTC Report Progress: {done}/{total} files processed " + f"Pre-Doczy Report Progress: {done}/{total} files processed " f"({done / total * 100:.1f}%)" ) return filename, result except Exception as e: - logging.error(f"Error processing {filename} for DTC report: {e}") - error_result = { - "IS_CONTRACT": "ERROR", - "DOCUMENT_TYPE": "ERROR", - "TIN_FOUND": "ERROR", - "TIN_PAGES": "ERROR", - "EFFECTIVE_DATE_FOUND": "ERROR", - "EFFECTIVE_DATE_PAGES": "ERROR", - "LOB_FOUND": "ERROR", - "LOB_PAGES": "ERROR", - "LOB_KEYWORDS_MATCHED": "ERROR", - "LOB_DERIVED": "ERROR", - } + logging.error(f"Error processing {filename} for Pre-Doczy report: {e}") + error_result = dict(_PRE_DOCZY_ERROR_RESULT) io_utils.save_result_to_json(filename, error_result, json_folder) return filename, error_result -# ── DTC Report: Main Entry Point ─────────────────────────────────────────── +# ── Pre-Doczy Report: Main Entry Point ────────────────────────────────────── -def generate_dtc_report(input_dict, run_timestamp): - """Generate a DTC report combining document classification with keyword detection. +def generate_pre_doczy_report(input_dict, run_timestamp): + """Generate a Pre-Doczy report combining document classification with keyword detection. + + Filters out non-contract documents before Doczy processing. For each file, produces a row with: - FILE_NAME: Original filename @@ -579,25 +606,25 @@ def generate_dtc_report(input_dict, run_timestamp): (format: run_YYYYMMDD_HH-MM_BATCHID). Returns: - pd.DataFrame: The complete DTC report DataFrame. + pd.DataFrame: The complete Pre-Doczy report DataFrame. Side effects: - Saves CSV results to S3 (if config.WRITE_TO_S3) or local filesystem. - Saves individual JSON files to config.DTC_JSON_OUTPUT_FOLDER. """ - logging.info(f"Starting DTC Report generation for {len(input_dict)} files...") + logging.info(f"Starting Pre-Doczy Report generation for {len(input_dict)} files...") logging.info( f"Max pages to check for classification: {config.DTC_MAX_PAGES_TO_CHECK}" ) # Initialize progress - _report_progress["total"] = len(input_dict) - _report_progress["completed"] = 0 + _pre_doczy_progress["total"] = len(input_dict) + _pre_doczy_progress["completed"] = 0 # Create output folder for individual JSONs - json_folder = config.DTC_JSON_OUTPUT_FOLDER + "_report" + json_folder = config.DTC_JSON_OUTPUT_FOLDER + "_pre_doczy_report" os.makedirs(json_folder, exist_ok=True) - logging.info(f"DTC Report JSON output folder: {json_folder}") + logging.info(f"Pre-Doczy Report JSON output folder: {json_folder}") start_time = datetime.now() results = {} @@ -605,7 +632,9 @@ def generate_dtc_report(input_dict, run_timestamp): # Process files concurrently with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: future_to_file = { - executor.submit(_process_file_for_report, fname, ftext, json_folder): fname + executor.submit( + _process_file_for_pre_doczy_report, fname, ftext, json_folder + ): fname for fname, ftext in input_dict.items() } @@ -615,22 +644,11 @@ def generate_dtc_report(input_dict, run_timestamp): result_fname, result_data = future.result() results[result_fname] = result_data except Exception as e: - logging.error(f"Exception for {fname} in DTC Report: {e}") - results[fname] = { - "IS_CONTRACT": "ERROR", - "DOCUMENT_TYPE": "ERROR", - "TIN_FOUND": "ERROR", - "TIN_PAGES": "ERROR", - "EFFECTIVE_DATE_FOUND": "ERROR", - "EFFECTIVE_DATE_PAGES": "ERROR", - "LOB_FOUND": "ERROR", - "LOB_PAGES": "ERROR", - "LOB_KEYWORDS_MATCHED": "ERROR", - "LOB_DERIVED": "ERROR", - } + logging.error(f"Exception for {fname} in Pre-Doczy Report: {e}") + results[fname] = dict(_PRE_DOCZY_ERROR_RESULT) duration = (datetime.now() - start_time).total_seconds() - logging.info(f"DTC Report generation complete in {duration:.2f} seconds") + logging.info(f"Pre-Doczy Report generation complete in {duration:.2f} seconds") if len(input_dict) > 0: logging.info(f"Average time per file: {duration / len(input_dict):.2f} seconds") @@ -666,7 +684,7 @@ def generate_dtc_report(input_dict, run_timestamp): ) logging.info( - f"DTC Report Summary: {total_files} files | " + f"Pre-Doczy Report Summary: {total_files} files | " f"TIN detected: {tin_count} | " f"Effective Date detected: {eff_date_count} | " f"LOB detected: {lob_count} | " @@ -675,21 +693,762 @@ def generate_dtc_report(input_dict, run_timestamp): # Save output using existing io_utils pattern if config.WRITE_TO_S3: - io_utils.write_s3(df, "", run_timestamp, "dtc") + io_utils.write_s3(df, "", run_timestamp, "pre_doczy") logging.info( - f"DTC Report uploaded to S3: " - f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-DTC.csv" + f"Pre-Doczy Report uploaded to S3: " + f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-PRE-DOCZY.csv" ) else: - io_utils.write_local(df, "", run_timestamp, "dtc") + io_utils.write_local(df, "", run_timestamp, "pre_doczy") logging.info( - f"DTC Report saved locally: " - f"{config.CONSOLIDATED_OUTPUT_DIRECTORY}/{run_timestamp}/{config.BATCH_ID}-DTC.csv" + f"Pre-Doczy Report saved locally: " + f"{config.CONSOLIDATED_OUTPUT_DIRECTORY}/{run_timestamp}/{config.BATCH_ID}-PRE-DOCZY.csv" ) return df +# ── Post-Doczy Report: Helpers ─────────────────────────────────────────────── + + +def _search_signature_in_page(page_text): + """Check if any signature keyword is present in a single page of text.""" + return any(p.search(page_text) for p in _SIGNATURE_COMPILED) + + +def _search_lob_in_filename(filename): + """Find LOB keywords present in the filename.""" + return [ + (kw, normalized_lob) + for kw, normalized_lob, pattern in _LOB_COMPILED + if pattern.search(filename) + ] + + +def _search_date_in_filename(filename): + """Extract first date pattern found in the filename, or None.""" + for p in _DATE_FORMAT_COMPILED: + match = p.search(filename) + if match: + return match.group(0) + return None + + +def _search_tin_in_filename(filename): + """Check if a TIN pattern is present in the filename.""" + return bool(_FILE_NAME_TIN_COMPILED.search(filename)) + + +def _extract_first_date_from_page(page_text): + """Extract the first date found in a page of text, or None.""" + for p in _DATE_FORMAT_COMPILED: + match = p.search(page_text) + if match: + return match.group(0) + return None + + +def _is_doczy_field_present(value): + """Check if a Doczy extracted field has a meaningful value. + + Handles str and list[str] column types from Doczy output. + Returns False for empty, N/A, NaN, None, or empty-list values. + """ + if value is None or (isinstance(value, float) and pd.isna(value)): + return False + if isinstance(value, str): + stripped = value.strip() + return stripped != "" and stripped.upper() not in ("N/A", "NONE", "[]", "NA") + if isinstance(value, list): + return len(value) > 0 and any( + str(v).strip() not in ("", "N/A", "None") for v in value + ) + return bool(value) + + +def _determine_status(doczy_extracted, in_contract, in_filename): + """Determine the status for a field based on three sources. + + Returns one of: + - "Extracted" → Doczy got it + - "In Contract, Not Extracted" → regex found it in text, Doczy missed + - "In Filename, Not Extracted" → regex found it in filename, Doczy missed + - "Not Present" → not found anywhere + """ + if doczy_extracted: + return "Extracted" + if in_contract: + return "In Contract, Not Extracted" + if in_filename: + return "In Filename, Not Extracted" + return "Not Present" + + +# Separate progress tracker for Post-Doczy report +_post_doczy_progress_lock = Lock() +_post_doczy_progress = {"completed": 0, "total": 0} + + +# ── Post-Doczy Report: Per-file Processing ────────────────────────────────── + + +def _process_file_for_post_doczy_report(filename, file_text, doczy_row, json_folder): + """Process a single file for the Post-Doczy validation report. + + Compares Doczy extraction results against what's actually in the document + (raw text + filename) to determine the status of each required field. + + Args: + filename: Name of the file being processed. + file_text: Raw text content (Textract output with page markers). + doczy_row: dict of Doczy extraction results for this file, or None + if file not found in Doczy output. + json_folder: Path to folder for saving individual JSON results. + + Returns: + tuple: (filename, result_dict) where result_dict contains all report columns. + """ + try: + cleaned_text = clean_law_symbols(clean_newlines(file_text)) + text_dict = split_text(cleaned_text) + + # ── 1. What Doczy extracted ── + if doczy_row is not None: + doczy_tin = doczy_row.get("PROV_GROUP_TIN", None) + doczy_filename_tin = doczy_row.get("FILENAME_TIN", None) + doczy_lob = doczy_row.get("AARETE_DERIVED_LOB", None) or doczy_row.get( + "LOB", None + ) + doczy_eff_dt = doczy_row.get( + "AARETE_DERIVED_EFFECTIVE_DT", None + ) or doczy_row.get("EFFECTIVE_DT", None) + doczy_signature = doczy_row.get( + "AARETE_DERIVED_SIGNATORY_COMPLETE_IND", None + ) + doczy_signed_count = doczy_row.get("NUM_SIGNED_SIGNATORY_LINE_COUNT", None) + + has_doczy_tin = _is_doczy_field_present( + doczy_tin + ) or _is_doczy_field_present(doczy_filename_tin) + has_doczy_lob = _is_doczy_field_present(doczy_lob) + has_doczy_eff_dt = _is_doczy_field_present(doczy_eff_dt) + has_doczy_signature = _is_doczy_field_present(doczy_signature) or ( + _is_doczy_field_present(doczy_signed_count) + and str(doczy_signed_count).strip() not in ("0", "0.0") + ) + else: + has_doczy_tin = False + has_doczy_lob = False + has_doczy_eff_dt = False + has_doczy_signature = False + doczy_tin = None + doczy_lob = None + doczy_eff_dt = None + doczy_signature = None + + # ── 2. What's in the raw document (regex search) ── + tin_in_filename = _search_tin_in_filename(filename) + lob_in_filename = _search_lob_in_filename(filename) + date_in_filename = _search_date_in_filename(filename) + + tin_in_contract = False + eff_date_in_contract = False + eff_date_detected_value = None + signature_in_contract = False + lob_in_contract = False + lob_derived_all = set() + + for page_key in sorted(text_dict.keys(), key=lambda x: int(x)): + page_text = text_dict[page_key] + + if not tin_in_contract and _search_tin_in_page(page_text): + tin_in_contract = True + + if not eff_date_in_contract and _search_effective_date_in_page(page_text): + eff_date_in_contract = True + if eff_date_detected_value is None: + eff_date_detected_value = _extract_first_date_from_page(page_text) + + if not signature_in_contract and _search_signature_in_page(page_text): + signature_in_contract = True + + matched_lob = _search_lob_in_page(page_text) + if matched_lob: + lob_in_contract = True + for _kw, normalized_lob in matched_lob: + lob_derived_all.add(normalized_lob) + + # Include filename LOB in detected values + if lob_in_filename: + for _kw, normalized_lob in lob_in_filename: + lob_derived_all.add(normalized_lob) + + # ── 3. Determine status per field ── + tin_status = _determine_status(has_doczy_tin, tin_in_contract, tin_in_filename) + lob_status = _determine_status( + has_doczy_lob, lob_in_contract, bool(lob_in_filename) + ) + eff_date_status = _determine_status( + has_doczy_eff_dt, eff_date_in_contract, bool(date_in_filename) + ) + signature_status = _determine_status( + has_doczy_signature, signature_in_contract, False + ) + + # ── 4. Format display values ── + # TIN value: prefer Doczy extracted, else show detected source + tin_doczy_display = str(doczy_tin) if has_doczy_tin else "" + + # LOB value: prefer Doczy extracted, else show regex-detected + lob_doczy_display = str(doczy_lob) if has_doczy_lob else "" + lob_detected_display = ( + ", ".join(sorted(lob_derived_all)) if lob_derived_all else "" + ) + + # Effective date value: prefer Doczy extracted, else show regex-detected + eff_date_doczy_display = str(doczy_eff_dt) if has_doczy_eff_dt else "" + eff_date_detected_display = ( + eff_date_detected_value + if eff_date_detected_value + else (date_in_filename if date_in_filename else "") + ) + + # Signature value: prefer Doczy extracted + signature_doczy_display = str(doczy_signature) if has_doczy_signature else "" + + # Pricing readiness: all 4 fields must be "Extracted" + pricing_ready = all( + s == "Extracted" + for s in [tin_status, lob_status, eff_date_status, signature_status] + ) + + result = { + "TIN_STATUS": tin_status, + "TIN_DOCZY_VALUE": tin_doczy_display, + "LOB_STATUS": lob_status, + "LOB_DOCZY_VALUE": lob_doczy_display, + "LOB_DETECTED_VALUE": lob_detected_display, + "EFF_DATE_STATUS": eff_date_status, + "EFF_DATE_DOCZY_VALUE": eff_date_doczy_display, + "EFF_DATE_DETECTED_VALUE": eff_date_detected_display, + "SIGNATURE_STATUS": signature_status, + "SIGNATURE_DOCZY_VALUE": signature_doczy_display, + "PRICING_READY": "Y" if pricing_ready else "N", + } + + io_utils.save_result_to_json(filename, result, json_folder) + + with _post_doczy_progress_lock: + _post_doczy_progress["completed"] += 1 + done = _post_doczy_progress["completed"] + total = _post_doczy_progress["total"] + if total > 0 and (done % 5 == 0 or done == total): + logging.info( + f"Post-Doczy Report Progress: {done}/{total} files processed " + f"({done / total * 100:.1f}%)" + ) + + return filename, result + + except Exception as e: + logging.error(f"Error processing {filename} for Post-Doczy report: {e}") + error_result = dict(_POST_DOCZY_ERROR_RESULT) + io_utils.save_result_to_json(filename, error_result, json_folder) + return filename, error_result + + +# ── Post-Doczy Report: Main Entry Point ───────────────────────────────────── + + +def generate_post_doczy_report(input_dict, doczy_results_df, run_timestamp): + """Generate a Post-Doczy validation report for contract pricing readiness. + + Compares Doczy extraction output against raw document content to determine + whether each required field was: + - "Extracted" by Doczy (ready for pricing) + - "In Contract, Not Extracted" (Doczy missed it — needs review) + - "In Filename, Not Extracted" (in filename but Doczy missed it) + - "Not Present" (not found anywhere — contract missing info) + + Output columns: + FILE_NAME, TIN_STATUS, TIN_DOCZY_VALUE, + LOB_STATUS, LOB_DOCZY_VALUE, LOB_DETECTED_VALUE, + EFF_DATE_STATUS, EFF_DATE_DOCZY_VALUE, EFF_DATE_DETECTED_VALUE, + SIGNATURE_STATUS, SIGNATURE_DOCZY_VALUE, PRICING_READY + + Args: + input_dict: Dictionary mapping filename -> file_text (raw contract text). + doczy_results_df: pd.DataFrame of Doczy extraction output with FILE_NAME column. + run_timestamp: Timestamp string for organizing output files. + + Returns: + pd.DataFrame: The complete Post-Doczy validation report DataFrame. + """ + logging.info( + f"Starting Post-Doczy Report generation for {len(input_dict)} files..." + ) + logging.info(f"Doczy results loaded: {len(doczy_results_df)} rows") + + # Initialize progress + _post_doczy_progress["total"] = len(input_dict) + _post_doczy_progress["completed"] = 0 + + # Create output folder for individual JSONs + json_folder = config.POST_DOCZY_JSON_OUTPUT_FOLDER + "_report" + os.makedirs(json_folder, exist_ok=True) + logging.info(f"Post-Doczy Report JSON output folder: {json_folder}") + + # Build a case-insensitive lookup dict from Doczy results: filename -> row dict + # Doczy postprocessing strips .txt from FILE_NAME, so we try multiple + # matching strategies: exact, with .txt appended, and case-insensitive. + doczy_lookup = {} + if "FILE_NAME" in doczy_results_df.columns: + for _, row in doczy_results_df.iterrows(): + fname = str(row.get("FILE_NAME", "")).strip() + if fname: + row_dict = row.to_dict() + # Store with original case and lowercased for case-insensitive matching + doczy_lookup[fname] = row_dict + doczy_lookup[fname.lower()] = row_dict + # Also store with .txt suffix since input_dict keys include .txt + if not fname.lower().endswith(".txt"): + doczy_lookup[fname + ".txt"] = row_dict + doczy_lookup[fname.lower() + ".txt"] = row_dict + + # Log sample filenames from both sides for debugging + doczy_samples = list(doczy_results_df["FILE_NAME"].head(3)) + input_samples = list(input_dict.keys())[:3] + logging.info(f"Doczy FILE_NAME samples: {doczy_samples}") + logging.info(f"Input dict key samples: {input_samples}") + else: + logging.warning( + "Doczy results DataFrame has no FILE_NAME column. " + "All files will show as 'Not Extracted'." + ) + logging.info(f"Doczy results columns: {list(doczy_results_df.columns)}") + + matched = sum( + 1 for f in input_dict if f in doczy_lookup or f.lower() in doczy_lookup + ) + logging.info( + f"Matched {matched}/{len(input_dict)} input files to Doczy results " + f"(doczy_lookup has {len(doczy_lookup)} entries)" + ) + + start_time = datetime.now() + results = {} + + # Process files concurrently + with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: + future_to_file = { + executor.submit( + _process_file_for_post_doczy_report, + fname, + ftext, + doczy_lookup.get(fname) or doczy_lookup.get(fname.lower()), + json_folder, + ): fname + for fname, ftext in input_dict.items() + } + + for future in concurrent.futures.as_completed(future_to_file): + fname = future_to_file[future] + try: + result_fname, result_data = future.result() + results[result_fname] = result_data + except Exception as e: + logging.error(f"Exception for {fname} in Post-Doczy Report: {e}") + results[fname] = dict(_POST_DOCZY_ERROR_RESULT) + + duration = (datetime.now() - start_time).total_seconds() + logging.info(f"Post-Doczy Report generation complete in {duration:.2f} seconds") + if len(input_dict) > 0: + logging.info(f"Average time per file: {duration / len(input_dict):.2f} seconds") + + # Build DataFrame + df = pd.DataFrame([{"FILE_NAME": k, **v} for k, v in results.items()]) + + col_order = [ + "FILE_NAME", + "TIN_STATUS", + "TIN_DOCZY_VALUE", + "LOB_STATUS", + "LOB_DOCZY_VALUE", + "LOB_DETECTED_VALUE", + "EFF_DATE_STATUS", + "EFF_DATE_DOCZY_VALUE", + "EFF_DATE_DETECTED_VALUE", + "SIGNATURE_STATUS", + "SIGNATURE_DOCZY_VALUE", + "PRICING_READY", + ] + df = df[[c for c in col_order if c in df.columns]] + + # Summary statistics + total_files = len(df) + error_count = ( + (df["TIN_STATUS"] == "ERROR").sum() if "TIN_STATUS" in df.columns else 0 + ) + + status_cols = ["TIN_STATUS", "LOB_STATUS", "EFF_DATE_STATUS", "SIGNATURE_STATUS"] + status_labels = ["TIN", "LOB", "Effective Date", "Signature"] + summary_parts = [] + + for col, label in zip(status_cols, status_labels): + if col in df.columns: + extracted = (df[col] == "Extracted").sum() + in_contract = (df[col] == "In Contract, Not Extracted").sum() + in_filename = (df[col] == "In Filename, Not Extracted").sum() + not_present = (df[col] == "Not Present").sum() + summary_parts.append( + f"{label}: {extracted} extracted, " + f"{in_contract} in-contract-missed, " + f"{in_filename} in-filename-missed, " + f"{not_present} not-present" + ) + + pricing_ready_count = ( + (df["PRICING_READY"] == "Y").sum() if "PRICING_READY" in df.columns else 0 + ) + + logging.info( + f"Post-Doczy Report Summary: {total_files} files | " + f"Pricing Ready: {pricing_ready_count} | " + f"Errors: {error_count}" + ) + for part in summary_parts: + logging.info(f" {part}") + + # Create summary DataFrame + summary_data = { + "total_files": [total_files], + "pricing_ready": [pricing_ready_count], + "not_pricing_ready": [total_files - pricing_ready_count - error_count], + "errors": [error_count], + "pricing_ready_rate": [ + ( + f"{(pricing_ready_count / total_files * 100):.1f}%" + if total_files > 0 + else "0%" + ) + ], + "run_timestamp": [run_timestamp], + "batch_id": [config.BATCH_ID], + } + + # Add per-field breakdown to summary + for col, label in zip(status_cols, status_labels): + if col in df.columns: + summary_data[f"{label.lower().replace(' ', '_')}_extracted"] = [ + (df[col] == "Extracted").sum() + ] + summary_data[f"{label.lower().replace(' ', '_')}_missed"] = [ + (df[col] == "In Contract, Not Extracted").sum() + + (df[col] == "In Filename, Not Extracted").sum() + ] + summary_data[f"{label.lower().replace(' ', '_')}_not_present"] = [ + (df[col] == "Not Present").sum() + ] + + summary_df = pd.DataFrame(summary_data) + + # Save output + if config.WRITE_TO_S3: + io_utils.write_s3(df, "", run_timestamp, "post_doczy") + logging.info( + f"Post-Doczy Report uploaded to S3: " + f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-POST-DOCZY.csv" + ) + io_utils.write_s3(summary_df, "", run_timestamp, "post_doczy_summary") + logging.info( + f"Post-Doczy Summary uploaded to S3: " + f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-POST-DOCZY-SUMMARY.csv" + ) + else: + io_utils.write_local(df, "", run_timestamp, "post_doczy") + logging.info( + f"Post-Doczy Report saved locally: " + f"{config.CONSOLIDATED_OUTPUT_DIRECTORY}/{run_timestamp}/" + f"{config.BATCH_ID}-POST-DOCZY.csv" + ) + io_utils.write_local(summary_df, "", run_timestamp, "post_doczy_summary") + logging.info( + f"Post-Doczy Summary saved locally: " + f"{config.CONSOLIDATED_OUTPUT_DIRECTORY}/{run_timestamp}/" + f"{config.BATCH_ID}-POST-DOCZY-SUMMARY.csv" + ) + + return df + + +# ── Post-Doczy Report from Excel: Standalone Entry Point ───────────────────── + + +def _determine_location(doczy_value_present, in_filename): + """Determine where a field's information was found. + + Returns one of: + - "Contract Content" → value exists in Doczy extraction (from document) + - "File Name" → value detected in the filename + - "Not Present" → not found anywhere + """ + if doczy_value_present: + return "Contract Content" + if in_filename: + return "File Name" + return "Not Present" + + +def _clean_list_field(value): + """Clean Doczy list-formatted field values like '["Medicaid"]' to 'Medicaid'. + + Handles string representations of JSON lists by parsing and joining values. + """ + if value is None or (isinstance(value, float) and pd.isna(value)): + return None + s = str(value).strip() + if s.startswith("[") and s.endswith("]"): + try: + parsed = json.loads(s) + if isinstance(parsed, list): + cleaned = [str(v).strip() for v in parsed if str(v).strip()] + return ", ".join(cleaned) if cleaned else None + except (json.JSONDecodeError, ValueError): + pass + return s if s and s.upper() not in ("N/A", "NONE", "NAN", "[]", "NA") else None + + +def generate_post_doczy_report_from_excel(excel_path, run_timestamp=None): + """Generate a Post-Doczy Validation Report directly from a Doczy output Excel file. + + This is a standalone version that works without raw document text files. + It reads the Doczy extraction output and checks whether each contract has + the minimum information required for contract pricing: + - TIN (Y/N) and where it was found + - LOB (Insurance Type value) + - Effective Date (Y/N), where found, and actual date value + - Signature Indicator (Y/N) + + Output columns: + Contract Name | TIN | TIN Where | LOB | EFF Date | Where Eff | + Eff date | Signature ind + + Since the Doczy output can have multiple rows per contract (one per + exhibit/reimbursement term), this function groups by FILE_NAME and + produces one row per unique contract file. + + Args: + excel_path: S3 URI or local path to the Doczy output Excel file. + run_timestamp: Optional timestamp string for organizing output files. + + Returns: + pd.DataFrame: The Post-Doczy validation report. + """ + if run_timestamp is None: + run_timestamp = ( + f"run_{datetime.now().strftime('%Y%m%d_%H-%M')}_{config.BATCH_ID}" + ) + + logging.info(f"Loading Doczy results from: {excel_path}") + doczy_df = io_utils.read_DataFrame(excel_path) + + if doczy_df is None or doczy_df.empty: + logging.error(f"Failed to load Doczy results from: {excel_path}") + return pd.DataFrame() + + logging.info( + f"Loaded Doczy results: {len(doczy_df)} rows, " + f"{len(doczy_df.columns)} columns" + ) + + # Group by FILE_NAME to produce one row per contract file. + # Uses first non-null value per field (a contract may have multiple rows + # with different values across exhibits/reimbursement terms). + # Note: .first() is order-dependent — result depends on row ordering in the + # Doczy output. If exhibits have conflicting values (e.g., different TINs), + # the first encountered row wins. + contract_level_cols = [ + "FILE_NAME", + "FILENAME_TIN", + "PROV_GROUP_TIN", + "AARETE_DERIVED_LOB", + "LOB", + "AARETE_DERIVED_EFFECTIVE_DT", + "EFFECTIVE_DT", + "AARETE_DERIVED_SIGNATORY_COMPLETE_IND", + "NUM_SIGNED_SIGNATORY_LINE_COUNT", + ] + available_cols = [c for c in contract_level_cols if c in doczy_df.columns] + contracts_df = ( + doczy_df[available_cols].groupby("FILE_NAME", sort=False).first().reset_index() + ) + + logging.info( + f"Unique contracts: {len(contracts_df)} " f"(from {len(doczy_df)} total rows)" + ) + + report_rows = [] + + for _, row in contracts_df.iterrows(): + filename = str(row.get("FILE_NAME", "")).strip() + if not filename: + continue + + # ── TIN ── + doczy_tin = row.get("PROV_GROUP_TIN", None) + filename_tin = row.get("FILENAME_TIN", None) + has_contract_tin = _is_doczy_field_present(doczy_tin) + has_filename_tin = _is_doczy_field_present(filename_tin) + has_tin = has_contract_tin or has_filename_tin + + tin_indicator = "Y" if has_tin else "N" + + # TIN Where: where TIN was found + tin_where_parts = [] + if has_filename_tin: + tin_where_parts.append("Filename") + if has_contract_tin: + tin_where_parts.append("within contract") + tin_where = ", ".join(tin_where_parts) + + # ── LOB ── + # Use AARETE_DERIVED_LOB first; fall back to LOB if derived is empty/NaN + doczy_lob_raw = row.get("AARETE_DERIVED_LOB", None) + if not _is_doczy_field_present(doczy_lob_raw): + doczy_lob_raw = row.get("LOB", None) + has_doczy_lob = _is_doczy_field_present(doczy_lob_raw) + cleaned_lob = _clean_list_field(doczy_lob_raw) if has_doczy_lob else None + + lob_value = cleaned_lob if cleaned_lob else "" + + # ── Effective Date ── + # Use AARETE_DERIVED_EFFECTIVE_DT first; fall back to EFFECTIVE_DT + doczy_eff_dt = row.get("AARETE_DERIVED_EFFECTIVE_DT", None) + if not _is_doczy_field_present(doczy_eff_dt): + doczy_eff_dt = row.get("EFFECTIVE_DT", None) + + # Validate the date is actually parseable (not NaT/nan) + has_doczy_eff_dt = False + eff_date_value = "" + if _is_doczy_field_present(doczy_eff_dt): + try: + dt = pd.to_datetime(doczy_eff_dt) + if not pd.isna(dt): + has_doczy_eff_dt = True + eff_date_value = f"{dt.month}/{dt.day}/{dt.year}" + except (ValueError, TypeError): + raw = str(doczy_eff_dt).strip() + if raw not in ("NaT", "nan", "None", ""): + has_doczy_eff_dt = True + eff_date_value = raw + + eff_date_indicator = "Y" if has_doczy_eff_dt else "N" + where_eff = "In contract" if has_doczy_eff_dt else "" + + # ── Signature ── + doczy_signature = row.get("AARETE_DERIVED_SIGNATORY_COMPLETE_IND", None) + doczy_signed_count = row.get("NUM_SIGNED_SIGNATORY_LINE_COUNT", None) + + # Check signatory indicator — treat "N", "No", "False" as no signature + sig_negative_values = ("N", "NO", "FALSE", "0", "0.0") + has_signature_ind = ( + _is_doczy_field_present(doczy_signature) + and str(doczy_signature).strip().upper() not in sig_negative_values + ) + has_signed_count = _is_doczy_field_present(doczy_signed_count) and str( + doczy_signed_count + ).strip() not in ("0", "0.0") + has_signature = has_signature_ind or has_signed_count + signature_ind = "Y" if has_signature else "N" + + report_rows.append( + { + "Contract Name": filename, + "TIN": tin_indicator, + "TIN Where": tin_where, + "LOB": lob_value, + "EFF Date": eff_date_indicator, + "Where Eff": where_eff, + "Eff date": eff_date_value, + "Signature ind": signature_ind, + } + ) + + report_df = pd.DataFrame(report_rows) + + # ── Summary statistics ── + total = len(report_df) + has_tin_count = (report_df["TIN"] == "Y").sum() + has_lob_count = (report_df["LOB"] != "").sum() + has_eff_date_count = (report_df["EFF Date"] == "Y").sum() + has_signature_count = (report_df["Signature ind"] == "Y").sum() + + # Pricing ready: has TIN + LOB + Effective Date + Signature + pricing_ready = ( + (report_df["TIN"] == "Y") + & (report_df["LOB"] != "") + & (report_df["EFF Date"] == "Y") + & (report_df["Signature ind"] == "Y") + ).sum() + missing_info = total - pricing_ready + + logging.info("Post-Doczy Report from Excel — Summary:") + logging.info(f" Total contracts: {total}") + logging.info(f" TIN found: {has_tin_count}/{total}") + logging.info(f" LOB found: {has_lob_count}/{total}") + logging.info(f" Effective Date found: {has_eff_date_count}/{total}") + logging.info(f" Signature found: {has_signature_count}/{total}") + logging.info(f" Usable for pricing: {pricing_ready}") + logging.info(f" Missing information: {missing_info}") + + # ── Summary DataFrame ── + summary_data = { + "total_contracts": [total], + "usable_for_pricing": [pricing_ready], + "missing_information": [missing_info], + "tin_found": [has_tin_count], + "lob_found": [has_lob_count], + "effective_date_found": [has_eff_date_count], + "signature_found": [has_signature_count], + "pricing_ready_rate": [ + f"{(pricing_ready / total * 100):.1f}%" if total > 0 else "0%" + ], + "run_timestamp": [run_timestamp], + "batch_id": [config.BATCH_ID], + } + summary_df = pd.DataFrame(summary_data) + + # ── Save output ── + # Uses "post_doczy_excel" key to avoid overwriting the full Post-Doczy report + # when both PERFORM_POST_DOCZY and PERFORM_POST_DOCZY_EXCEL are enabled. + if config.WRITE_TO_S3: + io_utils.write_s3(report_df, "", run_timestamp, "post_doczy_excel") + logging.info( + f"Post-Doczy Excel Report uploaded to S3: " + f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-POST-DOCZY-EXCEL.csv" + ) + io_utils.write_s3(summary_df, "", run_timestamp, "post_doczy_excel_summary") + logging.info( + f"Post-Doczy Excel Summary uploaded to S3: " + f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-POST-DOCZY-EXCEL-SUMMARY.csv" + ) + else: + io_utils.write_local(report_df, "", run_timestamp, "post_doczy_excel") + logging.info( + f"Post-Doczy Excel Report saved locally: " + f"{config.CONSOLIDATED_OUTPUT_DIRECTORY}/{run_timestamp}/" + f"{config.BATCH_ID}-POST-DOCZY-EXCEL.csv" + ) + io_utils.write_local(summary_df, "", run_timestamp, "post_doczy_excel_summary") + logging.info( + f"Post-Doczy Excel Summary saved locally: " + f"{config.CONSOLIDATED_OUTPUT_DIRECTORY}/{run_timestamp}/" + f"{config.BATCH_ID}-POST-DOCZY-EXCEL-SUMMARY.csv" + ) + + return report_df + + if __name__ == "__main__": logging.basicConfig( level=logging.INFO, @@ -704,11 +1463,56 @@ if __name__ == "__main__": # Generate timestamp for this run run_timestamp = f"run_{datetime.now().strftime('%Y%m%d_%H-%M')}_{config.BATCH_ID}" - # Load input files - logging.info("Loading input files...") - input_dict = io_utils.read_input() + # Run Post-Doczy Excel-only Report (no raw text files needed) + if config.PERFORM_POST_DOCZY_EXCEL: + if not config.DOCZY_RESULTS_PATH: + logging.error( + "Post-Doczy Excel report requires doczy_results_path argument " + "(S3 URI or local path to Doczy output Excel/CSV)." + ) + else: + post_doczy_df = generate_post_doczy_report_from_excel( + config.DOCZY_RESULTS_PATH, run_timestamp + ) + logging.info( + f"Post-Doczy Excel Report complete. " + f"Processed {len(post_doczy_df)} contracts." + ) - # Run DTC Report (classification + keyword search) - report_df = generate_dtc_report(input_dict, run_timestamp) + # Load input files (raw contract text) — only needed for Pre-Doczy or full Post-Doczy + input_dict = {} + if config.PERFORM_PRE_DOCZY or config.PERFORM_POST_DOCZY: + logging.info("Loading input files...") + input_dict = io_utils.read_input() - logging.info(f"DTC Report complete. Processed {len(report_df)} files.") + # Run Pre-Doczy Report (classification + keyword search) + if config.PERFORM_PRE_DOCZY: + pre_doczy_df = generate_pre_doczy_report(input_dict, run_timestamp) + logging.info(f"Pre-Doczy Report complete. Processed {len(pre_doczy_df)} files.") + + # Run Post-Doczy Validation Report (with raw document text) + if config.PERFORM_POST_DOCZY: + if not config.DOCZY_RESULTS_PATH: + logging.error( + "Post-Doczy report requires doczy_results_path argument " + "(path to Doczy extraction output CSV/Excel)." + ) + else: + logging.info(f"Loading Doczy results from: {config.DOCZY_RESULTS_PATH}") + doczy_results_df = io_utils.read_DataFrame(config.DOCZY_RESULTS_PATH) + if doczy_results_df is None or doczy_results_df.empty: + logging.error( + f"Failed to load Doczy results from: {config.DOCZY_RESULTS_PATH}. " + f"File not found or empty. Skipping Post-Doczy report." + ) + else: + logging.info( + f"Loaded Doczy results: {len(doczy_results_df)} rows, " + f"{len(doczy_results_df.columns)} columns" + ) + post_doczy_df = generate_post_doczy_report( + input_dict, doczy_results_df, run_timestamp + ) + logging.info( + f"Post-Doczy Report complete. Processed {len(post_doczy_df)} files." + ) diff --git a/src/utils/io_utils.py b/src/utils/io_utils.py index 59c2b93..717841f 100644 --- a/src/utils/io_utils.py +++ b/src/utils/io_utils.py @@ -4,7 +4,6 @@ import os import re import tempfile from io import StringIO -import json from threading import Lock import pandas as pd @@ -336,6 +335,47 @@ def read_s3_csv(csv_path) -> pd.DataFrame | None: return None +def read_s3_excel(excel_path) -> pd.DataFrame | None: + """Reads an Excel (.xlsx) file from an S3 bucket and returns it as a DataFrame. + + Downloads the file to a temporary location since pd.read_excel requires + a file-like object or path, not a raw byte stream with S3 chunking. + + Args: + excel_path (str): The S3 URL of the Excel file + (e.g., s3://bucket-name/path/to/file.xlsx). + + Returns: + pd.DataFrame | None: The contents of the Excel file as a pandas DataFrame, + or None if an error occurs. + """ + s3_client = config.S3_CLIENT + + if not excel_path.startswith("s3://"): + logging.error(f"Invalid S3 path: {excel_path}") + return None + + s3_parts = excel_path[5:].split("/", 1) + bucket = s3_parts[0] + key = s3_parts[1] if len(s3_parts) > 1 else "" + + try: + with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=True) as tmp: + s3_client.download_file(bucket, key, tmp.name) + df = pd.read_excel(tmp.name) + logging.info( + f"Read Excel from S3: {excel_path} " + f"({len(df)} rows, {len(df.columns)} columns)" + ) + return df + except ClientError as e: + logging.error(f"Error downloading Excel from S3: {e}") + return None + except Exception as e: + logging.error(f"Error reading Excel file from {excel_path}: {e}") + return None + + def read_input(local_path=config.LOCAL_PATH) -> dict: """Reads input files from the specified local path or S3 bucket. @@ -796,6 +836,33 @@ def write_local( quoting=1, ) return None + elif output_type == "pre_doczy": + output_dir = os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp) + os.makedirs(output_dir, exist_ok=True) + df.to_csv( + os.path.join(output_dir, f"{config.BATCH_ID}-PRE-DOCZY.csv"), + index=False, + quoting=1, + ) + return None + elif output_type == "post_doczy": + output_dir = os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp) + os.makedirs(output_dir, exist_ok=True) + df.to_csv( + os.path.join(output_dir, f"{config.BATCH_ID}-POST-DOCZY.csv"), + index=False, + quoting=1, + ) + return None + elif output_type == "post_doczy_summary": + output_dir = os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp) + os.makedirs(output_dir, exist_ok=True) + df.to_csv( + os.path.join(output_dir, f"{config.BATCH_ID}-POST-DOCZY-SUMMARY.csv"), + index=False, + quoting=1, + ) + return None elif output_type == "qc_qa_validated": # QC/QA validated output goes to outputs/qc_qa folder output_dir = os.path.join(config.QC_QA_OUTPUT_DIRECTORY, run_timestamp) @@ -842,7 +909,7 @@ def write_local( return None else: logging.error( - f"Unknown output_type: {output_type}. Valid types are: 'final', 'individual', 'error', 'usage', 'usage_summary', 'dtc', 'dtc_summary', 'qc_qa_validated', 'qc_qa_stats', 'cc_results_full', 'dashboard_results_full', 'qc_qa_cc_full', 'qc_qa_error', 'parent_child', 'individual_cc'" + f"Unknown output_type: {output_type}. Valid types are: 'final', 'individual', 'error', 'usage', 'usage_summary', 'dtc', 'dtc_summary', 'pre_doczy', 'post_doczy', 'post_doczy_summary', 'qc_qa_validated', 'qc_qa_stats', 'cc_results_full', 'dashboard_results_full', 'qc_qa_cc_full', 'qc_qa_error', 'parent_child', 'individual_cc'" ) return None @@ -960,6 +1027,28 @@ def write_s3( config.S3_CLIENT.put_object( Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer ) + elif output_type == "pre_doczy": + output_path = ( + f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-PRE-DOCZY.csv" + ) + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + elif output_type == "post_doczy": + output_path = ( + f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-POST-DOCZY.csv" + ) + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + elif output_type == "post_doczy_summary": + output_path = f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-POST-DOCZY-SUMMARY.csv" + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) elif output_type == "qc_qa_validated": output_path = f"{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/{config.BATCH_ID}-QC-QA-VALIDATED.csv" csv_buffer = df.to_csv(index=False).encode() @@ -1039,6 +1128,12 @@ def write_s3( logging.info(f"Saved DTC output file: {output_path}") elif output_type == "dtc_summary": logging.info(f"Saved DTC summary file: {output_path}") + elif output_type == "pre_doczy": + logging.info(f"Saved Pre-Doczy report file: {output_path}") + elif output_type == "post_doczy": + logging.info(f"Saved Post-Doczy report file: {output_path}") + elif output_type == "post_doczy_summary": + logging.info(f"Saved Post-Doczy summary file: {output_path}") elif output_type == "qc_qa_validated": logging.info(f"Saved QC/QA validated data file: {output_path}") elif output_type == "qc_qa_stats": @@ -1157,14 +1252,15 @@ def save_result_to_json(filename, result, json_folder): def read_DataFrame(file_path: str) -> pd.DataFrame: """ Takes in a file path and returns a pandas DataFrame. - Supports reading from local file system and only CSV from S3. + Supports reading from local file system, and CSV/Excel from S3. """ is_s3 = file_path.startswith("s3://") if is_s3: - file = read_s3_csv(file_path) - return file + if file_path.lower().endswith((".xlsx", ".xls")): + return read_s3_excel(file_path) + return read_s3_csv(file_path) file = read_local(file_path) return file