Validating Low-Confidence AWS Textract Extractions with Amazon Bedrock

"""
Validate low-confidence Textract extractions using Bedrock (Claude) with image crops.

This script:
1. Loads the Textract response JSON
2. Finds all WORD blocks with confidence below 70%
3. Crops the relevant region from the PDF using bounding box coordinates
4. Sends each crop to Bedrock Claude for visual validation
5. Updates the analyzeDocResponseNEW.json with corrected text
6. Writes a change log with all modifications made
"""

import json
import base64
import datetime
from io import BytesIO
from pathlib import Path

import boto3
from pdf2image import convert_from_path
from PIL import Image

# --- Configuration ---
CONFIDENCE_THRESHOLD = 70.0
PDF_PATH = Path(__file__).parent / "YOURPDF.pdf"
TEXTRACT_RESPONSE_PATH = Path(__file__).parent / "analyzeDocResponse.json"
OUTPUT_PATH = Path(__file__).parent / "analyzeDocResponse.json" # overwrite in place
LOG_PATH = Path(__file__).parent / "validation_change_log.json"
BEDROCK_MODEL_ID = "us.anthropic.claude-opus-4-8"
BEDROCK_REGION = "us-west-2"
PADDING_PX = 30 # pixels of padding around the bounding box crop

def load_textract_response(path: Path) -\> dict:
"""Load the Textract JSON response."""
with open(path) as f:
return json.load(f)

def get_low_confidence_blocks(textract_response: dict, threshold: float) -\> list[dict]:
"""Filter WORD blocks below the confidence threshold."""
return [
block for block in textract_response["Blocks"]
if block["BlockType"] == "WORD"
and block.get("Confidence", 100) \< threshold
]

def convert_pdf_to_images(pdf_path: Path, dpi: int = 300) -\> list[Image.Image]:
"""Convert PDF pages to PIL images."""
print(f"Converting PDF to images at {dpi} DPI...")
return convert_from_path(str(pdf_path), dpi=dpi)

def crop_region(page_image: Image.Image, bounding_box: dict, padding: int = PADDING_PX) -\> Image.Image:
"""Crop a region from the page image using Textract bounding box coordinates."""
width, height = page_image.size
left = int(bounding_box["Left"] * width) - padding
top = int(bounding_box["Top"] * height) - padding
right = int((bounding_box["Left"] + bounding_box["Width"]) * width) + padding
bottom = int((bounding_box["Top"] + bounding_box["Height"]) * height) + padding
# Clamp to image bounds
left = max(0, left)
top = max(0, top)
right = min(width, right)
bottom = min(height, bottom)
return page_image.crop((left, top, right, bottom))

def image_to_base64(image: Image.Image) -\> str:
"""Convert a PIL image to a base64-encoded PNG string."""
buffer = BytesIO()
image.save(buffer, format="PNG")
return base64.standard_b64encode(buffer.getvalue()).decode("utf-8")

def validate_with_bedrock(bedrock_client, cropped_image_b64: str, textract_text: str) -\> dict:
"""Send cropped image + Textract text to Bedrock for validation."""
prompt = (
f'AWS Textract extracted the following text from this image region: "{textract_text}"\n\n'
f"Textract flagged this as low confidence. Please look at the image carefully and tell me:\n"
f"1. What does the text actually say in this image?\n"
f"2. Pay special attention to mathematical symbols: is it a less-than sign (\<), "
f"less-than-or-equal-to sign (≤), greater-than-or-equal-to sign (≥), or something else?\n"
f"3. Pay attention to superscripts (like ² in cm² or m²) and special characters.\n\n"
f"Respond ONLY with valid JSON in this exact format, no other text:\n"
f'{{"corrected_text": "...", "is_different": true/false, "explanation": "..."}}'
)

response = bedrock_client.converse(
modelId=BEDROCK_MODEL_ID,
messages=[
{
"role": "user",
"content": [
{
"image": {
"format": "png",
"source": {"bytes": base64.standard_b64decode(cropped_image_b64)}
}
},
{
"text": prompt
}
]
}
],
inferenceConfig={"maxTokens": 256}
)

response_text = response["output"]["message"]["content"][0]["text"]

# Parse the JSON response from Bedrock
try:
# Handle case where model wraps in markdown code block
if "```" in response_text:
response_text = response_text.split("```json")[-1].split("```")[0].strip()
if not response_text:
response_text = response_text.split("```")[-2].strip()
return json.loads(response_text)
except json.JSONDecodeError:
return {
"corrected_text": textract_text,
"is_different": False,
"explanation": f"Could not parse Bedrock response: {response_text}"
}

def main():
print("=" * 60)
print("Textract Low-Confidence Validation with Bedrock")
print("=" * 60)

# Load Textract response
print(f"\nLoading Textract response from: {TEXTRACT_RESPONSE_PATH}")
textract_response = load_textract_response(TEXTRACT_RESPONSE_PATH)

# Find low-confidence blocks
low_conf_blocks = get_low_confidence_blocks(textract_response, CONFIDENCE_THRESHOLD)
print(f"Found {len(low_conf_blocks)} WORD blocks with confidence \< {CONFIDENCE_THRESHOLD}%")

if not low_conf_blocks:
print("Nothing to validate. Exiting.")
return

# Convert PDF to images
pages = convert_pdf_to_images(PDF_PATH)
print(f"Converted {len(pages)} pages from PDF")

# Initialize Bedrock client
bedrock = boto3.client("bedrock-runtime", region_name=BEDROCK_REGION)

# Process each low-confidence block
change_log = []
changes_made = 0

print(f"\nValidating {len(low_conf_blocks)} blocks...\n")
print("-" * 60)

for i, block in enumerate(low_conf_blocks, 1):
block_id = block["Id"]
original_text = block["Text"]
confidence = block["Confidence"]
page_num = block.get("Page", 1)
bbox = block["Geometry"]["BoundingBox"]

print(f"[{i}/{len(low_conf_blocks)}] Page {page_num}: \"{original_text}\" "
f"(confidence: {confidence:.1f}%)")

# Crop the region from the PDF page
page_image = pages[page_num - 1]
cropped = crop_region(page_image, bbox)
cropped_b64 = image_to_base64(cropped)

# Send to Bedrock for validation
correction = validate_with_bedrock(bedrock, cropped_b64, original_text)

log_entry = {
"block_id": block_id,
"page": page_num,
"original_text": original_text,
"original_confidence": round(confidence, 2),
"corrected_text": correction["corrected_text"],
"is_different": correction["is_different"],
"explanation": correction["explanation"],
"bounding_box": bbox
}

if correction["is_different"]:
# Update the block in the Textract response
block["OriginalText"] = original_text
block["OriginalConfidence"] = confidence
block["Text"] = correction["corrected_text"]
block["BedrockCorrected"] = True
block["CorrectionExplanation"] = correction["explanation"]
changes_made += 1
print(f" ✓ CORRECTED: \"{original_text}\" → \"{correction['corrected_text']}\"")
print(f" Reason: {correction['explanation']}")
else:
print(f" — No change needed: {correction['explanation']}")

change_log.append(log_entry)
print()

print("-" * 60)
print(f"\nSummary: {changes_made} corrections made out of {len(low_conf_blocks)} blocks validated")

# Save updated Textract response
print(f"\nSaving updated response to: {OUTPUT_PATH}")
with open(OUTPUT_PATH, "w") as f:
json.dump(textract_response, f, indent=2, ensure_ascii=False)

# Save change log
log_output = {
"timestamp": datetime.datetime.now().isoformat(),
"pdf_file": str(PDF_PATH.name),
"confidence_threshold": CONFIDENCE_THRESHOLD,
"total_blocks_validated": len(low_conf_blocks),
"total_corrections_made": changes_made,
"bedrock_model": BEDROCK_MODEL_ID,
"changes": change_log
}

print(f"Saving change log to: {LOG_PATH}")
with open(LOG_PATH, "w") as f:
json.dump(log_output, f, indent=2, ensure_ascii=False)

print("\nDone!")

if __name__ == "__main__":
main()