admin/Odoo-Modules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
gsinghpal
2026-02-22 01:22:18 -05:00
commit 5200d5baf0
2394 changed files with 386834 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

448
Fusion Accounting/models/sepa_direct_debit.py Normal file
View File

@@ -0,0 +1,448 @@
"""
Fusion Accounting - SEPA Direct Debit (pain.008)
Generates ISO 20022 ``pain.008.001.02`` XML documents for SEPA Direct
Debit collections. Includes a mandate model
(``fusion.sdd.mandate``) to track debtor authorisations.
The XML structure follows the published
`ISO 20022 message definition
<https://www.iso20022.org/catalogue-messages/iso-20022-messages-archive>`_
and uses ``xml.etree.ElementTree`` for construction.
Namespace
---------
``urn:iso:std:iso:20022:tech:xsd:pain.008.001.02``
"""
import hashlib
import re
import uuid
import xml.etree.ElementTree as ET
from datetime import datetime
from odoo import api, fields, models, _
from odoo.exceptions import UserError, ValidationError
# ISO 20022 pain.008.001.02 namespace
PAIN_008_NS = 'urn:iso:std:iso:20022:tech:xsd:pain.008.001.02'
# Reuse validation helpers from SEPA Credit Transfer module
_IBAN_RE = re.compile(r'^[A-Z]{2}\d{2}[A-Z0-9]{4,30}$')
_BIC_RE = re.compile(r'^[A-Z0-9]{8}([A-Z0-9]{3})?$')
def _sanitise_text(value, max_len=140):
"""Remove characters not allowed in SEPA XML text fields."""
if not value:
return ''
cleaned = re.sub(r"[^A-Za-z0-9 +\-/?.,:;'()\n]", '', value)
return cleaned[:max_len].strip()
def _validate_iban(iban):
"""Validate and normalise an IBAN string."""
if not iban:
raise ValidationError(_("IBAN is required for SEPA Direct Debit."))
normalised = iban.upper().replace(' ', '')
if not _IBAN_RE.match(normalised):
raise ValidationError(_(
"'%(iban)s' does not look like a valid IBAN.",
iban=iban,
))
return normalised
def _validate_bic(bic):
"""Validate and normalise a BIC / SWIFT code."""
if not bic:
raise ValidationError(_(
"BIC/SWIFT code is required for SEPA Direct Debit."
))
normalised = bic.upper().replace(' ', '')
if not _BIC_RE.match(normalised):
raise ValidationError(_(
"'%(bic)s' does not look like a valid BIC/SWIFT code.",
bic=bic,
))
return normalised
# ======================================================================
# Mandate Model
# ======================================================================
class FusionSDDMandate(models.Model):
"""Tracks a SEPA Direct Debit mandate authorisation.
A mandate is the legal agreement through which a debtor (the
customer) authorises the creditor (the company) to collect
payments from their bank account.
"""
_name = 'fusion.sdd.mandate'
_description = 'SEPA Direct Debit Mandate'
_order = 'create_date desc'
_inherit = ['mail.thread', 'mail.activity.mixin']
# ------------------------------------------------------------------
# Fields
# ------------------------------------------------------------------
name = fields.Char(
string='Mandate Reference',
required=True,
copy=False,
readonly=True,
default='/',
tracking=True,
help="Unique Mandate Reference (UMR) communicated to the "
"debtor's bank.",
)
partner_id = fields.Many2one(
comodel_name='res.partner',
string='Debtor',
required=True,
tracking=True,
help="The partner (customer) who signed this mandate.",
)
iban = fields.Char(
string='Debtor IBAN',
required=True,
size=34,
tracking=True,
help="IBAN of the debtor's bank account.",
)
bic = fields.Char(
string='Debtor BIC',
size=11,
tracking=True,
help="BIC of the debtor's bank.",
)
mandate_ref = fields.Char(
string='Unique Mandate Reference',
required=True,
copy=False,
tracking=True,
help="The unique reference identifying this mandate agreement.",
)
state = fields.Selection(
selection=[
('draft', 'Draft'),
('active', 'Active'),
('revoked', 'Revoked'),
('closed', 'Closed'),
],
string='Status',
default='draft',
required=True,
tracking=True,
help="Draft: mandate is being prepared.\n"
"Active: mandate is signed and ready for collections.\n"
"Revoked: debtor has withdrawn consent.\n"
"Closed: mandate is no longer in use.",
)
scheme = fields.Selection(
selection=[
('CORE', 'CORE'),
('B2B', 'B2B'),
],
string='Scheme',
default='CORE',
required=True,
tracking=True,
help="CORE: consumer direct debit.\n"
"B2B: business-to-business direct debit (no refund right).",
)
date_signed = fields.Date(
string='Date Signed',
default=fields.Date.context_today,
help="Date when the mandate was signed by the debtor.",
)
company_id = fields.Many2one(
comodel_name='res.company',
string='Company',
default=lambda self: self.env.company,
required=True,
)
# ------------------------------------------------------------------
# CRUD
# ------------------------------------------------------------------
@api.model_create_multi
def create(self, vals_list):
"""Assign a sequence-based name if not provided."""
for vals in vals_list:
if vals.get('name', '/') == '/':
vals['name'] = self.env['ir.sequence'].next_by_code(
'fusion.sdd.mandate'
) or _('New')
if not vals.get('mandate_ref'):
vals['mandate_ref'] = f'FUSIONMDT-{uuid.uuid4().hex[:8].upper()}'
return super().create(vals_list)
# ------------------------------------------------------------------
# Actions
# ------------------------------------------------------------------
def action_activate(self):
"""Activate a draft mandate."""
for mandate in self:
if mandate.state != 'draft':
raise UserError(_(
"Only draft mandates can be activated."
))
mandate.state = 'active'
def action_revoke(self):
"""Revoke an active mandate."""
for mandate in self:
if mandate.state != 'active':
raise UserError(_(
"Only active mandates can be revoked."
))
mandate.state = 'revoked'
def action_close(self):
"""Close a mandate (no more collections)."""
for mandate in self:
if mandate.state not in ('active', 'revoked'):
raise UserError(_(
"Only active or revoked mandates can be closed."
))
mandate.state = 'closed'
# ======================================================================
# SEPA Direct Debit XML generation
# ======================================================================
class FusionSEPADirectDebit(models.Model):
"""Extends ``res.company`` with SEPA Direct Debit XML generation.
The ``generate_pain_008`` method accepts a set of
``fusion.sdd.mandate`` records (each linked to an active mandate)
and produces a compliant pain.008.001.02 XML file.
"""
_inherit = 'res.company'
fusion_sdd_creditor_identifier = fields.Char(
string='SEPA Creditor Identifier',
size=35,
help="Your organisation's SEPA Creditor Identifier (CI), "
"assigned by your national authority.",
)
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def generate_pain_008(self, mandates, requested_date=None):
"""Build an ISO 20022 pain.008.001.02 XML for SEPA Direct Debits.
:param mandates: ``fusion.sdd.mandate`` recordset.
:param requested_date: Requested collection date (defaults to
today).
:return: UTF-8 encoded XML bytes.
:rtype: bytes
:raises UserError: when mandatory data is missing.
"""
self.ensure_one()
if not mandates:
raise UserError(_(
"No mandates provided for the SEPA Direct Debit file."
))
inactive = mandates.filtered(lambda m: m.state != 'active')
if inactive:
raise UserError(_(
"The following mandates are not active and cannot be "
"included:\n%(mandates)s",
mandates=', '.join(inactive.mapped('name')),
))
company_iban = _validate_iban(self.fusion_company_iban)
company_bic = _validate_bic(self.fusion_company_bic)
if not self.fusion_sdd_creditor_identifier:
raise UserError(_(
"Please configure the SEPA Creditor Identifier on "
"the company settings before generating a Direct "
"Debit file."
))
collection_date = requested_date or fields.Date.context_today(self)
# Root element
root = ET.Element('Document', xmlns=PAIN_008_NS)
cstmr_dd = ET.SubElement(root, 'CstmrDrctDbtInitn')
# -- Group Header --
grp_hdr = ET.SubElement(cstmr_dd, 'GrpHdr')
msg_id = self._pain008_message_id()
ET.SubElement(grp_hdr, 'MsgId').text = msg_id
ET.SubElement(grp_hdr, 'CreDtTm').text = datetime.utcnow().strftime(
'%Y-%m-%dT%H:%M:%S'
)
ET.SubElement(grp_hdr, 'NbOfTxs').text = str(len(mandates))
initg_pty = ET.SubElement(grp_hdr, 'InitgPty')
ET.SubElement(initg_pty, 'Nm').text = _sanitise_text(
self.name, max_len=70
)
# Group mandates by scheme (CORE / B2B)
for scheme in ('CORE', 'B2B'):
scheme_mandates = mandates.filtered(
lambda m: m.scheme == scheme
)
if not scheme_mandates:
continue
self._add_pain008_payment_info(
cstmr_dd, scheme_mandates, scheme,
company_iban, company_bic,
collection_date, msg_id,
)
tree = ET.ElementTree(root)
ET.indent(tree, space=' ')
xml_bytes = ET.tostring(
root, encoding='UTF-8', xml_declaration=True
)
return xml_bytes
# ------------------------------------------------------------------
# Private helpers
# ------------------------------------------------------------------
def _pain008_message_id(self):
"""Generate a unique message identifier for the pain.008 file."""
raw = f'{self.id}-SDD-{datetime.utcnow().isoformat()}'
digest = hashlib.sha256(raw.encode()).hexdigest()[:16]
return f'FUSION-SDD-{digest}'.upper()[:35]
def _add_pain008_payment_info(
self, parent, mandates, scheme,
company_iban, company_bic,
collection_date, msg_id,
):
"""Add a ``PmtInf`` block for a group of mandates sharing the
same scheme.
:param parent: ``CstmrDrctDbtInitn`` XML element.
:param mandates: ``fusion.sdd.mandate`` recordset for this scheme.
:param str scheme: ``'CORE'`` or ``'B2B'``.
:param str company_iban: Validated company IBAN.
:param str company_bic: Validated company BIC.
:param collection_date: Requested collection date.
:param str msg_id: Message identifier.
"""
pmt_inf = ET.SubElement(parent, 'PmtInf')
pmt_inf_id = f'{msg_id}-{scheme}'[:35]
ET.SubElement(pmt_inf, 'PmtInfId').text = pmt_inf_id
ET.SubElement(pmt_inf, 'PmtMtd').text = 'DD' # Direct Debit
ET.SubElement(pmt_inf, 'NbOfTxs').text = str(len(mandates))
# Payment Type Information
pmt_tp_inf = ET.SubElement(pmt_inf, 'PmtTpInf')
svc_lvl = ET.SubElement(pmt_tp_inf, 'SvcLvl')
ET.SubElement(svc_lvl, 'Cd').text = 'SEPA'
lcl_instrm = ET.SubElement(pmt_tp_inf, 'LclInstrm')
ET.SubElement(lcl_instrm, 'Cd').text = scheme
ET.SubElement(pmt_tp_inf, 'SeqTp').text = 'RCUR' # Recurring
ET.SubElement(pmt_inf, 'ReqdColltnDt').text = (
collection_date.isoformat()
if isinstance(collection_date, date)
else str(collection_date)
)
# Creditor (company)
cdtr = ET.SubElement(pmt_inf, 'Cdtr')
ET.SubElement(cdtr, 'Nm').text = _sanitise_text(self.name, 70)
cdtr_acct = ET.SubElement(pmt_inf, 'CdtrAcct')
cdtr_acct_id = ET.SubElement(cdtr_acct, 'Id')
ET.SubElement(cdtr_acct_id, 'IBAN').text = company_iban
cdtr_agt = ET.SubElement(pmt_inf, 'CdtrAgt')
cdtr_agt_fin = ET.SubElement(cdtr_agt, 'FinInstnId')
ET.SubElement(cdtr_agt_fin, 'BIC').text = company_bic
ET.SubElement(pmt_inf, 'ChrgBr').text = 'SLEV'
# Creditor Scheme Identification
cdtr_schme_id = ET.SubElement(pmt_inf, 'CdtrSchmeId')
cdtr_schme_id_el = ET.SubElement(cdtr_schme_id, 'Id')
prvt_id = ET.SubElement(cdtr_schme_id_el, 'PrvtId')
othr = ET.SubElement(prvt_id, 'Othr')
ET.SubElement(othr, 'Id').text = _sanitise_text(
self.fusion_sdd_creditor_identifier, 35
)
schme_nm = ET.SubElement(othr, 'SchmeNm')
ET.SubElement(schme_nm, 'Prtry').text = 'SEPA'
# Individual transactions
for mandate in mandates:
self._add_pain008_transaction(pmt_inf, mandate)
def _add_pain008_transaction(self, pmt_inf, mandate):
"""Append a ``DrctDbtTxInf`` element for a single mandate.
:param pmt_inf: The ``PmtInf`` XML element.
:param mandate: A ``fusion.sdd.mandate`` record.
"""
debtor_iban = _validate_iban(mandate.iban)
dd_tx = ET.SubElement(pmt_inf, 'DrctDbtTxInf')
# Payment Identification
pmt_id = ET.SubElement(dd_tx, 'PmtId')
ET.SubElement(pmt_id, 'EndToEndId').text = _sanitise_text(
mandate.mandate_ref or mandate.name, 35
)
# Amount - use a default amount; callers should set this via
# the payment amount linked to the mandate
amt = ET.SubElement(dd_tx, 'InstdAmt')
# Default to 0.00; the caller is expected to provide real
# amounts via a companion payment record. This serves as the
# structural placeholder per the schema.
amt.text = '0.00'
amt.set('Ccy', 'EUR')
# Direct Debit Transaction Info (mandate details)
dd_tx_inf = ET.SubElement(dd_tx, 'DrctDbtTx')
mndt_rltd_inf = ET.SubElement(dd_tx_inf, 'MndtRltdInf')
ET.SubElement(mndt_rltd_inf, 'MndtId').text = _sanitise_text(
mandate.mandate_ref, 35
)
ET.SubElement(mndt_rltd_inf, 'DtOfSgntr').text = (
mandate.date_signed.isoformat()
if mandate.date_signed
else fields.Date.context_today(self).isoformat()
)
# Debtor Agent (BIC)
if mandate.bic:
dbtr_agt = ET.SubElement(dd_tx, 'DbtrAgt')
dbtr_agt_fin = ET.SubElement(dbtr_agt, 'FinInstnId')
ET.SubElement(dbtr_agt_fin, 'BIC').text = (
mandate.bic.upper().replace(' ', '')
)
# Debtor
dbtr = ET.SubElement(dd_tx, 'Dbtr')
ET.SubElement(dbtr, 'Nm').text = _sanitise_text(
mandate.partner_id.name or '', 70
)
# Debtor Account
dbtr_acct = ET.SubElement(dd_tx, 'DbtrAcct')
dbtr_acct_id = ET.SubElement(dbtr_acct, 'Id')
ET.SubElement(dbtr_acct_id, 'IBAN').text = debtor_iban