from functools import wraps from trytond.model import fields from trytond.report import Report from trytond.pool import Pool, PoolMeta from trytond.pyson import Bool, Eval, Id, If from trytond.model import (ModelSQL, ModelView) from trytond.tools import is_full_text, lstrip_wildcard from trytond.transaction import Transaction, inactive_records from decimal import getcontext, Decimal, ROUND_UP, ROUND_HALF_UP from sql.aggregate import Count, Max, Min, Sum, Avg, BoolOr from sql.conditionals import Case from sql import Column, Literal from sql.functions import CurrentTimestamp, DateTrunc from trytond.wizard import Button, StateTransition, StateView, Wizard from itertools import chain, groupby from operator import itemgetter import datetime import logging from collections import defaultdict from trytond.exceptions import UserWarning, UserError from trytond.modules.account.exceptions import PeriodNotFoundError logger = logging.getLogger(__name__) def filter_state(state): def filter(func): @wraps(func) def wrapper(cls, fees): fees = [f for f in fees if f.state == state] return func(cls, fees) return wrapper return filter class Fee(ModelSQL,ModelView): "Fee" __name__ = 'fee.fee' line = fields.Many2One('purchase.line',"Line") source_rule = fields.Many2One('fee.rule', "Source Rule", readonly=True) generated_by_rule = fields.Boolean("Generated by Rule", readonly=True) shipment_in = fields.Many2One('stock.shipment.in') shipment_out = fields.Many2One('stock.shipment.out') shipment_internal = fields.Many2One('stock.shipment.internal') currency = fields.Many2One('currency.currency',"Currency", required=True) supplier = fields.Many2One('party.party',"Supplier", required=True) type = fields.Selection([ ('budgeted', 'Budgeted'), ('ordered', 'Ordered'), ('actual', 'Actual'), ], "Type", required=True) p_r = fields.Selection([ ('pay', 'PAY'), ('rec', 'REC'), ], "P/R", required=True) product = fields.Many2One('product.product',"Product", required=True, domain=[('type', '=', 'service')]) price = fields.Numeric("Price",digits=(1,4)) mode = fields.Selection([ ('lumpsum', 'Lump sum'), ('perqt', 'Per qt'), ('pprice', '% price'), ('rate', '% rate'), ('pcost', '% cost price'), ('ppack', 'Per packing'), ], 'Mode', required=True) auto_calculation = fields.Boolean("Auto",states={'readonly': (Eval('mode') != 'ppack')}) inherit_qt = fields.Boolean("Inh Qt",states={'readonly': Eval('mode') != 'ppack'}) quantity = fields.Numeric("Qt",digits='unit',states={'readonly': (Eval('mode') != 'ppack') | Bool(Eval('auto_calculation'))}) unit = fields.Many2One('product.uom',"Unit",domain=[ If(Eval('mode') == 'ppack', ('category', '=', Eval('packing_category')), ()), ], states={ 'readonly': (Bool(Eval('mode') != 'ppack') & Bool(Eval('mode') != 'perqt')), }, depends=['mode', 'packing_category']) packing_category = fields.Function(fields.Many2One('product.uom.category',"Packing Category"),'on_change_with_packing_category') inherit_shipment = fields.Boolean("Inh Sh",states={ 'invisible': (Eval('shipment_in')), }) purchase = fields.Many2One('purchase.purchase',"Purchase", ondelete='CASCADE') qt_state = fields.Many2One('lot.qt.type',"Qt State") amount = fields.Function(fields.Numeric("Amount", digits='currency'),'get_amount') fee_lots = fields.Function(fields.Many2Many('lot.lot', None, None, "Lots"),'get_lots')#, searcher='search_lots') lots = fields.Many2Many('fee.lots', 'fee', 'lot',"Lots",domain=[('id', 'in', Eval('fee_lots',-1))] ) lots_cp = fields.Integer("Lots number") state = fields.Selection([ ('not invoiced', 'Not invoiced'), ('invoiced', 'Invoiced'), ], string='State', readonly=True) fee_landed_cost = fields.Function(fields.Boolean("Inventory"),'get_landed_status') inv = fields.Function(fields.Many2One('account.invoice',"Invoice"),'get_invoice') weight_type = fields.Selection([ ('net', 'Net'), ('brut', 'Gross'), ], string='W. type') fee_date = fields.Date("Date") @classmethod def default_fee_date(cls): Date = Pool().get('ir.date') return Date.today() @classmethod def default_generated_by_rule(cls): return False @classmethod def default_qt_state(cls): return None @fields.depends('mode','unit') def on_change_with_packing_category(self, name=None): UnitCategory = Pool().get('product.uom.category') packing = UnitCategory.search(['name','=','Packing']) if packing: return packing[0] @fields.depends('line','sale_line','shipment_in','lots','price','unit','auto_calculation','mode','_parent_line.unit','_parent_line.lots','_parent_sale_line.unit','_parent_sale_line.lots','_parent_shipment_in.id') def on_change_with_quantity(self, name=None): qt = None unit = None line = self.line logger.info("ON_CHANGE_WITH_LINE:%s",line) if not line: line = self.sale_line if line: if line.lots: qt = sum([e.get_current_quantity_converted(0,self.unit) for e in line.lots]) qt_ = sum([e.get_current_quantity_converted(0) for e in line.lots]) unit = line.lots[0].lot_unit or line.lots[0].lot_unit_line logger.info("ON_CHANGE_WITH_QT0:%s",qt) logger.info("ON_CHANGE_WITH_SI:%s",self.shipment_in) if self.shipment_in: Lot = Pool().get('lot.lot') lots = Lot.search([('lot_shipment_in','=',self.shipment_in.id)]) logger.info("ON_CHANGE_WITH_LOTS:%s",lots) if lots: qt = sum([e.get_current_quantity_converted(0,self.unit) for e in lots]) qt_ = sum([e.get_current_quantity_converted(0) for e in lots]) unit = lots[0].lot_unit or lots[0].lot_unit_line if not qt: logger.info("ON_CHANGE_WITH_QT1:%s",qt) LotQt = Pool().get('lot.qt') if self.shipment_in: lqts = LotQt.search(['lot_shipment_in','=',self.shipment_in.id]) if lqts: qt = Decimal(lqts[0].lot_quantity) qt_ = qt unit = lqts[0].lot_unit logger.info("ON_CHANGE_WITH_QT2:%s",qt) if self.mode != 'ppack': return qt else: if self.auto_calculation: logger.info("AUTOCALCULATION:%s",qt) logger.info("AUTOCALCULATION2:%s",qt_) logger.info("AUTOCALCULATION3:%s",Decimal(unit.factor)) logger.info("AUTOCALCULATION4:%s",Decimal(self.unit.factor)) return (qt_ * Decimal(unit.factor) / Decimal(self.unit.factor)).to_integral_value(rounding=ROUND_UP) @fields.depends('price','mode','_parent_line.lots','_parent_sale_line.lots','shipment_in') def on_change_with_unit(self, name=None): if self.mode != 'ppack' and self.mode != 'perqt': line = self.line if not line: line = self.sale_line if line: if line.lots: if len(line.lots) == 1: return line.lots[0].lot_unit_line else: return line.lots[1].lot_unit_line if self.shipment_in: Lot = Pool().get('lot.lot') lots = Lot.search([('lot_shipment_in','=',self.shipment_in.id)]) logger.info("ON_CHANGE_WITH_UNIT:%s",lots) if lots: return lots[0].lot_unit_line else: return self.unit def get_lots(self, name): logger.info("GET_LOTS_LINE:%s",self.line) logger.info("GET_LOTS_SHIPMENT_IN:%s",self.shipment_in) Lot = Pool().get('lot.lot') if self.line: return self.line.lots if self.shipment_in: lots = Lot.search([('lot_shipment_in','=',self.shipment_in.id)]) logger.info("LOTSDOMAIN:%s",lots) if lots: return lots + [lots[0].getVlot_p()] if self.shipment_internal: return Lot.search([('lot_shipment_internal','=',self.shipment_internal.id)]) if self.shipment_out: return Lot.search([('lot_shipment_out','=',self.shipment_out.id)]) return Lot.search(['id','>',0]) def get_cog(self,lot): MoveLine = Pool().get('account.move.line') Currency = Pool().get('currency.currency') Date = Pool().get('ir.date') AccountConfiguration = Pool().get('account.configuration') account_configuration = AccountConfiguration(1) Uom = Pool().get('product.uom') ml = MoveLine.search([ ('lot', '=', lot.id), ('fee', '=', self.id), ('account', '=', self.product.account_stock_in_used.id), ('origin', 'ilike', '%stock.move%'), ]) logger.info("GET_COG_FEE:%s",ml) if ml: return round(Decimal(sum([e.credit-e.debit for e in ml if e.description != 'Delivery fee'])),2) return Decimal(0) def get_non_cog(self,lot): MoveLine = Pool().get('account.move.line') Currency = Pool().get('currency.currency') Date = Pool().get('ir.date') AccountConfiguration = Pool().get('account.configuration') account_configuration = AccountConfiguration(1) Uom = Pool().get('product.uom') ml = MoveLine.search([ ('lot', '=', lot.id), ('fee', '=', self.id), ('account', '=', self.product.account_stock_in_used.id), ]) logger.info("GET_NON_COG_FEE:%s",ml) if ml: return round(Decimal(sum([e.credit-e.debit for e in ml])),2) return Decimal(0) @classmethod def __setup__(cls): super().__setup__() cls._buttons.update({ 'invoice': { 'invisible': (Eval('state') == 'invoiced'), 'depends': ['state'], }, }) @classmethod def default_state(cls): return 'not invoiced' @classmethod def default_p_r(cls): return 'pay' def get_unit(self, name=None): FeeLots = Pool().get('fee.lots') fl = FeeLots.search(['fee','=',self.id]) if fl: if fl[0].lot.line: return fl[0].lot.line.unit if fl[0].lot.sale_line: return fl[0].lot.sale_line.unit @classmethod @ModelView.button @filter_state('not invoiced') def invoice(cls, fees): Purchase = Pool().get('purchase.purchase') FeeLots = Pool().get('fee.lots') for fee in fees: if fee.purchase: fl = FeeLots.search([('fee','=',fee.id)]) logger.info("PROCESS_FROM_FEE:%s",fl) Purchase._process_invoice([fee.purchase],[e.lot for e in fl],'service') cls.write(fees, {'state': 'invoiced',}) @classmethod def default_type(cls): return 'budgeted' @classmethod def default_weight_type(cls): return 'brut' def get_price_per_qt(self): price = Decimal(0) if self.mode == 'lumpsum': if self.quantity: if self.quantity > 0: return round(self.price / self.quantity,4) elif self.mode == 'perqt': return self.price elif self.mode == 'ppack': unit = self.get_unit() if unit and self.unit: return round(self.price / Decimal(self.unit.factor) * Decimal(unit.factor),4) elif self.mode == 'pprice' or self.mode == 'pcost': if self.line and self.price: return round(self.price * Decimal(self.line.unit_price) / 100,4) if self.sale_line and self.price: return round(self.price * Decimal(self.sale_line.unit_price) / 100,4) if self.shipment_in: StockMove = Pool().get('stock.move') sm = StockMove.search(['shipment','=','stock.shipment.in,'+str(self.shipment_in.id)]) if sm: if sm[0].lot: return round(self.price * Decimal(sm[0].lot.get_lot_price()) / 100,4) return price def get_invoice(self,name): if self.purchase: if self.purchase.invoices: return self.purchase.invoices[0] def get_landed_status(self,name): if self.product: return self.product.template.landed_cost def get_quantity(self,name=None): qt = self.get_fee_lots_qt() if qt: return qt line = self.line or self.sale_line if line and line.lots: return sum( Decimal(lot.get_current_quantity_converted(0, self.unit)) for lot in line.lots) LotQt = Pool().get('lot.qt') if self.shipment_in: lqts = LotQt.search(['lot_shipment_in','=',self.shipment_in.id]) if lqts: return Decimal(lqts[0].lot_quantity) def _get_effective_fee_lots(self): FeeLots = Pool().get('fee.lots') fee_lots = FeeLots.search([('fee', '=', self.id)]) lots = [fee_lot.lot for fee_lot in fee_lots if fee_lot.lot] physical_lots = [ lot for lot in lots if getattr(lot, 'lot_type', None) == 'physic' ] if physical_lots: return physical_lots return [ lot for lot in lots if getattr(lot, 'lot_type', None) == 'virtual' ] def is_effective_fee_lot(self, lot): return bool(lot and lot in self._get_effective_fee_lots()) def _get_weight_basis_qt_type(self, lot=None): if self.line and self.line.purchase: return getattr(self.line.purchase.wb, 'qt_type', None) if self.sale_line and self.sale_line.sale: return getattr(self.sale_line.sale.wb, 'qt_type', None) if lot and lot.line and lot.line.purchase: return getattr(lot.line.purchase.wb, 'qt_type', None) if lot and lot.sale_line and lot.sale_line.sale: return getattr(lot.sale_line.sale.wb, 'qt_type', None) @staticmethod def _qt_type_sequence(qt_type): sequence = getattr(qt_type, 'sequence', None) if sequence is None: return None return Decimal(str(sequence)) @staticmethod def _same_qt_type(left, right): return ( getattr(left, 'id', left) and getattr(left, 'id', left) == getattr(right, 'id', right)) def _target_qt_type_for_lot(self, lot): weight_basis_qt_type = self._get_weight_basis_qt_type(lot) target_qt_type = self.qt_state or weight_basis_qt_type if not target_qt_type or not weight_basis_qt_type: return target_qt_type target_sequence = self._qt_type_sequence(target_qt_type) weight_basis_sequence = self._qt_type_sequence(weight_basis_qt_type) if ( target_sequence is not None and weight_basis_sequence is not None and target_sequence > weight_basis_sequence): return weight_basis_qt_type return target_qt_type def _select_lot_qt_type(self, lot): target_qt_type = self._target_qt_type_for_lot(lot) if not target_qt_type: return None lot_hist = list(getattr(lot, 'lot_hist', []) or []) for hist in lot_hist: if self._same_qt_type(hist.quantity_type, target_qt_type): return hist.quantity_type target_sequence = self._qt_type_sequence(target_qt_type) if target_sequence is None: return None previous = [] for hist in lot_hist: hist_sequence = self._qt_type_sequence(hist.quantity_type) if hist_sequence is not None and hist_sequence <= target_sequence: previous.append((hist_sequence, hist.quantity_type)) if previous: previous.sort(key=lambda item: item[0]) return previous[-1][1] return None def _get_lot_fee_quantity(self, lot): state = self._select_lot_qt_type(lot) target_qt_type = self._target_qt_type_for_lot(lot) if target_qt_type and not state: return None state_id = getattr(state, 'id', 0) if state else 0 if self.weight_type == 'brut': return Decimal( lot.get_current_gross_quantity_converted(state_id, self.unit) or 0) return Decimal( lot.get_current_quantity_converted(state_id, self.unit) or 0) def _get_effective_fee_lots_quantity(self): lots = self._get_effective_fee_lots() if not lots: return None if self.mode == 'ppack': packing_qty = sum( Decimal(str(getattr(lot, 'lot_qt', 0) or 0)) for lot in lots) if packing_qty: return packing_qty return self.quantity quantities = [self._get_lot_fee_quantity(lot) for lot in lots] if any(quantity is None for quantity in quantities): return None return sum(quantities) def sync_quantity_from_lots(self): if self.mode == 'ppack' and self.auto_calculation: self.assert_quantity_consistency() return quantity = self._get_effective_fee_lots_quantity() if quantity is None: self.assert_quantity_consistency() return quantity = round(Decimal(quantity), 5) if self.mode == 'ppack': quantity = quantity.to_integral_value(rounding=ROUND_UP) if Decimal(self.quantity or 0) != quantity: self.quantity = quantity with Transaction().set_context(_purchase_trade_skip_fee_sync=True): self.__class__.save([self]) self.assert_quantity_consistency() def assert_quantity_consistency(self): if self.mode == 'ppack': return expected = self._get_effective_fee_lots_quantity() if expected is None: if self._get_effective_fee_lots(): raise UserError( "Fee quantity consistency error on fee %s: no usable " "quantity state found on linked lots" % ( getattr(self, 'id', None),)) return expected = round(Decimal(expected), 5) observed = round(Decimal(self.quantity or 0), 5) diff = observed - expected if diff: raise UserError( "Fee quantity consistency error on fee %s: fee quantity " "must equal linked lots quantity (observed=%s, expected=%s, " "diff=%s)" % ( getattr(self, 'id', None), observed, expected, diff)) @classmethod def assert_quantities_consistency(cls, fees): for fee in fees: fee.assert_quantity_consistency() def _get_amount_quantity(self): quantity = self.quantity if quantity is None: quantity = self.get_quantity() return Decimal(quantity or 0) @staticmethod def _unsigned_amount(amount): if amount is None: return None return abs(Decimal(amount)) def get_amount(self,name=None): sign = Decimal(1) if self.price: # if self.p_r: # if self.p_r == 'pay': # sign = -1 if self.mode == 'lumpsum': return self._unsigned_amount(self.price * sign) elif self.mode == 'ppack': return self._unsigned_amount( round(self.price * self.quantity,2)) elif self.mode == 'rate': if self.line: if self.line.estimated_date: est_lines = [dd for dd in self.line.estimated_date if dd.trigger == 'bldate'] est_line = est_lines[0] if est_lines else None if est_line and est_line.fin_int_delta: factor = ( Decimal(self.price) / Decimal(100) * Decimal(est_line.fin_int_delta) / Decimal(360) ) return self._unsigned_amount(round( factor * self.line.unit_price * self._get_amount_quantity() * sign,2)) if self.sale_line: if self.sale_line.estimated_date: est_lines = [ dd for dd in self.sale_line.estimated_date if dd.trigger == 'bldate' ] est_line = est_lines[0] if est_lines else None if est_line and est_line.fin_int_delta: factor = ( Decimal(self.price) / Decimal(100) * Decimal(est_line.fin_int_delta) / Decimal(360) ) return self._unsigned_amount(round( factor * self.sale_line.unit_price * self._get_amount_quantity() * sign, 2)) elif self.mode == 'perqt': if self.shipment_in: StockMove = Pool().get('stock.move') sm = StockMove.search(['shipment','=','stock.shipment.in,'+str(self.shipment_in.id)]) if sm: unique_lots = {e.lot for e in sm if e.lot} return self._unsigned_amount(round(self.price * Decimal(sum([e.get_current_quantity_converted(0,self.unit) for e in unique_lots])) * sign,2)) LotQt = Pool().get('lot.qt') lqts = LotQt.search(['lot_shipment_in','=',self.shipment_in.id]) if lqts: return self._unsigned_amount(round(self.price * Decimal(lqts[0].lot_quantity) * sign,2)) return self._unsigned_amount(round((self.quantity if self.quantity else 0) * self.price * sign,2)) elif self.mode == 'pprice': if self.line: return self._unsigned_amount(round(self.price / 100 * self.line.unit_price * (self.quantity if self.quantity else 0) * sign,2)) if self.sale_line: return self._unsigned_amount(round(self.price / 100 * self.sale_line.unit_price * (self.quantity if self.quantity else 0) * sign,2)) if self.shipment_in: StockMove = Pool().get('stock.move') sm = StockMove.search(['shipment','=','stock.shipment.in,'+str(self.shipment_in.id)]) if sm: if sm[0].lot: return self._unsigned_amount(round(self.price * Decimal(sum([e.lot.get_lot_price() for e in sm if e.lot])) / 100 * self.quantity * sign,2)) LotQt = Pool().get('lot.qt') lqts = LotQt.search(['lot_shipment_in','=',self.shipment_in.id]) if lqts: return self._unsigned_amount(round(self.price * Decimal(lqts[0].lot_p.get_lot_price()) / 100 * lqts[0].lot_quantity * sign,2)) @classmethod def copy(cls, fees, default=None): if default is None: default = {} else: default = default.copy() # Important : on vide le champ 'lots' default.setdefault('lots', []) return super().copy(fees, default=default) def get_fee_lots_qt(self,state_id=0): qt = Decimal(0) lots = self._get_effective_fee_lots() if lots: if state_id: qt = sum([ e.get_current_quantity_converted(state_id,self.unit) for e in lots]) else: qt = self._get_effective_fee_lots_quantity() or Decimal(0) logger.info("GET_FEE_LOTS_QT:%s",qt) return qt def adjust_purchase_values(self): Purchase = Pool().get('purchase.purchase') PurchaseLine = Pool().get('purchase.line') logger.info("ADJUST_PURCHASE_VALUES:%s",self) if self.type == 'ordered' and self.state == 'not invoiced' and self.purchase: logger.info("ADJUST_PURCHASE_VALUES_QT:%s",self.purchase.lines[0].quantity) if self.mode == 'lumpsum': if self.amount != self.purchase.lines[0].unit_price: self.purchase.lines[0].unit_price = self.amount elif self.mode == 'ppack': if self.amount != self.purchase.lines[0].amount: self.purchase.lines[0].unit_price = self.price self.purchase.lines[0].quantity = self.quantity else: if self.get_price_per_qt() != self.purchase.lines[0].unit_price: self.purchase.lines[0].unit_price = self.get_price_per_qt() if self.quantity != self.purchase.lines[0].quantity: self.purchase.lines[0].quantity = self.quantity if self.product != self.purchase.lines[0].product: self.purchase.lines[0].product = self.product PurchaseLine.save([self.purchase.lines[0]]) if self.supplier != self.purchase.party: self.purchase.party = self.supplier if self.currency != self.purchase.currency: self.purchase.currency = self.currency Purchase.save([self.purchase]) @classmethod def _add_lot_pnl_lines(cls, lot, purchase_lines, sale_lines): if not lot: return if getattr(lot, 'line', None): purchase_lines[lot.line.id] = lot.line if getattr(lot, 'sale_line', None): sale_lines[lot.sale_line.id] = lot.sale_line @classmethod def _collect_pnl_lines(cls, fees=None, lots=None): purchase_lines = {} sale_lines = {} for lot in lots or []: cls._add_lot_pnl_lines(lot, purchase_lines, sale_lines) for fee in fees or []: if getattr(fee, 'line', None): purchase_lines[fee.line.id] = fee.line if getattr(fee, 'sale_line', None): sale_lines[fee.sale_line.id] = fee.sale_line for lot in getattr(fee, 'lots', None) or []: cls._add_lot_pnl_lines(lot, purchase_lines, sale_lines) shipment = getattr(fee, 'shipment_in', None) if not shipment: continue for move in getattr(shipment, 'incoming_moves', None) or []: cls._add_lot_pnl_lines( getattr(move, 'lot', None), purchase_lines, sale_lines) for lqt in getattr(shipment, 'lotqt', None) or []: cls._add_lot_pnl_lines( getattr(lqt, 'lot_p', None), purchase_lines, sale_lines) cls._add_lot_pnl_lines( getattr(lqt, 'lot_s', None), purchase_lines, sale_lines) return purchase_lines.values(), sale_lines.values() @classmethod def _regenerate_fee_pnl(cls, fees=None, lots=None): purchase_lines, sale_lines = cls._collect_pnl_lines(fees, lots) cls._regenerate_fee_pnl_for_lines(purchase_lines, sale_lines) @classmethod def _regenerate_fee_pnl_for_lines(cls, purchase_lines, sale_lines): purchase_lines = cls._unique_records(purchase_lines) sale_lines = cls._unique_records(sale_lines) if not purchase_lines and not sale_lines: return Valuation = Pool().get('valuation.valuation') for line in purchase_lines: Valuation.generate(line, valuation_type='fees') for sale_line in sale_lines: Valuation.generate_from_sale_line( sale_line, valuation_type='fees') @staticmethod def _unique_records(records): unique = {} for record in records: if not record: continue unique[getattr(record, 'id', id(record))] = record return list(unique.values()) # @classmethod # def validate(cls, fees): # super(Fee, cls).validate(fees) @classmethod def _save_fee_accrual_or_warn(cls, account_move, fee, lot): AccountMove = Pool().get('account.move') if account_move: AccountMove.save([account_move]) return Warning = Pool().get('res.user.warning') warning_name = Warning.format( "Fee accrual not generated", [fee.product, lot]) if Warning.check(warning_name): raise UserWarning( warning_name, "The fee can be saved, but the accrual entry could not be " "generated because the stock accounting configuration or " "accounting period is missing or incomplete. Do you want to " "continue without creating the accrual entry?") @classmethod def create(cls, vlist): vlist = [x.copy() for x in vlist] fees = super(Fee, cls).create(vlist) qt_sh = Decimal(0) qt_line = Decimal(0) unit = None for fee in fees: FeeLots = Pool().get('fee.lots') Lots = Pool().get('lot.lot') LotQt = Pool().get('lot.qt') if fee.line: for l in fee.line.lots: if (l.lot_type == 'virtual' and len(fee.line.lots)==1) or (l.lot_type == 'physic' and len(fee.line.lots)>1): fl = FeeLots() fl.fee = fee.id fl.lot = l.id fl.line = l.line.id FeeLots.save([fl]) qt_line += l.get_current_quantity_converted() unit = l.line.unit if fee.sale_line: for l in fee.sale_line.lots: if (l.lot_type == 'virtual' and len(fee.sale_line.lots)==1) or (l.lot_type == 'physic' and len(fee.sale_line.lots)>1): fl = FeeLots() fl.fee = fee.id fl.lot = l.id fl.sale_line = l.sale_line.id FeeLots.save([fl]) qt_line += l.get_current_quantity_converted() unit = l.sale_line.unit if fee.shipment_in: if fee.shipment_in.state == 'draft'or fee.shipment_in.state == 'started': lots = Lots.search(['lot_shipment_in','=',fee.shipment_in.id]) if lots: for l in lots: fl = FeeLots() fl.fee = fee.id fl.lot = l.id FeeLots.save([fl]) qt_sh += l.get_current_quantity_converted() unit = l.line.unit else: lqts = LotQt.search(['lot_shipment_in','=',fee.shipment_in.id]) if lqts: for l in lqts: if l.lot_p: exist = FeeLots.search([ ('fee','=',fee.id), ('lot','=',l.lot_p.id), ]) if not exist: fl = FeeLots() fl.fee = fee.id fl.lot = l.lot_p.id FeeLots.save([fl]) qt_sh += l.lot_p.get_current_quantity_converted() unit = l.lot_p.line.unit else: raise UserError("You cannot add fee on received shipment!") type = fee.type if type == 'ordered': Purchase = Pool().get('purchase.purchase') PurchaseLine = Pool().get('purchase.line') pl = PurchaseLine() pl.product = fee.product if fee.line or fee.sale_line: pl.quantity = round(qt_line,5) if fee.shipment_in: pl.quantity = round(qt_sh,5) logger.info("CREATE_PURHCASE_FOR_FEE_QT:%s",pl.quantity) pl.unit = unit pl.fee_ = fee.id if fee.price: fee_price = fee.get_price_per_qt() logger.info("GET_FEE_PRICE_PER_QT:%s",fee_price) pl.unit_price = round(Decimal(fee_price),4) if fee.mode == 'lumpsum': pl.quantity = 1 pl.unit_price = round(Decimal(fee.amount),4) elif fee.mode == 'ppack': pl.unit_price = fee.price p = Purchase() p.lines = [pl] p.party = fee.supplier if p.party.addresses: p.invoice_address = p.party.addresses[0] p.currency = fee.currency p.line_type = 'service' p.from_location = fee.shipment_in.from_location if fee.shipment_in else (fee.line.purchase.from_location if fee.line else fee.sale_line.sale.from_location) p.to_location = fee.shipment_in.to_location if fee.shipment_in else (fee.line.purchase.to_location if fee.line else fee.sale_line.sale.to_location) if fee.shipment_in and fee.shipment_in.lotqt: p.payment_term = fee.shipment_in.lotqt[0].lot_p.line.purchase.payment_term elif fee.line: p.payment_term = fee.line.purchase.payment_term elif fee.sale_line: p.payment_term = fee.sale_line.sale.payment_term Purchase.save([p]) #if reception of moves done we need to generate accrual for fee if not fee.sale_line: feelots = FeeLots.search(['fee','=',fee.id]) for fl in feelots: if fee.product.template.landed_cost: move = fl.lot.get_received_move() if move: Warning = Pool().get('res.user.warning') warning_name = Warning.format("Lot ever received", []) if Warning.check(warning_name): raise UserWarning(warning_name, "By clicking yes, an accrual for this fee will be created") account_move = move._get_account_stock_move_fee(fee) cls._save_fee_accrual_or_warn( account_move, fee, fl.lot) else: account_move = fee._get_account_move_fee(fl.lot) cls._save_fee_accrual_or_warn( account_move, fee, fl.lot) for fee in fees: fee.sync_quantity_from_lots() cls.assert_quantities_consistency(fees) cls._regenerate_fee_pnl(fees=fees) return fees @classmethod def write(cls, *args): old_purchase_lines, old_sale_lines = cls._collect_pnl_lines( fees=sum(args[::2], [])) old_purchase_lines = list(old_purchase_lines) old_sale_lines = list(old_sale_lines) super().write(*args) if Transaction().context.get('_purchase_trade_skip_fee_sync'): return fees = sum(args[::2], []) for fee in fees: fee.sync_quantity_from_lots() fee.adjust_purchase_values() cls.assert_quantities_consistency(fees) purchase_lines, sale_lines = cls._collect_pnl_lines(fees=fees) cls._regenerate_fee_pnl_for_lines( old_purchase_lines + list(purchase_lines), old_sale_lines + list(sale_lines)) @classmethod def delete(cls, fees): purchase_lines, sale_lines = cls._collect_pnl_lines(fees=fees) purchases = cls._get_generated_purchases_to_delete(fees) super().delete(fees) if purchases: Purchase = Pool().get('purchase.purchase') Purchase.delete(purchases) cls._regenerate_fee_pnl_for_lines(purchase_lines, sale_lines) @classmethod def _get_generated_purchases_to_delete(cls, fees): purchases = [] seen = set() for fee in fees: purchase = getattr(fee, 'purchase', None) if not cls._is_generated_purchase_for_fee(fee, purchase): continue purchase_id = getattr(purchase, 'id', id(purchase)) if purchase_id in seen: continue seen.add(purchase_id) purchases.append(purchase) return purchases @staticmethod def _is_generated_purchase_for_fee(fee, purchase): if getattr(fee, 'type', None) != 'ordered' or not purchase: return False if getattr(purchase, 'line_type', 'service') != 'service': return False lines = list(getattr(purchase, 'lines', []) or []) if not lines: return False fee_id = getattr(fee, 'id', fee) for line in lines: line_fee = getattr(line, 'fee_', None) if getattr(line_fee, 'id', line_fee) != fee_id: return False return True def _get_account_move_fee(self,lot,in_out='in',amt = None): pool = Pool() AccountMove = pool.get('account.move') Date = pool.get('ir.date') Period = pool.get('account.period') AccountConfiguration = pool.get('account.configuration') if self.product.type != 'service': return today = Date.today() company = lot.line.purchase.company if lot.line else lot.sale_line.sale.company for date in [today]: try: period = Period.find(company, date=date, test_state=False) except PeriodNotFoundError: if date < today: return continue break else: return if period.state != 'open': date = today period = Period.find(company, date=date) AccountMoveLine = pool.get('account.move.line') Currency = pool.get('currency.currency') move_line = AccountMoveLine() move_line.lot = lot move_line.fee = self move_line.origin = None move_line_ = AccountMoveLine() move_line_.lot = lot move_line_.fee = self move_line_.origin = None amount = amt if amt else self.amount if self.currency != company.currency: with Transaction().set_context(date=today): amount_converted = amount amount = Currency.compute(self.currency, amount, company.currency) move_line.second_currency = self.currency if self.p_r == 'pay': move_line.debit = amount move_line.credit = Decimal(0) move_line.account = self.product.account_stock_used if in_out == 'in' else self.product.account_cogs_used if hasattr(move_line, 'second_currency') and move_line.second_currency: move_line.amount_second_currency = amount_converted move_line_.debit = Decimal(0) move_line_.credit = amount move_line_.account = self.product.account_stock_in_used if in_out == 'in' else self.product.account_stock_out_used if hasattr(move_line_, 'second_currency') and move_line_.second_currency: move_line_.amount_second_currency = -amount_converted else: move_line.debit = Decimal(0) move_line.credit = amount move_line.account = self.product.account_stock_used if in_out == 'in' else self.product.account_cogs_used if hasattr(move_line, 'second_currency') and move_line.second_currency: move_line.amount_second_currency = -amount_converted move_line_.debit = amount move_line_.credit = Decimal(0) move_line_.account = self.product.account_stock_in_used if in_out == 'in' else self.product.account_stock_out_used if hasattr(move_line_, 'second_currency') and move_line_.second_currency: move_line_.amount_second_currency = amount_converted logger.info("FEE_MOVELINES_1:%s",move_line) logger.info("FEE_MOVELINES_2:%s",move_line_) AccountJournal = Pool().get('account.journal') journal = AccountJournal.search(['type','=','expense']) if journal: journal = journal[0] description = None description = 'Fee' return AccountMove( journal=journal, period=period, date=date, origin=None, description=description, lines=[move_line,move_line_], ) class FeeRule(ModelSQL, ModelView): "Fee Rule" __name__ = 'fee.rule' name = fields.Char("Name", required=True) active = fields.Boolean("Active") sequence = fields.Integer("Sequence") apply_on = fields.Selection([ ('purchase_line', "Purchase Line"), ('sale_line', "Sale Line"), ], "Apply On", required=True) auto_apply = fields.Boolean("Auto Apply") party = fields.Many2One('party.party', "Counterparty") party_category = fields.Many2One( 'party.category', "Counterparty Category") line_product = fields.Many2One('product.product', "Line Product") incoterm_code = fields.Char("Incoterm Code") from_location = fields.Many2One('stock.location', "From Location") to_location = fields.Many2One('stock.location', "To Location") execution_plan = fields.Many2One('workflow.plan', "Execution Plan") fee_product = fields.Many2One( 'product.product', "Fee Product", required=True, domain=[('type', '=', 'service')]) supplier = fields.Many2One('party.party', "Supplier") p_r = fields.Selection([ ('pay', 'PAY'), ('rec', 'REC'), ], "P/R", required=True) mode = fields.Selection([ ('lumpsum', 'Lump sum'), ('perqt', 'Per qt'), ('pprice', '% price'), ('rate', '% rate'), ('pcost', '% cost price'), ('ppack', 'Per packing'), ], "Mode", required=True) price = fields.Numeric("Price", digits=(16, 4)) quantity = fields.Numeric("Qt", digits='unit') currency = fields.Many2One('currency.currency', "Currency") unit = fields.Many2One('product.uom', "Unit") qt_state = fields.Many2One('lot.qt.type', "Qt State") weight_type = fields.Selection([ ('net', 'Net'), ('brut', 'Gross'), ], "W. type") @classmethod def default_active(cls): return True @classmethod def default_sequence(cls): return 10 @classmethod def default_auto_apply(cls): return False @classmethod def default_p_r(cls): return 'pay' @classmethod def default_mode(cls): return 'lumpsum' @classmethod def default_weight_type(cls): return 'brut' @staticmethod def _same_record(left, right): return bool( left and right and getattr(left, 'id', left) == getattr(right, 'id', right)) @classmethod def _category_matches(cls, category, target): while target: if cls._same_record(category, target): return True target = getattr(target, 'parent', None) return False @classmethod def _party_has_category(cls, party, category): if not category: return True if not party: return False return any( cls._category_matches(category, party_category) for party_category in (getattr(party, 'categories', []) or [])) @classmethod def _rules_for_lines(cls, apply_on, auto_only=False): domain = [ ('active', '=', True), ('apply_on', '=', apply_on), ] if auto_only: domain.append(('auto_apply', '=', True)) return cls.search(domain, order=[('sequence', 'ASC'), ('id', 'ASC')]) def _matches_line(self, line, apply_on): contract = getattr(line, 'purchase', None) or getattr(line, 'sale', None) if not contract or self.apply_on != apply_on: return False checks = [ (self.party, getattr(contract, 'party', None)), (self.line_product, getattr(line, 'product', None)), (self.from_location, getattr(contract, 'from_location', None)), (self.to_location, getattr(contract, 'to_location', None)), ] if any(rule_value and not self._same_record(rule_value, actual) for rule_value, actual in checks): return False if not self._party_has_category( getattr(contract, 'party', None), self.party_category): return False if self.incoterm_code: incoterm = getattr(contract, 'incoterm', None) if (getattr(incoterm, 'code', '') or '').upper() != ( self.incoterm_code or '').upper(): return False if self.execution_plan and not self._same_record( self.execution_plan, getattr(contract, 'plan', None)): return False return True def _existing_fee(self, line, apply_on): Fee = Pool().get('fee.fee') field_name = 'line' if apply_on == 'purchase_line' else 'sale_line' existing = Fee.search([ ('source_rule', '=', self.id), (field_name, '=', line.id), ], limit=1) return existing[0] if existing else None def _fee_values(self, line, apply_on): contract = getattr(line, 'purchase', None) or getattr(line, 'sale', None) values = { 'source_rule': self.id, 'generated_by_rule': True, 'type': 'budgeted', 'p_r': self.p_r, 'product': self.fee_product.id, 'supplier': ( getattr(self.supplier, 'id', None) or getattr(getattr(contract, 'party', None), 'id', None)), 'mode': self.mode, 'price': self.price if self.price is not None else Decimal(0), 'currency': ( getattr(self.currency, 'id', None) or getattr(getattr(contract, 'currency', None), 'id', None)), 'unit': ( getattr(self.unit, 'id', None) or getattr(getattr(line, 'unit', None), 'id', None)), 'quantity': ( self.quantity if self.quantity is not None else getattr(line, 'quantity_theorical', None) or getattr(line, 'quantity', None)), 'qt_state': getattr(self.qt_state, 'id', None), 'weight_type': self.weight_type, } values['line' if apply_on == 'purchase_line' else 'sale_line'] = line.id return values @classmethod def apply_to_lines(cls, lines, apply_on, auto_only=False): Fee = Pool().get('fee.fee') rules = cls._rules_for_lines(apply_on, auto_only=auto_only) created = [] for line in lines: if not getattr(line, 'id', None): continue for rule in rules: if not rule._matches_line(line, apply_on): continue if rule._existing_fee(line, apply_on): continue values = rule._fee_values(line, apply_on) if not ( values.get('supplier') and values.get('currency') and values.get('unit')): logger.info( "FEE_RULE_SKIPPED_MISSING_VALUES rule=%s line=%s", rule.id, line.id) continue with Transaction().set_context( _purchase_trade_skip_fee_rules=True): created.extend(Fee.create([values])) return created class FeeLots(ModelSQL,ModelView): "Fee lots" __name__ = 'fee.lots' fee = fields.Many2One('fee.fee',"Fee",required=True, ondelete='CASCADE') lot = fields.Many2One('lot.lot',"Lot",required=True, ondelete='CASCADE') @classmethod def create(cls, vlist): records = super().create(vlist) Fee = Pool().get('fee.fee') fee_ids = { record.fee.id for record in records if getattr(record, 'fee', None)} for fee in Fee.browse(list(fee_ids)): fee.sync_quantity_from_lots() fee.adjust_purchase_values() Fee.assert_quantities_consistency(Fee.browse(list(fee_ids))) Fee._regenerate_fee_pnl( fees=Fee.browse(list(fee_ids)), lots=[record.lot for record in records if getattr(record, 'lot', None)]) return records @classmethod def write(cls, *args): old_lots = [ record.lot for record in sum(args[::2], []) if getattr(record, 'lot', None)] super().write(*args) records = sum(args[::2], []) Fee = Pool().get('fee.fee') fee_ids = { record.fee.id for record in records if getattr(record, 'fee', None)} for fee in Fee.browse(list(fee_ids)): fee.sync_quantity_from_lots() fee.adjust_purchase_values() Fee.assert_quantities_consistency(Fee.browse(list(fee_ids))) Fee._regenerate_fee_pnl( fees=Fee.browse(list(fee_ids)), lots=old_lots + [ record.lot for record in records if getattr(record, 'lot', None)]) @classmethod def delete(cls, records): Fee = Pool().get('fee.fee') fee_ids = { record.fee.id for record in records if getattr(record, 'fee', None)} lots = [record.lot for record in records if getattr(record, 'lot', None)] super().delete(records) for fee in Fee.browse(list(fee_ids)): fee.sync_quantity_from_lots() fee.adjust_purchase_values() Fee.assert_quantities_consistency(Fee.browse(list(fee_ids))) Fee._regenerate_fee_pnl( fees=Fee.browse(list(fee_ids)), lots=lots) class FeeReport( ModelSQL, ModelView): "Fee Report" __name__ = 'fee.report' r_purchase_line = fields.Many2One('purchase.line', "Purchase line") r_sale_line = fields.Many2One('sale.line', "Sale line") r_shipment_in = fields.Many2One('stock.shipment.in', "Shipment in") r_shipment_out = fields.Many2One('stock.shipment.out', "Shipment out") r_shipment_internal = fields.Many2One('stock.shipment.internal', "Shipment internal") r_fee_type = fields.Many2One('product.product', 'Fee type') r_fee_counterparty = fields.Many2One('party.party', "Counterparty", required=True) r_type = fields.Selection([ ('ordered', 'Ordered'), ('budgeted', 'Budgeted'), ('actual', 'Actual') ], 'Type') r_fee_paystatus = fields.Selection([ ('pay', 'PAY'), ('rec', 'REC') ], 'Pay status') r_mode = fields.Selection([ ('lumpsum', 'Lump sum'), ('perqt', 'Per qt'), ('pprice', '% price'), ('pcost', '% cost price'), ], 'Mode', required=True) r_fee_quantity = fields.Function(fields.Numeric("Qt",digits=(1,4)),'get_quantity') r_fee_unit = fields.Function(fields.Many2One('product.uom',"Unit"),'get_unit') r_purchase = fields.Many2One('purchase.purchase',"Purchase", ondelete='CASCADE') r_fee_amount = fields.Function(fields.Numeric("Amount", digits=(1,4)),'get_amount') r_inv = fields.Function(fields.Many2One('account.invoice',"Invoice"),'get_invoice') r_state = fields.Selection([ ('not invoiced', 'Not invoiced'), ('invoiced', 'Invoiced'), ], string='State', readonly=True) #r_fee_lots = fields.Function(fields.Many2Many('lot.lot', None, None, "Lots"),'get_lots')#, searcher='search_lots') #r_lots = fields.Many2Many('fee.lots', 'fee', 'lot',"Lots",domain=[('id', 'in', Eval('r_fee_lots',[]))] ) r_fee_currency = fields.Many2One('currency.currency',"Currency") r_fee_price = fields.Numeric("Price",digits=(1,4)) r_shipment_origin = fields.Function( fields.Reference( selection=[ ("stock.shipment.in", "In"), ("stock.shipment.out", "Out"), ("stock.shipment.internal", "Internal"), ], string="Shipment", ), "get_shipment_origin", ) def get_invoice(self,name): if self.r_purchase: if self.r_purchase.invoices: return self.r_purchase.invoices[0] def get_shipment_origin(self, name): if self.r_shipment_in: return 'stock.shipment.in,' + str(self.r_shipment_in.id) elif self.r_shipment_out: return 'stock.shipment.out,' + str(self.r_shipment_out.id) elif self.r_shipment_internal: return 'stock.shipment.internal,' + str(self.r_shipment_internal.id) return None # def get_lots(self, name): # if self.r_purchase_line: # return self.r_purchase_line.lots def get_unit(self, name): if self.r_purchase_line: return self.r_purchase_line.unit def get_quantity(self,name=None): Fee = Pool().get('fee.fee') fee = Fee(self.id) return fee.get_quantity() def get_amount(self,name=None): Fee = Pool().get('fee.fee') fee = Fee(self.id) return fee.get_amount() @classmethod def table_query(cls): FeeReport = Pool().get('fee.fee') fr = FeeReport.__table__() Purchase = Pool().get('purchase.purchase') pu = Purchase.__table__() PurchaseLine = Pool().get('purchase.line') pl = PurchaseLine.__table__() Sale = Pool().get('sale.sale') sa = Sale.__table__() SaleLine = Pool().get('sale.line') sl = SaleLine.__table__() context = Transaction().context party = context.get('party') fee_type = context.get('fee_type') purchase = context.get('purchase') sale = context.get('sale') shipment_in = context.get('shipment_in') shipment_out = context.get('shipment_out') shipment_internal = context.get('shipment_internal') asof = context.get('asof') todate = context.get('todate') wh = ((fr.create_date >= asof) & ((fr.create_date-datetime.timedelta(1)) <= todate)) if party: wh &= (fr.fee_counterparty == party) if fee_type: wh &= (fr.fee_type == fee_type) if purchase: wh &= (pu.id == purchase) if sale: wh &= (sa.id == sale) if shipment_in: wh &= (fr.shipment_in == shipment_in) # if shipment_out: # wh &= (fr.shipment_out == shipment_out) query = fr.join(pl,'LEFT',condition=fr.line == pl.id).join(pu,'LEFT', condition=pl.purchase == pu.id).select( Literal(0).as_('create_uid'), CurrentTimestamp().as_('create_date'), Literal(None).as_('write_uid'), Literal(None).as_('write_date'), fr.id.as_('id'), fr.line.as_('r_purchase_line'), Literal(None).as_('r_sale_line'), fr.shipment_in.as_('r_shipment_in'), Literal(None).as_('r_shipment_out'), fr.shipment_internal.as_('r_shipment_internal'), fr.product.as_('r_fee_type'), fr.supplier.as_('r_fee_counterparty'), fr.type.as_('r_type'), fr.p_r.as_('r_fee_paystatus'), fr.mode.as_('r_mode'), fr.state.as_('r_state'), fr.purchase.as_('r_purchase'), #fr.amount.as_('r_fee_amount'), fr.price.as_('r_fee_price'), fr.currency.as_('r_fee_currency'), #fr.fee_lots.as_('r_fee_lots'), #fr.lots.as_('r_lots'), where=wh) return query @classmethod def search_rec_name(cls, name, clause): _, operator, operand, *extra = clause if operator.startswith('!') or operator.startswith('not '): bool_op = 'AND' else: bool_op = 'OR' code_value = operand if operator.endswith('like') and is_full_text(operand): code_value = lstrip_wildcard(operand) return [bool_op, ('r_fee_type', operator, operand, *extra), ('r_fee_counterparty', operator, operand, *extra), ] class FeeContext(ModelView): "Fee Context" __name__ = 'fee.context' asof = fields.Date("As of") todate = fields.Date("To") party = fields.Many2One('party.party', "Counterparty") fee_type = fields.Many2One('product.product', 'Fee type') purchase = fields.Many2One('purchase.purchase', "Purchase") sale = fields.Many2One('sale.sale', "Sale") shipment_in = fields.Many2One('stock.shipment.in',"Shipment In") shipment_out = fields.Many2One('stock.shipment.out',"Shipment Out") shipment_internal = fields.Many2One('stock.shipment.internal',"Shipment Internal") @classmethod def default_asof(cls): pool = Pool() Date = pool.get('ir.date') return Date.today().replace(day=1,month=1,year=1999) @classmethod def default_todate(cls): pool = Pool() Date = pool.get('ir.date') return Date.today()