547 lines
20 KiB
Python
547 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
|
from decimal import Decimal
|
|
import logging
|
|
|
|
from trytond.pool import Pool
|
|
from trytond.transaction import Transaction
|
|
from trytond.exceptions import UserError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ContractFactory:
|
|
"""
|
|
Factory métier pour créer des Purchase depuis Sale
|
|
ou des Sale depuis Purchase.
|
|
|
|
Compatible :
|
|
- Wizard (n contrats)
|
|
- Appel direct depuis un modèle (1 contrat)
|
|
"""
|
|
|
|
@classmethod
|
|
def create_contracts(cls, contracts, *, type_, ct):
|
|
"""
|
|
:param contracts: iterable de contracts (wizard lines)
|
|
:param type_: 'Purchase' ou 'Sale'
|
|
:param ct: objet contenant le contexte (lot, product, unit, matched...)
|
|
:return: liste des contracts créés
|
|
"""
|
|
pool = Pool()
|
|
|
|
Sale = pool.get('sale.sale')
|
|
Purchase = pool.get('purchase.purchase')
|
|
SaleLine = pool.get('sale.line')
|
|
PurchaseLine = pool.get('purchase.line')
|
|
Lot = pool.get('lot.lot')
|
|
Date = pool.get('ir.date')
|
|
|
|
created = []
|
|
lines_to_check = []
|
|
sources = cls._get_sources(ct, type_)
|
|
cls._validate_requested_quantity(contracts, sources, ct)
|
|
base_contract = cls._get_base_contract(sources, ct, type_)
|
|
|
|
for c in contracts:
|
|
contract = Purchase() if type_ == 'Purchase' else Sale()
|
|
|
|
# ---------- CONTRACT ----------
|
|
parts = cls._get_currency_unit_parts(c, base_contract)
|
|
contract.currency = int(parts[0]) or 1
|
|
contract.party = c.party
|
|
contract.crop = cls._get_crop(c, base_contract)
|
|
contract.tol_min = cls._get_tolerance(c, 'tol_min')
|
|
contract.tol_max = cls._get_tolerance(c, 'tol_max')
|
|
contract.payment_term = cls._get_payment_term(
|
|
c, base_contract, type_)
|
|
contract.reference = c.reference
|
|
from_location, to_location = cls._get_locations(
|
|
c, base_contract, type_)
|
|
contract.from_location = from_location
|
|
contract.to_location = to_location
|
|
context = Transaction().context
|
|
contract.company = context.get('company') if context else None
|
|
if type_ == 'Purchase':
|
|
contract.purchase_date = Date.today()
|
|
else:
|
|
contract.sale_date = Date.today()
|
|
|
|
cls._apply_party_data(contract, c.party, base_contract, type_)
|
|
if type_ == 'Sale':
|
|
contract.product_origin = getattr(base_contract, 'product_origin', None)
|
|
|
|
contract.incoterm = cls._get_contract_value(
|
|
c, base_contract, 'incoterm')
|
|
|
|
addresses = getattr(c.party, 'addresses', None)
|
|
if addresses:
|
|
contract.invoice_address = addresses[0]
|
|
if type_ == 'Sale':
|
|
contract.shipment_address = addresses[0]
|
|
|
|
contract.save()
|
|
|
|
line_sources = cls._get_line_sources(c, sources, ct)
|
|
for source in line_sources:
|
|
line = PurchaseLine() if type_ == 'Purchase' else SaleLine()
|
|
|
|
# ---------- LINE ----------
|
|
line.quantity = source['quantity']
|
|
line.quantity_theorical = source['quantity']
|
|
line.product = ct.product
|
|
line.unit = ct.unit
|
|
line.price_type = c.price_type
|
|
line.created_by_code = ct.matched
|
|
line.premium = Decimal(0)
|
|
|
|
if type_ == 'Purchase':
|
|
line.purchase = contract.id
|
|
else:
|
|
line.sale = contract.id
|
|
|
|
cls._apply_price(line, c, parts)
|
|
cls._apply_delivery(line, c, source)
|
|
|
|
line.save()
|
|
|
|
logger.info("CREATE_ID:%s", contract.id)
|
|
logger.info("CREATE_LINE_ID:%s", line.id)
|
|
|
|
if ct.matched:
|
|
cls._create_lot(line, c, source, type_)
|
|
lines_to_check.append(line)
|
|
if source.get('trade_line'):
|
|
lines_to_check.append(source['trade_line'])
|
|
|
|
created.append(contract)
|
|
|
|
Lot.assert_lines_quantity_consistency(lines_to_check)
|
|
return created
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Helpers
|
|
# -------------------------------------------------------------------------
|
|
|
|
@classmethod
|
|
def _apply_locations(cls, contract, base, type_):
|
|
from_location, to_location = cls._get_mirror_locations(base, type_)
|
|
if from_location:
|
|
contract.from_location = from_location
|
|
if to_location:
|
|
contract.to_location = to_location
|
|
|
|
@staticmethod
|
|
def _get_mirror_locations(base, type_):
|
|
from_location = getattr(base, 'from_location', None)
|
|
to_location = getattr(base, 'to_location', None)
|
|
if not (from_location or to_location):
|
|
return None, None
|
|
|
|
if (from_location and to_location
|
|
and getattr(from_location, 'type', None) == 'supplier'
|
|
and getattr(to_location, 'type', None) == 'customer'):
|
|
return from_location, to_location
|
|
|
|
if type_ == 'Purchase':
|
|
if getattr(from_location, 'type', None) == 'storage':
|
|
return None, from_location
|
|
elif type_ == 'Sale':
|
|
if getattr(to_location, 'type', None) == 'storage':
|
|
return to_location, None
|
|
return None, None
|
|
|
|
@classmethod
|
|
def _get_locations(cls, contract_detail, base, type_):
|
|
mirror_from, mirror_to = cls._get_mirror_locations(base, type_)
|
|
return (
|
|
getattr(contract_detail, 'from_location', None)
|
|
or mirror_from
|
|
or getattr(base, 'from_location', None),
|
|
getattr(contract_detail, 'to_location', None)
|
|
or mirror_to
|
|
or getattr(base, 'to_location', None),
|
|
)
|
|
|
|
@classmethod
|
|
def _apply_party_data(cls, contract, party, base, type_):
|
|
Contract = Pool().get(
|
|
'purchase.purchase' if type_ == 'Purchase' else 'sale.sale')
|
|
contract.wb = cls._get_header_value(
|
|
'wb', party, base, Contract.default_wb)
|
|
contract.association = cls._get_header_value(
|
|
'association', party, base, Contract.default_association)
|
|
contract.certif = cls._get_header_value(
|
|
'certif', party, base, Contract.default_certif)
|
|
|
|
@staticmethod
|
|
def _get_header_value(field, party, base, default_getter=None):
|
|
value = getattr(party, field, None)
|
|
if value:
|
|
return value
|
|
value = getattr(base, field, None)
|
|
if value:
|
|
return value
|
|
if default_getter:
|
|
return default_getter()
|
|
|
|
@staticmethod
|
|
def _get_payment_term(contract_detail, base, type_):
|
|
value = getattr(contract_detail, 'payment_term', None)
|
|
if value:
|
|
return value
|
|
party = getattr(contract_detail, 'party', None)
|
|
field = (
|
|
'supplier_payment_term'
|
|
if type_ == 'Purchase' else 'customer_payment_term')
|
|
value = getattr(party, field, None)
|
|
if value:
|
|
return value
|
|
return getattr(base, 'payment_term', None)
|
|
|
|
@staticmethod
|
|
def _get_contract_value(contract_detail, base, field):
|
|
return (
|
|
getattr(contract_detail, field, None)
|
|
or getattr(base, field, None))
|
|
|
|
@staticmethod
|
|
def _get_currency_unit_parts(contract_detail, base):
|
|
value = getattr(contract_detail, 'currency_unit', None)
|
|
if value:
|
|
return value.split("_")
|
|
currency = (
|
|
getattr(contract_detail, 'currency', None)
|
|
or getattr(base, 'currency', None))
|
|
unit = getattr(contract_detail, 'unit', None)
|
|
currency_id = getattr(currency, 'id', currency)
|
|
unit_id = getattr(unit, 'id', unit)
|
|
if currency_id and unit_id:
|
|
return [str(currency_id), str(unit_id)]
|
|
return ["1", str(unit_id or 1)]
|
|
|
|
@staticmethod
|
|
def _apply_price(line, c, parts):
|
|
if int(parts[0]) == 0:
|
|
line.enable_linked_currency = True
|
|
line.linked_currency = 1
|
|
line.linked_unit = int(parts[1])
|
|
line.linked_price = c.price
|
|
line.unit_price = line.get_price_linked_currency()
|
|
else:
|
|
line.unit_price = c.price if c.price else Decimal(0)
|
|
|
|
@staticmethod
|
|
def _apply_delivery(line, c, source):
|
|
source_line = source.get('trade_line')
|
|
if source.get('use_source_delivery') and source_line:
|
|
line.del_period = getattr(source_line, 'del_period', None)
|
|
line.from_del = getattr(source_line, 'from_del', None)
|
|
line.to_del = getattr(source_line, 'to_del', None)
|
|
return
|
|
line.del_period = c.del_period
|
|
line.from_del = (
|
|
getattr(c, 'from_del', None)
|
|
or getattr(getattr(c, 'del_period', None), 'beg_date', None))
|
|
line.to_del = (
|
|
getattr(c, 'to_del', None)
|
|
or getattr(getattr(c, 'del_period', None), 'end_date', None))
|
|
|
|
@staticmethod
|
|
def _get_tolerance(contract_detail, field):
|
|
value = getattr(contract_detail, field, None)
|
|
if value is not None:
|
|
return value
|
|
party = getattr(contract_detail, 'party', None)
|
|
value = getattr(party, field, None)
|
|
return value if value is not None else Decimal(0)
|
|
|
|
@staticmethod
|
|
def _get_crop(contract_detail, base_contract):
|
|
crop = getattr(contract_detail, 'crop', None)
|
|
if crop:
|
|
return crop
|
|
return getattr(base_contract, 'crop', None)
|
|
|
|
@staticmethod
|
|
def _normalize_quantity(quantity):
|
|
return abs(Decimal(str(quantity or 0))).quantize(Decimal('0.00001'))
|
|
|
|
@classmethod
|
|
def _get_base_contract(cls, sources, ct, type_):
|
|
if sources:
|
|
source_lot = sources[0]['lot']
|
|
return (
|
|
source_lot.sale_line.sale
|
|
if type_ == 'Purchase'
|
|
else source_lot.line.purchase
|
|
)
|
|
return (
|
|
ct.lot.sale_line.sale
|
|
if type_ == 'Purchase'
|
|
else ct.lot.line.purchase
|
|
)
|
|
|
|
@classmethod
|
|
def _get_sources(cls, ct, type_):
|
|
pool = Pool()
|
|
LotQt = pool.get('lot.qt')
|
|
context = Transaction().context or {}
|
|
active_ids = context.get('active_ids') or []
|
|
sources = []
|
|
if active_ids:
|
|
for record_id in active_ids:
|
|
if record_id < 10000000:
|
|
continue
|
|
lqt = LotQt(record_id - 10000000)
|
|
lot = lqt.lot_p or lqt.lot_s
|
|
if not lot:
|
|
continue
|
|
trade_line = (
|
|
lot.sale_line if type_ == 'Purchase' else lot.line
|
|
)
|
|
sources.append({
|
|
'lqt': lqt,
|
|
'lot': lot,
|
|
'trade_line': trade_line,
|
|
'quantity': cls._normalize_quantity(lqt.lot_quantity),
|
|
'shipment_origin': lqt.lot_shipment_origin,
|
|
})
|
|
elif getattr(ct, 'lot', None):
|
|
lot = ct.lot
|
|
trade_line = (
|
|
lot.sale_line if type_ == 'Purchase' else lot.line
|
|
)
|
|
sources.append({
|
|
'lqt': None,
|
|
'lot': lot,
|
|
'trade_line': trade_line,
|
|
'quantity': cls._normalize_quantity(getattr(ct, 'quantity', 0)),
|
|
'shipment_origin': cls._get_shipment_origin(ct),
|
|
})
|
|
|
|
cls._validate_sources(sources, type_)
|
|
return sources
|
|
|
|
@classmethod
|
|
def _validate_sources(cls, sources, type_):
|
|
if not sources:
|
|
return
|
|
first_line = sources[0]['trade_line']
|
|
for source in sources[1:]:
|
|
line = source['trade_line']
|
|
if bool(getattr(line, 'sale', None)) != bool(getattr(first_line, 'sale', None)):
|
|
raise UserError('Selected lots must all come from the same side.')
|
|
if getattr(line.product, 'id', None) != getattr(first_line.product, 'id', None):
|
|
raise UserError('Selected lots must share the same product.')
|
|
if getattr(line.unit, 'id', None) != getattr(first_line.unit, 'id', None):
|
|
raise UserError('Selected lots must share the same unit.')
|
|
|
|
@classmethod
|
|
def _validate_requested_quantity(cls, contracts, sources, ct):
|
|
if not getattr(ct, 'matched', False) or not sources:
|
|
return
|
|
available = sum(source['quantity'] for source in sources)
|
|
requested = sum(
|
|
cls._normalize_quantity(contract.quantity)
|
|
for contract in contracts)
|
|
if requested > available:
|
|
raise UserError(
|
|
'The requested quantity exceeds the selected open quantity.')
|
|
|
|
@classmethod
|
|
def _get_line_sources(cls, contract_detail, sources, ct):
|
|
if not ct.matched or len(sources) <= 1:
|
|
quantity = cls._normalize_quantity(contract_detail.quantity)
|
|
source = sources[0] if sources else {
|
|
'lot': getattr(ct, 'lot', None),
|
|
'trade_line': None,
|
|
'shipment_origin': cls._get_shipment_origin(ct),
|
|
}
|
|
return [{
|
|
**source,
|
|
'quantity': quantity,
|
|
'use_source_delivery': False,
|
|
}]
|
|
|
|
selected_total = sum(source['quantity'] for source in sources)
|
|
requested = cls._normalize_quantity(contract_detail.quantity)
|
|
if requested != selected_total:
|
|
raise UserError(
|
|
'For multi-lot matched creation, quantity must equal the total selected open quantity.'
|
|
)
|
|
return [{
|
|
**source,
|
|
'use_source_delivery': True,
|
|
} for source in sources]
|
|
|
|
# -------------------------------------------------------------------------
|
|
# LOT / MATCHING (repris tel quel du wizard)
|
|
# -------------------------------------------------------------------------
|
|
|
|
@classmethod
|
|
def _create_lot(cls, line, c, source, type_):
|
|
pool = Pool()
|
|
Lot = pool.get('lot.lot')
|
|
LotQtHist = pool.get('lot.qt.hist')
|
|
LotQtType = pool.get('lot.qt.type')
|
|
|
|
lot = Lot()
|
|
|
|
if type_ == 'Purchase':
|
|
lot.line = line.id
|
|
else:
|
|
lot.sale_line = line.id
|
|
|
|
lot.lot_qt = None
|
|
lot.lot_unit = None
|
|
lot.lot_unit_line = line.unit
|
|
lot.lot_quantity = round(line.quantity, 5)
|
|
lot.lot_gross_quantity = None
|
|
lot.lot_status = 'forecast'
|
|
lot.lot_type = 'virtual'
|
|
lot.lot_product = line.product
|
|
|
|
lqtt = LotQtType.search([('sequence', '=', 1)])
|
|
if lqtt:
|
|
lqh = LotQtHist()
|
|
lqh.quantity_type = lqtt[0]
|
|
lqh.quantity = round(lot.lot_quantity, 5)
|
|
lqh.gross_quantity = round(lot.lot_quantity, 5)
|
|
lot.lot_hist = [lqh]
|
|
|
|
with Lot.skip_quantity_consistency():
|
|
lot.save()
|
|
|
|
vlot = source['lot']
|
|
shipment_origin = source.get('shipment_origin')
|
|
qt = source['quantity']
|
|
|
|
if type_ == 'Purchase':
|
|
if cls._attach_or_split_source_lot_qt(source, lot, type_):
|
|
return
|
|
if not lot.updateVirtualPart(qt, shipment_origin, vlot):
|
|
lot.createVirtualPart(qt, shipment_origin, vlot)
|
|
|
|
# Decrease forecasted virtual part non matched
|
|
lot.updateVirtualPart(-qt, shipment_origin, vlot, 'only sale')
|
|
else:
|
|
if cls._attach_or_split_source_lot_qt(source, lot, type_):
|
|
return
|
|
if not vlot.updateVirtualPart(qt, shipment_origin, lot):
|
|
vlot.createVirtualPart(qt, shipment_origin, lot)
|
|
|
|
# Decrease forecasted virtual part non matched
|
|
vlot.updateVirtualPart(-qt, shipment_origin, None)
|
|
|
|
@classmethod
|
|
def _attach_or_split_source_lot_qt(cls, source, new_virtual_lot, type_):
|
|
"""Reuse the selected forecast line instead of create/decrease."""
|
|
pool = Pool()
|
|
LotQt = pool.get('lot.qt')
|
|
|
|
quantity = cls._normalize_quantity(source['quantity'])
|
|
source_lqt = source.get('lqt')
|
|
if not source_lqt:
|
|
source_lqt = cls._find_source_lot_qt(source, type_)
|
|
if not source_lqt:
|
|
return False
|
|
|
|
remaining = cls._normalize_quantity(source_lqt.lot_quantity) - quantity
|
|
if remaining < 0:
|
|
return False
|
|
|
|
if remaining:
|
|
matched_lqt = LotQt()
|
|
matched_lqt.lot_p = (
|
|
new_virtual_lot.id
|
|
if type_ == 'Purchase'
|
|
else source_lqt.lot_p)
|
|
matched_lqt.lot_s = (
|
|
source_lqt.lot_s
|
|
if type_ == 'Purchase'
|
|
else new_virtual_lot.id)
|
|
matched_lqt.lot_shipment_in = source_lqt.lot_shipment_in
|
|
matched_lqt.lot_shipment_internal = source_lqt.lot_shipment_internal
|
|
matched_lqt.lot_shipment_out = source_lqt.lot_shipment_out
|
|
matched_lqt.lot_move = source_lqt.lot_move
|
|
matched_lqt.lot_av = source_lqt.lot_av
|
|
matched_lqt.lot_status = source_lqt.lot_status
|
|
matched_lqt.lot_unit = source_lqt.lot_unit
|
|
matched_lqt.lot_quantity = quantity
|
|
source_lqt.lot_quantity = remaining
|
|
LotQt.save([source_lqt, matched_lqt])
|
|
else:
|
|
if type_ == 'Purchase':
|
|
source_lqt.lot_p = new_virtual_lot.id
|
|
else:
|
|
source_lqt.lot_s = new_virtual_lot.id
|
|
LotQt.save([source_lqt])
|
|
return True
|
|
|
|
@classmethod
|
|
def _find_source_lot_qt(cls, source, type_):
|
|
pool = Pool()
|
|
LotQt = pool.get('lot.qt')
|
|
lot = source.get('lot')
|
|
if not lot:
|
|
return None
|
|
|
|
domain = []
|
|
if type_ == 'Purchase':
|
|
domain.extend([
|
|
('lot_s', '=', lot.id),
|
|
('lot_p', '=', None),
|
|
])
|
|
else:
|
|
domain.extend([
|
|
('lot_p', '=', lot.id),
|
|
('lot_s', '=', None),
|
|
])
|
|
cls._add_shipment_domain(domain, source.get('shipment_origin'))
|
|
domain.append(('lot_quantity', '!=', 0))
|
|
lqts = LotQt.search(domain)
|
|
if not lqts and source.get('shipment_origin'):
|
|
domain = domain[:2] + [
|
|
('lot_quantity', '!=', 0),
|
|
]
|
|
lqts = LotQt.search(domain)
|
|
quantity = cls._normalize_quantity(source['quantity'])
|
|
for lqt in lqts:
|
|
if cls._normalize_quantity(lqt.lot_quantity) == quantity:
|
|
return lqt
|
|
for lqt in lqts:
|
|
if cls._normalize_quantity(lqt.lot_quantity) >= quantity:
|
|
return lqt
|
|
return None
|
|
|
|
@staticmethod
|
|
def _add_shipment_domain(domain, shipment_origin):
|
|
domain.extend([
|
|
('lot_shipment_in', '=', None),
|
|
('lot_shipment_internal', '=', None),
|
|
('lot_shipment_out', '=', None),
|
|
])
|
|
if not shipment_origin:
|
|
return
|
|
model, record_id = shipment_origin.split(',', 1)
|
|
record_id = int(record_id)
|
|
if model == 'stock.shipment.in':
|
|
domain[-3] = ('lot_shipment_in', '=', record_id)
|
|
elif model == 'stock.shipment.internal':
|
|
domain[-2] = ('lot_shipment_internal', '=', record_id)
|
|
elif model == 'stock.shipment.out':
|
|
domain[-1] = ('lot_shipment_out', '=', record_id)
|
|
|
|
@staticmethod
|
|
def _get_shipment_origin(ct):
|
|
shipment_in = getattr(ct, 'shipment_in', None)
|
|
shipment_internal = getattr(ct, 'shipment_internal', None)
|
|
shipment_out = getattr(ct, 'shipment_out', None)
|
|
if shipment_in:
|
|
return 'stock.shipment.in,%s' % shipment_in.id
|
|
if shipment_internal:
|
|
return 'stock.shipment.internal,%s' % shipment_internal.id
|
|
if shipment_out:
|
|
return 'stock.shipment.out,%s' % shipment_out.id
|
|
return None
|