"""DDEX manifestor module.""" from xml.dom.minidom import DOMImplementation from datetime import datetime from vector_utils import queries from src import config # TODO: Create an XML generator class that hides the verbosity of # node and attribute operations. python-vector-utils would be the # best place to implement this class. class DDEXManifestor: """DDEX XML generator.""" def __init__(self, upcs, dms_id, remote_folder, remote_initial_dir): """Construct for DDEXManifestor. Args: upcs: Getting track info from relevant UPCs dms_id: the recipient store ID remote_folder: folder that resides in the store's storage remote_initial_dir: initial directory where folder lives on the store's storage """ self.upcs = {upc['upc']: upc['meta_update'] for upc in upcs} tracks_info = queries.get_release_track_info( self.upcs.keys(), conn_info=config.AR_MYSQL_CONN_INFO) self.msg_in_batch_upcs = { item['upc']: item['display_upc'] for item in tracks_info } self.xml_doc = DOMImplementation.createDocument( DOMImplementation(), namespaceURI=None, qualifiedName=None, doctype=None) self.root_node = self.xml_doc.createElement('ManifestMessage') self.store_info = queries.get_store_info( dms_id, config.AR_MYSQL_CONN_INFO) self.remote_folder = remote_folder self.remote_initial_dir = remote_initial_dir def output_xml(self): """Parent method to build out the xml file.""" self._generate_ddex_manifest_message() self._generate_message_header() # Create IsTestFlag element is_test_flag = self.xml_doc.createElement('IsTestFlag') test_flag_value = 'true' if config.ENVIRONMENT == 'dev' else 'false' is_test_flag.appendChild(self.xml_doc.createTextNode(test_flag_value)) self.root_node.appendChild(is_test_flag) # Create RootDirectory element root_directory_node = self.xml_doc.createElement('RootDirectory') root_directory_data = self.remote_folder.strip('/') if self.remote_initial_dir: root_directory_data = '/'.join( [self.remote_initial_dir.strip('/'), root_directory_data]) root_directory_data = '/' + root_directory_data root_directory_node.appendChild( self.xml_doc.createTextNode(root_directory_data)) self.root_node.appendChild(root_directory_node) # Create NumberOfMessages element num_of_messages = self.xml_doc.createElement('NumberOfMessages') num_of_messages.appendChild( self.xml_doc.createTextNode(str(len(self.upcs)))) self.root_node.appendChild(num_of_messages) self._generate_ddex_message_in_batch(root_directory_data) return self.xml_doc.toxml(encoding='UTF-8') def _generate_ddex_manifest_message(self): """Set attributes for ManifestMessage element.""" self.root_node = self.xml_doc.createElement('ManifestMessage') # Set ManifestMessage attributes self._append_attribute( self.root_node, 'xmlns:ern-c', 'http://ddex.net/xml/ern-c-sftp/17', 'http://ddex.net/xml/ern-c-sftp/17') self._append_attribute( self.root_node, 'MessageVersionId', 'http://www.w3.org/2001/XMLSchema-instance', '1.7') self._append_attribute( self.root_node, 'xs:schemaLocation', 'http://www.w3.org/2001/XMLSchema-instance', 'http://ddex.net/xml/ern-c-sftp/17/ern-choreography-sftp.xsd') self._append_attribute( self.root_node, 'xmlns:xs', 'http://www.w3.org/2001/XMLSchema-instance', 'http://www.w3.org/2001/XMLSchema-instance') self.xml_doc.appendChild(self.root_node) def _generate_message_header(self): """Generate the message header for xml file.""" # Begin MessageHeader elements msg_header = self.xml_doc.createElement('MessageHeader') # Begin MessageSender elements msg_sender = self.xml_doc.createElement('MessageSender') sender_party_id = self.xml_doc.createElement('PartyId') # Begin MessageSender PartyId self._append_attribute(sender_party_id, 'IsDPID', value='true') sender_party_id.appendChild( self.xml_doc.createTextNode('PADPIDA2012041004F')) # Begin MessageSender PartyName sender_party_name = self.xml_doc.createElement('PartyName') sender_party_full_name = self.xml_doc.createElement('FullName') sender_party_full_name.appendChild( self.xml_doc.createTextNode('The Orchard Enterprises')) sender_party_name.appendChild(sender_party_full_name) # Begin MessageSender TradingName sender_trading_name = self.xml_doc.createElement('TradingName') sender_trading_name.appendChild( self.xml_doc.createTextNode('The Orchard Enterprises, LLC')) # Insert MessageSender elements msg_sender.appendChild(sender_party_id) msg_sender.appendChild(sender_party_name) msg_sender.appendChild(sender_trading_name) # Begin MessageRecipient msg_recipient = self.xml_doc.createElement('MessageRecipient') # Begin MessageRecipient PartyId recipient_party_id = self.xml_doc.createElement('PartyId') recipient_party_id.appendChild(self.xml_doc.createTextNode( self.store_info['ddex_party_id'])) self._append_attribute(recipient_party_id, 'IsDPID', value='true') # Begin MessageRecipient PartyName recipient_party_name = self.xml_doc.createElement('PartyName') recipient_party_full_name = self.xml_doc.createElement('FullName') recipient_party_full_name.appendChild(self.xml_doc.createTextNode( self.store_info['customer_name'])) recipient_party_name.appendChild(recipient_party_full_name) # Insert MessageRecipient elements msg_recipient.appendChild(recipient_party_id) msg_recipient.appendChild(recipient_party_name) # Begin MessageCreatedDateTime element msg_created_datetime = self.xml_doc.createElement( 'MessageCreatedDateTime') created_datetime = datetime.utcnow().replace( microsecond=0).isoformat() + 'Z' msg_created_datetime.appendChild( self.xml_doc.createTextNode(created_datetime)) msg_header.appendChild(msg_sender) msg_header.appendChild(msg_recipient) msg_header.appendChild(msg_created_datetime) self.root_node.appendChild(msg_header) def _generate_ddex_message_in_batch(self, root_directory_data): """Dynamically generate MessageInBatch entries.""" for upc, display_upc in self.msg_in_batch_upcs.items(): msg_in_batch = self.xml_doc.createElement('MessageInBatch') msg_type = self.xml_doc.createElement('MessageType') # MessageType is always NewReleaseMessage msg_type.appendChild( self.xml_doc.createTextNode('NewReleaseMessage')) # MessageId will be UPCID_date_time format msg_id = self.xml_doc.createElement('MessageId') created_datetime = datetime.utcnow().strftime('%Y%m%d_%H%M%S') msg_id_data = '_'.join([str(upc), created_datetime]) msg_id.appendChild(self.xml_doc.createTextNode(msg_id_data)) # For URL we can use msg_url = '/'.join( [root_directory_data, display_upc, display_upc + '.xml']) msg_url_node = self.xml_doc.createElement('URL') msg_url_node.appendChild(self.xml_doc.createTextNode(msg_url)) # ICPN is display_upc included_release_id = self.xml_doc.createElement( 'IncludedReleaseId') msg_icpn = self.xml_doc.createElement('ICPN') msg_icpn.appendChild( self.xml_doc.createTextNode(display_upc)) included_release_id.appendChild(msg_icpn) # DeliveryType can be determined via the # batch_info's meta_update attribute delivery_type_node = self.xml_doc.createElement('DeliveryType') delivery_type = 'ReDelivery' # Check the meta_update flag if self.upcs.get(upc): delivery_type = 'NewReleaseDelivery' delivery_type_node.appendChild( self.xml_doc.createTextNode(delivery_type)) # ProductType can be hardcoded for now but we need to investigate # what other product types we need to support product_type = self.xml_doc.createElement('ProductType') product_type.appendChild( self.xml_doc.createTextNode('AudioProduct')) msg_in_batch.appendChild(msg_type) msg_in_batch.appendChild(msg_id) msg_in_batch.appendChild(msg_url_node) msg_in_batch.appendChild(included_release_id) msg_in_batch.appendChild(delivery_type_node) msg_in_batch.appendChild(product_type) self.root_node.appendChild(msg_in_batch) def _append_attribute( self, parentNode, qName, namespaceURI=None, value=None): """Append an attribute to an element.""" element = self.xml_doc.createAttribute(qName) if namespaceURI: element = self.xml_doc.createAttributeNS( namespaceURI=namespaceURI, qualifiedName=qName) if value: text_node = self.xml_doc.createTextNode(value) element.nodeValue = text_node.nodeValue parentNode.setAttributeNode(element)