account-closing/account_cutoff_accrual_picking/models/account_cutoff.py

# Copyright 2013-2021 Akretion France (http://www.akretion.com/)
# @author: Alexis de Lattre <alexis.delattre@akretion.com>
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).

from dateutil.relativedelta import relativedelta

from flectra import _, api, fields, models
from flectra.exceptions import UserError
from flectra.tools import float_compare, float_is_zero


class AccountCutoff(models.Model):
    _inherit = "account.cutoff"

    picking_interval_days = fields.Integer(
        string="Picking Analysis Interval",
        default=lambda self: self._default_picking_interval_days(),
        help="To generate the accruals based on pickings, Flectra will "
        "analyse all the pickings between the cutoff date and N "
        "days before. N is the Picking Analysis Interval.",
    )

    _sql_constraints = [
        (
            "picking_interval_days_positive",
            "CHECK(picking_interval_days > 0)",
            "The value of the field 'Picking Analysis Interval' must "
            "be strictly positive.",
        )
    ]

    @api.model
    def _default_picking_interval_days(self):
        return self.env.company.default_cutoff_accrual_picking_interval_days

    def picking_prepare_cutoff_line(self, vdict, account_mapping):
        dpo = self.env["decimal.precision"]
        assert self.cutoff_type in (
            "accrued_expense",
            "accrued_revenue",
        ), "The field 'cutoff_type' has a wrong value"
        qty_prec = dpo.precision_get("Product Unit of Measure")
        qty = vdict["precut_delivered_qty"] - vdict["precut_invoiced_qty"]
        if float_is_zero(qty, precision_digits=qty_prec):
            return False

        company_currency = self.company_currency_id
        currency = vdict["currency"]
        sign = self.cutoff_type == "accrued_expense" and -1 or 1
        amount = qty * vdict["price_unit"] * sign
        amount_company_currency = vdict["currency"]._convert(
            amount, company_currency, self.company_id, self.cutoff_date
        )

        # Use account mapping
        account_id = vdict["account_id"]
        if account_id in account_mapping:
            accrual_account_id = account_mapping[account_id]
        else:
            accrual_account_id = account_id
        vals = {
            "parent_id": self.id,
            "partner_id": vdict["partner"].id,
            "name": vdict["name"],
            "account_id": account_id,
            "cutoff_account_id": accrual_account_id,
            "analytic_account_id": vdict["analytic_account_id"],
            "currency_id": vdict["currency"].id,
            "quantity": qty,
            "price_unit": vdict["price_unit"],
            "amount": amount,
            "cutoff_amount": amount_company_currency,
            "price_origin": vdict.get("price_origin"),
        }

        if vdict["taxes"] and self.company_id.accrual_taxes:
            # vdict["price_unit"] is a price without tax,
            # so I set handle_price_include=False
            tax_compute_all_res = vdict["taxes"].compute_all(
                vdict["price_unit"],
                currency=currency,
                quantity=qty,
                product=vdict["product"],
                partner=vdict["partner"],
                handle_price_include=False,
            )
            vals["tax_line_ids"] = self._prepare_tax_lines(
                tax_compute_all_res, self.company_currency_id
            )
        return vals

    def order_line_update_oline_dict(self, order_line, order_type, oline_dict):
        assert order_line not in oline_dict
        dpo = self.env["decimal.precision"]
        qty_prec = dpo.precision_get("Product Unit of Measure")
        # These fields have the same name on PO and SO
        order = order_line.order_id
        product = order_line.product_id
        product_uom = product.uom_id
        moves = order_line.move_ids
        ilines = order_line.invoice_lines
        oline_dict[order_line] = {
            "precut_delivered_qty": 0.0,  # in product_uom
            "precut_invoiced_qty": 0.0,  # in product_uom
            "name": _("%s: %s") % (order.name, order_line.name),
            "product": product,
            "partner": order.partner_id.commercial_partner_id,
        }
        if order_type == "purchase":
            invoice_type = "in_invoice"
        elif order_type == "sale":
            invoice_type = "out_invoice"
        for move in moves:
            # TODO: improve comparaison of date and datetime
            # for our friends far away from GMT
            if move.state == "done" and move.date.date() <= self.cutoff_date:
                move_qty = move.product_uom._compute_quantity(
                    move.product_uom_qty, product_uom
                )
                oline_dict[order_line]["precut_delivered_qty"] += move_qty
        price_origin = False
        for iline in ilines:
            invoice = iline.move_id
            if (
                invoice.move_type == invoice_type
                and float_compare(iline.quantity, 0, precision_digits=qty_prec) > 0
            ):
                iline_qty_puom = iline.product_uom_id._compute_quantity(
                    iline.quantity, product_uom
                )
                if invoice.date <= self.cutoff_date:
                    oline_dict[order_line]["precut_invoiced_qty"] += iline_qty_puom
                # Most recent invoice line used for price_unit, account,...
                price_unit = iline.price_subtotal / iline_qty_puom
                price_origin = invoice.name
                currency = invoice.currency_id
                account_id = iline.account_id.id
                analytic_account_id = iline.analytic_account_id.id
                taxes = iline.tax_ids
        if not price_origin:
            if order_type == "purchase":
                oline_qty_puom = order_line.product_uom._compute_quantity(
                    order_line.product_qty, product_uom
                )
                price_unit = order_line.price_subtotal / oline_qty_puom
                price_origin = order.name
                currency = order.currency_id
                analytic_account_id = order_line.account_analytic_id.id
                taxes = order_line.taxes_id
                account = product._get_product_accounts()["expense"]
                if not account:
                    raise UserError(
                        _(
                            "Missing expense account on product '%s' or on its "
                            "related product category '%s'."
                        )
                        % (product.display_name, product.categ_id.display_name)
                    )
                account_id = order.fiscal_position_id.map_account(account).id
            elif order_type == "sale":
                oline_qty_puom = order_line.product_uom._compute_quantity(
                    order_line.product_uom_qty, product_uom
                )
                price_unit = order_line.price_subtotal / oline_qty_puom
                price_origin = order.name
                currency = order.currency_id
                analytic_account_id = order.analytic_account_id.id
                taxes = order_line.tax_id
                account = product._get_product_accounts()["income"]
                if not account:
                    raise UserError(
                        _(
                            "Missing income account on product '%s' or on its "
                            "related product category '%s'."
                        )
                        % (product.display_name, product.categ_id.display_name)
                    )
                account_id = order.fiscal_position_id.map_account(account).id

        oline_dict[order_line].update(
            {
                "price_unit": price_unit,
                "price_origin": price_origin,
                "currency": currency,
                "analytic_account_id": analytic_account_id,
                "account_id": account_id,
                "taxes": taxes,
            }
        )

    def stock_move_update_oline_dict(self, move_line, oline_dict):
        dpo = self.env["decimal.precision"]
        qty_prec = dpo.precision_get("Product Unit of Measure")
        if self.cutoff_type == "accrued_expense":
            if (
                move_line.purchase_line_id
                and move_line.purchase_line_id not in oline_dict
                and not float_is_zero(
                    move_line.purchase_line_id.product_qty, precision_digits=qty_prec
                )
            ):
                self.order_line_update_oline_dict(
                    move_line.purchase_line_id, "purchase", oline_dict
                )
        elif self.cutoff_type == "accrued_revenue":
            if (
                move_line.sale_line_id
                and move_line.sale_line_id not in oline_dict
                and not float_is_zero(
                    move_line.sale_line_id.product_uom_qty, precision_digits=qty_prec
                )
            ):
                self.order_line_update_oline_dict(
                    move_line.sale_line_id, "sale", oline_dict
                )

    def get_lines(self):
        res = super().get_lines()
        spo = self.env["stock.picking"]
        aclo = self.env["account.cutoff.line"]

        pick_type_map = {
            "accrued_revenue": "outgoing",
            "accrued_expense": "incoming",
        }
        cutoff_type = self.cutoff_type
        if cutoff_type not in pick_type_map:
            return res

        # Create account mapping dict
        account_mapping = self._get_mapping_dict()

        min_date_dt = self.cutoff_date - relativedelta(days=self.picking_interval_days)

        # TODO date_done is a Datetime field, so maybe we need more clever code
        # for our friends which are far away from GMT
        pickings = spo.search(
            [
                ("picking_type_code", "=", pick_type_map[cutoff_type]),
                ("state", "=", "done"),
                ("date_done", "<=", self.cutoff_date),
                ("date_done", ">=", min_date_dt),
                ("company_id", "=", self.company_id.id),
            ]
        )

        oline_dict = {}  # order line dict
        # key = PO line or SO line recordset
        # value = {
        #   'precut_delivered_qty': 1.0,
        #   'precut_invoiced_qty': 0.0,
        #   'price_unit': 12.42,
        #   }
        # -> we use precut_delivered_qty - precut_invoiced_qty
        for p in pickings:
            for move in p.move_lines.filtered(lambda m: m.state == "done"):
                self.stock_move_update_oline_dict(move, oline_dict)

        # from pprint import pprint
        # pprint(oline_dict)
        for vdict in oline_dict.values():
            vals = self.picking_prepare_cutoff_line(vdict, account_mapping)
            if vals:
                aclo.create(vals)
        return res