import concurrent.futures import json import logging import os import re from datetime import datetime from threading import Lock 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, split_text, clean_law_symbols, ) from src.prompts.fieldset import Field 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) # Load prompts at module level (cached for reuse across threads) DOCUMENT_TYPE_CLASSIFICATION_PROMPT = Field.load_from_file( config.DTC_PROMPTS_JSON_PATH, "document_type_classification" ).prompt CONTRACT_TYPE_PROMPT = Field.load_from_file( config.DTC_PROMPTS_JSON_PATH, "contract_type" ).prompt NON_CONTRACT_TYPE_PROMPT = Field.load_from_file( config.DTC_PROMPTS_JSON_PATH, "non_contract_type" ).prompt # Thread-safe counter for progress tracking progress_lock = Lock() progress_counter = {"completed": 0, "total": 0} # ── Pre-compiled patterns for DTC Report keyword search ───────────────────── _TIN_COMPILED = re.compile(TIN_PATTERN) _EFFECTIVE_DATE_COMPILED = [ re.compile(p) 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") _LOB_JSON_FILES = [ "crosswalk_lob.json", "crosswalk_program_lob.json", "crosswalk_product_lob.json", ] def _load_lob_keywords_from_mappings(): """Load LOB keywords and their normalized LOB values from mapping JSON files. Reads crosswalk_lob.json, crosswalk_program_lob.json, and crosswalk_product_lob.json and extracts all keyword -> normalized_LOB pairs from their mapping, state_mapping, and client_mapping sections. Returns: dict: Mapping of keyword (str) -> normalized LOB category (str). e.g. {"Medicaid": "Medicaid", "CHIP": "Medicaid", "TENNCARE": "Medicaid", ...} """ keyword_to_lob = {} for json_file in _LOB_JSON_FILES: filepath = os.path.join(_MAPPINGS_DIR, json_file) if not os.path.exists(filepath): logging.warning(f"LOB mapping file not found: {filepath}") continue with open(filepath, "r") as f: data = json.load(f) # Extract from "mapping" section (key=keyword, value=normalized LOB) if "mapping" in data: for keyword, normalized_lob in data["mapping"].items(): if keyword and normalized_lob: keyword_to_lob[keyword] = normalized_lob # Extract from "state_mapping" section (state -> {keyword: normalized LOB}) if "state_mapping" in data: for _state, state_keywords in data["state_mapping"].items(): if isinstance(state_keywords, dict): for keyword, normalized_lob in state_keywords.items(): if keyword and normalized_lob: keyword_to_lob[keyword] = normalized_lob # Extract from "client_mapping" section (client -> {keyword: normalized LOB}) if "client_mapping" in data: for _client, client_keywords in data["client_mapping"].items(): if isinstance(client_keywords, dict): for keyword, normalized_lob in client_keywords.items(): if keyword and normalized_lob: keyword_to_lob[keyword] = normalized_lob logging.info( f"Loaded {len(keyword_to_lob)} LOB keywords from " f"{len(_LOB_JSON_FILES)} mapping files" ) return keyword_to_lob # Load at module level: keyword -> normalized LOB _LOB_KEYWORD_TO_NORMALIZED = _load_lob_keywords_from_mappings() # Pre-compile regex for each LOB keyword (case-insensitive, word-boundary) _LOB_COMPILED = [ (kw, normalized_lob, re.compile(r"(?i)\b" + re.escape(kw) + r"\b")) for kw, normalized_lob in _LOB_KEYWORD_TO_NORMALIZED.items() ] # ── 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): """Call Claude through llm_utils.invoke_claude (unified LLM interface) Args: filename: Name of the file being processed (for logging/tracking) context: Document text context question: The prompt/question to ask the LLM Returns: str: The LLM's response, or "ERROR" if the call fails """ # Prepare the full prompt with context prompt = f"Document text:\n{context}\n\n{question}" try: # Use the unified llm_utils interface (supports both local and EC2 modes) answer = llm_utils.invoke_claude( prompt=prompt, model_id="sonnet_latest", # Uses config.MODEL_ALIASES["sonnet_latest"] filename=filename, max_tokens=8192, ) return answer.strip() except Exception as e: logging.error(f"Error calling Claude for {filename}: {str(e)}") return "ERROR" def process_single_file(filename, contract_text, json_folder): """Process a single file to determine if it's a contract and classify it. Args: filename: Name of the file being processed contract_text: Raw text content of the file json_folder: Path to folder for saving individual JSON results Returns: tuple: (filename, result_dict) """ try: contract_text = clean_newlines(contract_text) contract_text = clean_law_symbols(contract_text) text_dict = split_text(contract_text) # Loop through n pages is_contract = "NO" contract_type = "N/A" non_contract_type = "N/A" found_on_page = "N/A" for i in range(config.DTC_MAX_PAGES_TO_CHECK): page_key = str(i + 1) # Handle edge case: document has fewer pages than config.DTC_MAX_PAGES_TO_CHECK if page_key not in text_dict: break context = text_dict[page_key] # Ask if this page is a contract is_contract_answer = call_llm( filename, context, DOCUMENT_TYPE_CLASSIFICATION_PROMPT, ) if is_contract_answer.strip().upper() == "YES": # If contract found, get contract type type_of_contract_answer = call_llm( filename, context, CONTRACT_TYPE_PROMPT ) is_contract = "YES" contract_type = type_of_contract_answer found_on_page = i + 1 break # Stop checking further pages once contract is found # If not a contract, determine non-contract document type if is_contract == "NO": # Use the first page for non-contract classification if "1" in text_dict: context = text_dict["1"] non_contract_type_answer = call_llm( filename, context, NON_CONTRACT_TYPE_PROMPT ) non_contract_type = non_contract_type_answer found_on_page = 1 result = { "is_contract": is_contract, "contract_type": contract_type, "non_contract_type": non_contract_type, "found_on_page": found_on_page, } # Save result to JSON immediately io_utils.save_result_to_json(filename, result, json_folder) # Update progress counter with progress_lock: progress_counter["completed"] += 1 completed = progress_counter["completed"] total = progress_counter["total"] if total > 0 and ( completed % 5 == 0 or completed == total ): # Log every 5 files or at completion logging.info( f"Progress: {completed}/{total} files processed ({completed/total*100:.1f}%)" ) return filename, result except Exception as e: logging.error(f"Error processing {filename}: {str(e)}") error_result = { "is_contract": "ERROR", "contract_type": "ERROR", "non_contract_type": "ERROR", "found_on_page": "ERROR", } # Save error result to JSON io_utils.save_result_to_json(filename, error_result, json_folder) return filename, error_result def main(input_dict, run_timestamp): """Run document type classification on a dictionary of files. This function is designed to be called from investment/main.py as a pre-filter. It classifies all files and returns a dict mapping filenames to classification results. Args: input_dict: Dictionary mapping filename -> file_text run_timestamp: Timestamp string for organizing output files (format: run_YYYYMMDD_HH-MM_BATCHID) Returns: dict: Mapping of filename -> classification result dict with keys: - is_contract: "YES" or "NO" - contract_type: Type if contract, otherwise "N/A" - non_contract_type: Type if non-contract, otherwise "N/A" - found_on_page: Page number where classification was determined Side effects: - Saves 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 classification for {len(input_dict)} files...") logging.info(f"Max pages to check: {config.DTC_MAX_PAGES_TO_CHECK}") logging.info(f"Results will be saved to: {config.DTC_OUTPUT_FILE}") # Initialize progress counter progress_counter["total"] = len(input_dict) progress_counter["completed"] = 0 # Create output folder os.makedirs(config.DTC_JSON_OUTPUT_FOLDER, exist_ok=True) logging.info(f"JSON output folder: {config.DTC_JSON_OUTPUT_FOLDER}") # Record start time start_time = datetime.now() answer_dict = {} # Process files concurrently with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: future_to_file = {} for filename, contract_text in input_dict.items(): future = executor.submit( process_single_file, filename, contract_text, config.DTC_JSON_OUTPUT_FOLDER, ) future_to_file[future] = filename # Collect results as they complete for future in concurrent.futures.as_completed(future_to_file): filename = future_to_file[future] try: result_filename, result_data = future.result() answer_dict[result_filename] = result_data except Exception as e: logging.error(f"Exception occurred for {filename}: {str(e)}") answer_dict[filename] = { "is_contract": "ERROR", "contract_type": "ERROR", "non_contract_type": "ERROR", "found_on_page": "ERROR", } # Record end time end_time = datetime.now() duration = (end_time - start_time).total_seconds() logging.info(f"DTC classification complete in {duration:.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()]) # Log summary statistics contracts_found = len(df[df["is_contract"] == "YES"]) non_contracts = len(df[df["is_contract"] == "NO"]) errors = len(df[df["is_contract"] == "ERROR"]) logging.info( f"DTC Summary: {contracts_found} contracts, {non_contracts} non-contracts, {errors} errors" ) # Create summary DataFrame with aggregate statistics summary_data = { "total_files": [len(df)], "contracts": [contracts_found], "non_contracts": [non_contracts], "errors": [errors], "contract_rate": [ f"{(contracts_found/len(df)*100):.1f}%" if len(df) > 0 else "0%" ], "run_timestamp": [run_timestamp], "batch_id": [config.BATCH_ID], } # Add contract type breakdown if contracts_found > 0: contract_types = df[df["is_contract"] == "YES"]["contract_type"].value_counts() summary_data["most_common_contract_type"] = [ contract_types.index[0] if len(contract_types) > 0 else "N/A" ] summary_data["contract_type_breakdown"] = [contract_types.to_dict()] else: summary_data["most_common_contract_type"] = ["N/A"] summary_data["contract_type_breakdown"] = [{}] # Add non-contract type breakdown if non_contracts > 0: non_contract_types = df[df["is_contract"] == "NO"][ "non_contract_type" ].value_counts() summary_data["most_common_non_contract_type"] = [ non_contract_types.index[0] if len(non_contract_types) > 0 else "N/A" ] summary_data["non_contract_type_breakdown"] = [non_contract_types.to_dict()] else: summary_data["most_common_non_contract_type"] = ["N/A"] summary_data["non_contract_type_breakdown"] = [{}] summary_df = pd.DataFrame(summary_data) # Save detailed results using same pattern as investment pipeline if config.WRITE_TO_S3: io_utils.write_s3(df, "", run_timestamp, "dtc") logging.info( f"DTC results uploaded to S3: {config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-DTC.csv" ) # Save summary io_utils.write_s3(summary_df, "", run_timestamp, "dtc_summary") logging.info( f"DTC summary uploaded to S3: {config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-DTC-SUMMARY.csv" ) else: io_utils.write_local(df, "", run_timestamp, "dtc") logging.info( f"DTC results saved locally: {config.CONSOLIDATED_OUTPUT_DIRECTORY}/{run_timestamp}/{config.BATCH_ID}-DTC.csv" ) # Save summary io_utils.write_local(summary_df, "", run_timestamp, "dtc_summary") logging.info( f"DTC summary saved locally: {config.CONSOLIDATED_OUTPUT_DIRECTORY}/{run_timestamp}/{config.BATCH_ID}-DTC-SUMMARY.csv" ) return answer_dict # ── Pre-Doczy Report: Keyword Search Helpers ──────────────────────────────── def _search_tin_in_page(page_text): """Check if TIN pattern is present in a single page of text. Args: page_text: Text content of a single page. Returns: bool: True if a TIN-like pattern was found. """ return bool(_TIN_COMPILED.search(page_text)) def _search_effective_date_in_page(page_text): """Check if any effective date keyword or date format is present in a single page. Args: page_text: Text content of a single page. Returns: bool: True if an effective date indicator was found. """ return any(p.search(page_text) for p in _EFFECTIVE_DATE_COMPILED) def _search_lob_in_page(page_text): """Find all LOB keywords present in a single page of text. Args: page_text: Text content of a single page. Returns: list[tuple[str, str]]: List of (keyword, normalized_lob) tuples matched on this page. """ return [ (kw, normalized_lob) for kw, normalized_lob, pattern in _LOB_COMPILED if pattern.search(page_text) ] # ── Pre-Doczy Report: Per-file Processing ─────────────────────────────────── 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. Args: filename: Name of the file being processed. file_text: Raw text content (Textract output with page markers). 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) # ── Step 1: Document Type Classification (LLM-based) ── is_contract = "NO" document_type = "N/A" for i in range(config.DTC_MAX_PAGES_TO_CHECK): page_key = str(i + 1) if page_key not in text_dict: break context = text_dict[page_key] is_contract_answer = call_llm( filename, context, DOCUMENT_TYPE_CLASSIFICATION_PROMPT ) if is_contract_answer.strip().upper() == "YES": is_contract = "YES" document_type = call_llm(filename, context, CONTRACT_TYPE_PROMPT) break if is_contract == "NO" and "1" in text_dict: document_type = call_llm(filename, text_dict["1"], NON_CONTRACT_TYPE_PROMPT) # ── Step 2: Keyword Search Across ALL Pages ── tin_pages = [] eff_date_pages = [] lob_pages = [] lob_keywords_all = set() lob_derived_all = set() for page_key in sorted(text_dict.keys(), key=lambda x: int(x)): page_text = text_dict[page_key] if _search_tin_in_page(page_text): tin_pages.append(page_key) if _search_effective_date_in_page(page_text): eff_date_pages.append(page_key) matched_lob = _search_lob_in_page(page_text) if matched_lob: lob_pages.append(page_key) for kw, normalized_lob in matched_lob: lob_keywords_all.add(kw) lob_derived_all.add(normalized_lob) result = { "IS_CONTRACT": is_contract, "DOCUMENT_TYPE": document_type, "TIN_FOUND": "YES" if tin_pages else "NO", "TIN_PAGES": ", ".join(tin_pages) if tin_pages else "N/A", "EFFECTIVE_DATE_FOUND": "YES" if eff_date_pages else "NO", "EFFECTIVE_DATE_PAGES": ( ", ".join(eff_date_pages) if eff_date_pages else "N/A" ), "LOB_FOUND": "YES" if lob_pages else "NO", "LOB_PAGES": ", ".join(lob_pages) if lob_pages else "N/A", "LOB_KEYWORDS_MATCHED": ( ", ".join(sorted(lob_keywords_all)) if lob_keywords_all else "N/A" ), "LOB_DERIVED": ( ", ".join(sorted(lob_derived_all)) if lob_derived_all else "N/A" ), } # Save individual result to JSON io_utils.save_result_to_json(filename, result, json_folder) # Update progress 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"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 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 # ── Pre-Doczy Report: Main Entry Point ────────────────────────────────────── 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 - IS_CONTRACT: YES/NO from LLM classification - DOCUMENT_TYPE: Contract type or non-contract type - TIN_FOUND / TIN_PAGES: Whether TIN was detected and on which pages - EFFECTIVE_DATE_FOUND / EFFECTIVE_DATE_PAGES: Whether effective date was found - LOB_FOUND / LOB_PAGES / LOB_KEYWORDS_MATCHED: LOB detection results Args: input_dict: Dictionary mapping filename -> file_text. run_timestamp: Timestamp string for organizing output files (format: run_YYYYMMDD_HH-MM_BATCHID). Returns: 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 Pre-Doczy Report generation for {len(input_dict)} files...") logging.info( 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 # Create output folder for individual JSONs json_folder = config.DTC_JSON_OUTPUT_FOLDER + "_pre_doczy_report" os.makedirs(json_folder, exist_ok=True) logging.info(f"Pre-Doczy Report JSON output folder: {json_folder}") 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_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): 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 Pre-Doczy Report: {e}") results[fname] = dict(_PRE_DOCZY_ERROR_RESULT) duration = (datetime.now() - start_time).total_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") # 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", "TIN_PAGES", "EFFECTIVE_DATE_FOUND", "EFFECTIVE_DATE_PAGES", "LOB_FOUND", "LOB_PAGES", "LOB_KEYWORDS_MATCHED", "LOB_DERIVED", ] df = df[[c for c in col_order if c in df.columns]] # Log summary statistics total_files = len(df) tin_count = (df["TIN_FOUND"] == "YES").sum() if "TIN_FOUND" in df.columns else 0 eff_date_count = ( (df["EFFECTIVE_DATE_FOUND"] == "YES").sum() if "EFFECTIVE_DATE_FOUND" in df.columns else 0 ) lob_count = (df["LOB_FOUND"] == "YES").sum() if "LOB_FOUND" in df.columns else 0 error_count = ( (df["IS_CONTRACT"] == "ERROR").sum() if "IS_CONTRACT" in df.columns else 0 ) logging.info( 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} | " f"Errors: {error_count}" ) # Save output using existing io_utils pattern if config.WRITE_TO_S3: io_utils.write_s3(df, "", run_timestamp, "pre_doczy") logging.info( 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, "pre_doczy") logging.info( 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_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_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 # ── 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 = "Y" 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 Count (0,1,2,...) 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", "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_signed_count = row.get("NUM_SIGNED_SIGNATORY_LINE_COUNT", None) has_signed_count = _is_doczy_field_present(doczy_signed_count) and str( doczy_signed_count ).strip() not in ("0", "0.0") signature_ind = "Y" if has_signed_count 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, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", force=True, ) # Suppress AWS SDK logging logging.getLogger("botocore").setLevel(logging.WARNING) logging.getLogger("boto3").setLevel(logging.WARNING) # Generate timestamp for this run run_timestamp = f"run_{datetime.now().strftime('%Y%m%d_%H-%M')}_{config.BATCH_ID}" # 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." ) # 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() # 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." )