mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-08-18 21:49:37 -04:00
big commit with letters removal
This commit is contained in:
@@ -1,97 +0,0 @@
|
||||
import json
|
||||
import urllib
|
||||
|
||||
import botocore
|
||||
from boto3 import resource
|
||||
from flask import current_app
|
||||
from notifications_utils.s3 import s3upload as utils_s3upload
|
||||
|
||||
|
||||
class LetterNotFoundError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def get_transient_letter_file_location(service_id, upload_id):
|
||||
return 'service-{}/{}.pdf'.format(service_id, upload_id)
|
||||
|
||||
|
||||
def backup_original_letter_to_s3(
|
||||
data,
|
||||
upload_id,
|
||||
):
|
||||
utils_s3upload(
|
||||
filedata=data,
|
||||
region=current_app.config['AWS_REGION'],
|
||||
bucket_name=current_app.config['PRECOMPILED_ORIGINALS_BACKUP_LETTERS'],
|
||||
file_location=f'{upload_id}.pdf',
|
||||
)
|
||||
|
||||
|
||||
def upload_letter_to_s3(
|
||||
data,
|
||||
*,
|
||||
file_location,
|
||||
status,
|
||||
page_count,
|
||||
filename,
|
||||
message=None,
|
||||
invalid_pages=None,
|
||||
recipient=None
|
||||
):
|
||||
# Use of urllib.parse.quote encodes metadata into ascii, which is required by s3.
|
||||
# Making sure data for displaying to users is decoded is taken care of by LetterMetadata
|
||||
metadata = {
|
||||
'status': status,
|
||||
'page_count': str(page_count),
|
||||
'filename': urllib.parse.quote(filename),
|
||||
}
|
||||
if message:
|
||||
metadata['message'] = message
|
||||
if invalid_pages:
|
||||
metadata['invalid_pages'] = json.dumps(invalid_pages)
|
||||
if recipient:
|
||||
metadata['recipient'] = urllib.parse.quote(recipient)
|
||||
|
||||
utils_s3upload(
|
||||
filedata=data,
|
||||
region=current_app.config['AWS_REGION'],
|
||||
bucket_name=current_app.config['TRANSIENT_UPLOADED_LETTERS'],
|
||||
file_location=file_location,
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
|
||||
class LetterMetadata:
|
||||
KEYS_TO_DECODE = ["filename", "recipient"]
|
||||
|
||||
def __init__(self, metadata):
|
||||
self._metadata = metadata
|
||||
|
||||
def get(self, key, default=None):
|
||||
value = self._metadata.get(key, default)
|
||||
if value and key in self.KEYS_TO_DECODE:
|
||||
value = urllib.parse.unquote(value)
|
||||
return value
|
||||
|
||||
|
||||
def get_letter_s3_object(service_id, file_id):
|
||||
try:
|
||||
file_location = get_transient_letter_file_location(service_id, file_id)
|
||||
s3 = resource('s3')
|
||||
return s3.Object(current_app.config['TRANSIENT_UPLOADED_LETTERS'], file_location).get()
|
||||
except botocore.exceptions.ClientError as e:
|
||||
if e.response['Error']['Code'] == 'NoSuchKey':
|
||||
raise LetterNotFoundError(f'Letter not found for service {service_id} and file {file_id}')
|
||||
|
||||
raise
|
||||
|
||||
|
||||
def get_letter_pdf_and_metadata(service_id, file_id):
|
||||
s3_object = get_letter_s3_object(service_id, file_id)
|
||||
pdf = s3_object['Body'].read()
|
||||
return pdf, LetterMetadata(s3_object['Metadata'])
|
||||
|
||||
|
||||
def get_letter_metadata(service_id, file_id):
|
||||
s3_object = get_letter_s3_object(service_id, file_id)
|
||||
return LetterMetadata(s3_object['Metadata'])
|
||||
@@ -8,9 +8,6 @@ from app.s3_client import get_s3_object
|
||||
|
||||
TEMP_TAG = 'temp-{user_id}_'
|
||||
EMAIL_LOGO_LOCATION_STRUCTURE = '{temp}{unique_id}-{filename}'
|
||||
LETTER_PREFIX = 'letters/static/images/letter-template/'
|
||||
LETTER_TEMP_TAG = LETTER_PREFIX + TEMP_TAG
|
||||
LETTER_TEMP_LOGO_LOCATION = 'letters/static/images/letter-template/temp-{user_id}_{unique_id}-{filename}'
|
||||
|
||||
|
||||
def get_logo_location(filename=None):
|
||||
@@ -53,10 +50,6 @@ def get_temp_truncated_filename(filename, user_id):
|
||||
return filename[len(TEMP_TAG.format(user_id=user_id)):]
|
||||
|
||||
|
||||
def get_letter_filename_with_no_path_or_extension(filename):
|
||||
return filename[len(LETTER_PREFIX):-4]
|
||||
|
||||
|
||||
def upload_email_logo(filename, filedata, user_id):
|
||||
upload_file_name = EMAIL_LOGO_LOCATION_STRUCTURE.format(
|
||||
temp=TEMP_TAG.format(user_id=user_id),
|
||||
@@ -77,26 +70,6 @@ def upload_email_logo(filename, filedata, user_id):
|
||||
return upload_file_name
|
||||
|
||||
|
||||
def upload_letter_temp_logo(filename, filedata, user_id):
|
||||
upload_filename = LETTER_TEMP_LOGO_LOCATION.format(
|
||||
user_id=user_id,
|
||||
unique_id=str(uuid.uuid4()),
|
||||
filename=filename
|
||||
)
|
||||
bucket_name = bucket_creds('bucket')
|
||||
utils_s3upload(
|
||||
filedata=filedata,
|
||||
region=bucket_creds('region'),
|
||||
bucket_name=bucket_name,
|
||||
file_location=upload_filename,
|
||||
content_type='image/svg+xml',
|
||||
access_key=bucket_creds('access_key_id'),
|
||||
secret_key=bucket_creds('secret_access_key'),
|
||||
)
|
||||
|
||||
return upload_filename
|
||||
|
||||
|
||||
def permanent_email_logo_name(filename, user_id):
|
||||
if filename.startswith(TEMP_TAG.format(user_id=user_id)):
|
||||
return get_temp_truncated_filename(filename=filename, user_id=user_id)
|
||||
@@ -104,38 +77,13 @@ def permanent_email_logo_name(filename, user_id):
|
||||
return filename
|
||||
|
||||
|
||||
def permanent_letter_logo_name(filename, extension):
|
||||
return LETTER_PREFIX + filename + '.' + extension
|
||||
|
||||
|
||||
def letter_filename_for_db(filename, user_id):
|
||||
filename = get_letter_filename_with_no_path_or_extension(filename)
|
||||
|
||||
if filename.startswith(TEMP_TAG.format(user_id=user_id)):
|
||||
filename = get_temp_truncated_filename(filename=filename, user_id=user_id)
|
||||
|
||||
return filename
|
||||
|
||||
|
||||
def delete_email_temp_files_created_by(user_id):
|
||||
for obj in get_s3_objects_filter_by_prefix(TEMP_TAG.format(user_id=user_id)):
|
||||
delete_s3_object(obj.key)
|
||||
|
||||
|
||||
def delete_letter_temp_files_created_by(user_id):
|
||||
for obj in get_s3_objects_filter_by_prefix(LETTER_TEMP_TAG.format(user_id=user_id)):
|
||||
delete_s3_object(obj.key)
|
||||
|
||||
|
||||
def delete_email_temp_file(filename):
|
||||
if not filename.startswith(TEMP_TAG[:5]):
|
||||
raise ValueError('Not a temp file: {}'.format(filename))
|
||||
|
||||
delete_s3_object(filename)
|
||||
|
||||
|
||||
def delete_letter_temp_file(filename):
|
||||
if not filename.startswith(LETTER_TEMP_TAG[:43]):
|
||||
raise ValueError('Not a temp file: {}'.format(filename))
|
||||
|
||||
delete_s3_object(filename)
|
||||
|
||||
Reference in New Issue
Block a user