"""Module that creates HTML and PDF file by product data from database.
generate_pdf takes message from the SQS Queue, pulls release_id from it,
makes requests to art_relations database to get all necessary data
and then calls generation of HTML file from this data and convert
that HTML to PDF file.
"""
from collections import OrderedDict
from datetime import datetime
import os
import zipfile
import json
from oto import response
from ddtrace import tracer
from PyPDF2 import PdfMerger
from PyPDF2.errors import PageSizeNotDefinedError
from PyPDF2.errors import PdfReadError
from PyPDF2.errors import PdfStreamError
from PyPDF2.errors import PyPdfError
from salessheets import config
from salessheets import features
from salessheets.connectors import loggly
from salessheets.connectors import sqs
from salessheets.connectors.s3 import upload_file_to_s3
from salessheets.connectors.sentry import sentry_capture_exception
from salessheets.constants import barcode
from salessheets.constants import errors
from salessheets.constants import field_const
from salessheets.constants import localized_template
from salessheets.constants import ows_services
from salessheets.constants import pdf
from salessheets.constants import salessheets_history
from salessheets.logic import template_rendering
from salessheets.models import history
from salessheets.models import ows_assets
from salessheets.models import ows_marketing
from salessheets.models import ows_pricing
from salessheets.models import ows_product
from salessheets.models import ows_sales_goals
from salessheets.models import salessheets
from salessheets.models import template_details
from salessheets.utils import barcode_generator
from salessheets.utils import generation_utils
from salessheets.utils import htmltopdf
from salessheets.utils import misc
from salessheets.validation.validate_pdf import validate_sqs_message
logger = loggly.get_current_logger()
class GettingDataException(Exception):
"""
Exceptions that raises if any database-related func returns error status.
Attributes:
func (str): failed function name.
error (str): error information.
"""
def __init__(self, func, error):
"""Show exception message."""
Exception.__init__(
self,
'Error while {func}, code: {code}, error message: {msg}'.format(
func=func, msg=error['message'], code=error['code'])
)
class GeneratingBarcodeException(Exception):
"""
Exceptions that is raised if barcode_generator returns any error status.
Attributes:
func (str): failed function name.
error (str): error information.
"""
def __init__(self, func, error):
"""Show exception message."""
Exception.__init__(
self,
'Error while {func}, code: {code}, error message: {msg}'.format(
func=func, msg=error['message'], code=error['code'])
)
@tracer.wrap()
def generate_pdf(message):
"""
Main function, calls fetching release data and html and pdf generation.
Args:
message (JSONMessageExt): message pulled from SQS-message.
Returns:
Response
"""
try:
files_to_cleanup = []
render_data_list = []
validated_message = validate_sqs_message(message)
if not validated_message:
return response.create_error_response(
code=errors.ERROR_PDF_GENERATION_CODE,
message=errors.ERROR_INVALID_SQS_MESSAGE)
message_body = json.loads(message['Body'])
product_ids = message_body[field_const.PRODUCT_ID]
output_type = message_body[field_const.OUTPUT_TYPE]
job_id = message_body[field_const.JOB_ID]
set_job_in_progress = history.update_job(
job_id, salessheets_history.IN_PROGRESS)
if not set_job_in_progress:
return set_job_in_progress
localized_template_type_id = message_body[
field_const.LOCALIZED_TEMPLATE_TYPE_ID]
for product_id in product_ids:
pdf_creation_result = create_pdf(
product_id,
localized_template_type_id,
user_id=get_user_id(message))
if not pdf_creation_result:
message = sqs.JSONMessageExt(message)
message.logger.info(
'Error creating pdf: {}'.format(
pdf_creation_result.errors))
set_job_error = history.update_job(
job_id, salessheets_history.ERROR)
if not set_job_error:
return set_job_error
return pdf_creation_result
pdf_creation_result = pdf_creation_result.message
generated_files = pdf_creation_result[
field_const.GENERATED_FILES]
data_for_rendering = pdf_creation_result[
field_const.DATA_FOR_RENDERING]
render_data_list.append(data_for_rendering)
files_to_cleanup.extend(generated_files)
filename = create_output_file(
output_type, product_ids, job_id)
if not filename:
message = sqs.JSONMessageExt(message)
message.logger.info(
'Error creating output file: {}'.format(
filename.errors))
set_job_error = history.update_job(
job_id, salessheets_history.ERROR)
if not set_job_error:
return set_job_error
return filename
filename = filename.message
(files_to_cleanup
.append(os.path.join(config.PDF_RENDERING_DIR, filename)))
uploading_result = upload_file_to_s3(filename)
for data_for_rendering in render_data_list:
if (data_for_rendering.message[field_const.CONTEXT_TYPE] ==
field_const.PHYSICAL_CONTEXT_TYPE):
rendered_barcode_file = (
'{dir}/{upc}.png'
.format(
dir=config.BARCODES_RENDERING_DIR,
upc=data_for_rendering.message[field_const.UPC]))
misc.remove_file(rendered_barcode_file)
cleanup_files(files_to_cleanup)
if not uploading_result:
message = sqs.JSONMessageExt(message)
message.logger.info(
'Failed upload {file} to S3 bucket.'.format(
file=filename))
set_job_error = history.update_job(
job_id, salessheets_history.ERROR)
if not set_job_error:
return set_job_error
return uploading_result
message = sqs.JSONMessageExt(message)
message.logger.info(
'{file} successfully generated and uploaded to S3.'.format(
file=filename))
set_job_completed = history.update_job(
job_id, salessheets_history.COMPLETED)
if not set_job_completed:
return set_job_completed
return response.Response()
except Exception as e:
if sentry_capture_exception:
sentry_capture_exception()
message = sqs.JSONMessageExt(message)
message.logger.info(
'Uncaught exception while generating PDF: {}'.format(e))
@tracer.wrap()
def format_tracks_by_cd(tracks):
"""
Converter for tracks.
Function that converts track list to dict keyed by cd number
with tracks from this cd as value.
Args:
tracks (dict): track list pulled from database.
Returns:
Response: Response with formatted dict of tracks message.
"""
def process_side(formatted_tracks, track, disc):
if track[field_const.SIDE] is not None:
if track[field_const.SIDE] not in formatted_tracks[disc]:
formatted_tracks[disc][track[field_const.SIDE]] = []
formatted_tracks[disc][track[field_const.SIDE]].append(track)
else:
if 'side' not in formatted_tracks[disc]:
formatted_tracks[disc]['side'] = []
formatted_tracks[disc]['side'].append(track)
return formatted_tracks
tracks = sorted(
tracks.values(),
key=lambda x: (x['cd'], x['side'] or '', x['track_number']))
formatted_tracks = OrderedDict()
for track in tracks:
if track[field_const.CD] is not None:
if track[field_const.CD] not in formatted_tracks:
formatted_tracks[track[field_const.CD]] = OrderedDict()
formatted_tracks = process_side(
formatted_tracks, track, track[field_const.CD])
else:
if field_const.CD not in formatted_tracks:
formatted_tracks[field_const.CD] = OrderedDict()
formatted_tracks = process_side(
formatted_tracks, track, field_const.CD)
return response.Response(message=formatted_tracks)
@tracer.wrap()
def get_album_data_for_product(product_id):
"""
Function that pulls album data from db and formats it to the single dict.
Query the database depending on the context type and status of product-
for the digital product which is not in content the
get_album_for_digital_not_in_content is called.
Args:
product_id (str): release id pulled from sqs message.
Raises:
GettingDataException.
Returns:
Response: Response with dictionary with release data in message.
"""
product_type_and_status = salessheets.get_context_type_and_status(
product_id)
if not product_type_and_status:
return product_type_and_status
product_type_and_status = product_type_and_status.message
if (product_type_and_status.get(field_const.CONTEXT_TYPE) ==
field_const.DIGITAL_CONTEXT_TYPE):
album_data = salessheets.get_album_for_digital(product_id)
genre_data = salessheets.get_genre_and_subgenre(product_id)
if genre_data:
album_data.message.update(genre_data.message)
else:
album_data = salessheets.get_album(product_id)
if album_data:
album_info = album_data.message
elif album_data.status == 404:
return response.create_not_found_response(
message=errors.ERROR_MESSAGE_INVALID_PRODUCT_ID)
else:
raise GettingDataException(
salessheets.get_album,
album_data.errors)
return response.Response(message=album_info)
@tracer.wrap()
def get_pdf_data_from_db(release_id, localized_template_type_id):
"""
Function that pulls release data from db and formats it to the single dict.
Args:
release_id (str): release id pulled from sqs message.
localized_template_type_id (int): id of region specific template which
should be used for this sales sheet.
Raises:
GettingDataException.
Returns:
Response: Response with dictionary with release data in message.
"""
try:
product_type = salessheets.get_product_type_id(release_id)
if product_type or product_type.status == 404:
product_type = product_type.message or {}
else:
raise GettingDataException(
salessheets.get_product_type_id,
product_type.errors)
product_type_validation = validate_product_data(product_type)
if not product_type_validation:
return product_type_validation
album_data = get_album_data_for_product(release_id)
if not album_data:
return album_data
album_info = album_data.message
artists_data = salessheets.get_artists(release_id)
if artists_data or artists_data.status == 404:
artists_list = artists_data.message or []
else:
raise GettingDataException(
salessheets.get_artists,
artists_data.errors)
tracks_data = salessheets.get_tracks(release_id)
if tracks_data or tracks_data.status == 404:
tracks = tracks_data.message or {}
else:
raise GettingDataException(
salessheets.get_tracks,
tracks_data.errors)
tracks_list = format_tracks_by_cd(
tracks
).message
upc = album_info[field_const.DISPLAY_UPC]
data_from_db = {
field_const.RELEASE_ID: release_id,
field_const.PROJECT_ID: album_info[field_const.PROJECT_ID],
field_const.CONTEXT_TYPE: album_info[field_const.CONTEXT_TYPE],
field_const.LABEL: album_info[field_const.LABEL],
field_const.RELEASE_UPC: album_info[field_const.UPC],
field_const.UPC: upc,
field_const.PRODUCT_CODE: album_info[field_const.PRODUCT_CODE],
field_const.GENRE: album_info.get(field_const.GENRE),
field_const.SUBGENRE: album_info.get(field_const.SUBGENRE),
field_const.WHOLESALE_PRICE:
album_info[field_const.WHOLESALE_PRICE],
field_const.HEADER_MAIN_COLOR: pdf.ORCHARD_HEADER_MAIN_COLOR,
field_const.SALE_START_DATE:
album_info[field_const.SALE_START_DATE],
field_const.PRODUCT_NAME: album_info[field_const.PRODUCT_NAME],
field_const.TRACK_LIST: tracks_list,
field_const.ARTIST_LIST: artists_list,
field_const.DESCRIPTION: album_info[field_const.DESCRIPTION],
field_const.VENDOR_NAME: album_info[field_const.VENDOR_NAME],
field_const.VENDOR_OWNER: album_info[field_const.VENDOR_OWNER],
field_const.IMAGE_LOGO_ORCHARD: pdf.PDF_IMAGE_LOGO_ORCHARD,
field_const.ORCHARD_CONTACT: pdf.ORCHARD_HARDCODED_CONTACT_DATA
}
if (album_info[field_const.CONTEXT_TYPE] ==
field_const.DIGITAL_CONTEXT_TYPE):
data_from_db[field_const.EXPLICIT] = (
field_const.EXPLICIT_YES
if is_digital_product_explicit(tracks)
else field_const.EXPLICIT_NO)
if (album_info[field_const.CONTEXT_TYPE] ==
field_const.PHYSICAL_CONTEXT_TYPE):
fields_to_update = (
field_const.PRICING, field_const.BOX_LOT,
field_const.EXPORTABLE, field_const.EXPLICIT,
field_const.EDITION)
data_from_db.update({k: album_info[k] for k in fields_to_update})
barcode_generation_result = (
barcode_generator.generate_barcode(
data_from_db[field_const.UPC]))
if (barcode_generation_result or
barcode_generation_result.status == 400):
barcode_data = barcode_generation_result.message or {}
else:
raise GeneratingBarcodeException(
barcode_generator.generate_barcode,
barcode_generation_result.errors)
if barcode_data:
# The barcode path is constructed relative to the location of
# rendered template file orch_template.html where the barcode
# image is inserted.
# Path example: '../results/barcodes/111111111111.png'
barcode_path = os.path.join(
'../', barcode_data[barcode.BARCODE])
data_from_db.update({
field_const.BARCODE: barcode_path})
display_configuration = (
salessheets.get_distribution_format(release_id))
if display_configuration or (
display_configuration.status == 404):
display_configuration = display_configuration.message or {}
else:
raise GettingDataException(
salessheets.get_distribution_format,
display_configuration.errors)
if display_configuration:
data_from_db.update({
field_const.CONFIGURATION:
display_configuration[
field_const.DISPLAY_DISTRIBUTION_FORMAT]
})
template_details_response = (
template_details.get_by_template_type_id(
localized_template_type_id))
if not template_details_response:
return template_details_response
template_details_data = template_details_response.message
data_from_db[field_const.REGIONS] = template_details_data[
field_const.REGIONS]
data_from_db[
field_const.TEMPLATE_DETAILS_DATA] = template_details_data
contact_data = {
contact_field: template_details_data.get(contact_field)
if template_details_data.get(contact_field) else ''
for contact_field in field_const.CONTACT_FIELDS
}
template_name = template_details_data[field_const.NAME]
localized_contact_data = (
generation_utils.compose_address_for_localized(
template_name,
localized_template.CONTACT_DATA_FORMAT_MAPPING,
contact_data))
data_from_db[field_const.ORCHARD_CONTACT] = contact_data
data_from_db[
localized_template.LOCALIZED_ORCHARD_CONTACT] = (
localized_contact_data)
return response.Response(message=data_from_db)
except KeyError as ex:
message = errors.ERROR_GETTING_DATA_MESSAGE
if sentry_capture_exception:
sentry_capture_exception()
return response.create_error_response(
code=errors.ERROR_PDF_GENERATION_CODE,
message={'exception': ex.args,
'description': message})
except GettingDataException as ex:
if sentry_capture_exception:
sentry_capture_exception()
return response.create_fatal_response(
message={'exception': ex.args})
@tracer.wrap()
def get_sales_goals(
product_id,
project_id,
regions,
):
"""Get sales goals."""
sales_goals_response = ows_sales_goals\
.get_marketing_highlights_for_project(project_id)
success = sales_goals_response.status == 200
if success:
sales_goals = []
for sales_goal in sales_goals_response.message:
country_id = sales_goal.get('country_id')
country_name = sales_goal.get('country_name')
marketing_highlight = sales_goal['marketing_highlight'].strip()
slogan = sales_goal.get('slogan')
valid_region = False
if country_id in regions:
valid_region = True
# The logic here is comparing regions to countries
# which is not always a one-to-one match
# thus this patching
if country_id == field_const.JAPAN_COUNTRY_ID and \
field_const.JAPAN_REGION_ID in regions:
valid_region = True
if not valid_region:
continue
if not marketing_highlight and not slogan:
continue
sales_goals.append({
'country': {
'country_id': country_id,
'name': country_name
},
'marketing_drivers': {
'value': marketing_highlight,
'slogan': slogan
},
})
return response.Response(message=sales_goals)
if sales_goals_response.status == 404:
return response.Response()
return sales_goals_response
@tracer.wrap()
def get_data_from_microservices(
pdf_data,
should_fetch_phonetic_translations=False):
"""Collect data from ows-microservices.
Pull data from different microservices and format it to the single dict.
Args:
pdf_data (dict): release data pulled from the database.
should_fetch_phonetic_translations (bool): Wether or not we should
fetch the phonetic translations from ows-product.
Returns:
Response: Response with dictionary with release data in message.
"""
data = {}
product_id = pdf_data[field_const.RELEASE_ID]
project_id = pdf_data[field_const.PROJECT_ID]
image_path_response = ows_assets.get_large_cover_image_url(product_id)
if image_path_response:
data[field_const.LARGE_COVER_IMAGE] = image_path_response.message
mkt_highlights_for_project = ows_marketing.get_marketing_highlights(
ows_services.PROJECT, project_id)
if mkt_highlights_for_project:
filtered_marketing_highlights = (
ows_marketing.filter_marketing_highlights(
mkt_highlights_for_project.message,
mkt_program_id=ows_services
.ID_FOR_MARKETING_HIGHLIGHTS_PROGRAM_NAME,
scope=ows_services.MARKETING_HIGHLIGHTS_PUBLIC_SCOPE
))
data[field_const.MARKETING_HIGHLIGHTS] = (
filtered_marketing_highlights)
mkt_highlights_for_release = ows_marketing.get_marketing_highlights(
ows_services.RELEASE, product_id)
if mkt_highlights_for_release:
product_highlights = (
ows_marketing.filter_marketing_highlights(
mkt_highlights_for_release.message,
mkt_program_id=ows_services
.ID_FOR_MARKETING_HIGHLIGHTS_PROGRAM_NAME,
scope=ows_services.PUBLIC))
data[field_const.PRODUCT_HIGHLIGHTS] = product_highlights
sales_goals_response = get_sales_goals(
product_id,
project_id,
pdf_data[field_const.REGIONS],
)
if sales_goals_response.status == 200:
data[field_const.SALES_GOALS] = sales_goals_response.message
# The price should only be shown for physical products for now.
template_details_data = pdf_data.get(field_const.TEMPLATE_DETAILS_DATA)
store_id = template_details_data.get(localized_template.STORE_ID)
price_response = (
ows_pricing
.get_pricing_for_pricing_family_by_product_and_store_ids(
product_id=pdf_data.get(field_const.RELEASE_ID),
pricing_family_id=localized_template.PHYS_PRICING_FAMILY_ID,
store_id=store_id))
if not price_response:
return price_response
price_field_name, price_field_value = (
generation_utils
.extract_price_data_for_localized_template(
pdf_data.get(field_const.TEMPLATE_DETAILS_DATA),
price_response.message))
price_data = {
localized_template.SHEET_LABEL: price_field_name,
localized_template.SHEET_PRICE_VALUE: price_field_value}
data[field_const.PRICE] = price_data
if should_fetch_phonetic_translations:
translations_repsonse = ows_product.get_japan_phonetic_translations(
product_id)
if not translations_repsonse:
return translations_repsonse
(japan_release_name, japan_artist_name) = extract_japan_translations(
translations_repsonse.message['items'])
data[field_const.JAPAN_RELEASE_NAME] = japan_release_name
data[field_const.JAPAN_ARTIST_NAME] = japan_artist_name
return response.Response(message=data)
@tracer.wrap()
def validate_product_data(product_data):
"""Validate album data pulled from database.
Release should be an audio-type (releases.product_type_id == 1).
Args:
product_data (dict): product_data pulled from the database.
Returns:
Response: Response of dictionary with error or empty message if valid.
"""
if not product_data:
return response.create_not_found_response(
message=errors.ERROR_EMPTY_PRODUCT_TYPE_FROM_DB)
if product_data[pdf.PRODUCT_TYPE_ID] != pdf.PRODUCT_TYPE_MUSIC:
return response.create_error_response(
code=errors.ERROR_INVALID_PRODUCT_TYPE_CODE,
message=errors.ERROR_INVALID_PRODUCT_TYPE_MESSAGE,
status=400)
return response.Response()
def cleanup_files(files_to_cleanup):
"""Remove unecessary files with this function.
Args:
files_to_cleanup (list): generated files to remove.
"""
for file in files_to_cleanup:
misc.remove_file(file)
def create_output_file(output_type, product_ids, job_id):
"""Create bulk file for list of ids.
Args:
output_type (str): what extension resulting file
will have.
product_ids (list of str): list of numeric strings
that represent ids.
job_id (str): job id needed for bulk file name
generation.
Returns:
Response: response contains filename of generated
file in same directory where individual
files were generated.
"""
if output_type == pdf.SINGLE_PDF:
result = create_bulk_pdf(product_ids, job_id)
elif output_type == pdf.ZIPPED_PDF:
result = create_bulk_zip(product_ids, job_id)
return result
@tracer.wrap()
def create_bulk_pdf(product_ids, job_id):
"""Create bulk pdf file."""
rendering_dir_abspath = os.path.abspath(config.PDF_RENDERING_DIR)
merger = PdfMerger()
try:
for product_id in product_ids:
file_name = (
'{dir}/{product_id}.pdf'
.format(
dir=rendering_dir_abspath,
product_id=product_id))
if not os.path.isfile(file_name):
continue
merger.append(file_name)
output_file = (
generate_bulk_sales_sheets_filename(
pdf.PDF_EXTENSION, job_id))
merger.write(os.path.join(rendering_dir_abspath, output_file))
except (TypeError, PdfReadError, PageSizeNotDefinedError,
PdfStreamError, PyPdfError) as e:
if sentry_capture_exception:
sentry_capture_exception()
return response.create_error_response(
code=errors.ERROR_PDF_GENERATION_CODE,
message=e.args)
finally:
merger.close()
return response.Response(output_file)
@tracer.wrap()
def create_bulk_zip(product_ids, job_id):
"""Create bulk zip file."""
rendering_dir_abspath = os.path.abspath(config.PDF_RENDERING_DIR)
try:
output_file = (
generate_bulk_sales_sheets_filename(
pdf.ZIP_EXTENSION, job_id))
zip_file = zipfile.ZipFile(
'{dir}/{file}'
.format(
dir=rendering_dir_abspath,
file=output_file), 'w')
for product_id in product_ids:
file_name = (
'{dir}/{product_id}.pdf'
.format(
dir=config.PDF_RENDERING_DIR,
product_id=product_id))
if not os.path.isfile(file_name):
continue
zip_file.write(file_name)
except (AttributeError, OSError, RuntimeError) as e:
if sentry_capture_exception:
sentry_capture_exception()
return response.create_error_response(
code=errors.ERROR_PDF_GENERATION_CODE,
message=e.args)
finally:
zip_file.close()
return response.Response(output_file)
def generate_bulk_sales_sheets_filename(extension, job_id):
"""Generate filename for file.
This function generates filename in format
YYYYMMDD_{job_id}_prefix.
Args:
extension (str): either zip or pdf.
job_id (str): job id retrieved from message.
Returns:
str: generated filename for bulk sales sheet.
"""
files_timestamp = misc.datetime_to_str(
datetime.today(),
pdf.BULK_SALES_SHEETS_TIMESTAMP_FORMAT)
output_file_name = (
'{timestamp}_{job_id}{suffix}.{extension}'
.format(
timestamp=files_timestamp,
job_id=job_id,
suffix=pdf.BULK_SALES_SHEETS_FILENAME_SUFFIX,
extension=extension))
return output_file_name
@tracer.wrap()
def create_pdf(product_id, localized_template_type_id, user_id=None):
"""Create pdf from id.
Args:
product_id (str): numeric string that corresponds id.
localized_template_type_id (int): id of region specific template which
should be used for this sales sheet.
user_id (str): The Orchard User ID.
Returns:
Response: contains dict list of files to be removed
and data_for_rendering.
"""
user_features_response = features.check_user_features(user_id)
if not user_features_response:
return user_features_response
user_features = user_features_response.message
pdf_data = get_pdf_data_from_db(product_id, localized_template_type_id)
if not pdf_data:
return pdf_data
japan_template_id = localized_template.TEMPLATE_DETAILS_IDS['JP']
should_fetch_phonetic_translations = (
localized_template_type_id == japan_template_id
and
user_features.get('is_japan_sales_sheets_enabled'))
data_from_microservices = get_data_from_microservices(
pdf_data.message,
should_fetch_phonetic_translations)
if data_from_microservices:
pdf_data.message.update(data_from_microservices.message)
data_for_rendering = template_rendering.prepare_template_data(
pdf_data.message)
if not data_for_rendering:
return data_for_rendering
data_for_rendering.message[field_const.USER_FEATURES] = user_features
html_render_response = template_rendering.render_template(
'orch_template.html', data_for_rendering.message)
if not html_render_response:
return html_render_response
html_data = html_render_response.message
if not os.path.exists(config.PDF_RENDERING_DIR + '/'):
os.makedirs(config.PDF_RENDERING_DIR)
body_template_file = '{dir}/{id}.html'.format(
dir=config.PDF_RENDERING_DIR, id=product_id)
header_template_file = '{dir}/{id}_header.html'.format(
dir=config.PDF_RENDERING_DIR, id=product_id)
rendered_pdf_file = '{dir}/{id}.pdf'.format(
dir=config.PDF_RENDERING_DIR, id=product_id)
with open(body_template_file, 'w+', encoding='utf8') as f:
f.write(html_data)
html_render_header_response = template_rendering.render_template(
'orch_header.html', data_for_rendering.message)
if not html_render_header_response:
return html_render_header_response
html_header_data = html_render_header_response.message
with open(header_template_file, 'w+', encoding='utf8') as f:
f.write(html_header_data)
htmltopdf.convert_html_to_pdf(
body_template_file,
rendered_pdf_file,
{
'header-spacing': pdf.HEADER_SPACING,
'header-html': header_template_file
}
)
return response.Response({
field_const.GENERATED_FILES:
[body_template_file,
header_template_file,
rendered_pdf_file],
field_const.DATA_FOR_RENDERING:
data_for_rendering
})
def is_digital_product_explicit(tracks):
"""
Checks whether digital product is explicit or not.
Returns true if the product contains at least one track with explicit equal
to true.
Args:
tracks (dict): track list pulled from database.
Returns:
Boolean.
"""
return any((track['explicit'] == 'Y' for track in tracks.values()))
def get_user_id(message):
"""Extract the Orchard User Id from the message.
Args:
message (JSONMessageExt): message pulled from SQS-message.
Returns:
str or None: The user ID or None if not present.
"""
if message['MessageAttributes']:
return message['MessageAttributes']['feature_flag_user_context']['StringValue'] # noqa
return None
def extract_japan_translations(translations):
"""Extract the translated release_name and artist_name.
Args:
translations ([dict]): The list of phonetic translations.
Returns:
tuple: With the translated release_name and artist_name.
"""
japan_release_name = None
japan_artist_name = None
for t in translations:
if t['field_name'] == 'release_name':
japan_release_name = t['phonetic_translation']
elif t['field_name'] == 'artist_name':
japan_artist_name = t['phonetic_translation']
return (japan_release_name, japan_artist_name)