PAIN creation file

pain.py

from lxml import etree, objectify
import datetime
import time
import os
import random

PAIN_IDENTIFIER = 'IA'
XSD_PATH = 'xsd'

class PainError(Exception):
    pass

class PainXMLError(PainError):
    pass

class Pain(object):
    def _get_timestamp(self):
        """Returns now as xml timestamp"""
        t = datetime.datetime.utcnow()
        return t.strftime("%Y-%m-%dT%H:%M:%S")
    
    def _get_today(self):
        """Returns today as yyyy-mm-dd"""
        t = datetime.datetime.utcnow()
        return t.strftime("%Y-%m-%d")
    
    def _get_date_in_future(self, business_days):
        """Gets a day business_days working days in the future. Note that this
        only works for """
        
        t = datetime.date.today()
        while business_days > 0:
            t += datetime.timedelta(days=1)
            if t.weekday() < 5: # only count if day is not sat or sun
                business_days -= 1
        return t.strftime("%Y-%m-%d")
    
    def _get_random_identifier(self, addit=PAIN_IDENTIFIER):
        """Returns a random identifier"""
        
        utcdate = time.strftime('%Y%m%d', time.gmtime(time.time()))
        randint = random.randrange(1000)
        return '%s.%s.%s' % (addit, utcdate, randint)
    
    def _get_xml_builder(self):
        """Returns an ElementMaker for the pain namespace"""
        return objectify.ElementMaker(annotate=False, namespace=self.NS, nsmap={None: self.NS})
    


class PainCreditTransfer(Pain):
    NS = "urn:iso:std:iso:20022:tech:xsd:pain.001.001.03"
    XSD = XSD_PATH + 'pain.001.001.03.xsd'
    
    transactions = []
    total_amount = 0
    
    def __init__(self, name, iban, bic=None, identifier=None):
        """Create a new credit transfer from the account indicated by the name,
        iban and opitonal bic.
        """
        
        self.transactions = []
        self.total_amount = 0
        
        self.name = name
        self.iban = iban
        self.bic = bic
        
        if identifier is None:
            identifier = self._get_random_identifier('%s.EXPORT' % PAIN_IDENTIFIER)
        self.identifier = identifier
        
        schema = etree.XMLSchema(file=self.XSD)
        self._parser = objectify.makeparser(schema=schema)
        
    
    def add_transaction(self, amount, end_to_end, description, name, iban, bic=None, internal_id=None):
        """Adds a transaction to the file.
        
        amount      -- the amount of the transaction
        end_to_end  -- the end to end id
        description -- The description of the payment
        name        -- The name of the receiver of the payment
        iban        -- The iban of the same
        bic         -- the optional bic of the same, optional
        internal_id -- internal id of the transaction, optional
        """
        
        b = self._get_xml_builder()
        
        self.total_amount += amount
        self.transactions.append(
            b.CdtTrfTxInf(
                b.PmtId(
                    internal_id and b.InstrId(internal_id),
                    b.EndToEndId(end_to_end)
                ),
                b.Amt(
                    b.InstdAmt(amount, Ccy='EUR')
                ),
                bic and b.CdtrAgt(
                    b.FinInstnId(
                        b.BIC(bic)
                    )
                ),
                b.Cdtr(
                    b.Nm(name)
                ),
                b.CdtrAcct(
                    b.Id(
                         b.IBAN(iban)
                    )
                ),
                b.RmtInf(
                    b.Ustrd(description)
                )
            )
        )
    
    def to_xml(self, execution_date=None, batch=False):
        """Generate the xml. Indicate with the string 'batch' whether a batch is
        required."""
        
        if execution_date is None:
            execution_date = self._get_today()
        
        b = self._get_xml_builder()
        
        doc = b.Document(
            b.CstmrCdtTrfInitn(
                b.GrpHdr(
                    b.MsgId(self.identifier),
                    b.CreDtTm(self._get_timestamp()),
                    b.NbOfTxs(len(self.transactions)),
                    b.CtrlSum(self.total_amount),
                    b.InitgPty(
                        b.Nm(self.name)
                    )
                ),
                b.PmtInf(
                    b.PmtInfId(self.identifier),
                    b.PmtMtd('TRF'),
                    b.BtchBookg(batch and 'true' or 'false'),
                    b.ReqdExctnDt(execution_date),
                    b.Dbtr(
                        b.Nm(self.name)
                    ),
                    b.DbtrAcct(
                        b.Id(
                            b.IBAN(self.iban)
                        )
                    ),
                    b.DbtrAgt(
                        b.FinInstnId(
                            b.BIC(self.bic)
                        )
                    ),
                    *self.transactions
                )
            )
        )
                
        string = etree.tostring(doc, pretty_print=True, encoding='utf-8', xml_declaration=True)
        
        try:
            objectify.fromstring(string, self._parser)
        except etree.XMLSyntaxError as e:
            raise PainXMLError(str(e)) 
        
        return string



class PainDirectDebit(Pain):
    NS = "urn:iso:std:iso:20022:tech:xsd:pain.008.001.02"
    XSD = XSD_PATH + 'pain.008.001.02.xsd'
    
    transactions = []
    total_amount = 0
    
    FIRST = 'FRST'
    RECURRING = 'RCUR'
    FINAL = 'FNAL'
    ONE_OFF = 'OOFF'
    
    PROCESSING_TIMES = {FIRST: 6,
                        RECURRING: 3,
                        FINAL: 3,
                        ONE_OFF: 6}
    
    def __init__(self, name, creditor_id, iban, bic=None, identifier=None):
        """Create a new direct debit for the account indicated by the name,
        iban and opitonal bic.
        """
        
        self.transactions = {}
        self.total_amount = 0
        
        self.name = name
        self.creditor_id = creditor_id
        self.iban = iban
        self.bic = bic
        
        if identifier is None:
            identifier = self._get_random_identifier('%s.DDEBIT' % PAIN_IDENTIFIER)
        self.identifier = identifier
        
        schema = etree.XMLSchema(file=self.XSD)
        self._parser = objectify.makeparser(schema=schema)
    
    def add_transaction(self, amount, sequence, mandate_id, mandate_date, 
                        end_to_end, description, name, iban, bic=None, internal_id=None):
        """Adds a transaction to the file.
        
        amount       -- The amount to debit.
        sequence     -- The batch type of this transaction. Must be:
                        PainDirectDebit.FIRST, PainDirectDebit.RECURRING,
                        PainDirectDebit.FINAL, PainDirectDebit.ONE_OFF
        mandate_id   -- Unique mandate identifier.
        mandate_date -- The date when the mandate was signed. 
        end_to_end   -- The end-to-end ID.
        description  -- The description of the transaction
        name         -- The name of the creditor
        iban         -- The IBAN of the same
        bic          -- Optional BIC
        internal_id  -- Optional internal ID
        """
        
        b = self._get_xml_builder()
        
        self.total_amount += amount
        
        # Create sequence group if not already exists
        if sequence not in self.transactions:
            self.transactions[sequence] = []
        
        self.transactions[sequence].append(
            b.DrctDbtTxInf(
                b.PmtId(
                    internal_id and b.InstrId(internal_id),
                    b.EndToEndId(end_to_end)
                ),
                b.InstdAmt(amount, Ccy='EUR'),
                b.DrctDbtTx(
                    b.MndtRltdInf(
                        b.MndtId(mandate_id),
                        b.DtOfSgntr(mandate_date),
                    ),
                ),
                b.DbtrAgt(
                    b.FinInstnId() if not bic else b.FinInstnId(
                        b.BIC(bic)
                    )
                ),
                b.Dbtr(
                    b.Nm(name)
                ),
                b.DbtrAcct(
                    b.Id(
                         b.IBAN(iban)
                    )
                ),
                b.RmtInf(
                    b.Ustrd(description)
                )
            )
        )
    
    def to_xml(self, collection_date=None, batch=True, local_instrument='CORE'):
        """Generate the xml. Indicate with the string 'batch' whether a batch is
        required. local_instrument depends on bank, and may be CORE or B2B
        """
        if collection_date is None:
            collection_date = self._get_date_in_future(PainDirectDebit.PROCESSING_TIMES[sequence_type])
        
        b = self._get_xml_builder()
        
        # We group all transactions by their sequence type and every sequence is
        # a separate batch. We count the transactions separately.
        sequences = []
        transactions_length = 0
        for sequence_type,transactions in self.transactions.items():
            seq_identification = self.identifier.replace('DDEBIT', sequence_type)
            
            sequences.append(
                b.PmtInf(
                    b.PmtInfId(seq_identification),
                    b.PmtMtd('DD'),
                    b.BtchBookg(batch and 'true' or 'false'),
                    b.PmtTpInf(
                        b.SvcLvl(
                            b.Cd('SEPA')
                        ),
                        b.LclInstrm(
                            b.Cd(local_instrument)
                        ),
                        b.SeqTp(sequence_type)
                    ),
                    b.ReqdColltnDt(collection_date),
                    b.Cdtr(
                        b.Nm(self.name)
                    ),
                    b.CdtrAcct(
                        b.Id(
                            b.IBAN(self.iban)
                        )
                    ),
                    b.CdtrAgt(
                        b.FinInstnId(
                            b.BIC(self.bic)
                        )
                    ),
                    b.CdtrSchmeId(
                        b.Id(
                            b.PrvtId(
                                b.Othr(
                                    b.Id(self.creditor_id),
                                    b.SchmeNm(
                                        b.Prtry('SEPA')
                                    )
                                )
                            )
                        )
                    ),
                    *transactions
                )
            )
            
            transactions_length += len(transactions)
        
        # Full document.
        doc = b.Document(
            b.CstmrDrctDbtInitn(
                b.GrpHdr(
                    b.MsgId(self.identifier),
                    b.CreDtTm(self._get_timestamp()),
                    b.NbOfTxs(transactions_length),
                    b.CtrlSum(self.total_amount),
                    b.InitgPty(
                        b.Nm(self.name)
                    )
                ),
                *sequences
            )
        )
                
        string = etree.tostring(doc, pretty_print=True, encoding='utf-8', xml_declaration=True)
        
        try:
            objectify.fromstring(string, self._parser)
        except etree.XMLSyntaxError as e:
            raise PainXMLError(str(e)) 
        
        return string