Upload 8 files

anthropic_api_invoice_extractor.py

@@ -0,0 +1,238 @@
+import anthropic
+import os
+from remittance_pdf_processing_types import InvoiceNumbers, PaymentAmount
+from remittance_pdf_processing_utils import remittance_logger, remove_duplicate_lists
+from anthropic.types import ContentBlock, ImageBlockParam
+
+
+def extract_invoice_numbers_with_anthropic_ai(base64_images: list[str], multi_hop: bool = False) -> list[InvoiceNumbers]:
+ """
+ Extracts invoice numbers from one or more base64-encoded images using Anthropic's Claude 3.5 Sonnet model.
+ 
+ Args:
+ base64_images (list[str]): A list of base64-encoded image strings.
+ multi_hop (bool): Whether to use multi-hop processing.
+ 
+ Returns:
+ list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
+ """
+ if multi_hop:
+ return extract_invoice_numbers_with_anthropic_ai_multi_hop(base64_images)
+ else:
+ return extract_invoice_numbers_with_anthropic_ai_single_hop(base64_images)
+
+def extract_invoice_numbers_with_anthropic_ai_single_hop(base64_images: list[str]) -> list[InvoiceNumbers]:
+ client = anthropic.Anthropic(
+ api_key=os.environ.get("ANTHROPIC_API_KEY"),
+ )
+
+ content: list[ContentBlock] = [
+ {
+ "type": "image",
+ "source": {
+ "type": "base64",
+ "media_type": "image/png",
+ "data": image
+ }
+ } for image in base64_images
+ ]
+
+ message = client.messages.create(
+ model="claude-3-5-sonnet-20240620",
+ max_tokens=1024,
+ temperature=0,
+ system="Given the remittance letter images, extract all invoice numbers. Respond with a comma-separated list only.",
+ messages=[
+ {
+ "role": "user",
+ "content": content
+ }
+ ]
+ )
+
+ remittance_logger.debug(f'Anthropic (raw) response: {message.content}')
+
+ invoice_numbers = parse_anthropic_response(message.content[0].text)
+ return [invoice_numbers]
+
+def extract_invoice_numbers_with_anthropic_ai_multi_hop(base64_images: list[str]) -> list[InvoiceNumbers]:
+ # First hop: Extract column headers
+ column_headers = extract_column_headers_from_images(base64_images)
+ remittance_logger.debug(f"Extracted column headers: {column_headers}")
+
+ # Second hop: Extract invoice numbers for each column (up to 3 columns)
+ all_invoice_numbers = []
+ for column_name in column_headers[:3]:
+ invoice_numbers = extract_invoice_numbers_for_column_from_images(base64_images, column_name)
+ remittance_logger.debug(f"Extracted invoice numbers for column '{column_name}': {invoice_numbers}")
+ if invoice_numbers: # Only add non-empty lists
+ all_invoice_numbers.append(invoice_numbers)
+
+ # Remove duplicate lists using the utility function
+ unique_invoice_numbers = remove_duplicate_lists(all_invoice_numbers)
+ return unique_invoice_numbers
+
+def extract_column_headers_from_images(base64_images: list[str]) -> list[str]:
+ client = anthropic.Anthropic(
+ api_key=os.environ.get("ANTHROPIC_API_KEY"),
+ )
+
+ content: list[ContentBlock] = [
+ {
+ "type": "image",
+ "source": {
+ "type": "base64",
+ "media_type": "image/png",
+ "data": image
+ }
+ } for image in base64_images
+ ]
+ 
+ message = client.messages.create(
+ model="claude-3-5-sonnet-20240620",
+ max_tokens=1024,
+ temperature=0,
+ system="Given the remittance letter images, extract all column header names that could contain invoice numbers. Respond with a comma-separated list only.",
+ messages=[
+ {
+ "role": "user",
+ "content": content
+ }
+ ]
+ )
+
+ remittance_logger.debug(f'Anthropic (raw) response for column headers: {message.content}')
+
+ return parse_anthropic_response(message.content[0].text)
+
+def extract_invoice_numbers_for_column_from_images(base64_images: list[str], column_name: str) -> InvoiceNumbers:
+ client = anthropic.Anthropic(
+ api_key=os.environ.get("ANTHROPIC_API_KEY"),
+ )
+
+ content: list[ContentBlock] = [
+ {
+ "type": "image",
+ "source": {
+ "type": "base64",
+ "media_type": "image/png",
+ "data": image
+ }
+ } for image in base64_images
+ ]
+
+ message = client.messages.create(
+ model="claude-3-5-sonnet-20240620",
+ max_tokens=1024,
+ temperature=0,
+ system=f"Given the remittance letter images, extract all invoice numbers from the column '{column_name}'. Respond with a comma-separated list only.",
+ messages=[
+ {
+ "role": "user",
+ "content": content
+ }
+ ]
+ )
+
+ remittance_logger.debug(f'Anthropic (raw) response for invoice numbers in column {column_name}: {message.content}')
+
+ return parse_anthropic_response(message.content[0].text)
+
+def parse_anthropic_response(response: str) -> list[str]:
+ """
+ Parses the response from Claude 3.5 Sonnet model and extracts a list of items.
+ 
+ Args:
+ response (str): The response string from Claude 3.5 Sonnet model.
+ 
+ Returns:
+ list[str]: A list of extracted items (invoice numbers or column headers).
+ """
+ return [item.strip() for item in response.split(',') if item.strip()]
+
+def extract_invoice_numbers_from_single_base64_image(base64_image: str, multi_hop: bool = False) -> list[InvoiceNumbers]:
+ """
+ Extracts invoice numbers from a single base64-encoded image using Anthropic's Claude 3.5 Sonnet model.
+ 
+ Args:
+ base64_image (str): The base64-encoded image string.
+ multi_hop (bool): Whether to use multi-hop processing.
+ 
+ Returns:
+ list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
+ """
+ return extract_invoice_numbers_with_anthropic_ai([base64_image], multi_hop)
+
+def extract_invoice_numbers_from_multi_page_images(base64_images: list[str], multi_hop: bool = False) -> list[InvoiceNumbers]:
+ """
+ Extracts invoice numbers from multiple base64-encoded images using Anthropic's Claude 3.5 Sonnet model.
+ 
+ Args:
+ base64_images (list[str]): A list of base64-encoded image strings.
+ multi_hop (bool): Whether to use multi-hop processing.
+ 
+ Returns:
+ list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
+ """
+ return extract_invoice_numbers_with_anthropic_ai(base64_images, multi_hop)
+
+
+def extract_payment_amounts_with_anthropic_ai(base64_images: list[str]) -> list[PaymentAmount]:
+ """
+ Extracts payment amounts from one or more base64-encoded images using Anthropic's Claude 3.5 Sonnet model.
+ 
+ Args:
+ base64_images (list[str]): A list of base64-encoded image strings.
+ 
+ Returns:
+ list[PaymentAmount]: A list containing extracted payment amounts.
+ """
+ client = anthropic.Anthropic(
+ api_key=os.environ.get("ANTHROPIC_API_KEY"),
+ )
+ 
+ # Prepare the message content
+ content = []
+ for image in base64_images:
+ content.append({
+ "type": "image",
+ "source": {
+ "type": "base64",
+ "media_type": "image/png",
+ "data": image
+ }
+ })
+
+ # Call the Anthropic API
+ message = client.messages.create(
+ model="claude-3-5-sonnet-20240620",
+ max_tokens=1024,
+ temperature=0,
+ system="You are a precise payment amount extractor. Given remittance letter images, extract only the total payment amount. Respond with the numerical amount only, including any decimal places and currency symbols if present. Do not include any additional text or explanations.",
+ messages=[
+ {
+ "role": "user",
+ "content": content
+ }
+ ]
+ )
+
+ remittance_logger.debug(f'Anthropic (raw) response for payment amount: {message.content}')
+
+ # assert(isinstance(message.content, anthropic.TextBlock))
+ # Parse the response
+ payment_amount = parse_anthropic_payment_response(message.content[0].text)
+ return payment_amount
+
+def parse_anthropic_payment_response(response: str) -> list[PaymentAmount]:
+ """
+ Parses the response from Claude 3.5 Sonnet model and extracts the payment amount.
+ 
+ Args:
+ response (str): The response string from Claude 3.5 Sonnet model.
+ 
+ Returns:
+ list[PaymentAmount]: A list containing the extracted payment amount.
+ """
+ # Strip whitespace and return as a single-item list
+ return [response.strip()]

app.py

@@ -0,0 +1,79 @@
+import os
+import tempfile
+import gradio as gr
+from gradio_pdf import PDF
+from remittance_pdf_processor import process_pdf_with_flow
+from remittance_pdf_processing_types import Candidate, InvoiceNumbers, PaymentAmount, ProcessedPDFResult
+from remittance_pdf_processing_utils import remittance_logger, format_amount_str_to_decimal
+
+
+def custom_invoice_numbers_verifier(invoice_numbers: InvoiceNumbers, ground_truth_invoices: str) -> InvoiceNumbers:
+ ground_truth_set = set(map(str.lower, map(str.strip, ground_truth_invoices.split(','))))
+ return [num for num in invoice_numbers if num.lower() in ground_truth_set] 
+
+def custom_invoice_and_amount_verifier(invoice_numbers: InvoiceNumbers, amount: PaymentAmount, ground_truth_invoices: str, ground_truth_amount: str) -> bool:
+ return format_amount_str_to_decimal(amount) == format_amount_str_to_decimal(ground_truth_amount)
+
+def process_pdf_file(pdf_file_path: str, ground_truth_invoices: str, ground_truth_amount: str):
+ try: 
+ invoice_verifier = lambda x: custom_invoice_numbers_verifier(x, ground_truth_invoices)
+ invoice_and_amount_verifier = lambda x, y: custom_invoice_and_amount_verifier(x, y, ground_truth_invoices, ground_truth_amount)
+
+ extracted_result: ProcessedPDFResult = process_pdf_with_flow(
+ pdf_file_path, 
+ invoice_verifier=invoice_verifier,
+ invoice_and_amount_verifier=invoice_and_amount_verifier
+ )
+ 
+ verified_candidate, unverified_candidate = extracted_result
+
+ result = ""
+ result += f"Extracted invoice numbers and payment amounts:\n"
+ result += f"Ground Truth Invoices: {ground_truth_invoices}\n"
+ result += f"Ground Truth Amount: {ground_truth_amount}\n"
+ result += "-" * 40 + "\n"
+
+ # Print Invoice Numbers
+ result += "Invoice Numbers:\n"
+ for i, invoice_numbers in enumerate(unverified_candidate[0], 1):
+ status = "Verified" if any(set(invoice_numbers) == set(v) for v in verified_candidate[0]) else "Unverified"
+ result += f" Candidate {i} ({status}): {', '.join(invoice_numbers)}\n"
+ result += "\n"
+
+ # Print Payment Amounts
+ result += "Payment Amounts:\n"
+ for i, amount in enumerate(unverified_candidate[1], 1):
+ status = "Verified" if amount in verified_candidate[1] else "Unverified"
+ result += f" Candidate {i} ({status}): {amount}\n"
+
+ return result
+ except Exception as e:
+ error_message = f"An error occurred while processing the PDF: {str(e)}"
+ remittance_logger.error(error_message, exc_info=True)
+ return error_message
+
+# Create the Gradio interface
+with gr.Blocks() as iface:
+ gr.Markdown("# Remittance PDF Processor")
+ gr.Markdown("Upload a PDF file to extract invoice numbers and payment amounts. Provide ground truth data for comparison.")
+ 
+ with gr.Row():
+ with gr.Column(scale=1):
+ pdf_input = PDF(label="Upload PDF", visible=True)
+ 
+ with gr.Column(scale=1):
+ ground_truth_invoices = gr.Textbox(label="Ground Truth Invoice Numbers (comma-separated)")
+ ground_truth_amount = gr.Textbox(label="Ground Truth Payment Amount")
+ 
+ submit_button = gr.Button("Process PDF")
+ 
+ output = gr.Textbox(label="Processing Results", lines=20)
+ 
+ submit_button.click(
+ process_pdf_file,
+ inputs=[pdf_input, ground_truth_invoices, ground_truth_amount],
+ outputs=[output]
+ )
+
+# Launch the interface
+iface.launch()

remittance_pdf_processing_cli.py

@@ -0,0 +1,96 @@
+# File: remittance_pdf_processing_cli.py
+
+import argparse
+from remittance_pdf_processor import process_pdf, extract_text_from_pdf
+from ccmt_verifier_aus import verify_invoice_numbers
+from remittance_pdf_processing_types import Candidate, ProcessedPDFResult
+
+def main():
+ parser = argparse.ArgumentParser(description="Extract text and invoice numbers from a PDF file.")
+ parser.add_argument("pdf_file", help="Path to the PDF file")
+ parser.add_argument("-o", "--output", help="Output file path for extracted text (optional)")
+ parser.add_argument("-v", "--verbose", action="store_true", help="Print extracted text from PDF")
+ parser.add_argument("-f", "--force-image", action="store_true", help="Force processing PDF as image")
+ parser.add_argument("--verify", action="store_true", help="Enable invoice number verification")
+ parser.add_argument("--multi-hop", action="store_true", help="Force multi-hop processing")
+
+ args = parser.parse_args()
+
+ extracted_text = extract_text_from_pdf(args.pdf_file, wrap_pages=True)
+ 
+ # Determine the invoice verifier function
+ invoice_verifier = verify_invoice_numbers if args.verify else None
+
+ extracted_result: ProcessedPDFResult = process_pdf(args.pdf_file, 
+ force_image_processing=args.force_image,
+ force_multi_hop=args.multi_hop,
+ invoice_verifier=invoice_verifier)
+
+ verified_candidate, unverified_candidate = extracted_result
+
+ if args.verbose:
+ print("Extracted text from PDF:")
+ print("-" * 40)
+ print(extracted_text)
+ print("-" * 40)
+ print()
+
+ print("Extracted invoice numbers and payment amounts:")
+ print(f"Multi-hop processing: {'Enabled' if args.multi_hop else 'Disabled'}")
+ print(f"Verification: {'Enabled' if args.verify else 'Disabled'}")
+ print("-" * 40)
+ 
+ print("Verified Results:")
+ if len(verified_candidate[0]) == 1 and len(verified_candidate[1]) == 1:
+ print(f" Verified Invoice Numbers: {', '.join(verified_candidate[0][0])}")
+ print(f" Verified Amount: {verified_candidate[1][0]}")
+ elif len(verified_candidate[0]) == 1 and len(verified_candidate[1]) == 0:
+ print(f" Verified Invoice Numbers: {', '.join(verified_candidate[0][0])}")
+ print(" Amount doesn't match. This might be a partial payment.")
+ elif len(verified_candidate[0]) >= 2:
+ print(" Verified Invoice Numbers:")
+ for i, invoice_numbers in enumerate(verified_candidate[0], 1):
+ print(f" List {i}: {', '.join(invoice_numbers)}")
+ else:
+ print(" No verified results.")
+ 
+ print("\nUnverified Invoice Numbers:")
+ for i, invoice_numbers in enumerate(unverified_candidate[0], 1):
+ print(f" Candidate {i}: {', '.join(invoice_numbers)}")
+ 
+ print("\nUnverified Payment Amounts:")
+ for i, amount in enumerate(unverified_candidate[1], 1):
+ print(f" Candidate {i}: {amount}")
+ 
+ print("-" * 40)
+
+ if args.output:
+ with open(args.output, 'w', encoding='utf-8') as f:
+ f.write("Extracted text:\n")
+ f.write(extracted_text)
+ f.write("\n\nExtracted invoice numbers and payment amounts:\n")
+ f.write(f"Multi-hop processing: {'Enabled' if args.multi_hop else 'Disabled'}\n")
+ f.write(f"Verification: {'Enabled' if args.verify else 'Disabled'}\n")
+ f.write("Verified Results:\n")
+ if len(verified_candidate[0]) == 1 and len(verified_candidate[1]) == 1:
+ f.write(f" Verified Invoice Numbers: {', '.join(verified_candidate[0][0])}\n")
+ f.write(f" Verified Amount: {verified_candidate[1][0]}\n")
+ elif len(verified_candidate[0]) == 1 and len(verified_candidate[1]) == 0:
+ f.write(f" Verified Invoice Numbers: {', '.join(verified_candidate[0][0])}\n")
+ f.write(" Amount doesn't match. This might be a partial payment.\n")
+ elif len(verified_candidate[0]) >= 2:
+ f.write(" Verified Invoice Numbers:\n")
+ for i, invoice_numbers in enumerate(verified_candidate[0], 1):
+ f.write(f" List {i}: {', '.join(invoice_numbers)}\n")
+ else:
+ f.write(" No verified results.\n")
+ f.write("\nUnverified Invoice Numbers:\n")
+ for i, invoice_numbers in enumerate(unverified_candidate[0], 1):
+ f.write(f" Candidate {i}: {', '.join(invoice_numbers)}\n")
+ f.write("\nUnverified Payment Amounts:\n")
+ for i, amount in enumerate(unverified_candidate[1], 1):
+ f.write(f" Candidate {i}: {amount}\n")
+ print(f"Extracted text and invoice numbers written to {args.output}")
+
+if __name__ == "__main__":
+ main()

remittance_pdf_processing_types.py

@@ -0,0 +1,24 @@
+from typing import Callable, Optional, Literal
+
+# Type alias for invoice numbers
+InvoiceNumbers = list[str]
+
+# Type alias for invoice verifier function
+InvoiceVerifier = Callable[[InvoiceNumbers], InvoiceNumbers]
+
+# Type alias for document type
+DocumentType = Literal['single', 'multi']
+
+# Type alias for extractor function
+ExtractorFunction = Callable[[str], list[InvoiceNumbers]]
+
+# Add any other shared type definitions related to remittance PDF processing here
+PaymentAmount = str
+
+Candidate = tuple[list[InvoiceNumbers], list[PaymentAmount]]
+
+# Add this line to remittance_pdf_processing_types.py
+ProcessedPDFResult = tuple[Candidate, Candidate]
+
+# New type alias for invoice list and amount verifier function
+InvoiceListAndAmountVerifier = Callable[[InvoiceNumbers, PaymentAmount], bool]

remittance_pdf_processing_utils.py

@@ -0,0 +1,92 @@
+import logging
+
+def setup_logger():
+ # Create a logger
+ logger = logging.getLogger('remittance_processing')
+ logger.setLevel(logging.DEBUG)
+
+ # Create handlers
+ c_handler = logging.StreamHandler()
+ f_handler = logging.FileHandler('remittance_processing.log')
+ c_handler.setLevel(logging.INFO)
+ f_handler.setLevel(logging.DEBUG)
+
+ # Create formatters and add it to handlers
+ log_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+ c_handler.setFormatter(log_format)
+ f_handler.setFormatter(log_format)
+
+ # Add handlers to the logger
+ logger.addHandler(c_handler)
+ logger.addHandler(f_handler)
+
+ return logger
+
+remittance_logger = setup_logger()
+
+def remove_duplicate_lists(lists):
+ """
+ Remove duplicate lists from a list of lists.
+ Args:
+ lists:
+ a list of lists of strings
+ Returns:
+ a list of lists of strings, where each list is unique
+ """
+ seen = set()
+ unique_lists = []
+
+ for lst in lists:
+ sorted_list = tuple(sorted(lst))
+ if sorted_list not in seen:
+ seen.add(sorted_list)
+ unique_lists.append(lst)
+
+ return unique_lists
+
+
+import re
+from decimal import Decimal, ROUND_HALF_UP
+
+def format_amount_str_to_decimal(amount_str: str) -> str:
+ def standardize_number(s):
+ # Find the last occurrence of a comma or period
+ last_separator_index = max(s.rfind(','), s.rfind('.'))
+ if last_separator_index != -1:
+ # Split the string into two parts
+ before_separator = s[:last_separator_index]
+ after_separator = s[last_separator_index+1:]
+
+ # Clean the first part of any commas, periods, or whitespace
+ before_separator_cleaned = re.sub(r'[.,\s]', '', before_separator)
+
+ # Ensure the decimal part starts with a period, even if it was a comma
+ standardized_s = before_separator_cleaned + '.' + after_separator
+ else:
+ # If there's no separator, just remove commas, periods, or whitespace
+ standardized_s = re.sub(r'[.,\s]', '', s)
+
+ return standardized_s
+
+ def remove_chars_after_last_digit(s):
+ # Remove any non-digit characters following the last digit in the string
+ return re.sub(r'(?<=\d)[^\d]*$', '', s)
+
+ def clean_text(s):
+ # This pattern looks for:
+ # - Optional non-digit or non-negative sign characters followed by whitespace (if any)
+ # - Followed by any characters until a digit is found in the word
+ # It then replaces this matched portion with the remaining part of the word from the first digit
+ cleaned_s = re.sub(r'[^\d-]*\s?(\S*\d\S*)', r'\1', s)
+ return cleaned_s
+
+ # Run functions to format a text decimal
+ cleaned_amount = clean_text(remove_chars_after_last_digit(standardize_number(amount_str.strip().lower())))
+ 
+ # Convert to Decimal and round to 2 decimal places
+ try:
+ decimal_amount = Decimal(cleaned_amount).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)
+ return f"{decimal_amount:.2f}"
+ except:
+ # If conversion fails, return the cleaned string as is
+ return cleaned_amount

remittance_pdf_processor.py

@@ -0,0 +1,329 @@
+import pdfplumber
+from typing import Optional, Callable, Literal
+import base64
+import io
+from PIL import Image
+from remittance_pdf_processing_utils import remittance_logger, format_amount_str_to_decimal
+from vertex_api_invoice_extractor import extract_invoice_numbers_with_vertex_ai, extract_invoice_numbers_from_text_with_vertex_ai, extract_payment_amounts_with_vertex_ai, extract_payment_amounts_from_text_with_vertex_ai
+# from dspy_invoice_extractors import SinglePageInvoiceExtractor, MultiPageInvoiceExtractor
+from remittance_pdf_processing_types import InvoiceNumbers, InvoiceVerifier, DocumentType, ExtractorFunction, PaymentAmount, Candidate, ProcessedPDFResult, InvoiceListAndAmountVerifier
+from anthropic_api_invoice_extractor import extract_invoice_numbers_with_anthropic_ai, extract_payment_amounts_with_anthropic_ai
+
+def is_text_based_pdf(pdf: pdfplumber.PDF) -> bool:
+ text_threshold = 100 # Minimum number of characters to consider it text-based
+ for page in pdf.pages:
+ if len(page.extract_text()) > text_threshold:
+ return True
+ return False
+
+def determine_document_type(pdf: pdfplumber.PDF) -> DocumentType:
+ return 'single' if len(pdf.pages) == 1 else 'multi'
+
+def extract_text_from_pdf(pdf_path: str, wrap_pages: bool = False) -> str:
+ with pdfplumber.open(pdf_path) as pdf:
+ if not wrap_pages:
+ # Keep the current behavior
+ return "\n".join(page.extract_text() for page in pdf.pages)
+ else:
+ # Implement new wrapping behavior
+ pages_text = []
+ for i, page in enumerate(pdf.pages, start=1):
+ page_text = page.extract_text()
+ wrapped_page = f"<page_{i}>\n{page_text}\n</page_{i}>"
+ pages_text.append(wrapped_page)
+ 
+ all_pages_text = "\n".join(pages_text)
+ return f"<remittance>\n{all_pages_text}\n</remittance>"
+
+# def InvoiceExtractor(doc_type: DocumentType) -> ExtractorFunction:
+# if doc_type == 'single':
+# def single_page_extractor(text: str) -> list[InvoiceNumbers]:
+# return []
+# return single_page_extractor
+# else:
+# def multi_page_extractor(text: str) -> list[InvoiceNumbers]:
+# return []
+# return multi_page_extractor
+
+def extract_invoice_numbers_from_text(
+ text: str,
+ doc_type: DocumentType,
+ multi_hop: bool = False
+) -> list[InvoiceNumbers]:
+ remittance_logger.info(f"Extracting invoice numbers from {doc_type}-page text-based document (multi_hop: {multi_hop})")
+ 
+ # Call the Vertex AI extractor
+ return extract_invoice_numbers_from_text_with_vertex_ai(text, multi_hop)
+
+def extract_invoice_numbers_from_single_base64_image(base64_image: str, multi_hop: bool = False) -> list[InvoiceNumbers]:
+ remittance_logger.debug(f"Extracting invoice numbers from a single base64 image using Vertex AI (multi_hop: {multi_hop})")
+ return extract_invoice_numbers_with_vertex_ai(base64_image, multi_hop)
+
+def extract_invoice_numbers_from_multi_page_images(base64_images: list[str], multi_hop: bool = False) -> list[InvoiceNumbers]:
+ remittance_logger.debug(f"Extracting invoice numbers from {len(base64_images)} base64 images using Anthropic AI (multi_hop: {multi_hop})")
+ return extract_invoice_numbers_with_anthropic_ai(base64_images, multi_hop)
+
+def extract_invoice_numbers_from_base64_images(base64_images: list[str], multi_hop: bool = False) -> list[InvoiceNumbers]:
+ remittance_logger.info(f"Extracting invoice numbers from {len(base64_images)} base64 image(s) (multi_hop: {multi_hop})")
+ 
+ if len(base64_images) == 1:
+ return extract_invoice_numbers_from_single_base64_image(base64_images[0], multi_hop)
+ else:
+ return extract_invoice_numbers_from_multi_page_images(base64_images, multi_hop)
+
+def extract_invoice_numbers_from_image(
+ pdf: pdfplumber.PDF,
+ multi_hop: bool = False,
+ dpi: int = 257 # Number choosen for optimal resolution for Gemini Flash 1.5 model
+) -> list[InvoiceNumbers]:
+ remittance_logger.info(f"Extracting invoice numbers from {len(pdf.pages)}-page image-based document (multi_hop: {multi_hop})")
+ 
+ base64_images = []
+ for page in pdf.pages:
+ img = page.to_image(resolution=dpi)
+ img_bytes = io.BytesIO()
+ img.save(img_bytes, format='PNG')
+ img_base64 = base64.b64encode(img_bytes.getvalue()).decode('utf-8')
+ base64_images.append(img_base64)
+ 
+ return extract_invoice_numbers_from_base64_images(base64_images, multi_hop)
+
+def extract_invoices_from_pdf(pdf_path: str, force_image_processing: bool = False, invoice_verifier: InvoiceVerifier | None = None, force_multi_hop: bool = False) -> tuple[list[InvoiceNumbers], list[InvoiceNumbers]]:
+ with pdfplumber.open(pdf_path) as pdf:
+ doc_type = determine_document_type(pdf)
+ 
+ for multi_hop in [True] if force_multi_hop else [False, True]:
+ # if doc_type == 'single' or force_image_processing:
+ if force_image_processing:
+ invoice_numbers_candidates = extract_invoice_numbers_from_image(pdf, multi_hop=multi_hop)
+ else:
+ is_text_based = is_text_based_pdf(pdf)
+ if is_text_based:
+ text = extract_text_from_pdf(pdf_path, wrap_pages=True)
+ invoice_numbers_candidates = extract_invoice_numbers_from_text(text, doc_type, multi_hop=multi_hop)
+ else:
+ invoice_numbers_candidates = extract_invoice_numbers_from_image(pdf, multi_hop=multi_hop)
+
+ if invoice_verifier:
+ verified_invoices = [
+ invoice_verifier(invoice_numbers) or []
+ for invoice_numbers in invoice_numbers_candidates
+ ]
+ # Filter out empty lists for verified invoices
+ verified_result = [invoices for invoices in verified_invoices if invoices]
+ else:
+ verified_result = [] # When there's no verifier, the verified list should be empty
+
+ remittance_logger.info(f"Extracted invoice numbers (post verification, multi_hop={multi_hop}): {verified_result}")
+ # If we found invoices (either verified or unverified), return them
+ if verified_result or invoice_numbers_candidates:
+ return verified_result, invoice_numbers_candidates
+ 
+ # If we've tried both with and without multi_hop and found nothing, return empty lists
+ remittance_logger.warning("No invoice numbers found after trying both single-hop and multi-hop processing.")
+ return [], []
+
+def extract_payment_amounts_from_single_base64_image(base64_image: str) -> list[PaymentAmount]:
+ remittance_logger.debug("Extracting payment amounts from a single base64 image using Vertex AI")
+ return extract_payment_amounts_with_vertex_ai(base64_image)
+
+def extract_payment_amounts_from_multi_page_images(base64_images: list[str]) -> list[PaymentAmount]:
+ remittance_logger.debug(f"Extracting payment amounts from {len(base64_images)} base64 images using Anthropic AI")
+ return extract_payment_amounts_with_anthropic_ai(base64_images)
+
+def extract_payment_amounts_from_base64_images(base64_images: list[str]) -> list[PaymentAmount]:
+ remittance_logger.info(f"Extracting payment amounts from {len(base64_images)} base64 image(s)")
+ 
+ if len(base64_images) == 1:
+ return extract_payment_amounts_from_single_base64_image(base64_images[0])
+ else:
+ return extract_payment_amounts_from_multi_page_images(base64_images)
+
+def extract_payment_amounts_from_pdf(pdf_path: str, force_image_processing: bool = False, payment_amount_formatter: Callable[[str], str] | None = None) -> list[PaymentAmount]:
+ with pdfplumber.open(pdf_path) as pdf:
+ doc_type = determine_document_type(pdf)
+
+ if doc_type == 'single' or force_image_processing:
+ payment_amounts = extract_payment_amounts_from_image(pdf)
+ else:
+ is_text_based = is_text_based_pdf(pdf)
+ if is_text_based:
+ text = extract_text_from_pdf(pdf_path, wrap_pages=True)
+ payment_amounts = extract_payment_amounts_from_text(text, doc_type)
+ else:
+ payment_amounts = extract_payment_amounts_from_image(pdf)
+ 
+ if payment_amount_formatter:
+ payment_amounts = [payment_amount_formatter(amount) for amount in payment_amounts]
+ 
+ return payment_amounts
+
+def extract_payment_amounts_from_text(text: str, doc_type: DocumentType) -> list[PaymentAmount]:
+ remittance_logger.info(f"Extracting payment amounts from {doc_type}-page text-based document")
+ 
+ # Call the Vertex AI extractor
+ return extract_payment_amounts_from_text_with_vertex_ai(text)
+
+def extract_payment_amounts_from_image(pdf: pdfplumber.PDF, dpi: int = 257) -> list[PaymentAmount]:
+ remittance_logger.info(f"Extracting payment amounts from {len(pdf.pages)}-page image-based document")
+ 
+ base64_images = []
+ for page in pdf.pages:
+ img = page.to_image(resolution=dpi)
+ img_bytes = io.BytesIO()
+ img.save(img_bytes, format='PNG')
+ img_base64 = base64.b64encode(img_bytes.getvalue()).decode('utf-8')
+ base64_images.append(img_base64)
+ 
+ return extract_payment_amounts_from_base64_images(base64_images) 
+
+
+def process_pdf(pdf_path: str, force_image_processing: bool = False, force_multi_hop: bool = False, invoice_verifier: InvoiceVerifier | None = None, invoice_and_amount_verifier: InvoiceListAndAmountVerifier | None = None) -> ProcessedPDFResult:
+ verified_invoice_numbers, unverified_invoice_numbers = extract_invoices_from_pdf(
+ pdf_path, 
+ force_image_processing, 
+ invoice_verifier, 
+ force_multi_hop=force_multi_hop
+ )
+ payment_amounts = extract_payment_amounts_from_pdf(pdf_path, force_image_processing, payment_amount_formatter=format_amount_str_to_decimal)
+ remittance_logger.debug(f"Extracted payment amounts: {payment_amounts}")
+
+ verified_payment_amounts = [] 
+ if invoice_and_amount_verifier and len(verified_invoice_numbers) == 1:
+ for amount in payment_amounts:
+ if invoice_and_amount_verifier(verified_invoice_numbers[0], amount):
+ verified_payment_amounts = [amount]
+ break
+
+ verified_candidate = (verified_invoice_numbers, verified_payment_amounts)
+ unverified_candidate = (unverified_invoice_numbers, payment_amounts)
+ return verified_candidate, unverified_candidate
+
+# from typing import list, tuple
+
+def process_pdf_with_flow(
+ pdf_path: str,
+ invoice_verifier: InvoiceVerifier | None = None,
+ invoice_and_amount_verifier: InvoiceListAndAmountVerifier | None = None
+) -> ProcessedPDFResult:
+ """
+ Process a PDF file using a specific flow of extraction methods.
+ 
+ Args:
+ pdf_path (str): Path to the PDF file.
+ invoice_verifier (InvoiceVerifier | None): Function to verify invoice numbers.
+ invoice_and_amount_verifier (InvoiceListAndAmountVerifier | None): Function to verify invoice numbers and amount pairs.
+ 
+ Returns:
+ ProcessedPDFResult: A tuple containing verified and unverified candidates.
+ """
+ all_verified_invoices: list[InvoiceNumbers] = []
+ all_verified_amounts: list[PaymentAmount] = []
+ all_unverified_invoices: list[InvoiceNumbers] = []
+ all_unverified_amounts: list[PaymentAmount] = []
+
+ with pdfplumber.open(pdf_path) as pdf:
+ is_text_based = is_text_based_pdf(pdf)
+ 
+ if is_text_based:
+ # Try single hop text processing
+ text = extract_text_from_pdf(pdf_path, wrap_pages=True)
+ result = process_text_based(text, invoice_verifier, invoice_and_amount_verifier, multi_hop=False)
+ if has_single_verified_pair(result):
+ return result
+ accumulate_candidates(result, all_verified_invoices, all_verified_amounts, all_unverified_invoices, all_unverified_amounts)
+ remittance_logger.debug(f"Result snapshot - single hop text processing: {result}")
+
+ # Try multi hop text processing
+ result = process_text_based(text, invoice_verifier, invoice_and_amount_verifier, multi_hop=True)
+ if has_single_verified_pair(result):
+ return result
+ accumulate_candidates(result, all_verified_invoices, all_verified_amounts, all_unverified_invoices, all_unverified_amounts)
+ remittance_logger.debug(f"Result snapshot - multi hop text processing: {result}")
+ 
+ # Try single hop image processing
+ result = process_image_based(pdf, invoice_verifier, invoice_and_amount_verifier, multi_hop=False)
+ if has_single_verified_pair(result):
+ return result
+ accumulate_candidates(result, all_verified_invoices, all_verified_amounts, all_unverified_invoices, all_unverified_amounts)
+ remittance_logger.debug(f"Result snapshot - single hop image processing: {result}")
+ 
+ # Try multi hop image processing
+ result = process_image_based(pdf, invoice_verifier, invoice_and_amount_verifier, multi_hop=True)
+ if has_single_verified_pair(result):
+ return result
+ accumulate_candidates(result, all_verified_invoices, all_verified_amounts, all_unverified_invoices, all_unverified_amounts)
+ remittance_logger.debug(f"Result snapshot - multi hop image processing: {result}")
+ 
+ # If no single verified pair is found, return all accumulated candidates
+ return (all_verified_invoices, all_verified_amounts), (all_unverified_invoices, all_unverified_amounts)
+
+def process_text_based(
+ text: str,
+ invoice_verifier: InvoiceVerifier | None,
+ invoice_and_amount_verifier: InvoiceListAndAmountVerifier | None,
+ multi_hop: bool
+) -> ProcessedPDFResult:
+ invoice_numbers = extract_invoice_numbers_from_text(text, 'multi', multi_hop)
+ payment_amounts = extract_payment_amounts_from_text(text, 'multi')
+ 
+ return verify_candidates(invoice_numbers, payment_amounts, invoice_verifier, invoice_and_amount_verifier)
+
+def process_image_based(
+ pdf: pdfplumber.PDF,
+ invoice_verifier: InvoiceVerifier | None,
+ invoice_and_amount_verifier: InvoiceListAndAmountVerifier | None,
+ multi_hop: bool
+) -> ProcessedPDFResult:
+ invoice_numbers = extract_invoice_numbers_from_image(pdf, multi_hop)
+ payment_amounts = extract_payment_amounts_from_image(pdf)
+ 
+ return verify_candidates(invoice_numbers, payment_amounts, invoice_verifier, invoice_and_amount_verifier)
+
+def verify_candidates(
+ invoice_numbers: list[InvoiceNumbers],
+ payment_amounts: list[PaymentAmount],
+ invoice_verifier: InvoiceVerifier | None,
+ invoice_and_amount_verifier: InvoiceListAndAmountVerifier | None
+) -> ProcessedPDFResult:
+ verified_invoices = []
+ verified_amounts = []
+ 
+ if invoice_verifier:
+ verified_invoices = [invoice_verifier(inv) for inv in invoice_numbers if invoice_verifier(inv)]
+ 
+ if invoice_and_amount_verifier and len(verified_invoices) == 1:
+ for amount in payment_amounts:
+ if invoice_and_amount_verifier(verified_invoices[0], amount):
+ verified_amounts = [amount]
+ break
+ 
+ return (verified_invoices, verified_amounts), (invoice_numbers, payment_amounts)
+
+def has_single_verified_pair(result: ProcessedPDFResult) -> bool:
+ verified, _ = result
+ return len(verified[0]) == 1 and len(verified[1]) == 1
+
+def accumulate_candidates(
+ result: ProcessedPDFResult,
+ all_verified_invoices: list[InvoiceNumbers],
+ all_verified_amounts: list[PaymentAmount],
+ all_unverified_invoices: list[InvoiceNumbers],
+ all_unverified_amounts: list[PaymentAmount]
+) -> None:
+ verified, unverified = result
+ 
+ # Helper function to add unique items to a list
+ def add_unique(items: list, new_items: list) -> None:
+ for item in new_items:
+ if isinstance(item, list): # For invoice numbers
+ if not any(set(item) == set(existing) for existing in items):
+ items.append(item)
+ else: # For payment amounts
+ if item not in items:
+ items.append(item)
+
+ add_unique(all_verified_invoices, verified[0])
+ add_unique(all_verified_amounts, verified[1])
+ add_unique(all_unverified_invoices, unverified[0])
+ add_unique(all_unverified_amounts, unverified[1])

requirements.txt

@@ -0,0 +1,7 @@
+gradio==4.41.0
+gradio-pdf==0.0.12
+pdfplumber==0.11.3
+google-cloud-aiplatform==1.61.0
+google-auth==2.33.0
+google-api-python-client==2.140.0
+anthropic==0.33.1

vertex_api_invoice_extractor.py

@@ -0,0 +1,594 @@
+import base64
+import json
+import os
+from google.oauth2 import service_account
+import vertexai
+from remittance_pdf_processing_utils import remittance_logger
+from vertexai.generative_models import GenerativeModel, Part
+import vertexai.preview.generative_models as generative_models
+from remittance_pdf_processing_types import InvoiceNumbers,PaymentAmount
+from remittance_pdf_processing_utils import remove_duplicate_lists
+
+# Set up authentication
+def initialize_vertexai():
+ # Get the base64-encoded service account JSON from an environment variable
+ encoded_sa_json = os.environ.get('VERTEX_AI_SERVICE_ACCOUNT_JSON')
+ 
+ if not encoded_sa_json:
+ raise ValueError("VERTEX_AI_SERVICE_ACCOUNT_JSON environment variable is not set")
+ 
+ try:
+ # Decode the base64 string to get the JSON content
+ sa_json_str = base64.b64decode(encoded_sa_json).decode('utf-8')
+ sa_info = json.loads(sa_json_str)
+ 
+ # Create credentials object from the decoded JSON
+ credentials = service_account.Credentials.from_service_account_info(
+ sa_info,
+ scopes=['https://www.googleapis.com/auth/cloud-platform']
+ )
+ 
+ # Initialize Vertex AI with the credentials
+ vertexai.init(project="saltech-ai-sandbox", location="us-central1", credentials=credentials)
+ 
+ print("Vertex AI initialized successfully.")
+ except json.JSONDecodeError:
+ raise ValueError("Invalid JSON format in the decoded service account information")
+ except Exception as e:
+ raise Exception(f"Error initializing Vertex AI: {str(e)}")
+
+
+# Call this function at the start of your script or in your main function
+initialize_vertexai()
+
+def extract_invoice_numbers_with_vertex_ai(base64_image: str, multi_hop: bool = False) -> list[InvoiceNumbers]:
+ """
+ Dispatches the invoice number extraction to either single-hop or multi-hop method based on the multi_hop parameter.
+ 
+ Args:
+ base64_image (str): The base64-encoded image string.
+ multi_hop (bool): Whether to use multi-hop processing.
+ 
+ Returns:
+ list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
+ """
+ if multi_hop:
+ return extract_invoice_numbers_with_vertex_ai_multi_hop(base64_image)
+ else:
+ return extract_invoice_numbers_with_vertex_ai_single_hop(base64_image)
+
+def extract_invoice_numbers_with_vertex_ai_single_hop(base64_image: str) -> list[InvoiceNumbers]:
+ """
+ Extracts invoice numbers from a single base64-encoded image using Google's Gemini Flash model with single-hop processing.
+ 
+ Args:
+ base64_image (str): The base64-encoded image string.
+ 
+ Returns:
+ list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
+ """
+ vertexai.init(project="saltech-ai-sandbox", location="us-central1")
+ model = GenerativeModel("gemini-1.5-flash-001")
+
+ image_part = Part.from_data(
+ mime_type="image/png",
+ data=base64.b64decode(base64_image),
+ )
+
+ text_prompt = """Given the remittance letter image, extract all invoice numbers. 
+ Respond with a comma-separated list of invoice numbers only. 
+ If no invoice numbers are found, respond with 'No invoice numbers found'."""
+
+ generation_config = {
+ "max_output_tokens": 8192,
+ "temperature": 0.1,
+ "top_p": 0.95,
+ }
+
+ safety_settings = {
+ generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ }
+
+ responses = model.generate_content(
+ [image_part, text_prompt],
+ generation_config=generation_config,
+ safety_settings=safety_settings,
+ stream=True,
+ )
+
+ full_response = ""
+ for response in responses:
+ full_response += response.text
+
+ remittance_logger.debug(f"Extracted invoice numbers (raw model response): {full_response}")
+
+ extracted_numbers = parse_gemini_response(full_response)
+ return [extracted_numbers] # Wrap in a list to match the expected return type
+
+def extract_column_headers(base64_image: str) -> list[str]:
+ """
+ Extracts column header names that could contain invoice numbers from a base64-encoded image.
+
+ Args:
+ base64_image (str): The base64-encoded image string.
+
+ Returns:
+ list[str]: A list of column header names.
+ """
+ vertexai.init(project="saltech-ai-sandbox", location="us-central1")
+ model = GenerativeModel("gemini-1.5-flash-001")
+
+ image_part = Part.from_data(
+ mime_type="image/png",
+ data=base64.b64decode(base64_image),
+ )
+
+ text_prompt = """Given the remittance letter image, extract all column header names that could contain invoice numbers. 
+ Respond with a comma-separated list only."""
+
+ generation_config = {
+ "max_output_tokens": 8192,
+ "temperature": 0.1,
+ "top_p": 0.95,
+ }
+
+ safety_settings = {
+ generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ }
+
+ responses = model.generate_content(
+ [image_part, text_prompt],
+ generation_config=generation_config,
+ safety_settings=safety_settings,
+ stream=True,
+ )
+
+ full_response = ""
+ for response in responses:
+ full_response += response.text
+
+ remittance_logger.debug(f"Extracted column headers (raw model response): {full_response}")
+
+ return [header.strip() for header in full_response.split(',')]
+
+def extract_invoice_numbers_for_column(base64_image: str, column_name: str) -> InvoiceNumbers:
+ """
+ Extracts invoice numbers from a specific column in a base64-encoded image.
+
+ Args:
+ base64_image (str): The base64-encoded image string.
+ column_name (str): The name of the column to extract invoice numbers from.
+
+ Returns:
+ InvoiceNumbers: A list of extracted invoice numbers for the specified column.
+ """
+ vertexai.init(project="saltech-ai-sandbox", location="us-central1")
+ model = GenerativeModel("gemini-1.5-flash-001")
+
+ image_part = Part.from_data(
+ mime_type="image/png",
+ data=base64.b64decode(base64_image),
+ )
+
+ text_prompt = f"""Given the remittance letter image, extract all invoice numbers from the column "{column_name}". 
+ Respond with a comma-separated list only."""
+
+ generation_config = {
+ "max_output_tokens": 8192,
+ "temperature": 0.1,
+ "top_p": 0.95,
+ }
+
+ safety_settings = {
+ generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ }
+
+ responses = model.generate_content(
+ [image_part, text_prompt],
+ generation_config=generation_config,
+ safety_settings=safety_settings,
+ stream=True,
+ )
+
+ full_response = ""
+ for response in responses:
+ full_response += response.text
+
+ remittance_logger.debug(f"Extracted invoice numbers for column '{column_name}' (raw model response): {full_response}")
+
+ return [number.strip() for number in full_response.split(',') if number.strip()]
+
+def extract_invoice_numbers_with_vertex_ai_multi_hop(base64_image: str) -> list[InvoiceNumbers]:
+ """
+ Extracts invoice numbers from a single base64-encoded image using Google's Gemini Flash model with multi-hop processing.
+
+ Args:
+ base64_image (str): The base64-encoded image string.
+
+ Returns:
+ list[InvoiceNumbers]: A list containing lists of extracted invoice numbers for each processed column.
+ """
+ # First hop: Extract column headers
+ column_headers = extract_column_headers(base64_image)
+ remittance_logger.debug(f"Extracted column headers: {column_headers}")
+
+ # Second hop: Extract invoice numbers for each column (up to 3 columns)
+ all_invoice_numbers = []
+ for column_name in column_headers[:3]:
+ invoice_numbers = extract_invoice_numbers_for_column(base64_image, column_name)
+ remittance_logger.debug(f"Extracted invoice numbers for column '{column_name}': {invoice_numbers}")
+ if invoice_numbers: # Only add non-empty lists
+ all_invoice_numbers.append(invoice_numbers)
+
+ # Remove duplicate lists using the utility function
+ unique_invoice_numbers = remove_duplicate_lists(all_invoice_numbers)
+ return unique_invoice_numbers
+
+# def extract_invoice_numbers_from_text_with_vertex_ai(text: str, multi_hop: bool = False) -> list[InvoiceNumbers]:
+# """
+# Extracts invoice numbers from text using Google's Gemini Flash model.
+ 
+# Args:
+# text (str): The text of the remittance letter.
+# multi_hop (bool): Whether to use multi-hop processing (not implemented yet).
+ 
+# Returns:
+# list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
+# """
+# vertexai.init(project="saltech-ai-sandbox", location="us-central1")
+# model = GenerativeModel("gemini-1.5-flash-001")
+
+# prompt = f"""Given the following remittance letter text, extract all invoice numbers. 
+# Respond with a comma-separated list of invoice numbers only. 
+# If no invoice numbers are found, respond with 'No invoice numbers found'.
+
+# Remittance letter text:
+# {text}
+# """
+
+# generation_config = {
+# "max_output_tokens": 8192,
+# "temperature": 0.1,
+# "top_p": 0.95,
+# }
+
+# safety_settings = {
+# generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+# generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+# generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+# generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+# }
+
+# responses = model.generate_content(
+# prompt,
+# generation_config=generation_config,
+# safety_settings=safety_settings,
+# stream=True,
+# )
+
+# full_response = ""
+# for response in responses:
+# full_response += response.text
+ 
+# remittance_logger.debug(f"Vertex AI invoice numbers full response: {full_response}")
+
+# extracted_numbers = parse_gemini_response(full_response)
+# return [extracted_numbers] # Wrap in a list to match the expected return type
+
+def parse_gemini_response(response: str) -> list[str]:
+ """
+ Parses the response from Gemini Flash model and extracts invoice numbers.
+ 
+ Args:
+ response (str): The response string from Gemini Flash model.
+ 
+ Returns:
+ list[str]: A list of extracted invoice numbers.
+ """
+ if response.strip().lower().startswith('no invoice numbers found'):
+ return []
+ 
+ # Split the comma-separated list and strip whitespace from each number
+ invoice_numbers = [num.strip() for num in response.split(',')]
+ return invoice_numbers
+
+# Note: You'll need to set up authentication for Google Cloud.
+# Typically, you'd set the GOOGLE_APPLICATION_CREDENTIALS environment variable
+# to point to your service account key file.
+
+
+def extract_invoice_numbers_from_text_with_vertex_ai(text: str, multi_hop: bool = False) -> list[InvoiceNumbers]:
+ """
+ Dispatches the invoice number extraction to either single-hop or multi-hop method based on the multi_hop parameter.
+ 
+ Args:
+ text (str): The text of the remittance letter.
+ multi_hop (bool): Whether to use multi-hop processing.
+ 
+ Returns:
+ list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
+ """
+ if multi_hop:
+ return extract_invoice_numbers_from_text_with_vertex_ai_multi_hop(text)
+ else:
+ return extract_invoice_numbers_from_text_with_vertex_ai_single_hop(text)
+
+def extract_invoice_numbers_from_text_with_vertex_ai_single_hop(text: str) -> list[InvoiceNumbers]:
+ """
+ Extracts invoice numbers from text using Google's Gemini Flash model with single-hop processing.
+ 
+ Args:
+ text (str): The text of the remittance letter.
+ 
+ Returns:
+ list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
+ """
+ vertexai.init(project="saltech-ai-sandbox", location="us-central1")
+ model = GenerativeModel("gemini-1.5-flash-001")
+
+ prompt = f"""Given the following remittance letter text, extract all invoice numbers. 
+ Respond with a comma-separated list of invoice numbers only. 
+ If no invoice numbers are found, respond with 'No invoice numbers found'.
+
+ Remittance letter text:
+ {text}
+ """
+
+ generation_config = {
+ "max_output_tokens": 8192,
+ "temperature": 0.1,
+ "top_p": 0.95,
+ }
+
+ safety_settings = {
+ generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ }
+
+ responses = model.generate_content(
+ prompt,
+ generation_config=generation_config,
+ safety_settings=safety_settings,
+ stream=True,
+ )
+
+ full_response = ""
+ for response in responses:
+ full_response += response.text
+ 
+ remittance_logger.debug(f"Vertex AI invoice numbers full response (single-hop): {full_response}")
+
+ extracted_numbers = parse_gemini_response(full_response)
+ return [extracted_numbers] # Wrap in a list to match the expected return type
+
+def extract_invoice_numbers_from_text_with_vertex_ai_multi_hop(text: str) -> list[InvoiceNumbers]:
+ """
+ Extracts invoice numbers from text using Google's Gemini Flash model with multi-hop processing.
+
+ Args:
+ text (str): The text of the remittance letter.
+
+ Returns:
+ list[InvoiceNumbers]: A list containing lists of extracted invoice numbers for each processed column.
+ """
+ # First hop: Extract column headers
+ column_headers = extract_column_headers_from_text(text)
+ remittance_logger.debug(f"Extracted column headers: {column_headers}")
+
+ # Second hop: Extract invoice numbers for each column (up to 3 columns)
+ all_invoice_numbers = []
+ for column_name in column_headers[:3]:
+ invoice_numbers = extract_invoice_numbers_for_column_from_text(text, column_name)
+ remittance_logger.debug(f"Extracted invoice numbers for column '{column_name}': {invoice_numbers}")
+ if invoice_numbers: # Only add non-empty lists
+ all_invoice_numbers.append(invoice_numbers)
+
+ # Remove duplicate lists using the utility function
+ unique_invoice_numbers = remove_duplicate_lists(all_invoice_numbers)
+ return unique_invoice_numbers
+
+def extract_column_headers_from_text(text: str) -> list[str]:
+ """
+ Extracts column header names that could contain invoice numbers from the text.
+
+ Args:
+ text (str): The text of the remittance letter.
+
+ Returns:
+ list[str]: A list of column header names.
+ """
+ vertexai.init(project="saltech-ai-sandbox", location="us-central1")
+ model = GenerativeModel("gemini-1.5-flash-001")
+
+ prompt = f"""Given the following remittance letter text, extract all column header names or section titles that could contain invoice numbers. 
+ Respond with a comma-separated list only.
+
+ Remittance letter text:
+ {text}
+ """
+
+ generation_config = {
+ "max_output_tokens": 8192,
+ "temperature": 0.1,
+ "top_p": 0.95,
+ }
+
+ safety_settings = {
+ generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ }
+
+ response = model.generate_content(
+ prompt,
+ generation_config=generation_config,
+ safety_settings=safety_settings,
+ )
+
+ remittance_logger.debug(f"Extracted column headers (raw model response): {response.text}")
+
+ return [header.strip() for header in response.text.split(',')]
+
+def extract_invoice_numbers_for_column_from_text(text: str, column_name: str) -> InvoiceNumbers:
+ """
+ Extracts invoice numbers from a specific column or section in the text.
+
+ Args:
+ text (str): The text of the remittance letter.
+ column_name (str): The name of the column or section to extract invoice numbers from.
+
+ Returns:
+ InvoiceNumbers: A list of extracted invoice numbers for the specified column.
+ """
+ vertexai.init(project="saltech-ai-sandbox", location="us-central1")
+ model = GenerativeModel("gemini-1.5-flash-001")
+
+ prompt = f"""Given the following remittance letter text, extract all invoice numbers from the column or section "{column_name}". 
+ Respond with a comma-separated list only. If no invoice numbers are found, respond with 'No invoice numbers found'.
+
+ Remittance letter text:
+ {text}
+ """
+
+ generation_config = {
+ "max_output_tokens": 8192,
+ "temperature": 0.1,
+ "top_p": 0.95,
+ }
+
+ safety_settings = {
+ generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ }
+
+ response = model.generate_content(
+ prompt,
+ generation_config=generation_config,
+ safety_settings=safety_settings,
+ )
+
+ remittance_logger.debug(f"Extracted invoice numbers for column '{column_name}' (raw model response): {response.text}")
+
+ return parse_gemini_response(response.text)
+
+def extract_payment_amounts_with_vertex_ai(base64_image: str) -> list[PaymentAmount]:
+ vertexai.init(project="saltech-ai-sandbox", location="us-central1")
+ model = GenerativeModel("gemini-1.5-flash-001")
+
+ image_part = Part.from_data(
+ mime_type="image/png",
+ data=base64.b64decode(base64_image),
+ )
+
+ text_prompt = """Given the remittance letter image, extract the total payment amount. 
+ Respond with the payment amount only. 
+ If no payment amounts are found, respond with 'No payment amounts found'."""
+
+ generation_config = {
+ "max_output_tokens": 256,
+ "temperature": 0.1,
+ "top_p": 0.95,
+ }
+
+ safety_settings = {
+ generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ }
+
+ responses = model.generate_content(
+ [image_part, text_prompt],
+ generation_config=generation_config,
+ safety_settings=safety_settings,
+ stream=True,
+ )
+
+ full_response = ""
+ for response in responses:
+ full_response += response.text
+
+ remittance_logger.debug(f"Vertex AI payment amount full response: {full_response}")
+
+ extracted_amounts = parse_gemini_payment_response(full_response)
+ return extracted_amounts
+
+def extract_payment_amounts_from_text_with_vertex_ai(text: str) -> list[PaymentAmount]:
+ """
+ Extracts payment amounts from text using Google's Gemini Flash model.
+ 
+ Args:
+ text (str): The text of the remittance letter.
+ 
+ Returns:
+ list[PaymentAmount]: A list of extracted payment amounts.
+ """
+ vertexai.init(project="saltech-ai-sandbox", location="us-central1")
+ model = GenerativeModel("gemini-1.5-flash-001")
+
+ prompt = f"""Given the following remittance letter text, extract the total payment amount. 
+ Respond with the payment amount only. 
+ If no payment amounts are found, respond with 'No payment amounts found'.
+
+ Remittance letter text:
+ {text}
+ """
+
+ generation_config = {
+ "max_output_tokens": 256,
+ "temperature": 0.1,
+ "top_p": 0.95,
+ }
+
+ safety_settings = {
+ generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+ }
+
+ response = model.generate_content(
+ prompt,
+ generation_config=generation_config,
+ safety_settings=safety_settings,
+ )
+
+ remittance_logger.debug(f"Vertex AI payment amount full response: {response.text}")
+
+ extracted_amounts = parse_gemini_payment_response(response.text)
+ return extracted_amounts
+
+def parse_gemini_payment_response(response: str) -> list[PaymentAmount]:
+ """
+ Parses the response from Gemini Flash model and extracts payment amounts.
+ 
+ Args:
+ response (str): The response string from Gemini Flash model.
+ 
+ Returns:
+ list[PaymentAmount]: A list of one extracted payment amount (or empty).
+ """
+ if response.strip().lower() == 'no payment amounts found':
+ return []
+ 
+ payment_amounts = [response.strip()]
+ return payment_amounts
+
+def extract_payment_amounts_from_base64_images(base64_images: list[str]) -> list[PaymentAmount]:
+ # Implementation similar to extract_invoice_numbers_with_vertex_ai
+ # but focused on extracting payment amounts
+ return []