"""For delivery manifest file generation.""" from xml.dom.minidom import DOMImplementation from datetime import datetime from vector_utils import queries from src.ddex_manifestor import DDEXManifestor import xmlschema from src import config class Manifestor: """Manifestor class.""" def __init__( self, manifest_format, remote_initial_dir, batch_info, manifest_url=None): """Init. Args: manifest_format (str): manifest_url (str): batch_info (dict): """ self._manifest_format = manifest_format self._remote_initial_dir = remote_initial_dir self._manifest_url = manifest_url self._upcs = batch_info['delivered_upcs'] self._remote_folder = batch_info['remote_folder'] self._dms_id = batch_info['dms_id'] @property def manifest_format(self): """Manifest format getter.""" return self._manifest_format @manifest_format.setter def manifest_format(self, value): self._manifest_format = value @property def remote_initial_dir(self): """Remote initial dir getter.""" return self._remote_initial_dir @remote_initial_dir.setter def remote_initial_dir(self, value): self._remote_initial_dir = value @property def manifest_url(self): """Manifest URL getter.""" return self._manifest_url @manifest_url.setter def manifest_url(self, value): self._manifest_url = value @property def upcs(self): """UPC s getter.""" return self._upcs @upcs.setter def upcs(self, value): self._upcs = value @property def remote_folder(self): """Remote folder getter.""" return self._remote_folder @remote_folder.setter def remote_folder(self, value): self._remote_folder = value @property def dms_id(self): """DMS ID getter.""" return self._dms_id @dms_id.setter def dms_id(self, value): self._remote_folder = value def _manifest_247(self): manifest_content = [] for upc in self.upcs: lines = [] if not upc['meta_update']: result = queries.get_release_track_info( [upc['upc']], False, conn_info=config.AR_MYSQL_CONN_INFO) for row in result: track_code = f"{row['upc']}/{row['upc']}_{row['cd']:02}_{row['track_id']:03}" # noqa if row['track_type'] == 'music': lines.append('{}_FULL.flac'.format(track_code)) else: lines.append('{}_1500_FULL.wmv'.format(track_code)) lines.append('{}_700_FULL.wmv'.format(track_code)) lines.append('{}_176_FULL.mp4'.format(track_code)) lines.append('{}_450_CLIP.wmv'.format(track_code)) lines.append( '{upc}/{upc}_COVER.jpg'.format(upc=upc['upc'])) lines.append('{upc}/{upc}.xml'.format(upc=upc['upc'])) manifest_content.append('\r\n'.join(line for line in lines)) return '\r\n'.join(lines for lines in manifest_content) def _manifest_amazon(self): upcs = [upc['upc'] for upc in self.upcs] result = queries.get_release_track_info( upcs, conn_info=config.AR_MYSQL_CONN_INFO) unique_upcs = sorted(set(str(row['content_id']) for row in result)) xml_doc = DOMImplementation.createDocument( DOMImplementation(), None, None, None) root_node = xml_doc.createElement('Feed') xml_doc.appendChild(root_node) feed_id_node = xml_doc.createElement('FeedId') feed_id_node.appendChild(xml_doc.createTextNode(self.remote_folder)) root_node.appendChild(feed_id_node) bundle_count_node = xml_doc.createElement('BundleCount') bundle_count_node.appendChild(xml_doc.createTextNode(str(len(upcs)))) root_node.appendChild(bundle_count_node) for upc in unique_upcs: album_bundle_node = xml_doc.createElement('AlbumBundle') root_node.appendChild(album_bundle_node) dir_name_node = xml_doc.createElement('DirectoryName') dir_name_node.appendChild(xml_doc.createTextNode(upc)) album_bundle_node.appendChild(dir_name_node) return xml_doc.toxml(encoding='UTF-8').decode() def _manifest_eins(self): """Return stringified xml manifest for eins.""" upcs = [upc['upc'] for upc in self.upcs] rows = queries.get_release_track_info( upcs, False, conn_info=config.AR_MYSQL_CONN_INFO) # Hydrate rows with data from query lines = '' for row in rows: line = f""" {row['artist_name']} {row['release_name']} {row['cd']} {row['track_id']} {row['track_name']} {row['upc']} {row['isrc']} """ lines += line with open('./src/eins_manifest.xml') as eins_manifest_template: eins_manifest = eins_manifest_template.read() # Replace all the placeholders in the xml template file return eins_manifest.replace( '{{datetime_iso}}', # The replaced date should be in an ISO 8601 format datetime.utcnow().replace(microsecond=0).isoformat() ).replace( '{{$rows}}', str(len(rows)) ).replace( '{{$lines}}', lines ) def _manifest_ddex(self): """Generate ddex format xml.""" ddex_file = DDEXManifestor( self.upcs, self.dms_id, self.remote_folder, self.remote_initial_dir) output = ddex_file.output_xml() schema = xmlschema.XMLSchema('src/ern-choreography-sftp.xsd') decoded_output = output.decode() schema.validate(decoded_output) return decoded_output def _manifest_display_upc(self): """Generate manifest with display UPC's. Returns (str): Manifest with display UPC's """ upcs = [upc['upc'] for upc in self.upcs] result = queries.get_release_track_info( upcs, conn_info=config.AR_MYSQL_CONN_INFO) return '\r\n'.join( sorted(set(str(row['display_upc']) for row in result))) def _manifest_default(self): return '\r\n'.join(str(upc['upc']) for upc in self.upcs) def generate_manifest(self): """Generate manifest.""" format_selection = { '247': self._manifest_247, 'amazon': self._manifest_amazon, 'eins': self._manifest_eins, 'ddex_ern_c_sftp_1_7': self._manifest_ddex, 'display_upc': self._manifest_display_upc } manifest_to_return = format_selection.get( self.manifest_format) or self._manifest_default return manifest_to_return() def notify_xml_error(self, manifest_data): """Check to see if it's valid XML data. Args: manifest_data (str): Returns: bool """ pass def get_post_data_from_upcs(self): """Get post data from UPC's.""" post_data = '' for upc in self.upcs: post_data += 'upc_arr[]={}&'.format(upc) return post_data.strip('&')