# 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'