175 lines
4.9 KiB
Python
175 lines
4.9 KiB
Python
import json
|
|
from datetime import datetime, timedelta
|
|
from typing import NotRequired, TypedDict
|
|
|
|
import requests
|
|
from aws_lambda_powertools import Logger, Tracer
|
|
from aws_lambda_powertools.utilities.data_classes import (
|
|
EventBridgeEvent,
|
|
event_source,
|
|
)
|
|
from aws_lambda_powertools.utilities.typing import LambdaContext
|
|
from layercake.dateutils import fromisoformat, now
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair, SortKey
|
|
|
|
from boto3clients import dynamodb_client, s3_client
|
|
from config import (
|
|
BUCKET_NAME,
|
|
COURSE_TABLE,
|
|
ENROLLMENT_TABLE,
|
|
PAPERFORGE_API,
|
|
)
|
|
|
|
tracer = Tracer()
|
|
logger = Logger(__name__)
|
|
dyn = DynamoDBPersistenceLayer(ENROLLMENT_TABLE, dynamodb_client)
|
|
|
|
|
|
@logger.inject_lambda_context
|
|
@tracer.capture_lambda_handler
|
|
@event_source(data_class=EventBridgeEvent)
|
|
def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
|
|
new_image = event.detail['new_image']
|
|
now_ = now()
|
|
enrollment_id = new_image['id']
|
|
course_id = new_image['course']['id']
|
|
cert = dyn.collection.get_item(
|
|
KeyPair(
|
|
pk=course_id,
|
|
sk=SortKey('0', path_spec='cert', rename_key='cert'),
|
|
table_name=COURSE_TABLE,
|
|
),
|
|
raise_on_error=False,
|
|
default=None,
|
|
)
|
|
|
|
if not cert:
|
|
logger.debug('Certificate not found')
|
|
# There is no certificate to issue from course
|
|
return False
|
|
|
|
started_at: datetime = fromisoformat(new_image['started_at']) # type: ignore
|
|
completed_at: datetime = fromisoformat(new_image['completed_at']) # type: ignore
|
|
# Certificate may have no expiration
|
|
expires_at = (
|
|
completed_at + timedelta(days=int(cert['exp_interval']))
|
|
if cert.get('exp_interval', 0) > 0
|
|
else None
|
|
)
|
|
s3_uri = _generate_cert(
|
|
enrollment_id,
|
|
cert=cert,
|
|
user=new_image['user'],
|
|
score=new_image['score'],
|
|
started_at=started_at,
|
|
completed_at=completed_at,
|
|
expires_at=expires_at,
|
|
)
|
|
|
|
update_expr = 'SET cert = :cert, updated_at = :now'
|
|
expr_attr_values = {
|
|
':now': now_,
|
|
':cert': {'issued_at': now_}
|
|
| ({'expires_at': expires_at} if expires_at else {})
|
|
| ({'s3_uri': s3_uri} if s3_uri else {}),
|
|
}
|
|
|
|
# Post-migration: remove the following lines
|
|
if expires_at:
|
|
update_expr = 'SET cert = :cert, \
|
|
cert_expires_at = :cert_expires_at, \
|
|
updated_at = :now'
|
|
expr_attr_values[':cert_expires_at'] = expires_at
|
|
|
|
return dyn.update_item(
|
|
key=KeyPair(pk=enrollment_id, sk='0'),
|
|
update_expr=update_expr,
|
|
expr_attr_values=expr_attr_values,
|
|
cond_expr='attribute_exists(sk)',
|
|
)
|
|
|
|
|
|
def _cpffmt(s: str) -> str:
|
|
return '{}.{}.{}-{}'.format(s[:3], s[3:6], s[6:9], s[9:])
|
|
|
|
|
|
def _datefmt(dt: datetime) -> str:
|
|
months = [
|
|
'Janeiro',
|
|
'Fevereiro',
|
|
'Março',
|
|
'Abril',
|
|
'Maio',
|
|
'Junho',
|
|
'Julho',
|
|
'Agosto',
|
|
'Setembro',
|
|
'Outubro',
|
|
'Novembro',
|
|
'Dezembro',
|
|
]
|
|
return f'{dt.day:02d} de {months[dt.month - 1]} de {dt.year}'
|
|
|
|
|
|
User = TypedDict('User', {'name': str, 'cpf': str})
|
|
Cert = TypedDict('Cert', {'s3_uri': NotRequired[str]})
|
|
|
|
|
|
def _generate_cert(
|
|
id: str,
|
|
*,
|
|
score: int | float,
|
|
cert: Cert,
|
|
user: User,
|
|
started_at: datetime,
|
|
completed_at: datetime,
|
|
expires_at: datetime | None = None,
|
|
) -> str | None:
|
|
now_ = now()
|
|
|
|
if 's3_uri' not in cert:
|
|
logger.debug('Template URI is missing')
|
|
return None
|
|
|
|
try:
|
|
# Send template URI and data to Paperforge API to generate a PDF
|
|
r = requests.post(
|
|
PAPERFORGE_API,
|
|
data=json.dumps(
|
|
{
|
|
'template_uri': cert['s3_uri'],
|
|
'args': {
|
|
'name': user['name'],
|
|
'cpf': _cpffmt(user['cpf']),
|
|
'score': score,
|
|
'started_at': started_at.strftime('%d/%m/%Y'),
|
|
'completed_at': completed_at.strftime('%d/%m/%Y'),
|
|
'today': _datefmt(now_),
|
|
'year': now_.strftime('%Y'),
|
|
'expires_at': expires_at.strftime('%d/%m/%Y')
|
|
if expires_at
|
|
else None,
|
|
},
|
|
},
|
|
),
|
|
timeout=6,
|
|
)
|
|
r.raise_for_status()
|
|
|
|
object_key = f'certs/{id}.pdf'
|
|
s3_uri = f's3://{BUCKET_NAME}/{object_key}'
|
|
|
|
s3_client.put_object(
|
|
Bucket=BUCKET_NAME,
|
|
Key=object_key,
|
|
Body=r.content,
|
|
ContentType='application/pdf',
|
|
)
|
|
|
|
logger.debug(f'PDF uploaded successfully to {s3_uri}')
|
|
except requests.exceptions.RequestException as exc:
|
|
logger.exception(exc)
|
|
raise
|
|
|
|
return s3_uri
|