Merge pull request #1 from 7SOATSquad30/feature/lambda

Feature/lambda

.github/workflows/deploy-lambda.yml

@@ -0,0 +1,50 @@
+name: Deploy AWS Lambda
+
+on:
+  push:
+    branches:
+      - main  # Defina o branch que acionará o deploy
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout do repositório
+        uses: actions/checkout@v4
+
+      - name: Configurar credenciais AWS
+        uses: aws-actions/configure-aws-credentials@v4
+        with:
+          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
+          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+          aws-region: us-east-1  # Substitua pela sua região AWS
+
+      - name: Instalar dependências necessárias
+        run: sudo apt-get update && sudo apt-get install -y xz-utils zip
+
+      - name: Construir pacote de implantação
+        run: make build
+
+      - name: Fazer upload do pacote para S3 (opcional)
+        run: |
+          aws s3 cp deployment_package.zip s3://seu-bucket-deploy-lambda/
+        if: success()
+
+      - name: Implantar a AWS Lambda
+        run: |
+          aws lambda update-function-code \
+            --function-name nome-da-sua-lambda \
+            --zip-file fileb://deployment_package.zip
+        if: success()
+
+      - name: Limpeza pós-deploy
+        run: make clean
+
+      - name: Notificar sucesso do deploy
+        if: success()
+        run: echo "Deploy concluído com sucesso!"
+
+      - name: Notificar falha do deploy
+        if: failure()
+        run: echo "Falha no deploy!"

.gitignore

@@ -0,0 +1,39 @@
+# Python
+__pycache__/
+*.py[cod]
+*.egg-info/
+dist/
+build/
+
+# Virtual environment
+.venv/
+venv/
+
+# AWS
+*.log
+*.out
+.aws-sam/
+
+# Terraform
+.terraform/
+*.tfstate
+*.tfstate.backup
+
+# Editor/OS
+.vscode/
+.idea/
+*.swp
+.DS_Store
+Thumbs.db
+
+# Dependencies
+requirements.txt
+requirements-dev.txt
+
+# Deployment packages
+deployment_package.zip
+target/
+
+# Environment variables
+.env
+.env.*

Makefile

@@ -0,0 +1,24 @@
+ZIP_FILE=deployment_package.zip
+FFMPEG_URL=https://johnvansickle.com/ffmpeg/releases/ffmpeg-release-amd64-static.tar.xz
+FFMPEG_DIR=ffmpeg
+
+.PHONY: build
+build:
+	@echo "Criando diretórios necessários..."
+	rm -rf package
+	mkdir -p package/tmp
+	@echo "Instalando dependências..."
+	pip install --target ./package -r requirements.txt
+	@echo "Baixando FFmpeg..."
+	curl -L $(FFMPEG_URL) | tar -xJ -C package/tmp --strip-components 1
+	cp package/tmp/ffmpeg package/
+	@echo "Compactando pacote de implantação..."
+	cd package && zip -r9 ../$(ZIP_FILE) .
+	zip -g $(ZIP_FILE) lambda_function.py src/**/*
+	@echo "Pacote de implantação criado: $(ZIP_FILE)"
+
+.PHONY: clean
+clean:
+	@echo "Limpando arquivos temporários..."
+	rm -rf package $(ZIP_FILE)
+	@echo "Limpeza concluída."

README.md

@@ -1,2 +1,114 @@
-# lambda-video-processing-challenge
-Lambda that generates video frames
+# AWS Lambda para Processamento de Vídeos com Extração de Frames
+
+Esta AWS Lambda tem como objetivo automatizar o processamento de arquivos de vídeo enviados para um bucket S3. A função é acionada por eventos provenientes de uma fila Amazon SQS, realiza o download do arquivo, extrai frames utilizando FFmpeg e disponibiliza os resultados em formato ZIP no S3 de saída.  
+
+---
+
+## 🛠️ Funcionalidades
+
+- **Processamento automatizado de vídeos:**  
+  A função recebe eventos do SQS contendo informações do vídeo a ser processado.  
+- **Extração de frames com FFmpeg:**  
+  Cada vídeo enviado é processado para gerar imagens frame a frame.  
+- **Compactação dos frames em ZIP:**  
+  Os frames extraídos são compactados para facilitar o armazenamento e compartilhamento.  
+- **Upload para S3:**  
+  O arquivo ZIP finalizado é enviado para o bucket S3 de saída.  
+- **Atualização de status em DynamoDB:**  
+  Registra o status de processamento do arquivo no DynamoDB.  
+- **Notificação de erros por e-mail:**  
+  Em caso de falha, um e-mail de alerta é enviado via Amazon SES.  
+
+---
+
+## Fluxograma
+
+```mermaid
+graph TD
+    A[Recebe evento do SQS] --> B[Atualiza status no DynamoDB para Em processamento]
+    B --> C[Baixa arquivo do S3]
+    C --> D[Extrai frames com FFmpeg]
+    D --> E[Cria arquivo ZIP com frames]
+    E --> F[Faz upload do ZIP para o S3 de saída]
+    F --> G[Atualiza status no DynamoDB para Processado]
+    G --> H[Retorna resposta de sucesso]
+
+    %% Tratamento de erros
+    A -->|Chave ausente| X1[Chave ausente no evento]
+    A -->|Erro JSON| X2[Erro ao decodificar JSON]
+    A -->|Erro inesperado| X3[Erro inesperado]
+
+    X1 --> Y1[Envia e-mail de erro]
+    X2 --> Y2[Envia e-mail de erro]
+    X3 --> Y3[Envia e-mail de erro]
+
+    X1 --> Z[Retorna erro 500]
+    X2 --> Z[Retorna erro 500]
+    X3 --> Z[Retorna erro 500]
+```
+
+---
+
+## ⚙️ Fluxo de Processamento
+
+1. **Recebimento do evento do SQS:**  
+   - A função é acionada quando uma nova mensagem chega na fila SQS.  
+   - A mensagem contém o nome do bucket e a chave do arquivo de vídeo.  
+
+2. **Atualização do status no DynamoDB:**  
+   - O status do arquivo é atualizado para "Em processamento".  
+
+3. **Download do arquivo do S3:**  
+   - O vídeo é baixado para o diretório temporário da Lambda (`/tmp`).  
+
+4. **Extração de frames com FFmpeg:**  
+   - Os frames do vídeo são extraídos e armazenados temporariamente.  
+
+5. **Criação do arquivo ZIP:**  
+   - Todos os frames são compactados em um único arquivo ZIP.  
+
+6. **Upload do arquivo ZIP para S3:**  
+   - O arquivo ZIP é carregado no bucket de saída, na pasta `processed/`.  
+
+7. **Atualização do status para "Processado":**  
+   - O status final é atualizado no DynamoDB.  
+
+8. **Resposta de sucesso:**  
+   - A Lambda retorna uma resposta indicando o sucesso do processamento.  
+
+---
+
+## 🛡️ Tratamento de Erros
+
+Caso ocorra algum problema durante o processamento, a função captura exceções e executa as seguintes ações:  
+
+- **Erros de chave ausente no evento SQS:**  
+  - Envia um e-mail de alerta e retorna erro 500.  
+
+- **Erro ao processar JSON da mensagem:**  
+  - Envia uma notificação por e-mail e retorna erro 500.  
+
+- **Falhas inesperadas:**  
+  - Notifica via e-mail e encerra o processamento com erro.  
+
+---
+
+## 🧰 Tecnologias Utilizadas
+
+- **AWS Lambda** – Processamento serverless do vídeo.  
+- **Amazon SQS** – Fila de mensagens para acionar o processamento.  
+- **Amazon S3** – Armazenamento de vídeos de entrada e saída.  
+- **Amazon DynamoDB** – Rastreamento do status do processamento.  
+- **Amazon SES** – Envio de notificações de erro.  
+- **FFmpeg** – Extração de frames dos vídeos.  
+- **Python** – Linguagem utilizada na implementação.  
+
+---
+
+## 📦 Implantação
+
+Para implantar esta função Lambda, siga as etapas abaixo:
+
+1. Instale as dependências listadas no `requirements.txt`:  
+   ```bash
+   pip install -r requirements.txt -t package/

app/lambda_function.py

@@ -0,0 +1,77 @@
+import json
+import os
+
+from src.service.s3_download_file import download_file_from_s3
+from src.service.dynamodb_update_status import update_status_in_dynamodb
+from src.service.s3_upload_file import upload_file_to_s3
+from src.service.ffmpeg_extract_frames import extract_frames_with_ffmpeg
+from src.service.create_zip_from_folder import create_zip_from_folder
+from src.service.send_email_notification import send_email_notification
+from src.config.config import logger
+from src.config.config import get_env_variable
+
+def lambda_handler(event, context):
+    """
+    Função Lambda para processar eventos do SQS e baixar arquivos do S3.
+    
+    :param event: Evento recebido do SQS.
+    :param context: Contexto da execução Lambda.
+    """
+    try:
+        table_name = get_env_variable('DYNAMODB_TABLE_NAME')
+        output_bucket_name = get_env_variable('OUTPUT_S3_BUCKET')
+        client_email = get_env_variable('CLIENT_EMAIL')
+        processed_files = []
+
+        for record in event['Records']:
+            message_body = json.loads(record['body'])
+            bucket_name = message_body['bucket_name']
+            object_key = message_body['object_key']
+            download_path = os.path.join('/tmp', os.path.basename(object_key))
+            output_frames_dir = os.path.join('/tmp', 'frames')
+            zip_file_path = os.path.join('/tmp', 'frames.zip')
+
+            # Atualiza status para "Em processamento"
+            update_status_in_dynamodb(table_name, object_key, 'Em processamento')
+
+            logger.info(f"Iniciando download do arquivo: {object_key} do bucket: {bucket_name}")
+            download_file_from_s3(bucket_name, object_key, download_path)
+            logger.info(f"Arquivo baixado com sucesso em: {download_path}")
+
+            # Extrai frames do vídeo usando FFmpeg
+            extract_frames_with_ffmpeg(download_path, output_frames_dir)
+
+            # Compacta os frames em um arquivo .zip
+            create_zip_from_folder(output_frames_dir, zip_file_path)
+
+            # Faz upload do arquivo zip para o bucket S3 de saída
+            upload_key = f"processed/{os.path.basename(zip_file_path)}"
+            upload_file_to_s3(zip_file_path, output_bucket_name, upload_key)
+            logger.info(f"Arquivo ZIP carregado com sucesso para: {output_bucket_name}/{upload_key}")
+
+            # Atualiza status para "Processado"
+            update_status_in_dynamodb(table_name, object_key, 'Processado')
+            processed_files.append(upload_key)
+
+        return {
+            'statusCode': 200,
+            'body': json.dumps({'message': 'Processamento concluido', 'files': processed_files})
+        }
+
+    except KeyError as e:
+        error_message = f"Erro: Chave ausente no evento - {str(e)}"
+        logger.error(error_message)
+        send_email_notification(client_email, error_message)
+    except json.JSONDecodeError as e:
+        error_message = f"Erro ao decodificar JSON - {str(e)}"
+        logger.error(error_message)
+        send_email_notification(client_email, error_message)
+    except Exception as e:
+        error_message = f"Erro inesperado: {str(e)}"
+        logger.error(error_message)
+        send_email_notification(client_email, error_message)
+
+    return {
+        'statusCode': 500,
+        'body': json.dumps('Erro interno no processamento')
+    }

app/src/config/config.py

@@ -0,0 +1,18 @@
+import os
+from aws_lambda_powertools import Logger
+
+def get_env_variable(var_name, default_value=None):
+    """
+    Recupera uma variável de ambiente, com a opção de fornecer um valor padrão.
+
+    :param var_name: Nome da variável de ambiente.
+    :param default_value: Valor padrão caso a variável não esteja definida.
+    :return: Valor da variável de ambiente ou o valor padrão.
+    """
+    return os.environ.get(var_name, default_value)
+
+# Configurações
+LOG_LEVEL = get_env_variable('LOG_LEVEL', 'INFO')
+
+# Configuração do logger
+logger = Logger(service="s3-file-processor", level=LOG_LEVEL)

app/src/service/create_zip_from_folder.py

@@ -0,0 +1,9 @@
+import os
+import zipfile
+
+
+def create_zip_from_folder(folder_path, zip_path):
+    with zipfile.ZipFile(zip_path, 'w') as zipf:
+        for root, _, files in os.walk(folder_path):
+            for file in files:
+                zipf.write(os.path.join(root, file), arcname=file)

app/src/service/dynamodb_update_status.py

@@ -0,0 +1,26 @@
+import boto3
+from src.config.config import logger
+
+dynamodb = boto3.client('dynamodb')
+
+def update_status_in_dynamodb(table_name, object_key, status):
+    """
+    Atualiza o status de processamento no DynamoDB.
+
+    :param table_name: Nome da tabela DynamoDB.
+    :param object_key: Chave do objeto no S3.
+    :param status: Novo status a ser atualizado.
+    """
+    try:
+        logger.info(f"Atualizando status para '{status}' no DynamoDB para {object_key}")
+        logger.info(f"Nome da tabela DynamoDB: {table_name}")
+        dynamodb.update_item(
+            TableName=table_name,
+            Key={'object_key': {'S': object_key}},
+            UpdateExpression='SET processing_status = :status',
+            ExpressionAttributeValues={':status': {'S': status}}
+        )
+        logger.info(f"Status atualizado para '{status}' no DynamoDB para {object_key}")
+    except Exception as e:
+        logger.error(f"Erro ao atualizar status no DynamoDB: {str(e)}")
+        raise e