Files
stock-rma/rma/models/rma_order_line.py
Aaron ForgeFlow e60e71e58a Merge pull request #539 from akretion/16-propagate-cancel-mutliple-step-rma-line
[16][IMP] rma : propagate cancelation
2024-10-22 17:56:04 +02:00

855 lines
29 KiB
Python

# Copyright (C) 2017-20 ForgeFlow S.L.
# License LGPL-3.0 or later (https://www.gnu.org/licenses/lgpl.html)
import operator
from odoo import _, api, fields, models
from odoo.exceptions import UserError, ValidationError
ops = {"=": operator.eq, "!=": operator.ne}
class RmaOrderLine(models.Model):
_name = "rma.order.line"
_description = "RMA"
_inherit = ["mail.thread"]
_order = "id desc"
@api.model
def _get_default_type(self):
if "supplier" in self.env.context:
return "supplier"
return "customer"
@api.model
def _default_warehouse_id(self):
rma_id = self.env.context.get("default_rma_id", False)
warehouse = self.env["stock.warehouse"]
if rma_id:
rma = self.env["rma.order"].browse(rma_id)
warehouse = self.env["stock.warehouse"].search(
[("company_id", "=", rma.company_id.id)], limit=1
)
return warehouse
@api.model
def _default_location_id(self):
wh = self._default_warehouse_id()
return wh.lot_rma_id
@api.onchange("partner_id")
def _onchange_delivery_address(self):
self.delivery_address_id = self.env["res.partner"].browse(
self.partner_id.address_get(["delivery"])["delivery"]
)
@api.model
def _get_in_pickings(self):
# We consider an in move one where the first origin is outside
# of the company and the final destination is outside. In case
# of 2 or 3 step pickings, we should categorize as in shipments
# even when they are technically internal transfers.
pickings = self.env["stock.picking"]
for move in self.move_ids:
first_usage = move._get_first_usage()
last_usage = move._get_last_usage()
if last_usage == "internal" and first_usage != "internal":
pickings |= move.picking_id
elif last_usage == "supplier" and first_usage == "customer":
pickings |= move.picking_id
return pickings
@api.model
def _get_in_moves(self):
moves = self.env["stock.move"]
for move in self.move_ids:
first_usage = move._get_first_usage()
last_usage = move._get_last_usage()
if last_usage == "internal" and first_usage != "internal":
moves |= move
elif last_usage == "supplier" and first_usage == "customer":
moves |= move
return moves
@api.model
def _get_out_pickings(self):
pickings = self.env["stock.picking"]
for move in self.move_ids:
first_usage = move._get_first_usage()
last_usage = move._get_last_usage()
if first_usage in ("internal", "production") and last_usage in (
"customer",
"supplier",
):
pickings |= move.picking_id
elif last_usage == "customer" and first_usage == "supplier":
pickings |= move.picking_id
return pickings
def _compute_in_shipment_count(self):
for line in self:
pickings = line._get_in_pickings()
line.in_shipment_count = len(pickings)
def _compute_out_shipment_count(self):
for line in self:
pickings = line._get_out_pickings()
line.out_shipment_count = len(pickings)
def _get_rma_move_qty(self, states, direction="in"):
for rec in self:
product_obj = self.env["uom.uom"]
qty = 0.0
if direction == "in":
moves = rec.move_ids.filtered(
lambda m: m.state in states
and (
m.location_id.usage == "supplier"
or m.location_id.usage == "customer"
)
and (
m.location_dest_id.usage == "internal"
or m.location_dest_id.usage == "supplier"
)
)
elif direction == "out":
moves = rec.move_ids.filtered(
lambda m: m.state in states
and (
m.location_dest_id.usage == "supplier"
or m.location_dest_id.usage == "customer"
)
and (
m.location_id.usage == "internal"
or m.location_id.usage == "supplier"
)
)
for move in moves:
# If the move is part of a chain don't count it
if direction == "out" and move.move_orig_ids:
continue
elif direction == "in" and move.move_dest_ids:
continue
qty += product_obj._compute_quantity(move.product_uom_qty, rec.uom_id)
return qty
@api.depends(
"move_ids",
"move_ids.state",
"qty_received",
"receipt_policy",
"product_qty",
"type",
)
def _compute_qty_to_receive(self):
for rec in self:
rec.qty_to_receive = 0.0
if rec.receipt_policy == "ordered":
rec.qty_to_receive = rec.product_qty - rec.qty_received
elif rec.receipt_policy == "delivered":
rec.qty_to_receive = rec.qty_delivered - rec.qty_received
@api.depends(
"move_ids",
"move_ids.state",
"delivery_policy",
"product_qty",
"type",
"qty_delivered",
"qty_received",
)
def _compute_qty_to_deliver(self):
for rec in self:
rec.qty_to_deliver = 0.0
if rec.delivery_policy == "ordered":
rec.qty_to_deliver = rec.product_qty - rec.qty_delivered
elif rec.delivery_policy == "received":
rec.qty_to_deliver = rec.qty_received - rec.qty_delivered
@api.depends("move_ids", "move_ids.state", "type")
def _compute_qty_incoming(self):
for rec in self:
qty = rec._get_rma_move_qty(
("draft", "confirmed", "assigned"), direction="in"
)
rec.qty_incoming = qty
@api.depends("move_ids", "move_ids.state", "type")
def _compute_qty_received(self):
for rec in self:
qty = rec._get_rma_move_qty("done", direction="in")
rec.qty_received = qty
@api.depends("move_ids", "move_ids.state", "type")
def _compute_qty_outgoing(self):
for rec in self:
qty = rec._get_rma_move_qty(
("draft", "confirmed", "assigned"), direction="out"
)
rec.qty_outgoing = qty
@api.depends("move_ids", "move_ids.state", "type")
def _compute_qty_delivered(self):
for rec in self:
qty = rec._get_rma_move_qty("done", direction="out")
rec.qty_delivered = qty
@api.model
def _get_supplier_rma_qty(self):
return sum(
self.supplier_rma_line_ids.filtered(lambda r: r.state != "cancel").mapped(
"product_qty"
)
)
@api.depends(
"customer_to_supplier",
"supplier_rma_line_ids",
"move_ids",
"move_ids.state",
"qty_received",
"receipt_policy",
"product_qty",
"type",
)
def _compute_qty_supplier_rma(self):
for rec in self:
if rec.customer_to_supplier:
supplier_rma_qty = rec._get_supplier_rma_qty()
rec.qty_to_supplier_rma = rec.product_qty - supplier_rma_qty
rec.qty_in_supplier_rma = supplier_rma_qty
else:
rec.qty_to_supplier_rma = 0.0
rec.qty_in_supplier_rma = 0.0
def _compute_rma_line_count(self):
for rec in self.filtered(lambda r: r.type == "customer"):
rec.rma_line_count = len(rec.supplier_rma_line_ids)
for rec in self.filtered(lambda r: r.type == "supplier"):
rec.rma_line_count = len(rec.customer_rma_id)
@api.model
def _default_date_rma(self):
return fields.Datetime.now()
delivery_address_id = fields.Many2one(
comodel_name="res.partner",
string="Partner delivery address",
readonly=True,
states={"draft": [("readonly", False)]},
help="This address will be used to deliver repaired or replacement "
"products.",
)
rma_id = fields.Many2one(
comodel_name="rma.order",
string="RMA Group",
tracking=True,
readonly=True,
)
name = fields.Char(
string="Reference",
required=True,
default="/",
readonly=True,
states={"draft": [("readonly", False)]},
help="Add here the supplier RMA #. Otherwise an internal code is" " assigned.",
copy=False,
)
description = fields.Text()
conditions = fields.Html(string="Terms and conditions")
origin = fields.Char(
string="Source Document",
readonly=True,
states={"draft": [("readonly", False)]},
help="Reference of the document that produced this rma.",
)
date_rma = fields.Datetime(
string="Order Date", index=True, default=lambda self: self._default_date_rma()
)
state = fields.Selection(
selection=[
("draft", "Draft"),
("to_approve", "To Approve"),
("approved", "Approved"),
("done", "Done"),
("canceled", "Canceled"),
],
default="draft",
tracking=True,
)
operation_id = fields.Many2one(
comodel_name="rma.operation",
required=True,
readonly=False,
tracking=True,
)
assigned_to = fields.Many2one(
comodel_name="res.users",
tracking=True,
default=lambda self: self.env.uid,
)
requested_by = fields.Many2one(
comodel_name="res.users",
tracking=True,
default=lambda self: self.env.uid,
)
partner_id = fields.Many2one(
comodel_name="res.partner",
required=True,
store=True,
tracking=True,
readonly=True,
states={"draft": [("readonly", False)]},
)
sequence = fields.Integer(
default=10, help="Gives the sequence of this line when displaying the rma."
)
product_id = fields.Many2one(
comodel_name="product.product",
ondelete="restrict",
required=True,
readonly=True,
states={"draft": [("readonly", False)]},
)
product_tracking = fields.Selection(related="product_id.tracking")
lot_id = fields.Many2one(
comodel_name="stock.lot",
string="Lot/Serial Number",
readonly=True,
states={"draft": [("readonly", False)]},
)
product_qty = fields.Float(
string="Return Qty",
copy=False,
default=1.0,
digits="Product Unit of Measure",
readonly=True,
states={"draft": [("readonly", False)]},
)
uom_id = fields.Many2one(
comodel_name="uom.uom",
string="Unit of Measure",
required=True,
readonly=True,
compute="_compute_uom_id",
precompute=True,
store=True,
states={"draft": [("readonly", False)]},
)
price_unit = fields.Monetary(
string="Unit cost",
readonly=True,
states={"draft": [("readonly", False)]},
help="Unit cost of the items under RMA",
)
in_shipment_count = fields.Integer(
compute="_compute_in_shipment_count", string="# of Shipments"
)
out_shipment_count = fields.Integer(
compute="_compute_out_shipment_count", string="# of Deliveries"
)
move_ids = fields.One2many(
"stock.move", "rma_line_id", string="Stock Moves", readonly=True, copy=False
)
reference_move_id = fields.Many2one(
comodel_name="stock.move",
string="Originating Stock Move",
copy=False,
readonly=True,
states={"draft": [("readonly", False)]},
)
currency_id = fields.Many2one(
"res.currency",
default=lambda self: self.env.company.currency_id,
)
company_id = fields.Many2one(
comodel_name="res.company",
required=True,
default=lambda self: self.env.company,
)
type = fields.Selection(
selection=[("customer", "Customer"), ("supplier", "Supplier")],
required=True,
default=lambda self: self._get_default_type(),
readonly=True,
)
customer_to_supplier = fields.Boolean(
"The customer will send to the supplier",
readonly=True,
states={"draft": [("readonly", False)]},
)
supplier_to_customer = fields.Boolean(
"The supplier will send to the customer",
readonly=True,
states={"draft": [("readonly", False)]},
)
receipt_policy = fields.Selection(
[
("no", "Not required"),
("ordered", "Based on Ordered Quantities"),
("delivered", "Based on Delivered Quantities"),
],
required=True,
default="no",
readonly=False,
)
delivery_policy = fields.Selection(
[
("no", "Not required"),
("ordered", "Based on Ordered Quantities"),
("received", "Based on Received Quantities"),
],
required=True,
default="no",
readonly=False,
ondelete="cascade",
)
in_route_id = fields.Many2one(
"stock.route",
string="Inbound Route",
required=True,
domain=[("rma_selectable", "=", True)],
readonly=True,
compute="_compute_in_route_id",
precompute=True,
store=True,
states={"draft": [("readonly", False)]},
)
out_route_id = fields.Many2one(
"stock.route",
string="Outbound Route",
required=True,
domain=[("rma_selectable", "=", True)],
readonly=True,
compute="_compute_out_route_id",
precompute=True,
store=True,
states={"draft": [("readonly", False)]},
)
in_warehouse_id = fields.Many2one(
comodel_name="stock.warehouse",
string="Inbound Warehouse",
required=True,
readonly=True,
states={"draft": [("readonly", False)]},
default=lambda self: self._default_warehouse_id(),
)
out_warehouse_id = fields.Many2one(
comodel_name="stock.warehouse",
string="Outbound Warehouse",
required=True,
readonly=True,
states={"draft": [("readonly", False)]},
default=lambda self: self._default_warehouse_id(),
)
location_id = fields.Many2one(
comodel_name="stock.location",
string="Send To This Company Location",
required=True,
readonly=True,
states={"draft": [("readonly", False)]},
default=lambda self: self._default_location_id(),
)
customer_rma_id = fields.Many2one(
"rma.order.line", string="Customer RMA line", ondelete="cascade"
)
supplier_rma_line_ids = fields.One2many("rma.order.line", "customer_rma_id")
rma_line_count = fields.Integer(
compute="_compute_rma_line_count", string="# of RMA lines associated"
)
supplier_address_id = fields.Many2one(
comodel_name="res.partner",
readonly=True,
states={"draft": [("readonly", False)]},
string="Supplier Address",
help="Address of the supplier in case of Customer RMA operation " "dropship.",
)
customer_address_id = fields.Many2one(
comodel_name="res.partner",
readonly=True,
states={"draft": [("readonly", False)]},
string="Customer Address",
help="Address of the customer in case of Supplier RMA operation " "dropship.",
)
qty_to_receive = fields.Float(
digits="Product Unit of Measure",
compute="_compute_qty_to_receive",
store=True,
)
qty_incoming = fields.Float(
string="Incoming Qty",
copy=False,
readonly=True,
digits="Product Unit of Measure",
compute="_compute_qty_incoming",
store=True,
)
qty_received = fields.Float(
copy=False,
digits="Product Unit of Measure",
compute="_compute_qty_received",
store=True,
)
qty_to_deliver = fields.Float(
copy=False,
digits="Product Unit of Measure",
readonly=True,
compute="_compute_qty_to_deliver",
store=True,
)
qty_outgoing = fields.Float(
string="Outgoing Qty",
copy=False,
readonly=True,
digits="Product Unit of Measure",
compute="_compute_qty_outgoing",
store=True,
)
qty_delivered = fields.Float(
copy=False,
digits="Product Unit of Measure",
readonly=True,
compute="_compute_qty_delivered",
store=True,
)
qty_to_supplier_rma = fields.Float(
string="Qty to send to Supplier RMA",
digits="Product Unit of Measure",
readonly=True,
compute="_compute_qty_supplier_rma",
store=True,
)
qty_in_supplier_rma = fields.Float(
string="Qty in Supplier RMA",
digits="Product Unit of Measure",
readonly=True,
compute="_compute_qty_supplier_rma",
store=True,
)
under_warranty = fields.Boolean(
string="Under Warranty?", readonly=True, states={"draft": [("readonly", False)]}
)
def _prepare_rma_line_from_stock_move(self, sm, lot=False):
if not self.type:
self.type = self._get_default_type()
if self.type == "customer":
operation = (
sm.product_id.rma_customer_operation_id
or sm.product_id.categ_id.rma_customer_operation_id
)
else:
operation = (
sm.product_id.rma_supplier_operation_id
or sm.product_id.categ_id.rma_supplier_operation_id
)
if not operation:
operation = self.env["rma.operation"].search(
[("type", "=", self.type)], limit=1
)
if not operation:
raise ValidationError(_("Please define an operation first."))
if not operation.in_route_id or not operation.out_route_id:
route = self.env["stock.route"].search(
[("rma_selectable", "=", True)], limit=1
)
if not route:
raise ValidationError(_("Please define an RMA route."))
if (
not operation.in_warehouse_id
or not operation.out_warehouse_id
or not (
operation.in_warehouse_id.lot_rma_id
or operation.out_warehouse_id.lot_rma_id
)
):
warehouse = self.env["stock.warehouse"].search(
[("company_id", "=", self.company_id.id), ("lot_rma_id", "!=", False)],
limit=1,
)
if not warehouse:
raise ValidationError(
_("Please define a warehouse with a default RMA location.")
)
data = {
"product_id": sm.product_id.id,
"lot_id": lot and lot.id or False,
"origin": sm.picking_id.name or sm.name,
"uom_id": sm.product_uom.id,
"product_qty": sm.product_uom_qty,
"delivery_address_id": sm.picking_id.partner_id.id,
"operation_id": operation.id,
"receipt_policy": operation.receipt_policy,
"delivery_policy": operation.delivery_policy,
"in_warehouse_id": operation.in_warehouse_id.id or warehouse.id,
"out_warehouse_id": operation.out_warehouse_id.id or warehouse.id,
"in_route_id": operation.in_route_id.id or route.id,
"out_route_id": operation.out_route_id.id or route.id,
"location_id": (
operation.location_id.id
or operation.in_warehouse_id.lot_rma_id.id
or operation.out_warehouse_id.lot_rma_id.id
or warehouse.lot_rma_id.id
),
}
return data
@api.onchange("reference_move_id")
def _onchange_reference_move_id(self):
self.ensure_one()
sm = self.reference_move_id
if not sm:
return
if sm.move_line_ids.lot_id:
if len(sm.move_line_ids.lot_id) > 1:
raise UserError(_("To manage lots use RMA groups."))
else:
data = self._prepare_rma_line_from_stock_move(
sm, lot=sm.move_line_ids.lot_id[0]
)
self.update(data)
else:
data = self._prepare_rma_line_from_stock_move(sm, lot=False)
self.update(data)
self._remove_other_data_origin("reference_move_id")
@api.constrains("reference_move_id", "partner_id")
def _check_move_partner(self):
for rec in self:
if (
rec.reference_move_id
and rec.reference_move_id.picking_id.partner_id != rec.partner_id
):
raise ValidationError(
_(
"RMA customer and originating stock move customer "
"doesn't match."
)
)
def _remove_other_data_origin(self, exception):
if not exception == "reference_move_id":
self.reference_move_id = False
return True
def _check_lot_assigned(self):
for rec in self:
if rec.product_id.tracking == "serial" and rec.product_qty != 1:
raise ValidationError(
_(
"Product %s has serial tracking configuration, "
"quantity to receive should be 1"
)
% (rec.product_id.display_name)
)
def action_rma_to_approve(self):
self._check_lot_assigned()
self.write({"state": "to_approve"})
for rec in self:
if rec.product_id.rma_approval_policy == "one_step":
rec.action_rma_approve()
return True
def action_rma_draft(self):
self.write({"state": "draft"})
return True
def action_rma_approve(self):
self.write({"state": "approved"})
return True
def action_rma_done(self):
self.write({"state": "done"})
return True
@api.model_create_multi
def create(self, vals_list):
for vals in vals_list:
if not vals.get("name") or vals.get("name") == "/":
if self.env.context.get("supplier"):
vals["name"] = self.env["ir.sequence"].next_by_code(
"rma.order.line.supplier"
)
else:
vals["name"] = self.env["ir.sequence"].next_by_code(
"rma.order.line.customer"
)
return super().create(vals_list)
def check_cancel(self):
for move in self.move_ids:
if move.state == "done":
raise UserError(
_("Unable to cancel %s as some receptions have already been done.")
% (self.name)
)
def action_rma_cancel(self):
for order in self:
order.check_cancel()
# cancel ongoing orig moves
# dest move cancelation can be managed with propagate_cancel option
# on stock rules.
moves = order.move_ids
to_cancel_orig_moves = self.env["stock.move"]
while moves:
moves = moves.move_orig_ids.filtered(
lambda m: m.state not in ("done", "cancel") and m.picking_id
)
to_cancel_orig_moves |= moves
to_cancel_orig_moves._action_cancel()
order.write({"state": "canceled"})
order.move_ids._action_cancel()
return True
def _get_price_unit(self):
"""The price unit corresponds to the cost of that product"""
self.ensure_one()
if self.reference_move_id:
price_unit = self.reference_move_id.price_unit
else:
price_unit = self.product_id.with_company(self.company_id).standard_price
return price_unit
@api.onchange("product_id")
def _onchange_product_id(self):
result = {}
if not self.product_id:
return result
self.uom_id = self.product_id.uom_id.id
if not self.type:
self.type = self._get_default_type()
self.price_unit = self._get_price_unit()
if self.type == "customer":
self.operation_id = (
self.product_id.rma_customer_operation_id
or self.product_id.categ_id.rma_customer_operation_id
)
else:
self.operation_id = (
self.product_id.rma_supplier_operation_id
or self.product_id.categ_id.rma_supplier_operation_id
)
if self.lot_id.product_id != self.product_id:
self.lot_id = False
return result
@api.onchange("operation_id")
def _onchange_operation_id(self):
result = {}
if not self.operation_id:
return result
self.receipt_policy = self.operation_id.receipt_policy
self.delivery_policy = self.operation_id.delivery_policy
self.customer_to_supplier = (
self.rma_id.customer_to_supplier or self.operation_id.customer_to_supplier
)
self.supplier_to_customer = (
self.rma_id.supplier_to_customer or self.operation_id.supplier_to_customer
)
self.in_warehouse_id = (
self.rma_id.in_warehouse_id or self.operation_id.in_warehouse_id
)
self.out_warehouse_id = (
self.rma_id.out_warehouse_id or self.operation_id.out_warehouse_id
)
self.location_id = (
self.rma_id.location_id
or self.operation_id.location_id
or self.in_warehouse_id.lot_rma_id
)
self.in_route_id = self.rma_id.in_route_id or self.operation_id.in_route_id
self.out_route_id = self.rma_id.out_route_id or self.operation_id.out_route_id
return result
@api.depends("operation_id")
def _compute_in_route_id(self):
for rec in self:
rec.in_route_id = rec.operation_id.in_route_id
@api.depends("operation_id")
def _compute_out_route_id(self):
for rec in self:
rec.out_route_id = rec.operation_id.out_route_id
@api.onchange("customer_to_supplier", "type")
def _onchange_receipt_policy(self):
if self.type == "supplier" and self.customer_to_supplier:
self.receipt_policy = "no"
elif self.type == "customer" and self.supplier_to_customer:
self.delivery_policy = "no"
@api.onchange("lot_id")
def _onchange_lot_id(self):
product = self.lot_id.product_id
if product:
self.product_id = product
@api.depends("product_id")
def _compute_uom_id(self):
for rec in self:
if rec.product_id:
rec.uom_id = rec.product_id.uom_id
def action_view_in_shipments(self):
action = self.env.ref("stock.action_picking_tree_all")
result = action.sudo().read()[0]
shipments = self.env["stock.picking"]
for line in self:
shipments |= line._get_in_pickings()
# choose the view_mode accordingly
if len(shipments) != 1:
result["domain"] = "[('id', 'in', " + str(shipments.ids) + ")]"
elif len(shipments) == 1:
res = self.env.ref("stock.view_picking_form", False)
result["views"] = [(res and res.id or False, "form")]
result["res_id"] = shipments.ids[0]
return result
def action_view_out_shipments(self):
action = self.env.ref("stock.action_picking_tree_all")
result = action.sudo().read()[0]
shipments = self.env["stock.picking"]
for line in self:
shipments |= line._get_out_pickings()
# choose the view_mode accordingly
if len(shipments) != 1:
result["domain"] = "[('id', 'in', " + str(shipments.ids) + ")]"
elif len(shipments) == 1:
res = self.env.ref("stock.view_picking_form", False)
result["views"] = [(res and res.id or False, "form")]
result["res_id"] = shipments.ids[0]
return result
def action_view_rma_lines(self):
if self.type == "customer":
# from customer we link to supplier rma
action = self.env.ref("rma.action_rma_supplier_lines")
rma_lines = self.supplier_rma_line_ids
res = self.env.ref("rma.view_rma_line_supplier_form", False)
else:
# from supplier we link to customer rma
action = self.env.ref("rma.action_rma_customer_lines")
rma_lines = self.customer_rma_id
res = self.env.ref("rma.view_rma_line_form", False)
result = action.sudo().read()[0]
# choose the view_mode accordingly
if rma_lines and len(rma_lines) != 1:
result["domain"] = [("id", "in", rma_lines.ids)]
elif len(rma_lines) == 1:
result["views"] = [(res and res.id or False, "form")]
result["res_id"] = rma_lines.id
return result
@api.constrains("partner_id", "rma_id")
def _check_partner_id(self):
if self.rma_id and self.partner_id != self.rma_id.partner_id:
raise ValidationError(
_("Group partner and RMA's partner must be the same.")
)