import logging from odoo import api, models, fields from odoo.exceptions import ValidationError, UserError class StockPicking(models.Model): _inherit = 'stock.picking' whether_show_quality_check = fields.Boolean('是否显示质量检测按钮', default=True) 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 picking.whether_show_quality_check = False break else: picking.whether_show_quality_check = True 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', '=', '出厂检验报告'), ('quality_state', '=', 'pass')]) # 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