1426 lines
58 KiB
Python
Executable File
1426 lines
58 KiB
Python
Executable File
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)
|
|
|
|
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)
|
|
|
|
@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):
|
|
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()
|