180 lines
9.7 KiB
Python
180 lines
9.7 KiB
Python
import logging
|
||
|
||
from odoo import api, models
|
||
from odoo.exceptions import ValidationError, UserError
|
||
|
||
|
||
class StockPicking(models.Model):
|
||
_inherit = 'stock.picking'
|
||
|
||
def _compute_check(self):
|
||
super()._compute_check()
|
||
for picking in self:
|
||
picking_to_quality = picking.get_picking_to_quality()
|
||
if not picking_to_quality:
|
||
picking.quality_check_todo = False
|
||
break
|
||
else:
|
||
need_quality_line = picking.get_need_quality_line(picking_to_quality)
|
||
if not need_quality_line or all(not line.get('need_done_check_ids') for line in need_quality_line):
|
||
picking.quality_check_todo = False
|
||
|
||
def check_quality(self):
|
||
self.ensure_one()
|
||
# checkable_products = self.mapped('move_line_ids').mapped('product_id')
|
||
# checks = self.check_ids.filtered(lambda check: check.quality_state == 'none' and (
|
||
# check.product_id in checkable_products or check.measure_on == 'operation'))
|
||
checks = self.env['quality.check']
|
||
picking_to_quality = self._get_picking_to_quality()
|
||
need_quality_line = self.get_need_quality_line(picking_to_quality)
|
||
if need_quality_line and any(line.get('need_done_check_ids') for line in need_quality_line):
|
||
for item in need_quality_line:
|
||
checks += item.get('need_done_check_ids')
|
||
if checks:
|
||
return checks.action_open_quality_check_wizard()
|
||
return False
|
||
|
||
def button_validate(self):
|
||
"""=
|
||
出厂检验报告上传
|
||
"""
|
||
|
||
out_quality_checks = self.env['quality.check'].search(
|
||
[('picking_id', '=', self.id), ('test_type_id.name', '=', '出厂检验报告')])
|
||
# out_quality_checks 可能存在多个
|
||
if out_quality_checks:
|
||
for out_quality_check in out_quality_checks:
|
||
if not out_quality_check.is_factory_report_uploaded:
|
||
if out_quality_check and self.state == 'assigned':
|
||
out_quality_check.upload_factory_report()
|
||
quality_action = self.pinking_checkout_quality()
|
||
if quality_action:
|
||
return quality_action
|
||
res = super(StockPicking, self).button_validate()
|
||
return res
|
||
|
||
def pinking_checkout_quality(self):
|
||
"""
|
||
调拨单若关联了质量检查单，验证调拨单时，应校验是否有不合格品，若存在，应弹窗提示：
|
||
“警告：存在不合格产品XXXX n 件、YYYYY m件，继续调拨请点“确认”，否则请取消？”
|
||
"""
|
||
try:
|
||
self.ensure_one()
|
||
context = self.env.context
|
||
if not context.get('pinking_checkout_quality'):
|
||
picking_to_quality = self._get_picking_to_quality()
|
||
if not picking_to_quality: return False
|
||
need_quality_val = self.get_need_quality_line(picking_to_quality)
|
||
if any(line.get('fail_check_ids') for line in need_quality_val):
|
||
# 回滚事务，为二次确认/取消做准备
|
||
self.env.cr.rollback()
|
||
# 获取存在失败的 质检单 调拨单明细行
|
||
check_list = [item for item in need_quality_val if item.get('fail_check_ids')]
|
||
fail_check_text = ''
|
||
for item in check_list:
|
||
move_id, pre_done_qty = item.get('move_id'), item.get('pre_done_qty')
|
||
fail_check_text = (f'{fail_check_text}、{move_id.product_id.display_name} {pre_done_qty}件'
|
||
if fail_check_text != '' else f'{move_id.product_id.display_name} {pre_done_qty}件')
|
||
return {
|
||
'type': 'ir.actions.act_window',
|
||
'res_model': 'picking.validate.check.wizard',
|
||
'name': '质检不合格提示',
|
||
'view_mode': 'form',
|
||
'target': 'new',
|
||
'context': {
|
||
'default_picking_id': self.id,
|
||
'default_fail_check_text': f'警告：存在不合格产品{fail_check_text}，继续调拨请点“确认”，否则请取消？',
|
||
'pinking_checkout_quality': True}
|
||
}
|
||
else:
|
||
return False
|
||
except Exception as e:
|
||
logging.info('pinking_checkout_quality()方法报错：%s' % e)
|
||
raise ValidationError('调拨单验证质检单是否合格时报错,请联系管理员处理！！')
|
||
|
||
def get_need_quality_line(self, picking_to_quality):
|
||
"""
|
||
# 对需要进行质检，还没有质检完成的明细行进行统计
|
||
# 1、当【质量标准_控制方式】=“产品、作业”，仅校验“预完成数量”>0的产品行对应的质检单必须处理。
|
||
2、当【质量标准_控制方式】=“数量”时，仅校验“预完成数量”>0的产品行对应的质检单必须处理：
|
||
1）每一类的“总单数”=【调拨单_需求】时，则已处理的质检单“单数”≥“预完成数量”时，可执行调拨单验证；
|
||
2）每一类的“总单数”<【调拨单_需求】时，则已处理的质检单“单数”≥0时，可执行调拨单验证；
|
||
"""
|
||
res = []
|
||
for item in picking_to_quality:
|
||
need_done_check_ids = self.env['quality.check']
|
||
fail_check_ids = self.env['quality.check']
|
||
move_id, pre_done_qty, check_ids = item.values()
|
||
check_ids_1 = check_ids.filtered(lambda qc: qc.measure_on in ('operation', 'product'))
|
||
if check_ids_1:
|
||
check_ids_1_done = check_ids_1.filtered(lambda qc: qc.quality_state in ('pass', 'fail'))
|
||
check_ids_1_fail = check_ids_1.filtered(lambda qc: qc.quality_state == 'fail')
|
||
check_ids_1_none = check_ids_1.filtered(lambda qc: qc.quality_state == 'none')
|
||
if check_ids_1 and not check_ids_1_done:
|
||
need_done_check_ids += check_ids_1_none
|
||
if check_ids_1_fail:
|
||
fail_check_ids += check_ids_1_fail
|
||
|
||
check_ids_2 = check_ids.filtered(lambda qc: qc.measure_on == 'move_line')
|
||
if check_ids_2:
|
||
check_ids_2_done = check_ids_2.filtered(lambda qc: qc.quality_state in ('pass', 'fail'))
|
||
check_ids_2_fail = check_ids_2.filtered(lambda qc: qc.quality_state == 'fail')
|
||
check_ids_2_none = check_ids_2.filtered(lambda qc: qc.quality_state == 'none')
|
||
# 每一类的“总单数”=【调拨单_需求】时，则已处理的质检单“单数”≥“预完成数量”时，可执行调拨单验证；
|
||
if len(check_ids_2) >= move_id.product_uom_qty and len(check_ids_2_done) < pre_done_qty:
|
||
need_done_check_ids += check_ids_2_none
|
||
# 每一类的“总单数”<【调拨单_需求】时，则已处理的质检单“单数”≥0时，可执行调拨单验证
|
||
elif len(check_ids_2) < move_id.product_uom_qty and len(check_ids_2_done) == 0:
|
||
need_done_check_ids += check_ids_2_none
|
||
if check_ids_2_fail:
|
||
fail_check_ids += check_ids_2_fail
|
||
|
||
if need_done_check_ids or fail_check_ids:
|
||
res.append({'move_id': move_id,
|
||
'pre_done_qty': pre_done_qty,
|
||
'check_ids': check_ids,
|
||
'fail_check_ids': fail_check_ids,
|
||
'need_done_check_ids': need_done_check_ids})
|
||
return res
|
||
|
||
def get_picking_to_quality(self):
|
||
self.ensure_one()
|
||
return self._get_picking_to_quality()
|
||
|
||
def _get_picking_to_quality(self):
|
||
"""
|
||
对需要质检的明细行进行统计(针对“预完成数量”>0的行)
|
||
"""
|
||
quality_piking_line_list = []
|
||
pre_done_qty_lines = self._get_pinking_pre_done_qty()
|
||
for line in pre_done_qty_lines:
|
||
move_id, pre_done_qty = line.values()
|
||
if pre_done_qty == 0:
|
||
continue
|
||
product_id = move_id.product_id
|
||
check_ids = self.check_ids.filtered(lambda c: c.product_id == product_id)
|
||
quality_piking_line_list.append({'move_id': move_id, 'pre_done_qty': pre_done_qty, 'check_ids': check_ids})
|
||
return quality_piking_line_list
|
||
|
||
def _get_pinking_pre_done_qty(self):
|
||
"""
|
||
return: 明细行 及 预完成数量
|
||
1、若调拨单所有明细行的【完成】=0，且任意行的【预留】<【需求】，则验证时将会话是否需创建欠单。
|
||
---->此时“预完成数量”=【预留】
|
||
2、若调拨单任意行的0<【完成】<【需求】，且则验证时将会话是否需创建欠单
|
||
---->此时“预完成数量”=【完成】
|
||
"""
|
||
# if all(move_id.quantity_done == 0 for move_id in self.move_ids_without_package):
|
||
# pre_done_qty = [{'move_id': move_id, 'pre_done_qty': move_id.reserved_availability} for move_id in
|
||
# self.move_ids_without_package]
|
||
# else:
|
||
# pre_done_qty = [{'move_id': move_id, 'pre_done_qty': move_id.quantity_done} for move_id in
|
||
# self.move_ids_without_package]
|
||
pre_done_qty = []
|
||
for move_id in self.move_ids_without_package:
|
||
if move_id.quantity_done > 0:
|
||
pre_done_qty.append({'move_id': move_id, 'pre_done_qty': move_id.quantity_done})
|
||
else:
|
||
pre_done_qty.append({'move_id': move_id, 'pre_done_qty': move_id.reserved_availability})
|
||
return pre_done_qty
|