813 lines
31 KiB
Python
813 lines
31 KiB
Python
# This file is part of Tryton. The COPYRIGHT file at the top level of
|
|
# this repository contains the full copyright notices and license terms.
|
|
from decimal import Decimal
|
|
from types import SimpleNamespace
|
|
|
|
from trytond.exceptions import UserError, UserWarning
|
|
from trytond.model import ModelView, fields
|
|
from trytond.pool import Pool
|
|
from trytond.pyson import Bool, Eval
|
|
from trytond.transaction import Transaction
|
|
from trytond.wizard import Button, StateAction, StateTransition, StateView, Wizard
|
|
|
|
from trytond.modules.purchase_trade.service import ContractFactory
|
|
|
|
|
|
PRICE_TYPES = [
|
|
('cash', 'Cash Price'),
|
|
('priced', 'Priced'),
|
|
('basis', 'Basis'),
|
|
('efp', 'EFP'),
|
|
]
|
|
|
|
MIRROR_MODES = [
|
|
('duplicate_pair', 'Duplicate matched pair'),
|
|
('new_mirror', 'Create new mirror contract'),
|
|
]
|
|
|
|
|
|
class TradeCustomDuplicateStart(ModelView):
|
|
"Custom Duplicate"
|
|
__name__ = 'trade.custom_duplicate.start'
|
|
|
|
source_model = fields.Selection([
|
|
('purchase.purchase', 'Purchase'),
|
|
('sale.sale', 'Sale'),
|
|
], "Source", readonly=True)
|
|
source = fields.Reference("Source Record", selection=[
|
|
('purchase.purchase', 'Purchase'),
|
|
('sale.sale', 'Sale'),
|
|
], readonly=True)
|
|
duplicated_record_id = fields.Integer("Duplicated Record ID", readonly=True)
|
|
duplicate_pair_available = fields.Boolean(
|
|
"Duplicate Pair Available", readonly=True)
|
|
party = fields.Many2One('party.party', "Counterparty", required=True)
|
|
currency = fields.Many2One('currency.currency', "Currency", required=True)
|
|
payment_term = fields.Many2One(
|
|
'account.invoice.payment_term', "Payment Term", required=True)
|
|
incoterm = fields.Many2One('incoterm.incoterm', "Incoterm", required=True)
|
|
quantity = fields.Numeric("Contractual Quantity", digits='unit',
|
|
required=True)
|
|
unit = fields.Many2One('product.uom', "Unit", readonly=True)
|
|
unit_price = fields.Numeric("Price", digits=(16, 4), required=True)
|
|
price_type = fields.Selection(PRICE_TYPES, "Price Type", required=True)
|
|
clear_fees = fields.Boolean("Clear Fees")
|
|
clear_pricing_components = fields.Boolean("Clear Pricing Components")
|
|
create_matched_mirror = fields.Boolean("Create matched mirror contract")
|
|
mirror_mode = fields.Selection(MIRROR_MODES, "Mirror Mode",
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': Bool(Eval('create_matched_mirror')),
|
|
'readonly': ~Bool(Eval('duplicate_pair_available')),
|
|
})
|
|
mirror_party = fields.Many2One('party.party', "Mirror Counterparty",
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': (
|
|
Bool(Eval('create_matched_mirror'))
|
|
& (Eval('mirror_mode') == 'new_mirror')),
|
|
})
|
|
mirror_currency = fields.Many2One('currency.currency', "Mirror Currency",
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': Bool(Eval('create_matched_mirror')),
|
|
})
|
|
mirror_payment_term = fields.Many2One(
|
|
'account.invoice.payment_term', "Mirror Payment Term",
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': Bool(Eval('create_matched_mirror')),
|
|
})
|
|
mirror_incoterm = fields.Many2One('incoterm.incoterm', "Mirror Incoterm",
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': Bool(Eval('create_matched_mirror')),
|
|
})
|
|
mirror_quantity = fields.Numeric("Mirror Contractual Quantity",
|
|
digits='unit',
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': Bool(Eval('create_matched_mirror')),
|
|
})
|
|
mirror_unit = fields.Many2One('product.uom', "Mirror Unit",
|
|
readonly=True,
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': Bool(Eval('create_matched_mirror')),
|
|
})
|
|
mirror_unit_price = fields.Numeric("Mirror Price", digits=(16, 4),
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': Bool(Eval('create_matched_mirror')),
|
|
})
|
|
mirror_price_type = fields.Selection(PRICE_TYPES, "Mirror Price Type",
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': Bool(Eval('create_matched_mirror')),
|
|
})
|
|
mirror_clear_fees = fields.Boolean("Mirror Clear Fees",
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
})
|
|
mirror_clear_pricing_components = fields.Boolean(
|
|
"Mirror Clear Pricing Components",
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
})
|
|
|
|
@fields.depends('source', 'mirror_mode')
|
|
def on_change_mirror_mode(self):
|
|
record = TradeCustomDuplicate._reference_record(self.source)
|
|
if not record or not self.mirror_mode:
|
|
return
|
|
line = TradeCustomDuplicate._trade_line(record)
|
|
if not line:
|
|
return
|
|
defaults = TradeCustomDuplicate._mirror_defaults(
|
|
record, line, self.mirror_mode, strict=False, as_ids=False)
|
|
for field, value in defaults.items():
|
|
setattr(self, field, value)
|
|
|
|
|
|
class TradeCustomDuplicate(Wizard):
|
|
"Custom Duplicate"
|
|
__name__ = 'trade.custom_duplicate'
|
|
|
|
start = StateView(
|
|
'trade.custom_duplicate.start',
|
|
'purchase_trade.trade_custom_duplicate_start_view_form', [
|
|
Button("Cancel", 'end', 'tryton-cancel'),
|
|
Button("Duplicate", 'duplicate', 'tryton-ok', default=True),
|
|
])
|
|
duplicate = StateTransition()
|
|
open_purchase = StateAction('purchase.act_purchase_form')
|
|
open_sale = StateAction('sale.act_sale_form')
|
|
|
|
@staticmethod
|
|
def _trade_line(record):
|
|
lines = [
|
|
line for line in (record.lines or [])
|
|
if getattr(line, 'type', 'line') == 'line' and line.product]
|
|
if len(lines) > 1:
|
|
raise UserError(
|
|
"Custom duplicate only supports contracts with one trade line.")
|
|
return lines[0] if lines else None
|
|
|
|
@staticmethod
|
|
def _source_model(record):
|
|
name = record.__name__
|
|
if name not in {'purchase.purchase', 'sale.sale'}:
|
|
raise UserError("Custom duplicate only supports purchases and sales.")
|
|
return name
|
|
|
|
@staticmethod
|
|
def _opposite_type(source_model):
|
|
return 'Sale' if source_model == 'purchase.purchase' else 'Purchase'
|
|
|
|
@staticmethod
|
|
def _record_from_lot_report(record):
|
|
if getattr(record, '__name__', None) != 'lot.report':
|
|
return record
|
|
if getattr(record, 'r_purchase', None):
|
|
return record.r_purchase
|
|
if getattr(record, 'r_sale', None):
|
|
return record.r_sale
|
|
lot = (
|
|
getattr(record, 'r_lot_p', None)
|
|
or getattr(record, 'r_lot_s', None)
|
|
or getattr(record, 'r_lot_himself', None))
|
|
if lot:
|
|
line = getattr(lot, 'line', None)
|
|
if line and getattr(line, 'purchase', None):
|
|
return line.purchase
|
|
sale_line = getattr(lot, 'sale_line', None)
|
|
if sale_line and getattr(sale_line, 'sale', None):
|
|
return sale_line.sale
|
|
raise UserError(
|
|
"Custom duplicate could not find a purchase or sale from this "
|
|
"Lots Management line.")
|
|
|
|
@staticmethod
|
|
def _source_record(record):
|
|
return TradeCustomDuplicate._record_from_lot_report(record)
|
|
|
|
@staticmethod
|
|
def _reference_record(reference):
|
|
if not reference:
|
|
return None
|
|
if isinstance(reference, str):
|
|
model, record_id = reference.split(',', 1)
|
|
return Pool().get(model)(int(record_id))
|
|
return reference
|
|
|
|
def default_start(self, fields):
|
|
if len(self.records) != 1:
|
|
raise UserError(
|
|
"Please select a single purchase, sale, or Lots Management "
|
|
"line.")
|
|
|
|
record = self._source_record(self.records[0])
|
|
source_model = self._source_model(record)
|
|
line = self._trade_line(record)
|
|
if not line:
|
|
raise UserError("The selected contract has no trade line to duplicate.")
|
|
matched_record = self._matched_counterpart_record(
|
|
record, line, strict=False)
|
|
duplicate_pair_available = bool(matched_record)
|
|
mirror_mode = 'duplicate_pair' if matched_record else 'new_mirror'
|
|
mirror_defaults = self._mirror_defaults(
|
|
record, line, mirror_mode, strict=False)
|
|
|
|
return {
|
|
'source_model': source_model,
|
|
'source': '%s,%s' % (source_model, record.id),
|
|
'party': record.party.id if record.party else None,
|
|
'currency': record.currency.id if record.currency else None,
|
|
'payment_term': (
|
|
record.payment_term.id if record.payment_term else None),
|
|
'incoterm': record.incoterm.id if record.incoterm else None,
|
|
'quantity': (
|
|
getattr(line, 'quantity_theorical', None)
|
|
or getattr(line, 'quantity', None)),
|
|
'unit': line.unit.id if line.unit else None,
|
|
'unit_price': getattr(line, 'unit_price', None),
|
|
'price_type': getattr(line, 'price_type', None) or 'priced',
|
|
'clear_fees': False,
|
|
'clear_pricing_components': False,
|
|
'create_matched_mirror': False,
|
|
'duplicate_pair_available': duplicate_pair_available,
|
|
'mirror_mode': mirror_mode,
|
|
**mirror_defaults,
|
|
}
|
|
|
|
@staticmethod
|
|
def _field_id(record, field):
|
|
value = getattr(record, field, None)
|
|
return value.id if value else None
|
|
|
|
@staticmethod
|
|
def _field_value(record, field, as_ids=True):
|
|
value = getattr(record, field, None)
|
|
if as_ids:
|
|
return value.id if value else None
|
|
return value
|
|
|
|
@staticmethod
|
|
def _line_quantity_value(line):
|
|
return (
|
|
getattr(line, 'quantity_theorical', None)
|
|
or getattr(line, 'quantity', None))
|
|
|
|
@staticmethod
|
|
def _mirror_defaults(record, line, mirror_mode, strict=True, as_ids=True):
|
|
mirror_record = record
|
|
mirror_line = line
|
|
if mirror_mode == 'duplicate_pair':
|
|
mirror_line, mirror_record = TradeCustomDuplicate._matched_counterpart(
|
|
line, record=record, strict=strict)
|
|
if not mirror_line or not mirror_record:
|
|
mirror_record = record
|
|
mirror_line = line
|
|
return {
|
|
'mirror_party': TradeCustomDuplicate._field_value(
|
|
mirror_record, 'party', as_ids=as_ids),
|
|
'mirror_currency': TradeCustomDuplicate._field_value(
|
|
mirror_record, 'currency', as_ids=as_ids),
|
|
'mirror_payment_term': TradeCustomDuplicate._field_value(
|
|
mirror_record, 'payment_term', as_ids=as_ids),
|
|
'mirror_incoterm': TradeCustomDuplicate._field_value(
|
|
mirror_record, 'incoterm', as_ids=as_ids),
|
|
'mirror_quantity': TradeCustomDuplicate._line_quantity_value(
|
|
mirror_line),
|
|
'mirror_unit': (
|
|
mirror_line.unit.id if (
|
|
as_ids and getattr(mirror_line, 'unit', None))
|
|
else getattr(mirror_line, 'unit', None)),
|
|
'mirror_unit_price': getattr(mirror_line, 'unit_price', None),
|
|
'mirror_price_type': (
|
|
getattr(mirror_line, 'price_type', None) or 'priced'),
|
|
'mirror_clear_fees': False,
|
|
'mirror_clear_pricing_components': False,
|
|
}
|
|
|
|
@staticmethod
|
|
def _copy_contract(record, options):
|
|
Model = Pool().get(record.__name__)
|
|
default = {
|
|
'party': options.party.id,
|
|
'currency': options.currency.id,
|
|
'payment_term': options.payment_term.id,
|
|
'incoterm': options.incoterm.id,
|
|
}
|
|
new_record, = Model.copy([record], default=default)
|
|
new_record = Model(new_record.id)
|
|
TradeCustomDuplicate._apply_party_addresses(new_record, options.party)
|
|
new_record.save()
|
|
return Model(new_record.id)
|
|
|
|
@staticmethod
|
|
def _mirror_options(options, fallback_record=None, fallback_line=None):
|
|
return SimpleNamespace(
|
|
party=(
|
|
options.mirror_party
|
|
or getattr(fallback_record, 'party', None)),
|
|
currency=(
|
|
options.mirror_currency
|
|
or getattr(fallback_record, 'currency', None)),
|
|
payment_term=(
|
|
options.mirror_payment_term
|
|
or getattr(fallback_record, 'payment_term', None)),
|
|
incoterm=(
|
|
options.mirror_incoterm
|
|
or getattr(fallback_record, 'incoterm', None)),
|
|
quantity=(
|
|
options.mirror_quantity
|
|
or TradeCustomDuplicate._line_quantity_value(fallback_line)),
|
|
unit=(
|
|
options.mirror_unit
|
|
or getattr(fallback_line, 'unit', None)),
|
|
unit_price=(
|
|
options.mirror_unit_price
|
|
if options.mirror_unit_price is not None
|
|
else getattr(fallback_line, 'unit_price', None)),
|
|
price_type=(
|
|
options.mirror_price_type
|
|
or getattr(fallback_line, 'price_type', None)
|
|
or 'priced'),
|
|
clear_fees=options.mirror_clear_fees,
|
|
clear_pricing_components=(
|
|
options.mirror_clear_pricing_components),
|
|
)
|
|
|
|
@staticmethod
|
|
def _apply_party_addresses(record, party):
|
|
addresses = list(getattr(party, 'addresses', None) or [])
|
|
address = addresses[0] if addresses else None
|
|
if hasattr(record, 'invoice_address'):
|
|
record.invoice_address = address
|
|
if hasattr(record, 'shipment_address'):
|
|
record.shipment_address = address
|
|
|
|
@staticmethod
|
|
def _apply_line_options(record, options):
|
|
line = TradeCustomDuplicate._trade_line(record)
|
|
if not line:
|
|
raise UserError("The duplicated contract has no trade line.")
|
|
|
|
line.quantity = options.quantity
|
|
line.quantity_theorical = options.quantity
|
|
line.unit_price = options.unit_price
|
|
line.price_type = options.price_type
|
|
if hasattr(line, 'finished'):
|
|
line.finished = False
|
|
if hasattr(line, 'currency'):
|
|
line.currency = options.currency
|
|
if options.clear_fees and hasattr(line, 'fees'):
|
|
line.fees = []
|
|
if (options.clear_pricing_components
|
|
and hasattr(line, 'price_components')):
|
|
line.price_components = []
|
|
line.save()
|
|
return line.__class__(line.id)
|
|
|
|
@staticmethod
|
|
def _first_virtual_lot(line):
|
|
for lot in line.lots or []:
|
|
if getattr(lot, 'lot_type', None) == 'virtual':
|
|
return lot
|
|
if hasattr(line, 'getVirtualLot'):
|
|
return line.getVirtualLot()
|
|
return None
|
|
|
|
@staticmethod
|
|
def _line_side(line):
|
|
if getattr(line, '__name__', None) == 'purchase.line':
|
|
return 'purchase'
|
|
if getattr(line, '__name__', None) == 'sale.line':
|
|
return 'sale'
|
|
if getattr(line, 'purchase', None):
|
|
return 'purchase'
|
|
if getattr(line, 'sale', None):
|
|
return 'sale'
|
|
|
|
@staticmethod
|
|
def _line_quantity(line):
|
|
quantity = (
|
|
getattr(line, 'quantity_theorical', None)
|
|
or getattr(line, 'quantity', None)
|
|
or Decimal(0))
|
|
return Decimal(str(quantity)).quantize(Decimal("0.00001"))
|
|
|
|
@staticmethod
|
|
def _ensure_open_virtual_lot(line):
|
|
product = getattr(line, 'product', None)
|
|
if not product or getattr(product, 'type', None) == 'service':
|
|
return None
|
|
|
|
quantity = TradeCustomDuplicate._line_quantity(line)
|
|
if quantity <= 0:
|
|
return None
|
|
|
|
side = TradeCustomDuplicate._line_side(line)
|
|
if side not in {'purchase', 'sale'}:
|
|
return None
|
|
|
|
lot = TradeCustomDuplicate._first_virtual_lot(line)
|
|
if not lot:
|
|
lot = TradeCustomDuplicate._create_open_virtual_lot(
|
|
line, side, quantity)
|
|
|
|
TradeCustomDuplicate._ensure_open_lot_qt(lot, side, quantity)
|
|
return lot
|
|
|
|
@staticmethod
|
|
def _create_open_virtual_lot(line, side, quantity):
|
|
pool = Pool()
|
|
Lot = pool.get('lot.lot')
|
|
LotQtHist = pool.get('lot.qt.hist')
|
|
LotQtType = pool.get('lot.qt.type')
|
|
|
|
lot = Lot()
|
|
if side == 'purchase':
|
|
lot.line = line.id
|
|
lot.lot_qt = None
|
|
lot.lot_unit = None
|
|
else:
|
|
lot.sale_line = line.id
|
|
lot.lot_qt = quantity
|
|
lot.lot_unit_line = line.unit
|
|
lot.lot_quantity = quantity
|
|
lot.lot_gross_quantity = None
|
|
lot.lot_status = 'forecast'
|
|
lot.lot_type = 'virtual'
|
|
lot.lot_product = line.product
|
|
|
|
lqtt = LotQtType.search([('sequence', '=', 1)])
|
|
if lqtt:
|
|
lqh = LotQtHist()
|
|
lqh.quantity_type = lqtt[0]
|
|
lqh.quantity = quantity
|
|
lqh.gross_quantity = quantity
|
|
lot.lot_hist = [lqh]
|
|
Lot.save([lot])
|
|
|
|
lot = Lot(lot.id)
|
|
TradeCustomDuplicate._sync_fee_lots(line, lot, side)
|
|
return lot
|
|
|
|
@staticmethod
|
|
def _sync_fee_lots(line, lot, side):
|
|
fees = list(getattr(line, 'fees', None) or [])
|
|
if not fees:
|
|
return
|
|
FeeLots = Pool().get('fee.lots')
|
|
line_field = 'line' if side == 'purchase' else 'sale_line'
|
|
for fee in fees:
|
|
domain = [
|
|
('fee', '=', fee.id),
|
|
('lot', '=', lot.id),
|
|
(line_field, '=', line.id),
|
|
]
|
|
if not FeeLots.search(domain):
|
|
fee_lot = FeeLots()
|
|
fee_lot.fee = fee.id
|
|
fee_lot.lot = lot.id
|
|
setattr(fee_lot, line_field, line.id)
|
|
FeeLots.save([fee_lot])
|
|
|
|
@staticmethod
|
|
def _ensure_open_lot_qt(lot, side, quantity):
|
|
LotQt = Pool().get('lot.qt')
|
|
if side == 'purchase':
|
|
domain = [
|
|
('lot_p', '=', lot.id),
|
|
('lot_s', '=', None),
|
|
('lot_shipment_in', '=', None),
|
|
('lot_shipment_internal', '=', None),
|
|
('lot_shipment_out', '=', None),
|
|
]
|
|
create = lambda: lot.createVirtualPart(quantity, None, None)
|
|
else:
|
|
domain = [
|
|
('lot_p', '=', None),
|
|
('lot_s', '=', lot.id),
|
|
('lot_shipment_in', '=', None),
|
|
('lot_shipment_internal', '=', None),
|
|
('lot_shipment_out', '=', None),
|
|
]
|
|
create = lambda: lot.createVirtualPart(
|
|
quantity, None, lot.id, 'only sale')
|
|
if not LotQt.search(domain):
|
|
create()
|
|
|
|
@staticmethod
|
|
def _open_lot_qt(lot, side):
|
|
LotQt = Pool().get('lot.qt')
|
|
if side == 'purchase':
|
|
domain = [
|
|
('lot_p', '=', lot.id),
|
|
('lot_s', '=', None),
|
|
('lot_shipment_in', '=', None),
|
|
('lot_shipment_internal', '=', None),
|
|
('lot_shipment_out', '=', None),
|
|
]
|
|
else:
|
|
domain = [
|
|
('lot_p', '=', None),
|
|
('lot_s', '=', lot.id),
|
|
('lot_shipment_in', '=', None),
|
|
('lot_shipment_internal', '=', None),
|
|
('lot_shipment_out', '=', None),
|
|
]
|
|
records = LotQt.search(domain)
|
|
return records[0] if records else None
|
|
|
|
@staticmethod
|
|
def _matched_counterpart_line(line, strict=True, warn=False):
|
|
side = TradeCustomDuplicate._line_side(line)
|
|
if side not in {'purchase', 'sale'}:
|
|
if strict:
|
|
raise UserError(
|
|
"Custom duplicate cannot determine the contract side.")
|
|
return None
|
|
|
|
lot_ids = [lot.id for lot in (getattr(line, 'lots', None) or [])]
|
|
if not lot_ids:
|
|
if strict:
|
|
raise UserError(
|
|
"Duplicate matched pair requires an existing matched lot.")
|
|
return None
|
|
|
|
LotQt = Pool().get('lot.qt')
|
|
if side == 'purchase':
|
|
lot_qts = LotQt.search([
|
|
('lot_p', 'in', lot_ids),
|
|
('lot_s', '!=', None),
|
|
('lot_quantity', '>', 0),
|
|
])
|
|
lines = [
|
|
lqt.lot_s.sale_line for lqt in lot_qts
|
|
if getattr(getattr(lqt, 'lot_s', None), 'sale_line', None)]
|
|
lines.extend(
|
|
lot.sale_line for lot in (getattr(line, 'lots', None) or [])
|
|
if (getattr(lot, 'lot_type', None) == 'physic'
|
|
and getattr(lot, 'sale_line', None)))
|
|
else:
|
|
lot_qts = LotQt.search([
|
|
('lot_s', 'in', lot_ids),
|
|
('lot_p', '!=', None),
|
|
('lot_quantity', '>', 0),
|
|
])
|
|
lines = [
|
|
lqt.lot_p.line for lqt in lot_qts
|
|
if getattr(getattr(lqt, 'lot_p', None), 'line', None)]
|
|
lines.extend(
|
|
lot.line for lot in (getattr(line, 'lots', None) or [])
|
|
if (getattr(lot, 'lot_type', None) == 'physic'
|
|
and getattr(lot, 'line', None)))
|
|
|
|
unique = []
|
|
seen = set()
|
|
for matched_line in lines:
|
|
matched_id = getattr(matched_line, 'id', id(matched_line))
|
|
if matched_id in seen:
|
|
continue
|
|
seen.add(matched_id)
|
|
unique.append(matched_line)
|
|
|
|
if unique:
|
|
if warn and len(unique) > 1:
|
|
TradeCustomDuplicate._warn_multiple_counterparts(line)
|
|
return unique[0]
|
|
if strict:
|
|
raise UserError(
|
|
"Duplicate matched pair requires an existing matched "
|
|
"counterpart.")
|
|
return None
|
|
|
|
@staticmethod
|
|
def _warn_multiple_counterparts(line):
|
|
Warning = Pool().get('res.user.warning')
|
|
warning_name = 'custom_duplicate_multiple_counterparts:%s' % (
|
|
getattr(line, 'id', id(line)))
|
|
if Warning.check(warning_name):
|
|
raise UserWarning(
|
|
warning_name,
|
|
"Several matched counterpart contracts were found. The custom "
|
|
"duplicate will use the first one by default. Continue only "
|
|
"if that is the expected sale/purchase to duplicate.")
|
|
|
|
@staticmethod
|
|
def _counterpart_record_from_line(matched_line):
|
|
return (
|
|
getattr(matched_line, 'sale', None)
|
|
or getattr(matched_line, 'purchase', None))
|
|
|
|
@staticmethod
|
|
def _btb_counterpart_line(record, strict=True, warn=False):
|
|
btb = getattr(record, 'btb', None)
|
|
if not btb:
|
|
if strict:
|
|
raise UserError(
|
|
"Duplicate matched pair requires an existing matched "
|
|
"lot or back-to-back counterpart.")
|
|
return None
|
|
|
|
source_model = TradeCustomDuplicate._source_model(record)
|
|
counterpart_records = (
|
|
getattr(btb, 'sale', None)
|
|
if source_model == 'purchase.purchase'
|
|
else getattr(btb, 'purchase', None))
|
|
lines = []
|
|
for counterpart in counterpart_records or []:
|
|
line = TradeCustomDuplicate._trade_line(counterpart)
|
|
if line:
|
|
lines.append(line)
|
|
|
|
unique = []
|
|
seen = set()
|
|
for line in lines:
|
|
line_id = getattr(line, 'id', id(line))
|
|
if line_id in seen:
|
|
continue
|
|
seen.add(line_id)
|
|
unique.append(line)
|
|
|
|
if unique:
|
|
if warn and len(unique) > 1:
|
|
TradeCustomDuplicate._warn_multiple_counterparts(
|
|
TradeCustomDuplicate._trade_line(record))
|
|
return unique[0]
|
|
if strict:
|
|
raise UserError(
|
|
"Duplicate matched pair requires an existing matched "
|
|
"counterpart.")
|
|
return None
|
|
|
|
@staticmethod
|
|
def _matched_counterpart(line, record=None, strict=True):
|
|
matched_line = TradeCustomDuplicate._matched_counterpart_line(
|
|
line, strict=False, warn=strict)
|
|
if not matched_line and record:
|
|
matched_line = TradeCustomDuplicate._btb_counterpart_line(
|
|
record, strict=False, warn=strict)
|
|
if not matched_line:
|
|
if strict:
|
|
raise UserError(
|
|
"Duplicate matched pair requires exactly one matched "
|
|
"counterpart.")
|
|
return None, None
|
|
return (
|
|
matched_line,
|
|
TradeCustomDuplicate._counterpart_record_from_line(matched_line))
|
|
|
|
@staticmethod
|
|
def _matched_counterpart_record(record, line, strict=True):
|
|
matched_line, matched_record = TradeCustomDuplicate._matched_counterpart(
|
|
line, record=record, strict=strict)
|
|
return matched_record
|
|
|
|
@staticmethod
|
|
def _pair_duplicate_options(matched_record, matched_line, options):
|
|
return TradeCustomDuplicate._mirror_options(
|
|
options, fallback_record=matched_record,
|
|
fallback_line=matched_line)
|
|
|
|
@staticmethod
|
|
def _match_duplicate_pair_lots(source_line, source_lot, counterpart_line,
|
|
counterpart_lot, quantity):
|
|
LotQt = Pool().get('lot.qt')
|
|
Lot = Pool().get('lot.lot')
|
|
source_side = TradeCustomDuplicate._line_side(source_line)
|
|
if source_side not in {'purchase', 'sale'}:
|
|
raise UserError(
|
|
"Custom duplicate cannot determine the contract side.")
|
|
if source_side == 'purchase':
|
|
purchase_line = source_line
|
|
purchase_lot = source_lot
|
|
sale_line = counterpart_line
|
|
sale_lot = counterpart_lot
|
|
else:
|
|
purchase_line = counterpart_line
|
|
purchase_lot = counterpart_lot
|
|
sale_line = source_line
|
|
sale_lot = source_lot
|
|
|
|
purchase_lqt = TradeCustomDuplicate._open_lot_qt(
|
|
purchase_lot, 'purchase')
|
|
sale_lqt = TradeCustomDuplicate._open_lot_qt(sale_lot, 'sale')
|
|
if not purchase_lqt or not sale_lqt:
|
|
raise UserError(
|
|
"Duplicate matched pair could not find the new open lot "
|
|
"quantities to match.")
|
|
|
|
purchase_lqt.lot_s = sale_lot.id
|
|
purchase_lqt.lot_quantity = Decimal(str(quantity)).quantize(
|
|
Decimal("0.00001"))
|
|
LotQt.save([purchase_lqt])
|
|
LotQt.delete([sale_lqt])
|
|
Lot.assert_lines_quantity_consistency([purchase_line, sale_line])
|
|
|
|
@staticmethod
|
|
def _duplicate_matched_pair(record, line, new_line, options):
|
|
matched_line, matched_record = TradeCustomDuplicate._matched_counterpart(
|
|
line, record=record)
|
|
if not matched_record:
|
|
raise UserError(
|
|
"Duplicate matched pair could not find the matched contract.")
|
|
|
|
mirror_options = TradeCustomDuplicate._pair_duplicate_options(
|
|
matched_record, matched_line, options)
|
|
new_matched_record = TradeCustomDuplicate._copy_contract(
|
|
matched_record, mirror_options)
|
|
new_matched_line = TradeCustomDuplicate._apply_line_options(
|
|
new_matched_record, mirror_options)
|
|
source_lot = TradeCustomDuplicate._ensure_open_virtual_lot(new_line)
|
|
counterpart_lot = TradeCustomDuplicate._ensure_open_virtual_lot(
|
|
new_matched_line)
|
|
if source_lot and counterpart_lot:
|
|
TradeCustomDuplicate._match_duplicate_pair_lots(
|
|
new_line, source_lot, new_matched_line, counterpart_lot,
|
|
mirror_options.quantity)
|
|
return new_matched_record
|
|
|
|
@staticmethod
|
|
def _create_matched_mirror(record, line, options):
|
|
mirror_options = TradeCustomDuplicate._mirror_options(
|
|
options, fallback_record=record, fallback_line=line)
|
|
source_model = record.__name__
|
|
mirror_type = TradeCustomDuplicate._opposite_type(source_model)
|
|
source_lot = TradeCustomDuplicate._first_virtual_lot(line)
|
|
if not source_lot:
|
|
raise UserError(
|
|
"The duplicated line has no virtual lot to match.")
|
|
if not mirror_options.unit:
|
|
raise UserError(
|
|
"The duplicated line has no unit for mirror contract creation.")
|
|
|
|
ContractDetail = Pool().get('contract.detail')
|
|
ContractsStart = Pool().get('contracts.start')
|
|
|
|
from_location, to_location = ContractFactory._get_mirror_locations(
|
|
record, mirror_type)
|
|
|
|
detail = ContractDetail()
|
|
detail.party = mirror_options.party
|
|
detail.currency = mirror_options.currency
|
|
detail.currency_unit = "%s_%s" % (
|
|
mirror_options.currency.id, mirror_options.unit.id)
|
|
detail.incoterm = mirror_options.incoterm
|
|
detail.payment_term = mirror_options.payment_term
|
|
detail.quantity = mirror_options.quantity
|
|
detail.unit = mirror_options.unit
|
|
detail.price = mirror_options.unit_price
|
|
detail.price_type = mirror_options.price_type
|
|
detail.reference = getattr(record, 'reference', None)
|
|
detail.from_location = from_location
|
|
detail.to_location = to_location
|
|
detail.tol_min = getattr(record, 'tol_min', None)
|
|
detail.tol_max = getattr(record, 'tol_max', None)
|
|
detail.crop = getattr(record, 'crop', None)
|
|
detail.del_period = getattr(record, 'del_period', None)
|
|
|
|
ct = ContractsStart()
|
|
ct.type = mirror_type
|
|
ct.matched = True
|
|
ct.product = line.product
|
|
ct.unit = mirror_options.unit
|
|
ct.quantity = mirror_options.quantity
|
|
ct.lot = source_lot
|
|
|
|
with Transaction().set_context(active_ids=[]):
|
|
ContractFactory.create_contracts([detail], type_=mirror_type, ct=ct)
|
|
|
|
def transition_duplicate(self):
|
|
record = self._source_record(self.records[0])
|
|
options = self.start
|
|
new_record = self._copy_contract(record, options)
|
|
new_line = self._apply_line_options(new_record, options)
|
|
TradeCustomDuplicate._ensure_open_virtual_lot(new_line)
|
|
self.start.duplicated_record_id = new_record.id
|
|
if options.create_matched_mirror:
|
|
if options.mirror_mode == 'duplicate_pair':
|
|
self._duplicate_matched_pair(record, self._trade_line(record),
|
|
new_line, options)
|
|
else:
|
|
self._create_matched_mirror(new_record, new_line, options)
|
|
if new_record.__name__ == 'purchase.purchase':
|
|
return 'open_purchase'
|
|
return 'open_sale'
|
|
|
|
def _open_duplicated_record(self, action):
|
|
action['views'].reverse()
|
|
return action, {'res_id': self.start.duplicated_record_id}
|
|
|
|
def do_open_purchase(self, action):
|
|
return self._open_duplicated_record(action)
|
|
|
|
def do_open_sale(self, action):
|
|
return self._open_duplicated_record(action)
|
|
|
|
def end(self):
|
|
return 'reload'
|