220 lines
13 KiB
Python
220 lines
13 KiB
Python
"""Apply quote edits in chronological order, including derived carton quantities."""
|
|
|
|
from copy import deepcopy
|
|
from decimal import Decimal
|
|
|
|
from .quote_intent import kg,MISSING_WEIGHT_UNIT
|
|
from .quote_edits import ADJUSTMENT_QUESTION,VOLUME_QUESTION
|
|
|
|
|
|
QUANTITY_QUESTION='箱数已更改，请确认新的总重或各箱重量，原分箱明细不能直接套用。'
|
|
|
|
|
|
def apply_quantity_change(result,current,previous):
|
|
from .quote_engine import exact_number
|
|
|
|
change=current.get('quantity_change')
|
|
if not change:return False
|
|
if change['kind']=='volume':
|
|
updated=Decimal(current['cbm']) if current['cbm'] is not None else None
|
|
mismatch=False
|
|
if not change['uncertain'] and previous['cbm'] is not None and not previous.get('volume_pending'):
|
|
expected=Decimal(previous['cbm'])+Decimal(change['delta'])
|
|
if updated is not None and updated!=expected:
|
|
mismatch=True
|
|
result['errors'].append('增减体积与您提供的新总体积不一致，请确认调整后的总体积。')
|
|
else:updated=expected
|
|
if change['uncertain'] or updated is None or updated<=0 or mismatch:
|
|
result['volume_pending']=True
|
|
result['errors'].append(VOLUME_QUESTION)
|
|
else:
|
|
result['cbm']=str(updated)
|
|
result['volume_pending']=False
|
|
result['quantity_note']=f'按本次增减更新，总体积为{exact_number(updated)}CBM。'
|
|
return False
|
|
question=ADJUSTMENT_QUESTION
|
|
result['quantity_question']=question
|
|
result['quantity_uncertain']=True
|
|
if change['kind']=='boxes' and (previous['cbm'] is not None or previous.get('volume_pending') or current['cbm'] is not None):
|
|
result['volume_pending']=True
|
|
if change['uncertain'] or previous.get('allocations'):
|
|
result['errors'].append(question)
|
|
return False
|
|
if change['kind']=='weight':
|
|
if previous.get('weight') is None or previous.get('quantity_uncertain') or previous.get('pending_weight'):
|
|
if current['weight'] is None:
|
|
result['errors'].append(question)
|
|
return False
|
|
updated=Decimal(current['weight'])
|
|
else:
|
|
updated=Decimal(previous['weight'])+Decimal(change['delta'])
|
|
if current['weight'] is not None and Decimal(current['weight'])!=updated:
|
|
result['errors'].append('增减量与您提供的新总重不一致，请确认调整后的总重。')
|
|
return False
|
|
if updated<=0:
|
|
result['errors'].append('调整后的总重需要大于0，请确认增减量或新的总重。')
|
|
return False
|
|
result['weight']=str(updated)
|
|
else:
|
|
pieces=deepcopy(previous.get('packages') or [])
|
|
if not pieces or previous.get('quantity_uncertain') or previous.get('pending_weight'):
|
|
result['errors'].append(question)
|
|
return False
|
|
for piece in pieces:
|
|
if not piece.get('dims') and previous['dims']:
|
|
piece.update(dims=previous['dims'],dim_unit=previous['dim_unit'])
|
|
first=pieces[0]
|
|
uniform=all(Decimal(piece['weight'])==Decimal(first['weight'])
|
|
and all(piece.get(key)==first.get(key) for key in ('dims','dim_unit')) for piece in pieces)
|
|
if uniform:
|
|
first['count']=sum(piece['count'] for piece in pieces)
|
|
pieces=[first]
|
|
delta=change['delta']
|
|
additions=deepcopy(current.get('packages') or [])
|
|
target_count=sum(piece['count'] for piece in pieces)+delta
|
|
declared_count=change.get('declared_boxes')
|
|
if declared_count is not None and declared_count!=target_count:
|
|
result['errors'].append('增减箱数与您提供的新总箱数不一致，请确认调整后的箱数。')
|
|
return False
|
|
if declared_count is not None and additions and sum(piece['count'] for piece in additions)==target_count:
|
|
pieces=additions
|
|
elif delta>0 and additions:
|
|
if sum(piece['count'] for piece in additions)!=delta:
|
|
result['errors'].append(question)
|
|
return False
|
|
pieces.extend(additions)
|
|
elif uniform:
|
|
if delta<0 and additions and any(Decimal(piece['weight'])!=Decimal(first['weight'])
|
|
or piece.get('dims') and (piece['dims']!=first.get('dims') or piece['dim_unit']!=first.get('dim_unit')) for piece in additions):
|
|
result['errors'].append('减少箱子的重量与原分箱明细不一致，请提供更新后的逐箱明细。')
|
|
return False
|
|
first['count']+=delta
|
|
if first['count']<=0:
|
|
result['errors'].append('调整后的箱数需要大于0，请确认增减量或新的箱数。')
|
|
return False
|
|
else:
|
|
result['errors'].append('原货物包含不同箱规，请提供更新后的逐箱数量和重量。')
|
|
return False
|
|
result['packages']=pieces
|
|
result['boxes']=sum(piece['count'] for piece in pieces)
|
|
result['weight']=str(sum(Decimal(piece['weight'])*piece['count'] for piece in pieces))
|
|
result['dims']=None
|
|
result['cbm']=None
|
|
if result['boxes']<=0 or any(piece['count']<=0 for piece in pieces):
|
|
result['errors'].append('调整后的箱数需要大于0，请确认增减量或新的箱数。')
|
|
return False
|
|
if change.get('declared_weight') is not None and Decimal(change['declared_weight'])!=Decimal(result['weight']):
|
|
result['errors'].append('增减箱数推算的总重与您提供的新总重不一致，请确认更新后的分箱明细。')
|
|
return False
|
|
result['quantity_uncertain']=False
|
|
result['quantity_question']=None
|
|
verb='增加' if Decimal(change['delta'])>0 else '减少'
|
|
delta=exact_number(abs(Decimal(change['delta'])))
|
|
if change['kind']=='boxes':
|
|
old_count=sum(piece['count'] for piece in previous['packages'])
|
|
result['quantity_note']=f'按原{old_count}箱{verb}{delta}箱核算，共{result["boxes"]}箱。'
|
|
elif previous['weight'] is not None and not previous.get('quantity_uncertain'):
|
|
result['quantity_note']=f'按原总重{exact_number(previous["weight"])}KG{verb}{delta}KG核算。'
|
|
else:result['quantity_note']='按您确认的新总重核算。'
|
|
return True
|
|
|
|
|
|
def merge_intent(current,previous):
|
|
result=deepcopy(current)
|
|
excluded={'mode','selection','clear_selection','clear_choice','compare_services','choice_pending','excluded_services','quantity_change','tax_pending','volume_pending','quantity_question',
|
|
'text','errors','dim_unit','quantity_note','quantity_uncertain','preserve_total','metal_weight_update','pending_weight','weight_unit'}
|
|
unit=current.get('weight_unit')
|
|
pending=previous.get('pending_weight')
|
|
weight_change=current['weight'] is not None or bool(unit and pending is not None)
|
|
weight_change=apply_quantity_change(result,current,previous) or weight_change
|
|
change=current.get('quantity_change')
|
|
if change and change['kind']=='boxes':excluded.update(('dims','cbm','allocations'))
|
|
if change and change['kind']=='volume':excluded.add('dims')
|
|
volume_confirmed=not (change and change['kind']=='boxes') and (current['cbm'] is not None or current['dims']
|
|
or current.get('packages') and all(piece.get('dims') for piece in current['packages']))
|
|
if previous.get('volume_pending') and not volume_confirmed:result['volume_pending']=True
|
|
if result.get('volume_pending'):
|
|
result['cbm']=None
|
|
excluded.add('cbm')
|
|
result['errors'].append(VOLUME_QUESTION)
|
|
if unit:
|
|
if pending is None:result['errors'].append('请同时提供重量数值和单位，例如800KG。')
|
|
else:result['weight']=str(kg(pending,unit))
|
|
elif pending is not None and current['weight'] is None and current.get('pending_weight') is None:
|
|
result['pending_weight']=pending
|
|
result['errors'].append(MISSING_WEIGHT_UNIT)
|
|
if not current.get('selection') and not current.get('clear_selection') and not current.get('clear_choice') and not current.get('services'):
|
|
result['selection']=previous.get('selection')
|
|
if current.get('clear_selection'):
|
|
excluded.update(('services','kinds','transport','air_only','air_requested'))
|
|
if current.get('compare_services'):excluded.add('services')
|
|
blocked=[] if current.get('clear_selection') else previous.get('excluded_services') or []
|
|
blocked=list(dict.fromkeys(blocked+(current.get('excluded_services') or [])))
|
|
if current.get('excluded_services'):
|
|
excluded.add('services')
|
|
if any(name in ('海派','海卡','海铁') for name in current['excluded_services']):excluded.add('kinds')
|
|
positives=(current.get('services') or [])+[name for name in ('海派','海卡','海铁') if name in current['text']]
|
|
result['excluded_services']=[name for name in blocked if not any(name in choice or choice in name for choice in positives)] or None
|
|
if current.get('tax_pending') or previous.get('tax_pending') and current['tax'] is None:
|
|
result['tax_pending']=True
|
|
result['tax']=None
|
|
excluded.add('tax')
|
|
result['errors'].append('请明确选择包税或不包税方案。')
|
|
if current.get('choice_pending') or (previous.get('choice_pending') and not current.get('selection')
|
|
and not current.get('services') and not current.get('clear_selection')):
|
|
result['choice_pending']=True
|
|
result['selection']=None
|
|
result['errors'].append('请指定希望使用的渠道，或选择最快、最慢、最便宜方案。')
|
|
if current.get('metal_weight_update') and previous.get('metal_chargeable_weight') is None:
|
|
result['metal_chargeable_weight']=None
|
|
result['errors'].append('请确认铁铝部分的计费重，不能把毛重直接当作计费重。')
|
|
if current.get('warehouse') or current.get('unknown_warehouses'):
|
|
excluded.update(('postal','zone','zones'))
|
|
elif current.get('postal'):
|
|
excluded.update(('warehouse','warehouses','unknown_warehouses','zone','zones'))
|
|
elif current.get('zone'):
|
|
excluded.update(('warehouse','warehouses','unknown_warehouses','postal'))
|
|
if any(current.get(k) for k in ('warehouse','postal','zone')) and not current.get('unknown_warehouses'):
|
|
excluded.add('unknown_warehouses')
|
|
if any(current.get(k) for k in ('warehouse','unknown_warehouses','postal','zone')) and not current.get('allocations'):
|
|
excluded.add('allocations')
|
|
if current.get('allocations'):excluded.update(('packages','boxes','dims','cbm'))
|
|
if current['transport'] or current['kinds'] or current['services']:excluded.update(('air_only','air_requested'))
|
|
if current['air_only'] or current['tax']=='excluded':excluded.add('kinds')
|
|
if weight_change:
|
|
if not current.get('packages'):excluded.add('packages')
|
|
if not current.get('allocations'):excluded.add('allocations')
|
|
if current['dims']:excluded.add('cbm')
|
|
if current['cbm'] is not None:excluded.add('dims')
|
|
if (weight_change or any(current.get(k) is not None for k in ('dims','cbm','boxes','packages','allocations','transport','kinds','services','tax'))) and current.get('metal_chargeable_weight') is None:
|
|
excluded.add('metal_chargeable_weight')
|
|
if current.get('material')=='ordinary':excluded.update(('metal_scope','metal_chargeable_weight'))
|
|
if current.get('metal_scope')=='all' and current.get('metal_chargeable_weight') is None:
|
|
excluded.add('metal_chargeable_weight')
|
|
for key,value in previous.items():
|
|
if result.get(key) is None and key not in excluded:result[key]=deepcopy(value)
|
|
if current['dims'] is None and 'dims' not in excluded and previous['dims']:
|
|
result['dim_unit']=previous['dim_unit']
|
|
pieces=result.get('packages') or []
|
|
if current['dims'] and not (change and change['kind']=='boxes'):
|
|
for piece in pieces:piece.update(dims=current['dims'],dim_unit=current['dim_unit'])
|
|
elif (current['cbm'] is not None or change and change['kind']=='volume') and not (change and change['kind']=='boxes'):
|
|
for piece in pieces:piece.update(dims=None,dim_unit='cm')
|
|
if previous.get('quantity_uncertain') and not weight_change and not current.get('preserve_total'):
|
|
result['quantity_uncertain']=True
|
|
result['quantity_question']=previous.get('quantity_question') or QUANTITY_QUESTION
|
|
result['errors'].append(result['quantity_question'])
|
|
if pieces and result['boxes']!=sum(p['count'] for p in pieces):
|
|
if current.get('preserve_total'):
|
|
result['packages']=None
|
|
elif len(pieces)==1:
|
|
pieces[0]['count']=result['boxes']
|
|
result['weight']=str(Decimal(pieces[0]['weight'])*result['boxes'])
|
|
result['quantity_note']=f'按每箱{pieces[0]["weight"]}KG不变，更新为{result["boxes"]}箱。'
|
|
else:
|
|
result['packages']=None
|
|
result['quantity_uncertain']=True
|
|
result['errors'].append(QUANTITY_QUESTION)
|
|
result['errors']=list(dict.fromkeys(result['errors']))
|
|
return result
|