Technical Implementation Deep Dive
Building this intelligent document engine required careful orchestration of multiple AWS services. Here's the step-by-step implementation using my actual Lambda functions:
1. PresignUpload Lambda - Secure File Upload
Handles CORS preflight requests and generates secure S3 presigned URLs for document uploads:
def lambda_handler(event, context):
# Preflight CORS request
if event.get("requestContext", {}).get("http", {}).get("method") == "OPTIONS":
return {
"statusCode": 200,
"headers": cors_headers(),
"body": json.dumps({"message": "CORS preflight success"})
}
try:
# Parse filename from query string or body
filename = None
if event.get("queryStringParameters") and "filename" in event["queryStringParameters"]:
filename = event["queryStringParameters"]["filename"]
elif event.get("body"):
body = json.loads(event["body"])
filename = body.get("filename")
if not filename:
return {
"statusCode": 400,
"headers": cors_headers(),
"body": json.dumps({"error": "Missing 'filename'"})
}
# URL encode for special characters
safe_filename = urllib.parse.quote_plus(filename)
# Generate presigned URL
url = s3.generate_presigned_url(
ClientMethod='put_object',
Params={
'Bucket': bucket_name,
'Key': safe_filename,
'ContentType': 'application/octet-stream'
},
ExpiresIn=300
)
return {
"statusCode": 200,
"headers": cors_headers(),
"body": json.dumps({"upload_url": url})
}
2. TextractProcessor Lambda - OCR Text Extraction
Triggered by S3 events to extract text using Amazon Textract with SNS notifications:
textract = boto3.client('textract')
def lambda_handler(event, context):
print("✓ ENV CHECK:", dict(os.environ))
record = event['Records'][0]
bucket = record['s3']['bucket']['name']
key = record['s3']['object']['key']
sns_topic_arn = os.environ['SNS_TOPIC_ARN']
textract_role_arn = os.environ['TEXTRACT_ROLE_ARN']
# Start Textract job with SNS notification
response = textract.start_document_text_detection(
DocumentLocation={
'S3Object': {
'Bucket': bucket,
'Name': key
}
},
NotificationChannel={
'RoleArn': textract_role_arn,
'SNSTopicArn': sns_topic_arn
}
)
print(f"✅ Textract job started: {response['JobId']}")
print(f"📄 New file uploaded: {key}")
print(f"🪣 Bucket: {bucket}")
return {
'statusCode': 200,
'body': json.dumps({
'message': 'TextractProcessor ran successfully',
'bucket': bucket,
'key': key
})
}
3. MLAnalyzer Lambda - SageMaker Integration
Processes SNS notifications from Textract and calls SageMaker endpoints for document analysis:
lambda_client = boto3.client('lambda')
textract = boto3.client('textract')
def call_sagemaker_model(text):
runtime = boto3.client('sagemaker-runtime')
response = runtime.invoke_endpoint(
EndpointName="intellidoc-distilbert-1748476781",
ContentType="application/x-text",
Body=text.encode("utf-8")
)
result = json.loads(response["Body"].read().decode())
probs = result["probabilities"]
topics = ["medical", "healthcare"]
confidence = max(probs)
sentiment = "positive" if confidence >= 0.5 else "negative"
return {
"insights": {
"sentiment": sentiment,
"confidence": round(confidence, 2),
"topics": topics
}
}
def lambda_handler(event, context):
print("MLAnalyzer triggered by SNS")
sns_record = event['Records'][0]['Sns']
message = json.loads(sns_record['Message'])
job_id = message.get("JobId")
if job_id:
print(f"Fetching Textract job results for JobId: {job_id}")
response = textract.get_document_text_detection(JobId=job_id)
blocks = response.get("Blocks", [])
text_lines = [block["Text"] for block in blocks if block["BlockType"] == "LINE"]
else:
print("No JobId provided - using mock extracted_text for testing")
text_lines = message.get("extracted_text", ["No text provided"])
print(f"Extracted {len(text_lines)} lines:")
for line in text_lines:
print("→", line)
joined_text = " ".join(text_lines)
4. SaveAndIndex Lambda - Dual Storage Implementation
Implements the critical dual-write pattern to maintain consistency between DynamoDB and OpenSearch:
def lambda_handler(event, context):
print("SaveAndIndex Lambda Invoked")
print("Raw event:", json.dumps(event))
if "body" in event and isinstance(event["body"], str):
event = json.loads(event["body"])
print("Unwrapped body:", json.dumps(event))
# Generate document metadata
document_id = "doc-" + datetime.now().strftime("%Y%m%d%H%M%S")
category = "Cardiology"
confidence = Decimal("0.99")
extracted_text = event.get("lines", ["No text provided"])
timestamp = datetime.now().isoformat()
payload = {
"DocumentId": document_id,
"category": category,
"confidence": str(confidence),
"extracted_text": extracted_text,
"timestamp": timestamp
}
print("Final payload:", json.dumps(payload, default=str))
# Define storage targets
index = "documents"
url = os.getenv("OPENSEARCH_ENDPOINT")
table_name = "IntelliDocMetadata"
# Perform dual write operation
save_to_opensearch(document_id, payload, index, url, awsauth)
save_to_dynamodb(payload, table_name)
return {
"statusCode": 200,
"body": json.dumps({"message": "✅ Data saved successfully"})
}
def save_to_opensearch(document_id, payload, index, url, auth):
try:
response = requests.put(
f"{url}/{index}/_doc/{document_id}",
auth=auth,
json=payload,
headers={"Content-Type": "application/json"}
)
print("🔍 OpenSearch response:", response.text)
5. SearchOpenSearch Lambda - Intelligent Query Processing
Processes search queries with advanced OpenSearch capabilities including highlighting and filtering:
def lambda_handler(event, context):
print("✅ SearchOpenSearch Lambda invoked")
query = event.get("queryStringParameters", {}).get("q", "")
print("🔍 Query received:", query)
# TEMP: run match_all to verify indexing is working
search_body = {
"query": {
"bool": {
"should": [
{"match_phrase": {"category": query}},
{"match_phrase": {"extracted_text": query}}
],
"minimum_should_match": 1
}
}
}
try:
response = requests.get(
f"{opensearch_url}/{index_name}/_search",
auth=awsauth,
headers={"Content-Type": "application/json"},
json=search_body
)
data = response.json()
print("📊 Raw OpenSearch response:")
print(json.dumps(data, indent=2)) # ✅ this will show in CloudWatch
hits = data.get("hits", {}).get("hits", [])
results = [hit["_source"] for hit in hits]
return {
"statusCode": 200,
"headers": cors_headers(),
"body": json.dumps({"results": results})
}
except Exception as e:
print("❌ ERROR:", str(e))
return {
"statusCode": 500,
"headers": cors_headers(),
"body": json.dumps({"error": str(e)})
}
6. Service Interconnection Flow
The services interconnect through the following event-driven architecture:
Document Processing Flow:
1. Frontend → API Gateway → PresignUpload Lambda → S3 presigned URL
2. Direct upload → S3 Bucket → S3 Event Trigger
3. S3 Event → TextractProcessor Lambda → Amazon Textract API
4. Textract → SNS Topic → MLAnalyzer Lambda → SageMaker Endpoint
5. ML Results → SaveAndIndex Lambda → DynamoDB + OpenSearch
Search Flow:
6. Frontend → API Gateway → SearchOpenSearch Lambda → OpenSearch
7. Query Processing → Results with highlighting → Frontend display
Key Features:
- SageMaker Endpoint: "intellidoc-distilbert-1748476781"
- OpenSearch Index: "documents"
- DynamoDB Table: "IntelliDocMetadata"
- SNS Topics for event coordination
- CORS handling for frontend integration
