"""DDEX Supply Chain Status Message XML generator.""" from datetime import datetime from xml.dom.minidom import DOMImplementation from ddex_ingester_common.constants.send_acknowledgement import \ ORCHARD_DDEX_PARTY_ID, SME_DDEX_PARTY_ID class DDEXSupplyChainStatusMessage: """DDEX XML generator.""" def __init__(self, grid, status): """Construct. Args: grid (str): GRid status (str): success of failure """ self.grid = grid self.status = status self.xml_doc = DOMImplementation.createDocument( DOMImplementation(), namespaceURI=None, qualifiedName=None, doctype=None) def output_xml(self): """Parent method to build out the xml file.""" self._generate_ddex_scsm_message() self._generate_message_header() self._generate_dsp() self._generate_release_id() status = self.xml_doc.createElement('Status') status.appendChild(self.xml_doc.createTextNode(self.status)) self.root_node.appendChild(status) return self.xml_doc.toxml(encoding='UTF-8') def _generate_ddex_scsm_message(self): """Set attributes for ManifestMessage element.""" self.root_node = self.xml_doc.createElement( 'echo:SupplyChainStatusMessage') # Set ManifestMessage attributes self._append_attribute( self.root_node, 'xmlns:echo', 'http://ddex.net/xml/ern-c/15', 'http://ddex.net/xml/ern-c/15') self._append_attribute( self.root_node, 'MessageVersionId', 'http://www.w3.org/2001/XMLSchema-instance', '1.5') self._append_attribute( self.root_node, 'xmlns:ds', 'http://www.w3.org/2000/09/xmldsig#', 'http://www.w3.org/2000/09/xmldsig#') self.xml_doc.appendChild(self.root_node) def _generate_message_header(self): """Generate the message header for xml file.""" created_datetime = datetime.utcnow().replace( microsecond=0).isoformat() + 'Z' # Begin MessageHeader elements msg_header = self.xml_doc.createElement('MessageHeader') # Begin WsMessageId ws_msg_id = self.xml_doc.createElement('WsMessageId') ws_msg_id.appendChild(self.xml_doc.createTextNode( f'{self.grid}_{created_datetime}')) msg_header.appendChild(ws_msg_id) # Begin MessageSender elements msg_sender = self.xml_doc.createElement('MessageSender') sender_party_id = self.xml_doc.createElement('PartyId') # Begin MessageSender PartyId sender_party_id.appendChild( self.xml_doc.createTextNode(ORCHARD_DDEX_PARTY_ID)) # 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) # Insert MessageSender elements msg_sender.appendChild(sender_party_id) msg_sender.appendChild(sender_party_name) msg_header.appendChild(msg_sender) # 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(SME_DDEX_PARTY_ID)) # 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('Sony Music Entertainment')) recipient_party_name.appendChild(recipient_party_full_name) # Insert MessageRecipient elements msg_recipient.appendChild(recipient_party_id) msg_recipient.appendChild(recipient_party_name) msg_header.appendChild(msg_recipient) # Begin MessageCreatedDateTime element msg_created_datetime = self.xml_doc.createElement( 'MessageCreatedDateTime') msg_created_datetime.appendChild( self.xml_doc.createTextNode(created_datetime)) msg_header.appendChild(msg_created_datetime) # Begin IsSymmetric element is_symmetric = self.xml_doc.createElement('IsSymmetric') is_symmetric.appendChild(self.xml_doc.createTextNode('false')) msg_header.appendChild(is_symmetric) # Begin Priority element priority = self.xml_doc.createElement('Priority') priority.appendChild(self.xml_doc.createTextNode('Normal')) msg_header.appendChild(priority) self.root_node.appendChild(msg_header) def _generate_dsp(self): """Generate DSP tag.""" # Begin DSP element dsp = self.xml_doc.createElement('DSP') party_id = self.xml_doc.createElement('PartyId') party_id.appendChild( self.xml_doc.createTextNode(ORCHARD_DDEX_PARTY_ID)) dsp.appendChild(party_id) self.root_node.appendChild(dsp) def _generate_release_id(self): """Generate ReleaseId tag.""" # Begin ReleaseId element release_id = self.xml_doc.createElement('ReleaseId') grid = self.xml_doc.createElement('GRid') grid.appendChild(self.xml_doc.createTextNode(self.grid)) release_id.appendChild(grid) self.root_node.appendChild(release_id) def _append_attribute( self, parent_node, q_name, namespace_uri=None, value=None): """Append an attribute to an element.""" element = self.xml_doc.createAttribute(q_name) if namespace_uri: element = self.xml_doc.createAttributeNS( namespaceURI=namespace_uri, qualifiedName=q_name) if value: text_node = self.xml_doc.createTextNode(value) element.nodeValue = text_node.nodeValue parent_node.setAttributeNode(element)