import os import re import sys import collections import pandas as pd import pypdf # Try to import OCR libraries if available try: import pytesseract from pdf2image import convert_from_path HAS_OCR = True except ImportError: HAS_OCR = False def normalize_doc_no(doc_no): """Normalize a document number by converting to uppercase and removing non-alphanumeric chars.""" if pd.isna(doc_no): return "" val_str = str(doc_no).strip().upper() return re.sub(r'[^A-Z0-9]', '', val_str) def get_doc_no_variants(doc_no): """Generate potential search variants for a document number to increase match rate.""" val_str = str(doc_no).strip() variants = {val_str, val_str.upper()} # Normalized full version norm = normalize_doc_no(val_str) if norm: variants.add(norm) # Split by common separators (/, -, _) and get part suffixes parts = re.split(r'[/\-_]', val_str) for p in parts: p_strip = p.strip() if len(p_strip) >= 4: variants.add(p_strip) variants.add(p_strip.upper()) variants.add(normalize_doc_no(p_strip)) return list(variants) def find_excel_headers_and_df(file_path): """ Detect the header row of an Excel sheet and return the cleaned DataFrame. Returns (header_row_index, df, detected_column_name) or (None, None, None) on error. """ try: df_raw = pd.read_excel(file_path, header=None) except Exception as e: print(f"Error reading Excel {file_path}: {e}") return None, None, None header_keywords = {'date', 'voucher', 'invoice', 'ledger', 'amount', 'pnr', 'passenger', 'ticket', 'flight'} best_row_idx = 0 max_matches = 0 # Scan the first 15 rows to find the headers row for idx in range(min(15, len(df_raw))): row_vals = [str(x).strip().lower() for x in df_raw.iloc[idx] if pd.notna(x)] matches = sum(1 for val in row_vals if any(kw in val for kw in header_keywords)) if matches > max_matches: max_matches = matches best_row_idx = idx # Set headers headers = [str(x).strip() for x in df_raw.iloc[best_row_idx]] df = df_raw.iloc[best_row_idx + 1:].reset_index(drop=True) df.columns = headers # Try to identify primary document reference column doc_col = None doc_col_keywords = [ 'voucher no', 'voucher number', 'voucher_no', 'invoice no', 'invoice number', 'invoice_no', 'document no', 'document number', 'document_no', 'reference no', 'reference number', 'reference_no', 'inv no', 'doc no', 'ref no', 'bill no', 'voucher', 'invoice', 'document', 'reference', 'bill' ] # Case-insensitive containment check for kw in doc_col_keywords: for col in df.columns: if pd.notna(col): col_str = str(col).strip().lower() if kw in col_str: doc_col = col break if doc_col: break # Fallback: check content of columns to find the best candidate if not doc_col: best_col = None max_score = 0 for col in df.columns: non_null = df[col].dropna() if len(non_null) == 0: continue score = 0 for val in non_null: val_str = str(val).strip() if len(val_str) > 3: has_digit = any(c.isdigit() for c in val_str) has_alpha = any(c.isalpha() for c in val_str) has_symbol = any(c in val_str for c in ['/', '-', '_']) if (has_digit and has_alpha) or (has_digit and has_symbol): score += 1 score_pct = score / len(non_null) if score_pct > max_score: max_score = score_pct best_col = col if best_col is not None and max_score > 0.3: doc_col = best_col # Absolute fallback: first non-date, non-amount column if not doc_col: for col in df.columns: col_str = str(col).lower() if 'date' not in col_str and 'amount' not in col_str and 'sum' not in col_str: doc_col = col break if not doc_col: doc_col = df.columns[0] return best_row_idx, df, doc_col def extract_excel_records(df, doc_col): """Extract and clean document records from the identified column, ignoring summary/total rows.""" raw_records = [] summary_keywords = {'total', 'outstanding', 'grand total', 'subtotal', 'balance', 'closing', 'opening'} for idx, row in df.iterrows(): val = row[doc_col] if pd.isna(val) or str(val).strip() == '': continue val_str = str(val).strip() # Check if the row represents a total/summary row is_summary = False for cell_val in row: if pd.notna(cell_val): cell_str = str(cell_val).strip().lower() if any(kw in cell_str for kw in summary_keywords): is_summary = True break if is_summary: continue raw_records.append(val_str) return raw_records def extract_pdf_pages_text(pdf_path): """Extract text page-by-page from a PDF file. Returns a list of strings.""" pages_text = [] try: reader = pypdf.PdfReader(pdf_path) for page in reader.pages: t = page.extract_text() pages_text.append(t if t else '') except Exception as e: print(f"Error reading PDF {pdf_path}: {e}") return pages_text def run_ocr_on_pdf(pdf_path): """Run OCR on a PDF file if it is scanned. Returns (pages_text, error_message).""" if not HAS_OCR: return None, "OCR libraries (pytesseract/pdf2image) are not installed." # Set tesseract path if found in standard Windows locations tess_paths = [ r"C:\Program Files\Tesseract-OCR\tesseract.exe", r"C:\Program Files (x86)\Tesseract-OCR\tesseract.exe" ] for p in tess_paths: if os.path.exists(p): pytesseract.pytesseract.tesseract_cmd = p break try: # pdf2image convert pages = convert_from_path(pdf_path, dpi=150) pages_text = [] for page in pages: text = pytesseract.image_to_string(page) pages_text.append(text) return pages_text, None except Exception as e: return None, str(e) def analyze_folder(folder_path, rel_folder_name=None): """Perform matching and reconciliation on a single folder.""" all_files = os.listdir(folder_path) # 1. Find Excel files (exclude temp files starting with ~$) excel_files = [f for f in all_files if f.endswith(('.xlsx', '.xls')) and not f.startswith('~$')] # 2. Find PDF files pdf_files = [f for f in all_files if f.endswith('.pdf')] excel_file = None excel_records = [] doc_col = None excel_duplicates = [] # Process Excel file if found if excel_files: # Prioritize files with 'updated' or 'final' in name, or pick largest size excel_files.sort(key=lambda f: ( -int('updated' in f.lower() or 'final' in f.lower()), -os.path.getsize(os.path.join(folder_path, f)) )) excel_file = excel_files[0] excel_path = os.path.join(folder_path, excel_file) header_row, df, doc_col = find_excel_headers_and_df(excel_path) if df is not None and doc_col is not None: excel_records = extract_excel_records(df, doc_col) # Find duplicate document numbers in Excel counts = collections.Counter(excel_records) excel_duplicates = [doc for doc, count in counts.items() if count > 1] # Extract texts from all PDF files in this folder pdf_texts = {} pdf_scanned_warnings = {} for pdf in pdf_files: pdf_path = os.path.join(folder_path, pdf) pages = extract_pdf_pages_text(pdf_path) # Check if PDF is likely scanned (empty or very short text) full_text = "".join(pages).strip() if len(full_text) < 50 and len(pdf_files) > 0: # Try OCR if possible ocr_pages, ocr_err = run_ocr_on_pdf(pdf_path) if ocr_pages and "".join(ocr_pages).strip(): pages = ocr_pages pdf_scanned_warnings[pdf] = "Processed using OCR." else: pdf_scanned_warnings[pdf] = f"Scanned/empty text. OCR could not run: {ocr_err if ocr_err else 'No text found'}" pdf_texts[pdf] = pages # Perform matching # Map Excel document -> list of matches: (pdf_filename, match_type, page_number, confidence) doc_matches = {doc: [] for doc in set(excel_records)} pdf_matched_docs = {pdf: set() for pdf in pdf_files} for doc in set(excel_records): variants = get_doc_no_variants(doc) # A. Try filename matching first for pdf in pdf_files: pdf_name_no_ext = os.path.splitext(pdf)[0] norm_pdf = normalize_doc_no(pdf_name_no_ext) matched_via_file = False for var in variants: # If variant is long, substring check is safe. If short, exact match only. if len(var) >= 5 and var in norm_pdf: doc_matches[doc].append((pdf, "Filename partial match", 1, "High")) pdf_matched_docs[pdf].add(doc) matched_via_file = True break elif len(var) < 5 and var == norm_pdf: doc_matches[doc].append((pdf, "Filename exact match", 1, "High")) pdf_matched_docs[pdf].add(doc) matched_via_file = True break if matched_via_file: continue # B. Try content matching page-by-page for pdf, pages in pdf_texts.items(): for page_idx, page_text in enumerate(pages): if not page_text.strip(): continue # Strip particulars block to avoid matching cross-references in table lower_text = page_text.lower() first_idx = len(lower_text) for kw in ["particulars", "description", "details"]: idx = lower_text.find(kw) if idx != -1 and idx < first_idx: first_idx = idx search_text = page_text[:first_idx] # Check for exact string in cleaned text if doc.lower() in search_text.lower(): match_type = "Exact string in content" conf = "High" # If PDF is scanned and matched via OCR if pdf in pdf_scanned_warnings and "OCR" in pdf_scanned_warnings[pdf]: conf = "Medium (OCR)" doc_matches[doc].append((pdf, match_type, page_idx + 1, conf)) pdf_matched_docs[pdf].add(doc) continue # Check for normalized variants in normalized cleaned text norm_search = normalize_doc_no(search_text) matched_var = False for var in variants: if len(var) >= 5 and var in norm_search: match_type = "Normalized variant in content" conf = "High" if pdf in pdf_scanned_warnings and "OCR" in pdf_scanned_warnings[pdf]: conf = "Medium (OCR)" doc_matches[doc].append((pdf, match_type, page_idx + 1, conf)) pdf_matched_docs[pdf].add(doc) matched_var = True break if matched_var: continue # Filter out cross-reference false positives (e.g. Sales Invoice number in Credit Note PDF, or Credit Note number in Sales Invoice PDF) has_credit_pdf = any(any(x in f.lower() for x in ["credit", "crn"]) for f in pdf_files) has_invoice_pdf = any(not any(x in f.lower() for x in ["credit", "crn"]) for f in pdf_files) if has_credit_pdf and has_invoice_pdf: filtered_doc_matches = {} pdf_matched_docs = {pdf: set() for pdf in pdf_files} for doc, matches in doc_matches.items(): doc_upper = str(doc).upper() is_doc_credit_note = "CRN" in doc_upper valid_matches = [] for match in matches: pdf = match[0] pdf_lower = pdf.lower() is_pdf_credit_note = any(x in pdf_lower for x in ["credit", "crn"]) if is_doc_credit_note: if is_pdf_credit_note: valid_matches.append(match) pdf_matched_docs[pdf].add(doc) else: if not is_pdf_credit_note: valid_matches.append(match) pdf_matched_docs[pdf].add(doc) filtered_doc_matches[doc] = valid_matches doc_matches = filtered_doc_matches # Compile findings matched_records_set = set() missing_records = [] duplicate_pdf_matches = [] # Excel records found in multiple PDFs or multiple times for doc in excel_records: matches = doc_matches.get(doc, []) if matches: matched_records_set.add(doc) # If doc is matched to multiple places (different files or different pages) unique_places = {(pdf, page) for pdf, type_, page, conf in matches} if len(unique_places) > 1: duplicate_pdf_matches.append(doc) else: missing_records.append(doc) # Extra PDFs: PDFs that didn't match any Excel record extra_pdfs = [pdf for pdf, matched in pdf_matched_docs.items() if len(matched) == 0] # Deduplicate lists for reporting missing_records = sorted(list(set(missing_records))) duplicate_pdf_matches = sorted(list(set(duplicate_pdf_matches))) excel_duplicates = sorted(list(set(excel_duplicates))) result = { "folder": rel_folder_name if rel_folder_name else os.path.basename(folder_path), "excel_file": excel_file if excel_file else "None", "total_excel": len(excel_records), "total_pdfs": len(pdf_files), "matched_count": len(matched_records_set), "missing_count": len(missing_records), "extra_count": len(extra_pdfs), "duplicate_excel_count": len(excel_duplicates), "missing_details": missing_records, "extra_details": extra_pdfs, "duplicate_excel_details": excel_duplicates, "duplicate_pdf_details": duplicate_pdf_matches, "all_matches": doc_matches, "warnings": pdf_scanned_warnings, "detected_col": doc_col if doc_col else "N/A" } return result def format_folder_report(res): """Format the reconciliation results of a folder as a string.""" report = [] report.append(f"Folder: {res['folder']}") report.append(f"Excel File: {res['excel_file']}") report.append(f"Total Excel Records: {res['total_excel']}") report.append(f"Total PDFs: {res['total_pdfs']}") report.append(f"Matched Records: {res['matched_count']}") report.append(f"Missing PDFs: {res['missing_count']}") report.append(f"Extra PDFs: {res['extra_count']}") report.append("") # Warnings (like OCR or scanned) if res['warnings']: report.append("PDF Processing Warnings:") for pdf, msg in res['warnings'].items(): report.append(f" - {pdf}: {msg}") report.append("") # Missing PDF Details report.append("Missing PDF Details:") if res['missing_details']: for doc in res['missing_details']: report.append(f"- Document No: {doc}") else: report.append("- None") report.append("") # Extra PDF Details report.append("Extra PDF Details:") if res['extra_details']: for pdf in res['extra_details']: report.append(f"- PDF Name: {pdf}") else: report.append("- None") report.append("") # Duplicate PDF Matches report.append("Duplicate PDF Match Details (Same Doc matched multiple PDFs/pages):") if res['duplicate_pdf_details']: for doc in res['duplicate_pdf_details']: matches = res['all_matches'][doc] match_str = ", ".join([f"{pdf} (Page {page})" for pdf, type_, page, conf in matches]) report.append(f"- Document No: {doc} found in: {match_str}") else: report.append("- None") report.append("\n" + "="*80 + "\n") return "\n".join(report) def main(): root_dir = "." if len(sys.argv) > 1: root_dir = sys.argv[1] print(f"Starting reconciliation analysis in: {os.path.abspath(root_dir)}") folders_to_process = [] for root, dirs, files in os.walk(root_dir): # Exclude system and special directories dirs[:] = [d for d in dirs if d not in ('.git', '.gemini', '.omd', 'brain', 'Filtered_Files', 'scratch')] # Check if there is an Excel file or PDF in this directory has_excel = any(f.endswith(('.xlsx', '.xls')) and not f.startswith('~$') for f in files) has_pdf = any(f.endswith('.pdf') for f in files) if has_excel or has_pdf: folders_to_process.append(root) folders_to_process = sorted(folders_to_process) print(f"Found {len(folders_to_process)} folders to process.\n") results = [] full_report_text = [] # Add assumptions and methodology full_report_text.append("="*80) full_report_text.append("RECONCILIATION VERIFICATION REPORT") full_report_text.append(f"Generated on: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}") full_report_text.append(f"Root Directory: {os.path.abspath(root_dir)}") full_report_text.append("="*80 + "\n") full_report_text.append("METHODOLOGY & ASSUMPTIONS:") full_report_text.append("1. Column Auto-Detection:") full_report_text.append(" - Excel header rows are identified by scanning the first 15 rows for header-like keywords.") full_report_text.append(" - The reference column is identified using key terms ('voucher no', 'invoice no', 'document no', etc.).") full_report_text.append(" - If no keyword matches, the column with the highest density of alphanumeric identifiers is used.") full_report_text.append("2. Clean Record Selection:") full_report_text.append(" - Empty rows and summary/total rows (containing 'total', 'outstanding', 'balance', etc.) are excluded.") full_report_text.append("3. Match Search Logic:") full_report_text.append(" - Exact String Matching: Searching for the exact Excel identifier inside the PDF's extracted text.") full_report_text.append(" - Normalized Matching: Both the PDF content/filename and Excel identifiers are normalized (removing non-alphanumeric chars and spaces) before comparison.") full_report_text.append(" - Suffix Matching: Splitting identifiers by symbols (-, /, _) and searching for unique suffixes of length >= 4.") full_report_text.append("4. Confidence Levels:") full_report_text.append(" - HIGH: Exact match found in filename or in PDF extracted text.") full_report_text.append(" - MEDIUM: Match found using normalized/suffix comparison, or matched via OCR text.") full_report_text.append(" - LOW: Ambiguous match or match with short characters (skipped to avoid false positives).") full_report_text.append("5. Scanned PDFs & OCR:") full_report_text.append(" - PDFs with extracted text length < 50 chars are treated as scanned. OCR (via pytesseract) is run if available.") full_report_text.append(" - If OCR is not configured/installed, matching falls back to filename-only and warnings are logged.") full_report_text.append("\n" + "="*80 + "\n") for idx, folder in enumerate(folders_to_process, 1): rel_folder = os.path.relpath(folder, root_dir) print(f"[{idx}/{len(folders_to_process)}] Analyzing: {rel_folder}...") res = analyze_folder(folder, rel_folder) results.append(res) has_mismatch = ( res['missing_count'] > 0 or res['extra_count'] > 0 or len(res['duplicate_pdf_details']) > 0 ) if has_mismatch: full_report_text.append(format_folder_report(res)) # Generate Master Summary Table full_report_text.append("="*80) full_report_text.append("MASTER RECONCILIATION SUMMARY (MISMATCHED FOLDERS ONLY)") full_report_text.append("="*80) # Table headers headers = [ "Folder Name", "Excel File", "Total Excel", "Total PDFs", "Matched", "Missing", "Extra", "Detected Col" ] col_widths = [35, 30, 12, 11, 8, 8, 7, 15] header_row = "".join(f"{h:<{w}}" for h, w in zip(headers, col_widths)) full_report_text.append(header_row) full_report_text.append("-" * sum(col_widths)) for res in results: has_mismatch = ( res['missing_count'] > 0 or res['extra_count'] > 0 or len(res['duplicate_pdf_details']) > 0 ) if not has_mismatch: continue f_name = res['folder'] if len(f_name) > 32: f_name = f_name[:29] + "..." e_file = res['excel_file'] if len(e_file) > 27: e_file = e_file[:24] + "..." row_str = ( f"{f_name:<35}" f"{e_file:<30}" f"{res['total_excel']:<12}" f"{res['total_pdfs']:<11}" f"{res['matched_count']:<8}" f"{res['missing_count']:<8}" f"{res['extra_count']:<7}" f"{res['detected_col']:<15}" ) full_report_text.append(row_str) full_report_text.append("-" * sum(col_widths)) # Totals row tot_excel = sum(r['total_excel'] for r in results) tot_pdfs = sum(r['total_pdfs'] for r in results) tot_matched = sum(r['matched_count'] for r in results) tot_missing = sum(r['missing_count'] for r in results) tot_extra = sum(r['extra_count'] for r in results) totals_row = ( f"{'TOTALS':<35}" f"{'':<30}" f"{tot_excel:<12}" f"{tot_pdfs:<11}" f"{tot_matched:<8}" f"{tot_missing:<8}" f"{tot_extra:<7}" f"{'':<15}" ) full_report_text.append(totals_row) full_report_text.append("="*80) # Write to file report_content = "\n".join(full_report_text) report_file_path = os.path.join(root_dir, "reconciliation_report.txt") with open(report_file_path, "w", encoding="utf-8") as f: f.write(report_content) print(f"\nAnalysis complete! Verification report saved to: {os.path.abspath(report_file_path)}") if __name__ == "__main__": main()