228 lines
8.6 KiB
Python
228 lines
8.6 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 trytond.exceptions import UserError
|
|
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, StateTransition, StateView, Wizard
|
|
|
|
from trytond.modules.purchase_trade.service import ContractFactory
|
|
|
|
|
|
PRICE_TYPES = [
|
|
('cash', 'Cash Price'),
|
|
('priced', 'Priced'),
|
|
('basis', 'Basis'),
|
|
('efp', 'EFP'),
|
|
]
|
|
|
|
|
|
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)
|
|
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_party = fields.Many2One('party.party', "Mirror Counterparty",
|
|
states={
|
|
'invisible': ~Bool(Eval('create_matched_mirror')),
|
|
'required': Bool(Eval('create_matched_mirror')),
|
|
})
|
|
|
|
|
|
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()
|
|
|
|
@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'
|
|
|
|
def default_start(self, fields):
|
|
if len(self.records) != 1:
|
|
raise UserError("Please select a single purchase or sale.")
|
|
|
|
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.")
|
|
|
|
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,
|
|
}
|
|
|
|
@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 _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, '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 _create_matched_mirror(record, line, options):
|
|
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 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 = options.mirror_party
|
|
detail.currency = options.currency
|
|
detail.currency_unit = "%s_%s" % (options.currency.id, options.unit.id)
|
|
detail.incoterm = options.incoterm
|
|
detail.payment_term = options.payment_term
|
|
detail.quantity = options.quantity
|
|
detail.unit = options.unit
|
|
detail.price = options.unit_price
|
|
detail.price_type = 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 = line.unit
|
|
ct.quantity = 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.records[0]
|
|
options = self.start
|
|
new_record = self._copy_contract(record, options)
|
|
new_line = self._apply_line_options(new_record, options)
|
|
if options.create_matched_mirror:
|
|
self._create_matched_mirror(new_record, new_line, options)
|
|
return 'end'
|
|
|
|
def end(self):
|
|
return 'reload'
|