[WIP] add l10n_it_fatturapa_govway

l10n_it_fatturapa_govway/__init__.py

@@ -0,0 +1,2 @@
+from . import controllers
+from . import models

l10n_it_fatturapa_govway/__manifest__.py

@@ -0,0 +1,28 @@
+# Copyright 2024 Sergio Corato <https://github.com/sergiocorato>
+# Copyright 2024 Marco Colombo <https://github.com/TheMule71>
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
+{
+    "name": "ITA - Fattura elettronica - Supporto GovWay",
+    "version": "14.0.1.0.0",
+    "category": "Localization/Italy",
+    "summary": "Invio fatture elettroniche tramite GovWay",
+    "author": "Odoo Community Association (OCA)",
+    "website": "https://github.com/OCA/l10n-italy",
+    "license": "AGPL-3",
+    "depends": [
+        "l10n_it_fatturapa_out",
+        "l10n_it_fatturapa_in",
+        "l10n_it_sdi_channel",
+        "web",
+    ],
+    "data": [
+        # "security/ir.model.access.csv",
+        # "views/fetchmail_view.xml",
+        "views/sdi_view.xml",
+        # "views/sdi_view.xml",
+        # "views/ir_mail_server.xml",
+        # "data/fetchmail_data.xml",
+        # "data/sdi_channel_demo.xml",
+    ],
+    "installable": True,
+}

l10n_it_fatturapa_govway/controllers/__init__.py

@@ -0,0 +1 @@
+from . import main

l10n_it_fatturapa_govway/controllers/main.py

@@ -0,0 +1,90 @@
+import logging
+
+from odoo.http import Controller, request, route
+
+_logger = logging.getLogger()
+
+
+class FatturaPAGovWay(Controller):
+    # incoming invoices
+    @route(["/fatturapa/govway/ricevi_fattura"], type="http", auth="user", website=True)
+    def ricevi_fattura(self, *args, **post):
+        # headers
+        # - GovWay-SDI-FormatoArchivioBase64
+        # - GovWay-SDI-FormatoArchivioInvioFattura
+        # - GovWay-SDI-FormatoFatturaPA
+        # - GovWay-SDI-IdentificativoSdI
+        # - GovWay-SDI-MessageId
+        # - GovWay-SDI-NomeFile
+        # - GovWay-SDI-NomeFileMetadati
+        # - GovWay-Transaction-ID
+        identificativo_sdi = request.httprequest.headers.get(
+            "GovWay-SDI-IdentificativoSdI", ""
+        )
+        sdi_nomefile = request.httprequest.headers.get("GovWay-SDI-NomeFile", "")
+        transaction_id = request.httprequest.headers.get("GovWay-Transaction-ID", "")
+
+        sdi_formatoarchiviobase64 = request.httprequest.headers.get(
+            "GovWay-SDI-FormatoArchivioBase64", ""
+        )
+        sdi_formatoarchiviiinviofattura = request.httprequest.headers.get(
+            "GovWay-SDI-FormatoArchivioInvioFattura", ""
+        )
+        sdi_formartofatturapa = request.httprequest.headers.get(
+            "GovWay-SDI-FormatoFatturaPA", ""
+        )
+        sdi_messageid = request.httprequest.headers.get("GovWay-SDI-MessageId", "")
+        sdi_nomefile_metadati = request.httprequest.headers.get(
+            "GovWay-SDI-NomeFileMetadati", ""
+        )
+
+        _logger.info(
+            f"ricevi_fattura(): {identificativo_sdi} {sdi_nomefile} {transaction_id}"
+        )
+        _logger.debug(f"ricevi_fattura(): args={repr(args)}")
+        _logger.debug(f"ricevi_fattura(): post={repr(post)}")
+
+    @route(["/fatturapa/govway/ricevi_ndt"], type="http", auth="user", website=True)
+    def ricevi_ndt(self, *args, **post):
+        # headers
+        # - GovWay-SDI-IdentificativoSdI
+        # - GovWay-SDI-NomeFile
+        # - GovWay-Transaction-ID
+        identificativo_sdi = request.httprequest.headers.get(
+            "GovWay-SDI-IdentificativoSdI", ""
+        )
+        sdi_nomefile = request.httprequest.headers.get("GovWay-SDI-NomeFile", "")
+        transaction_id = request.httprequest.headers.get("GovWay-Transaction-ID", "")
+
+        _logger.info(
+            f"ricevi_ndt(): {identificativo_sdi} {sdi_nomefile} {transaction_id}"
+        )
+        _logger.debug(f"ricevi_ndt(): args={repr(args)}")
+        _logger.debug(f"ricevi_ndt(): post={repr(post)}")
+
+    # outgoing invoices
+    @route(
+        ["/fatturapa/govway/ricevi_notifica"],
+        type="http",
+        auth="user",
+        methods=["POST"],
+        website=True,
+    )
+    def ricevi_notifica(self, *args, **post):
+        # headers:
+        # - GovWay-SDI-IdentificativoSdI
+        # - GovWay-SDI-NomeFile
+        # - GovWay-Transaction-ID
+        identificativo_sdi = request.httprequest.headers.get(
+            "GovWay-SDI-IdentificativoSdI", ""
+        )
+        sdi_nomefile = request.httprequest.headers.get("GovWay-SDI-NomeFile", "")
+        transaction_id = request.httprequest.headers.get("GovWay-Transaction-ID", "")
+
+        _logger.info(
+            f"ricevi_notifica(): {identificativo_sdi} {sdi_nomefile} {transaction_id}"
+        )
+        _logger.debug(f"ricevi_notifica(): args={repr(args)}")
+        _logger.debug(f"ricevi_notifica(): post={repr(post)}")
+        # request.env["sdi.channel"].sdi_channel_model.receive_notification(
+        # { sdi_nomefile: post })

l10n_it_fatturapa_govway/data/sdi_channel_demo.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<odoo>
+    <record id="sdi_channel_govway" model="sdi.channel">
+        <field name="name">GovWay</field>
+        <field name="channel_type">govway</field>
+    </record>
+</odoo>

l10n_it_fatturapa_govway/models/__init__.py

@@ -0,0 +1,5 @@
+# from . import fatturapa_attachment_out
+# from . import mail_thread
+# from . import ir_mail_server
+# from . import fetchmail
+from . import sdi

l10n_it_fatturapa_govway/models/fatturapa_attachment_out.py

@@ -0,0 +1,147 @@
+# Author(s): Andrea Colangelo ([email protected])
+# Copyright 2018 Openforce Srls Unipersonale (www.openforce.it)
+# Copyright 2018 Sergio Corato (https://efatto.it)
+# Copyright 2018-2019 Lorenzo Battistini <https://github.com/eLBati>
+
+import logging
+import re
+
+from odoo import _, fields, models
+from odoo.exceptions import UserError
+
+_logger = logging.getLogger(__name__)
+
+RESPONSE_MAIL_REGEX = (
+    "[A-Z]{2}[a-zA-Z0-9]{11,16}_[a-zA-Z0-9]{,5}_[A-Z]{2}_" "[a-zA-Z0-9]{,3}"
+)
+
+
+class FatturaPAAttachmentOut(models.Model):
+    _inherit = "fatturapa.attachment.out"
+
+    def _message_type_ns(
+        self, root, id_sdi, message_id, receipt_dt, fatturapa_attachment_out
+    ):
+        error_list = root.find("ListaErrori")
+        error_str = ""
+        for error in error_list:
+            error_str += "\n[%s] %s %s" % (
+                error.find("Codice").text if error.find("Codice") is not None else "",
+                error.find("Descrizione").text
+                if error.find("Descrizione") is not None
+                else "",
+                error.find("Suggerimento").text
+                if error.find("Suggerimento") is not None
+                else "",
+            )
+        fatturapa_attachment_out.write(
+            {
+                "state": "sender_error",
+                "last_sdi_response": f"SdI ID: {id_sdi}; "
+                f"Message ID: {message_id}; Receipt date: {receipt_dt}; "
+                f"Error: {error_str}",
+            }
+        )
+
+    def _message_type_mc(
+        self, root, id_sdi, message_id, receipt_dt, fatturapa_attachment_out
+    ):
+        missed_delivery_note = root.find("Descrizione").text
+        fatturapa_attachment_out.write(
+            {
+                "state": "recipient_error",
+                "last_sdi_response": f"SdI ID: {id_sdi}; "
+                f"Message ID: {message_id}; Receipt date: {receipt_dt}; "
+                f"Missed delivery note: {missed_delivery_note}",
+            }
+        )
+
+    def _message_type_rc(
+        self, root, id_sdi, message_id, receipt_dt, fatturapa_attachment_out
+    ):
+        delivery_dt = root.find("DataOraConsegna").text
+        fatturapa_attachment_out.write(
+            {
+                "state": "validated",
+                "delivered_date": fields.Datetime.now(),
+                "last_sdi_response": f"SdI ID: {id_sdi}; "
+                f"Message ID: {message_id}; Receipt date: {receipt_dt}; "
+                f"Delivery date: {delivery_dt}",
+            }
+        )
+
+    def _message_type_ne(self, root, id_sdi, message_id, fatturapa_attachment_out):
+        esito_committente = root.find("EsitoCommittente")
+        if esito_committente is not None:
+            # more than one esito?
+            esito = esito_committente.find("Esito")
+            state = ""
+            if esito is not None:
+                if esito.text == "EC01":
+                    state = "accepted"
+                elif esito.text == "EC02":
+                    state = "rejected"
+                fatturapa_attachment_out.write(
+                    {
+                        "state": state,
+                        "last_sdi_response": f"SdI ID: {id_sdi}; "
+                        f"Message ID: {message_id}; Response: {esito.text}; ",
+                    }
+                )
+
+    def _message_type_dt(
+        self, root, id_sdi, message_id, receipt_dt, fatturapa_attachment_out
+    ):
+        description = root.find("Descrizione")
+        if description is not None:
+            fatturapa_attachment_out.write(
+                {
+                    "state": "validated",
+                    "last_sdi_response": f"SdI ID: {id_sdi}; "
+                    f"Message ID: {message_id}; Receipt date: {receipt_dt}; "
+                    f"Description: {description.text}",
+                }
+            )
+
+    def _message_type_at(
+        self, root, id_sdi, message_id, receipt_dt, fatturapa_attachment_out
+    ):
+        description = root.find("Descrizione")
+        if description is not None:
+            fatturapa_attachment_out.write(
+                {
+                    "state": "accepted",
+                    "last_sdi_response": (
+                        "SdI ID: {}; Message ID: {}; Receipt date: {};"
+                        " Description: {}"
+                    ).format(id_sdi, message_id, receipt_dt, description.text),
+                }
+            )
+
+    def parse_pec_response(self, message_dict):
+        message_dict["model"] = self._name
+        message_dict["res_id"] = 0
+
+        regex = re.compile(RESPONSE_MAIL_REGEX)
+        notifications = [x for x in message_dict["attachments"] if regex.match(x.fname)]
+
+        if not notifications:
+            raise UserError(
+                _(
+                    'PEC message "%s" is coming from SDI but attachments do not '
+                    "match SDI response format. Please check."
+                )
+                % (message_dict["subject"])
+            )
+
+        sdi_channel_model = self.env["sdi.channel"]
+        attachments = sdi_channel_model.receive_notification(
+            {
+                notification.fname: notification.content
+                for notification in notifications
+            },
+        )
+
+        # Link the message to the last attachment updated
+        message_dict["res_id"] = attachments[-1].id
+        return message_dict