Ejercicio: Extracción de Texto con Textract y Lambda ​

import boto3
import json
import os
import uuid
from datetime import datetime

def lambda_handler(event, context):
    # Inicializar clientes
    textract = boto3.client('textract')
    dynamodb = boto3.client('dynamodb')
    sns = boto3.client('sns')
    
    # Obtener información del archivo S3
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    
    try:
        # Iniciar procesamiento asíncrono con Textract
        response = textract.start_document_text_detection(
            DocumentLocation={
                'S3Object': {
                    'Bucket': bucket,
                    'Name': key
                }
            },
            NotificationChannel={
                'SNSTopicArn': os.environ['SNS_TOPIC_ARN'],
                'RoleArn': os.environ['TEXTRACT_ROLE_ARN']
            }
        )
        
        # Guardar información del trabajo en DynamoDB
        document_id = str(uuid.uuid4())
        timestamp = int(datetime.now().timestamp())
        
        dynamodb.put_item(
            TableName=os.environ['DYNAMODB_TABLE'],
            Item={
                'document_id': {'S': document_id},
                'timestamp': {'N': str(timestamp)},
                'job_id': {'S': response['JobId']},
                'status': {'S': 'IN_PROGRESS'},
                'source_bucket': {'S': bucket},
                'source_key': {'S': key}
            }
        )
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Document processing started',
                'job_id': response['JobId'],
                'document_id': document_id
            })
        }
        
    except Exception as e:
        print(e)
        sns.publish(
            TopicArn=os.environ['SNS_TOPIC_ARN'],
            Message=f'Error processing document {key}: {str(e)}',
            Subject='Document Processing Error'
        )
        raise e

import boto3
import json
import os
from datetime import datetime

def lambda_handler(event, context):
    # Inicializar clientes
    textract = boto3.client('textract')
    s3 = boto3.client('s3')
    dynamodb = boto3.client('dynamodb')
    sns = boto3.client('sns')
    
    try:
        # Obtener mensaje SNS
        message = json.loads(event['Records'][0]['Sns']['Message'])
        job_id = message['JobId']
        status = message['Status']
        
        if status == 'SUCCEEDED':
            # Obtener resultados de Textract
            response = textract.get_document_text_detection(JobId=job_id)
            
            # Extraer texto
            text_blocks = []
            for block in response['Blocks']:
                if block['BlockType'] == 'LINE':
                    text_blocks.append(block['Text'])
            
            # Obtener información del documento de DynamoDB
            dynamo_response = dynamodb.query(
                TableName=os.environ['DYNAMODB_TABLE'],
                IndexName='JobIdIndex',
                KeyConditionExpression='job_id = :job_id',
                ExpressionAttributeValues={
                    ':job_id': {'S': job_id}
                }
            )
            
            document_info = dynamo_response['Items'][0]
            document_id = document_info['document_id']['S']
            source_bucket = document_info['source_bucket']['S']
            source_key = document_info['source_key']['S']
            
            # Guardar resultados en S3
            output_key = f'processed/{document_id}/extracted_text.txt'
            s3.put_object(
                Bucket=os.environ['OUTPUT_BUCKET'],
                Key=output_key,
                Body='\n'.join(text_blocks)
            )
            
            # Actualizar DynamoDB
            dynamodb.update_item(
                TableName=os.environ['DYNAMODB_TABLE'],
                Key={
                    'document_id': {'S': document_id},
                    'timestamp': {'N': document_info['timestamp']['N']}
                },
                UpdateExpression='SET #status = :status, output_key = :output_key',
                ExpressionAttributeNames={
                    '#status': 'status'
                },
                ExpressionAttributeValues={
                    ':status': {'S': 'COMPLETED'},
                    ':output_key': {'S': output_key}
                }
            )
            
            # Notificar completación
            sns.publish(
                TopicArn=os.environ['SNS_TOPIC_ARN'],
                Message=f'Document {source_key} processing completed. Results available at s3://{os.environ["OUTPUT_BUCKET"]}/{output_key}',
                Subject='Document Processing Completed'
            )
            
        else:
            # Manejar error
            sns.publish(
                TopicArn=os.environ['SNS_TOPIC_ARN'],
                Message=f'Document processing failed for job {job_id}',
                Subject='Document Processing Failed'
            )
            
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': f'Processing {status}',
                'job_id': job_id
            })
        }
        
    except Exception as e:
        print(e)
        sns.publish(
            TopicArn=os.environ['SNS_TOPIC_ARN'],
            Message=f'Error processing Textract results: {str(e)}',
            Subject='Document Processing Error'
        )
        raise e

# En las funciones Lambda
cloudwatch = boto3.client('cloudwatch')

# Métrica de tiempo de procesamiento
cloudwatch.put_metric_data(
    Namespace='DocumentProcessing',
    MetricData=[
        {
            'MetricName': 'ProcessingTime',
            'Value': processing_time,
            'Unit': 'Seconds'
        }
    ]
)