account-financial-tools/account_invoice_constraint_chronology/model/account_move.py

# Copyright 2015-2019 ACSONE SA/NV (<http://acsone.eu>)
# Copyright 2021 CorporateHub (https://corporatehub.eu)
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).

from odoo import _, models
from odoo.exceptions import UserError
from odoo.osv import expression
from odoo.tools.misc import format_date


class AccountMove(models.Model):
    _inherit = "account.move"

    def _get_conflicting_invoices_domain(self):
        self.ensure_one()
        domain = [
            ("journal_id", "=", self.journal_id.id),
            ("move_type", "!=", "entry"),
        ]
        if self.journal_id.refund_sequence:
            domain.append(("move_type", "=", self.move_type))
        return domain

    def _get_older_conflicting_invoices_domain(self):
        self.ensure_one()
        return expression.AND(
            [
                self._get_conflicting_invoices_domain(),
                [
                    ("state", "=", "draft"),
                    ("invoice_date", "!=", False),
                    ("invoice_date", "<", self.invoice_date),
                ],
            ]
        )

    def _raise_older_conflicting_invoices(self):
        self.ensure_one()
        raise UserError(
            _(
                "Chronology conflict: A conflicting draft invoice dated before "
                "{date_invoice} exists, please validate it first."
            ).format(date_invoice=format_date(self.env, self.invoice_date))
        )

    def _get_newer_conflicting_invoices_domain(self):
        self.ensure_one()
        return expression.AND(
            [
                self._get_conflicting_invoices_domain(),
                [("state", "=", "posted"), ("invoice_date", ">", self.invoice_date)],
            ]
        )

    def _raise_newer_conflicting_invoices(self):
        self.ensure_one()
        raise UserError(
            _(
                "Chronology conflict: A conflicting validated invoice dated after "
                "{date_invoice} exists."
            ).format(date_invoice=format_date(self.env, self.invoice_date))
        )

    def _get_sequence_order_conflicting_invoices_domain(self):
        self.ensure_one()

        if not self.name or self.name == "/":
            return expression.FALSE_DOMAIN

        last_sequence = self._get_last_sequence()
        if not last_sequence or self.name > last_sequence:
            return expression.FALSE_DOMAIN

        return expression.AND(
            [
                [("name", "=", last_sequence)],
                self._get_conflicting_invoices_domain(),
                [("state", "=", "posted"), ("invoice_date", "<", self.invoice_date)],
            ]
        )

    def _raise_sequence_ordering_conflict(self):
        self.ensure_one()
        raise UserError(
            _(
                "Chronology conflict: An invoice with a higher number {highest_name}"
                " dated before {date_invoice} exists."
            ).format(
                highest_name=self._get_last_sequence(),
                date_invoice=format_date(self.env, self.invoice_date),
            )
        )

    def _conflicting_inv_after_sequence_before_inv_date_domain(self):
        return expression.AND(
            [
                (
                    ("name", ">", self.name),
                    ("invoice_date", "<", self.invoice_date),
                )
            ]
        )

    def _conflicting_inv_before_sequence_after_inv_date_domain(self):
        return expression.AND(
            [
                (
                    ("name", "<", self.name),
                    ("invoice_date", ">", self.invoice_date),
                )
            ]
        )

    def _get_sequence_order_conflicting_previously_validated(self):
        self.ensure_one()
        return expression.AND(
            [
                self._get_conflicting_invoices_domain(),
                expression.OR(
                    [
                        self._conflicting_inv_after_sequence_before_inv_date_domain(),
                        self._conflicting_inv_before_sequence_after_inv_date_domain(),
                    ]
                ),
            ]
        )

    def _raise_sequence_order_conflicting_previously_validated(self):
        self.ensure_one()
        before_inv = self.search(
            self._conflicting_inv_after_sequence_before_inv_date_domain(), limit=1
        )
        after_inv = self.search(
            self._conflicting_inv_before_sequence_after_inv_date_domain(), limit=1
        )
        if after_inv:
            time = "before"
        else:
            time = "after"
        raise UserError(
            _(
                "Chronology conflict: Invoice {name} cannot be {time} "
                "invoice {inv_name}."
            ).format(
                name=self.name,
                time=time,
                inv_name=after_inv.name if after_inv else before_inv.name,
                date_invoice=format_date(self.env, self.invoice_date),
            )
        )

    def write(self, vals):
        if vals.get("state") != "posted":
            return super().write(vals)
        previously_validated = self.filtered(lambda m: m.name and m.name != "/")
        newly_posted = self.filtered(lambda move: move.state != "posted")
        res = super().write(vals)
        for move in newly_posted & self.filtered("journal_id.check_chronology"):
            if self.search(
                move._get_sequence_order_conflicting_invoices_domain(), limit=1
            ):
                move._raise_sequence_ordering_conflict()
            if self.search(move._get_older_conflicting_invoices_domain(), limit=1):
                move._raise_older_conflicting_invoices()
            if move in previously_validated:
                if self.search(
                    move._get_sequence_order_conflicting_previously_validated(), limit=1
                ):
                    move._raise_sequence_order_conflicting_previously_validated()
                continue
            if self.search(move._get_newer_conflicting_invoices_domain(), limit=1):
                move._raise_newer_conflicting_invoices()

        return res