tryton/modules/edocument_peppol_peppyrus/edocument.py

# This file is part of Tryton.  The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.

import uuid
from base64 import b64decode, b64encode
from functools import wraps
from io import BytesIO
from urllib.parse import quote, urljoin

import requests
from lxml import etree

from trytond.i18n import gettext
from trytond.model import fields
from trytond.modules.edocument_peppol.exceptions import PeppolServiceError
from trytond.pool import Pool, PoolMeta
from trytond.protocols.wrappers import HTTPStatus
from trytond.pyson import Eval
from trytond.transaction import Transaction
from trytond.url import http_host

from .exceptions import PeppyrusCredentialWarning, PeppyrusError

URLS = {
    'testing': 'https://api.test.peppyrus.be/v1/',
    'production': 'https://api.peppyrus.be/v1/',
    }

UBL_NAMESPACES = {
    'urn:oasis:names:specification:ubl:schema:xsd:Invoice-2',
    'urn:oasis:names:specification:ubl:schema:xsd:CreditNote-2',
    }
BIS_BILLING_3 = (
    'urn:cen.eu:en16931:2017#compliant'
    '#urn:fdc:peppol.eu:2017:poacc:billing:3.0')


def peppyrus_api(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except requests.HTTPError as e:
            error_message = e.response.text or e.args[0]
            raise PeppyrusError(
                gettext('edocument_peppol_peppyrus'
                    '.msg_peppyrus_webserver_error',
                    message=error_message)) from e
    return wrapper


class PeppolService(metaclass=PoolMeta):
    __name__ = 'edocument.peppol.service'

    peppyrus_api_key = fields.Char(
        "API Key",
        states={
            'invisible': Eval('service') != 'peppyrus',
            'required': Eval('service') == 'peppyrus',
            })
    peppyrus_server = fields.Selection([
            (None, ""),
            ('testing', "Testing"),
            ('production', "Production"),
            ], "Server",
        states={
            'invisible': Eval('service') != 'peppyrus',
            'required': Eval('service') == 'peppyrus',
            })
    peppyrus_identifier = fields.Char(
        "Identifier", readonly=True,
        states={
            'invisible': Eval('service') != 'peppyrus',
            })
    peppyrus_incoming_webhook = fields.Function(
        fields.Char(
            "Incoming webhook",
            help="The URL for the incoming documents webhook."),
        'on_change_with_peppyrus_incoming_webhook')
    peppyrus_outgoing_webhook = fields.Function(
        fields.Char(
            "Outgoing webhook",
            help="The URL for the outgoing documents webhook."),
        'on_change_with_peppyrus_outgoing_webhook')

    @classmethod
    def __setup__(cls):
        super().__setup__()
        cls.service.selection.append(('peppyrus', "Peppyrus"))
        cls._buttons.update(
            peppyrus_new_identifier={
                'invisible': Eval('service') != 'peppyrus',
                'icon': 'tryton-refresh',
                },
            )

    @fields.depends('service', 'peppyrus_server')
    def on_change_service(self):
        if (self.service == 'peppyrus'
                and not self.peppyrus_server):
            self.peppyrus_server = 'testing'

    @fields.depends('peppyrus_identifier')
    def on_change_with_peppyrus_incoming_webhook(self, name=None):
        if self.peppyrus_identifier:
            url_part = {
                'identifier': self.peppyrus_identifier,
                'database_name': Transaction().database.name,
                }
            return http_host() + (
                quote(
                    '/%(database_name)s/edocument_peppol_peppyrus/'
                    '%(identifier)s/in'
                    % url_part))

    @fields.depends('peppyrus_identifier')
    def on_change_with_peppyrus_outgoing_webhook(self, name=None):
        if self.peppyrus_identifier:
            url_part = {
                'identifier': self.peppyrus_identifier,
                'database_name': Transaction().database.name,
                }
            return http_host() + (
                quote(
                    '/%(database_name)s/edocument_peppol_peppyrus/'
                    '%(identifier)s/out'
                    % url_part))

    @classmethod
    def peppyrus_new_identifier(cls, services):
        for service in services:
            if service.peppyrus_identifier:
                service.peppyrus_identifier = None
            else:
                service.peppyrus_identifier = uuid.uuid4().hex
        cls.save(services)

    @peppyrus_api
    def _post_peppyrus(self, document):
        tree = etree.parse(BytesIO(document.data))
        if not (sender := self._peppyrus_sender(document.type, tree)):
            raise PeppolServiceError("Missing sender")
        if not (recipient := self._peppyrus_recipient(document.type, tree)):
            raise PeppolServiceError("Missing recipient")
        response = requests.post(
            urljoin(URLS[self.peppyrus_server], 'message'),
            json={
                'sender': sender,
                'recipient': recipient,
                'processType': (
                    self._peppyrus_process_type(document.type, tree)),
                'documentType': (
                    self._peppyrus_document_type(document.type, tree)),
                'fileContent': b64encode(document.data).decode(),
                },
            headers={
                'Accept': 'application/json',
                'X-Api-Key': self.peppyrus_api_key,
                })
        if response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY:
            raise PeppolServiceError(response.json())
        response.raise_for_status()
        return response.json()['id']

    def _peppyrus_sender(self, type, tree):
        if type == 'bis-billing-3':
            el = tree.find(
                './/{*}AccountingSupplierParty/{*}Party/{*}EndpointID')
            if el is not None:
                scheme = el.get('schemeID', '')
                return f'{scheme}:{el.text}'

    def _peppyrus_recipient(self, type, tree):
        if type == 'bis-billing-3':
            el = tree.find(
                './/{*}AccountingCustomerParty/{*}Party/{*}EndpointID')
            if el is not None:
                scheme = el.get('schemeID', '')
                return f'{scheme}:{el.text}'

    def _peppyrus_process_type(self, type, tree):
        if type == 'bis-billing-3':
            return 'cenbii-procid-ubl::' + tree.findtext('.//{*}ProfileID')

    def _peppyrus_document_type(self, type, tree):
        root = tree.getroot()
        namespace = root.nsmap.get(root.prefix)
        if type == 'bis-billing-3':
            return (
                'busdox-docid-qns::'
                f'{namespace}::{etree.QName(root).localname}##'
                + tree.findtext('.//{*}CustomizationID') + '::2.1')

    @peppyrus_api
    def _update_status_peppyrus(self, document):
        assert document.direction == 'out'
        if not document.transmission_id:
            return
        response = requests.get(
            urljoin(
                URLS[self.peppyrus_server],
                f'message/{document.transmission_id}'),
            headers={
                'Accept': 'application/json',
                'X-Api-Key': self.peppyrus_api_key,
                })
        response.raise_for_status()
        message = response.json()
        self.peppyrus_update(document, message)

    def peppyrus_update(self, document, message):
        assert message['id'] == document.transmission_id
        if message['folder'] == 'sent':
            document.succeed()
        elif message['folder'] == 'failed':
            document.fail(status=self._peppyrus_status(document))

    @peppyrus_api
    def _peppyrus_status(self, document):
        response = requests.get(
            urljoin(
                URLS[self.peppyrus_server],
                f'message/{document.transmission_id}/report'),
            headers={
                'Accept': 'application/json',
                'X-Api-Key': self.peppyrus_api_key,
                })
        response.raise_for_status()
        report = response.json()
        return report.get('transmissionRules')

    @classmethod
    def peppyrus_fetch(cls, services=None):
        if services is None:
            services = cls.search([('service', '=', 'peppyrus')])
        for service in services:
            service._peppyrus_fetch()

    @peppyrus_api
    def _peppyrus_fetch(self):
        assert self.service == 'peppyrus'
        response = requests.get(
            urljoin(URLS[self.peppyrus_server], 'message/list'),
            params={
                'folder': 'INBOX',
                'confirmed': 'false',
                'perPage': 100,
                },
            headers={
                'Accept': 'application/json',
                'X-Api-Key': self.peppyrus_api_key,
                })
        response.raise_for_status()
        for message in response.json()['items']:
            self.__class__.__queue__.peppyrus_store(self, message)

    def peppyrus_store(self, message):
        pool = Pool()
        Document = pool.get('edocument.peppol')
        transaction = Transaction()
        assert message['direction'] == 'IN'

        def confirm(server, api_key, message, silent=False):
            try:
                response = requests.patch(
                    urljoin(
                        URLS[server],
                        f"message/{message['id']}/confirm"),
                    headers={
                        'Accept': 'application/json',
                        'X-Api-Key': api_key,
                        })
                if response.status_code != HTTPStatus.NOT_FOUND:
                    response.raise_for_status()
            except Exception:
                if not silent:
                    raise

        server = self.peppyrus_server
        api_key = self.peppyrus_api_key
        if Document.search([
                    ('service', '=', self.id),
                    ('transmission_id', '=', message['id']),
                    ]):
            confirm(server, api_key, message)
            return
        document = Document(
            direction='in',
            company=self.company,
            type=self._peppyrus_type(message['documentType']),
            service=self,
            data=b64decode(message['fileContent']),
            transmission_id=message['id'],
            )
        document.save()
        document.submit()
        transaction.atexit(confirm, server, api_key, message, True)

    @classmethod
    def _peppyrus_type(cls, document_type):
        schema, document_identifier = document_type.split('::', 1)
        syntax, customization = document_identifier.split('##', 1)
        namespace, localname = syntax.split('::', 1)
        customization_id, version = customization.split('::', 1)
        if (schema in {'busdox-docid-qns', 'peppol-doctype-wildcard'}
                and namespace in UBL_NAMESPACES
                and customization_id in BIS_BILLING_3):
            return 'bis-billing-3'

    @classmethod
    def check_modification(cls, mode, services, values=None, external=False):
        pool = Pool()
        Warning = pool.get('res.user.warning')

        super().check_modification(
            mode, services, values=values, external=external)

        if mode == 'write' and external and 'peppyrus_api_key' in values:
            warning_name = Warning.format('peppyrus_credential', services)
            if Warning.check(warning_name):
                raise PeppyrusCredentialWarning(
                    warning_name,
                    gettext('edocument_peppol_peppyrus'
                        '.msg_peppyrus_credential_modified'))