# 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 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) 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')), }) 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() 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' 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.") matched_record = self._matched_counterpart_record(line, strict=False) mirror_mode = 'duplicate_pair' if matched_record else 'new_mirror' 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, 'mirror_mode': mirror_mode, 'mirror_party': ( matched_record.party.id if matched_record and matched_record.party else None), } @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, '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): 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)] 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)] 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 len(unique) == 1: return unique[0] if strict: raise UserError( "Duplicate matched pair requires exactly one matched " "counterpart.") return None @staticmethod def _matched_counterpart_record(line, strict=True): matched_line = TradeCustomDuplicate._matched_counterpart_line( line, strict=strict) if not matched_line: return None return ( getattr(matched_line, 'sale', None) or getattr(matched_line, 'purchase', None)) @staticmethod def _pair_duplicate_options(matched_record, matched_line, options): return SimpleNamespace( party=options.mirror_party or matched_record.party, currency=matched_record.currency or options.currency, payment_term=( matched_record.payment_term or options.payment_term), incoterm=matched_record.incoterm or options.incoterm, quantity=options.quantity, unit_price=getattr(matched_line, 'unit_price', None), price_type=( getattr(matched_line, 'price_type', None) or options.price_type), clear_fees=options.clear_fees, clear_pricing_components=options.clear_pricing_components, ) @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 = TradeCustomDuplicate._matched_counterpart_line(line) matched_record = ( getattr(matched_line, 'sale', None) or getattr(matched_line, 'purchase', None)) 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, options.quantity) return new_matched_record @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) 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'