from collections import defaultdict from odoo import _, api, fields, models, Command from odoo.exceptions import AccessError, UserError class IfcProject(models.Model): _inherit = ['grt.ifc.project', 'mail.thread', 'mail.activity.mixin'] document_ref = fields.Char( string='Document No', required=True, copy=False, default=lambda self: _('New'), readonly=True, tracking=True, ) uploaded_date = fields.Datetime( string='Uploaded On', default=fields.Datetime.now, readonly=True, tracking=True, ) uploaded_by_id = fields.Many2one( 'res.users', string='Uploaded By', default=lambda self: self.env.user, readonly=True, tracking=True, ) last_update = fields.Datetime(string='Last Update', related='write_date', store=True, readonly=True) revision_note_input = fields.Text(string='Revision Notes') revision_ids = fields.One2many( 'grt.ifc.document.revision', 'project_id', string='Document Revisions', readonly=True, copy=False, ) last_revision_id = fields.Many2one( 'grt.ifc.document.revision', string='Last Revision', compute='_compute_last_revision', store=True, ) last_revision_label = fields.Char(string='Last Revision Label', related='last_revision_id.revision_label', store=True) last_revision_note = fields.Text(string='Last Revision Notes', related='last_revision_id.revision_note', store=True) revised_by_id = fields.Many2one('res.users', string='Revised By', related='last_revision_id.revised_by_id', store=True) last_revision_date = fields.Datetime(string='Last Revision Date', related='last_revision_id.revision_date', store=True) approval_state = fields.Selection( [ ('draft', 'Draft'), ('in_validation', 'In Validation'), ('waiting_production', 'Waiting Production Approval'), ('waiting_provisioning', 'Waiting Provisioning Approval'), ('approved', 'Approved'), ('rejected', 'Rejected'), ], default='draft', required=True, tracking=True, index=True, string='Approval Status', ) approved_production_by_id = fields.Many2one('res.users', string='Approved Production By', readonly=True) approved_production_at = fields.Datetime(string='Production Approved At', readonly=True) approved_provisioning_by_id = fields.Many2one('res.users', string='Approved Provisioning By', readonly=True) approved_provisioning_at = fields.Datetime(string='Provisioning Approved At', readonly=True) rejected_by_id = fields.Many2one('res.users', string='Rejected By', readonly=True) rejected_at = fields.Datetime(string='Rejected At', readonly=True) history_ids = fields.One2many( 'grt.ifc.activity.history', 'project_id', string='Activity History', readonly=True, copy=False, ) estimated_product_tmpl_id = fields.Many2one('product.template', string='Estimated Main Product') estimated_bom_id = fields.Many2one('mrp.bom', string='Estimated BoM', copy=False, readonly=True) permanent_bom_id = fields.Many2one('mrp.bom', string='Permanent BoM', copy=False, readonly=True) estimated_bom_count = fields.Integer(compute='_compute_estimated_bom_count', string='Estimated BoM Count') manufacturing_order_id = fields.Many2one('mrp.production', string='Latest Manufacturing Order', copy=False, readonly=True) manufacturing_order_ids = fields.One2many('mrp.production', 'ifc_project_id', string='Manufacturing Orders', readonly=True) @api.depends('revision_ids.revision_number', 'revision_ids.is_current') def _compute_last_revision(self): for record in self: current = record.revision_ids.filtered(lambda rev: rev.is_current)[:1] if current: record.last_revision_id = current.id continue ordered = record.revision_ids.sorted(key=lambda rev: rev.revision_number, reverse=True) record.last_revision_id = ordered[:1].id if ordered else False @api.depends('estimated_bom_id') def _compute_estimated_bom_count(self): for record in self: record.estimated_bom_count = 1 if record.estimated_bom_id else 0 @api.model_create_multi def create(self, vals_list): for vals in vals_list: if not vals.get('document_ref') or vals['document_ref'] == _('New'): vals['document_ref'] = self.env['ir.sequence'].next_by_code('grt.ifc.project.document') or _('New') vals.setdefault('uploaded_date', fields.Datetime.now()) vals.setdefault('uploaded_by_id', self.env.user.id) records = super().create(vals_list) for record in records: if record.file_data: record._create_revision_entry(_('Dokumen awal diunggah.')) record._log_history('upload', _('Upload dokumen IFC awal.')) return records def write(self, vals): update_file = 'file_data' in vals and not self.env.context.get('skip_revision_auto') revision_note = vals.get('revision_note_input') or _('File IFC diperbarui.') result = super().write(vals) if update_file: for record in self: if record.file_data: record._create_revision_entry(revision_note) record._log_history('revision', revision_note) return result def action_parse_ifc(self): result = super().action_parse_ifc() for record in self: record._log_history('parse', _('IFC parsing dijalankan.')) return result def action_request_validation(self): for record in self: if record.approval_state not in ('draft', 'rejected'): raise UserError(_('Dokumen hanya dapat divalidasi dari status Draft atau Rejected.')) if record.state != 'parsed': raise UserError(_('Dokumen harus diparse dahulu sebelum validasi.')) record.write({'approval_state': 'in_validation'}) record._schedule_activity( record.uploaded_by_id, _('Lakukan validasi dokumen IFC %s') % (record.document_ref,), _('Validasi teknis dokumen dan pastikan data IFC siap untuk approval.'), ) record._log_history('validation', _('Permintaan validasi dikirim.')) def action_submit_for_approval(self): for record in self: if record.approval_state not in ('in_validation', 'draft', 'rejected'): raise UserError(_('Dokumen tidak dapat diajukan dari status saat ini.')) if record.state != 'parsed': raise UserError(_('Dokumen harus diparse sebelum diajukan approval.')) record.write({'approval_state': 'waiting_production'}) production_user = record._get_first_user_of_group('grt_product_engineering.group_ifc_manager_production') if production_user: record._schedule_activity( production_user, _('Approval Produksi untuk %s') % (record.document_ref,), _('Periksa dokumen IFC dan berikan approval tahap Produksi.'), ) record._log_history('validation', _('Dokumen diajukan ke approval Manager Produksi.')) def action_approve_production(self): self._ensure_group('grt_product_engineering.group_ifc_manager_production', _('Hanya Manager Produksi yang dapat melakukan approval tahap ini.')) for record in self: if record.approval_state != 'waiting_production': raise UserError(_('Dokumen tidak berada pada tahap approval Produksi.')) record._mark_user_activities_done(_('Approved by Production Manager.')) record.write( { 'approval_state': 'waiting_provisioning', 'approved_production_by_id': self.env.user.id, 'approved_production_at': fields.Datetime.now(), } ) provisioning_user = record._get_first_user_of_group('grt_product_engineering.group_ifc_manager_provisioning') if provisioning_user: record._schedule_activity( provisioning_user, _('Approval Provisioning untuk %s') % (record.document_ref,), _('Dokumen telah lolos approval Produksi. Lanjutkan approval Provisioning.'), ) record._log_history('approval_production', _('Dokumen disetujui Manager Produksi.')) def action_approve_provisioning(self): self._ensure_group('grt_product_engineering.group_ifc_manager_provisioning', _('Hanya Manager Provisioning yang dapat melakukan approval tahap ini.')) for record in self: if record.approval_state != 'waiting_provisioning': raise UserError(_('Dokumen tidak berada pada tahap approval Provisioning.')) record._mark_user_activities_done(_('Approved by Provisioning Manager.')) record.write( { 'approval_state': 'approved', 'approved_provisioning_by_id': self.env.user.id, 'approved_provisioning_at': fields.Datetime.now(), } ) record._log_history('approval_provisioning', _('Dokumen disetujui Manager Provisioning.')) def action_reject_document(self): self._ensure_group_any( [ 'grt_product_engineering.group_ifc_manager_production', 'grt_product_engineering.group_ifc_manager_provisioning', ], _('Hanya Manager Produksi atau Manager Provisioning yang dapat menolak dokumen.'), ) for record in self: if record.approval_state not in ('waiting_production', 'waiting_provisioning', 'in_validation'): raise UserError(_('Dokumen tidak berada pada tahap yang dapat ditolak.')) record._mark_user_activities_done(_('Dokumen ditolak dan perlu revisi.')) record.write( { 'approval_state': 'rejected', 'rejected_by_id': self.env.user.id, 'rejected_at': fields.Datetime.now(), } ) note = record.revision_note_input or _('Dokumen ditolak. Harap lakukan revisi.') record._log_history('rejected', note) if record.uploaded_by_id: record._schedule_activity( record.uploaded_by_id, _('Dokumen ditolak: %s') % (record.document_ref,), _('Lakukan revisi dokumen sesuai catatan penolakan.'), ) def action_back_to_draft(self): for record in self: record.write({'approval_state': 'draft'}) record._log_history('validation', _('Dokumen dikembalikan ke Draft untuk perbaikan.')) def action_create_bom_estimate(self): unit_uom = self.env.ref('uom.product_uom_unit') for record in self: if record.state != 'parsed': raise UserError(_('BoM estimasi hanya bisa dibuat untuk dokumen yang sudah parsed.')) if not record.bom_ids: raise UserError(_('Data komponen IFC belum ada. Jalankan parse IFC terlebih dahulu.')) main_template = record.estimated_product_tmpl_id or record._get_or_create_main_product_template(unit_uom) if not main_template: raise UserError(_('Produk utama estimasi tidak dapat dibuat.')) component_map = defaultdict(float) for bom_line in record.bom_ids: if bom_line.level == 0 and bom_line.parent_id: continue component_product = record._get_or_create_component_product(bom_line, unit_uom) if component_product: qty = bom_line.quantity if bom_line.quantity and bom_line.quantity > 0 else 1.0 component_map[component_product.id] += qty if not component_map: raise UserError(_('Komponen untuk BoM estimasi tidak ditemukan.')) line_commands = [Command.clear()] for product_id, quantity in component_map.items(): component = self.env['product.product'].browse(product_id) line_commands.append( Command.create( { 'product_id': component.id, 'product_qty': quantity, 'product_uom_id': component.uom_id.id, } ) ) bom_vals = { 'product_tmpl_id': main_template.id, 'product_id': main_template.product_variant_id.id, 'product_qty': 1.0, 'code': '%s-%s' % (record.document_ref, record.last_revision_label or 'R00'), 'type': 'normal', 'company_id': record.company_id.id, 'ifc_project_id': record.id, 'line_ids': line_commands, } if record.estimated_bom_id: record.estimated_bom_id.write(bom_vals) estimated_bom = record.estimated_bom_id else: estimated_bom = self.env['mrp.bom'].create(bom_vals) record.estimated_product_tmpl_id = main_template.id record.estimated_bom_id = estimated_bom.id record.permanent_bom_id = False record._log_history('bom_estimate', _('BoM estimasi dibuat/diperbarui.')) def action_create_permanent_bom_if_available(self): self.ensure_one() if not self.estimated_bom_id: raise UserError(_('Belum ada Estimated BoM. Jalankan generate BoM terlebih dahulu.')) availability = self._compute_inventory_availability() if not availability.get('all_available'): raise UserError(_('Estimated BoM belum terpenuhi availability-nya. Permanent BoM belum dapat dibuat.')) estimated_bom = self.estimated_bom_id product = estimated_bom.product_id or estimated_bom.product_tmpl_id.product_variant_id if not product: raise UserError(_('Produk utama pada Estimated BoM tidak ditemukan.')) line_commands = [Command.clear()] for line in estimated_bom.line_ids: line_commands.append( Command.create( { 'product_id': line.product_id.id, 'product_qty': line.product_qty, 'product_uom_id': line.product_uom_id.id, } ) ) permanent_vals = { 'product_tmpl_id': estimated_bom.product_tmpl_id.id, 'product_id': product.id, 'product_qty': estimated_bom.product_qty or 1.0, 'code': '%s-%s-PERM' % (self.document_ref, self.last_revision_label or 'R00'), 'type': estimated_bom.type, 'company_id': self.company_id.id, 'ifc_project_id': self.id, 'is_ifc_permanent': True, 'source_estimated_bom_id': estimated_bom.id, 'line_ids': line_commands, } if self.permanent_bom_id: self.permanent_bom_id.write(permanent_vals) permanent_bom = self.permanent_bom_id else: permanent_bom = self.env['mrp.bom'].create(permanent_vals) self.permanent_bom_id = permanent_bom.id self._log_history('bom_estimate', _('Permanent BoM dibuat/diperbarui dari Estimated BoM.')) return permanent_bom def action_view_estimated_bom(self): self.ensure_one() if not self.estimated_bom_id: raise UserError(_('Belum ada BoM estimasi.')) return { 'type': 'ir.actions.act_window', 'name': _('Estimated BoM'), 'res_model': 'mrp.bom', 'view_mode': 'form', 'res_id': self.estimated_bom_id.id, 'target': 'current', } def _compute_inventory_availability(self): self.ensure_one() if not self.estimated_bom_id: raise UserError(_('Belum ada Estimated BoM. Jalankan generate BoM terlebih dahulu.')) items = [] total_required = 0.0 total_on_hand = 0.0 total_shortage = 0.0 for line in self.estimated_bom_id.line_ids: product = line.product_id required_qty = float(line.product_qty or 0.0) on_hand_qty = float(product.qty_available or 0.0) if product else 0.0 forecast_qty = float(product.virtual_available or 0.0) if product else 0.0 shortage_qty = max(required_qty - on_hand_qty, 0.0) items.append( { 'bom_line_id': line.id, 'product_id': product.id if product else False, 'product_name': product.display_name if product else line.name, 'default_code': product.default_code if product else False, 'required_qty': required_qty, 'on_hand_qty': on_hand_qty, 'forecast_qty': forecast_qty, 'shortage_qty': shortage_qty, 'is_available': shortage_qty <= 0, 'uom_id': line.product_uom_id.id if line.product_uom_id else False, 'uom_name': line.product_uom_id.name if line.product_uom_id else False, } ) total_required += required_qty total_on_hand += on_hand_qty total_shortage += shortage_qty return { 'document_id': self.id, 'document_ref': self.document_ref, 'estimated_bom_id': self.estimated_bom_id.id, 'total_required_qty': total_required, 'total_on_hand_qty': total_on_hand, 'total_shortage_qty': total_shortage, 'all_available': total_shortage <= 0, 'items': items, } def action_check_inventory_availability(self): self.ensure_one() availability = self._compute_inventory_availability() self._log_history('validation', _('Cek ketersediaan inventory untuk Estimated BoM dilakukan.')) return availability def _select_purchase_vendor(self, product, forced_vendor=False): self.ensure_one() if forced_vendor: return forced_vendor seller = product.seller_ids[:1] return seller.partner_id if seller else False def action_create_draft_purchase_for_shortage(self, vendor_id=False, only_shortage=True): self.ensure_one() availability = self._compute_inventory_availability() purchase_orders_by_vendor = {} no_vendor_items = [] forced_vendor = self.env['res.partner'].browse(int(vendor_id)).exists() if vendor_id else False PurchaseOrder = self.env['purchase.order'] for item in availability['items']: shortage_qty = item['shortage_qty'] required_qty = item['required_qty'] if only_shortage and shortage_qty <= 0: continue order_qty = shortage_qty if only_shortage else required_qty if order_qty <= 0: continue product = self.env['product.product'].browse(item['product_id']).exists() if not product: continue vendor = self._select_purchase_vendor(product, forced_vendor=forced_vendor) if not vendor: no_vendor_items.append(product.display_name) continue po = purchase_orders_by_vendor.get(vendor.id) if not po: po = PurchaseOrder.create( { 'partner_id': vendor.id, 'company_id': self.company_id.id, 'origin': '%s - %s' % (self.document_ref, _('IFC Estimated BoM Shortage')), 'notes': _('Auto-generated dari Product Engineering IFC.'), } ) purchase_orders_by_vendor[vendor.id] = po seller = product.seller_ids.filtered(lambda s: s.partner_id.id == vendor.id)[:1] price_unit = seller.price if seller else (product.standard_price or 0.0) self.env['purchase.order.line'].create( { 'order_id': po.id, 'product_id': product.id, 'name': product.display_name, 'product_qty': order_qty, 'product_uom': product.uom_po_id.id or product.uom_id.id, 'price_unit': price_unit, 'date_planned': fields.Datetime.now(), } ) if no_vendor_items: raise UserError( _('Vendor belum ditentukan untuk beberapa produk: %s') % ', '.join(no_vendor_items) ) purchase_orders = list(purchase_orders_by_vendor.values()) if not purchase_orders: raise UserError(_('Tidak ada item yang perlu dibuatkan draft purchase.')) self._log_history('validation', _('Draft purchase dibuat dari shortage Estimated BoM.')) return purchase_orders def action_create_missing_product_for_ifc_bom(self, ifc_bom_line_id): self.ensure_one() if not ifc_bom_line_id: raise UserError(_('ifc_bom_line_id wajib diisi.')) bom_line = self.env['grt.ifc.bom'].browse(int(ifc_bom_line_id)).exists() if not bom_line or bom_line.project_id.id != self.id: raise UserError(_('Baris IFC BOM tidak valid untuk dokumen ini.')) default_code = bom_line.global_id or bom_line.ifc_key if not default_code: raise UserError(_('Baris IFC tidak memiliki global_id atau ifc_key sebagai kode produk.')) product = self.env['product.product'].search([('default_code', '=', default_code)], limit=1) if product: return product, False unit_uom = self.env.ref('uom.product_uom_unit') template = self.env['product.template'].create( { 'name': bom_line.name, 'default_code': default_code, 'detailed_type': 'consu', 'uom_id': unit_uom.id, 'uom_po_id': unit_uom.id, } ) self._log_history('validation', _('Produk baru dibuat dari IFC BOM line: %s') % bom_line.name) return template.product_variant_id, True def _get_default_mrp_picking_type(self): self.ensure_one() return self.env['stock.picking.type'].search( [ ('code', '=', 'mrp_operation'), ('company_id', 'in', [self.company_id.id, False]), ], order='company_id desc, id', limit=1, ) def action_create_mo_if_available(self): self.ensure_one() if not self.permanent_bom_id: self.action_create_permanent_bom_if_available() availability = self._compute_inventory_availability() if not availability.get('all_available'): raise UserError(_('Stok belum terpenuhi untuk semua komponen. MO belum dapat dibuat.')) bom = self.permanent_bom_id product = bom.product_id or bom.product_tmpl_id.product_variant_id if not product: raise UserError(_('Produk utama pada Estimated BoM tidak ditemukan.')) mo_vals = { 'product_id': product.id, 'product_uom_id': product.uom_id.id, 'product_qty': bom.product_qty or 1.0, 'bom_id': bom.id, 'company_id': self.company_id.id, 'origin': '%s - %s' % (self.document_ref, _('IFC Engineering')), 'ifc_project_id': self.id, } picking_type = self._get_default_mrp_picking_type() if picking_type: mo_vals['picking_type_id'] = picking_type.id manufacturing_order = self.env['mrp.production'].create(mo_vals) self.manufacturing_order_id = manufacturing_order.id self._log_history('manufacturing_order', _('Manufacturing Order dibuat: %s') % manufacturing_order.name) return manufacturing_order def _compute_mo_requirements(self, manufacturing_order): requirements = [] bom = manufacturing_order.bom_id if not bom: return requirements factor = (manufacturing_order.product_qty or 0.0) / (bom.product_qty or 1.0) for line in bom.line_ids: required_qty = (line.product_qty or 0.0) * factor product = line.product_id on_hand_qty = float(product.qty_available or 0.0) if product else 0.0 requirements.append( { 'product_id': product.id if product else False, 'product_name': product.display_name if product else False, 'default_code': product.default_code if product else False, 'required_qty': required_qty, 'on_hand_qty': on_hand_qty, 'shortage_qty': max(required_qty - on_hand_qty, 0.0), 'uom_id': line.product_uom_id.id if line.product_uom_id else False, 'uom_name': line.product_uom_id.name if line.product_uom_id else False, } ) return requirements def _build_all_mo_availability_summary(self, raise_if_empty=True, log=False): self.ensure_one() if not self.manufacturing_order_ids: if raise_if_empty: raise UserError(_('Belum ada Manufacturing Order untuk dokumen ini.')) return { 'document_id': self.id, 'document_ref': self.document_ref, 'mo_count': 0, 'all_available': False, 'total_shortage_qty': 0.0, 'mo_items': [], 'product_summary': [], } active_mos = self.manufacturing_order_ids.filtered(lambda mo: mo.state not in ('cancel',)) if not active_mos: if raise_if_empty: raise UserError(_('Tidak ada Manufacturing Order aktif untuk dicek.')) return { 'document_id': self.id, 'document_ref': self.document_ref, 'mo_count': 0, 'all_available': False, 'total_shortage_qty': 0.0, 'mo_items': [], 'product_summary': [], } by_product = {} mo_summaries = [] for mo in active_mos: mo_items = self._compute_mo_requirements(mo) mo_shortage = 0.0 for item in mo_items: product_id = item['product_id'] if not product_id: continue agg = by_product.setdefault( product_id, { 'product_id': product_id, 'product_name': item['product_name'], 'default_code': item['default_code'], 'required_qty': 0.0, 'on_hand_qty': item['on_hand_qty'], 'uom_id': item['uom_id'], 'uom_name': item['uom_name'], }, ) agg['required_qty'] += item['required_qty'] mo_shortage += item['shortage_qty'] mo_summaries.append( { 'mo_id': mo.id, 'mo_name': mo.name, 'state': mo.state, 'product_id': mo.product_id.id, 'product_name': mo.product_id.display_name, 'required_items': mo_items, 'is_available': mo_shortage <= 0, } ) total_shortage = 0.0 product_summary = [] for data in by_product.values(): shortage_qty = max(data['required_qty'] - data['on_hand_qty'], 0.0) total_shortage += shortage_qty product_summary.append( { **data, 'shortage_qty': shortage_qty, 'is_available': shortage_qty <= 0, } ) result = { 'document_id': self.id, 'document_ref': self.document_ref, 'mo_count': len(active_mos), 'all_available': total_shortage <= 0, 'total_shortage_qty': total_shortage, 'mo_items': mo_summaries, 'product_summary': product_summary, } if log: self._log_history('manufacturing_order', _('Cek summary availability seluruh MO dijalankan.')) return result def action_check_all_mo_availability_summary(self): self.ensure_one() return self._build_all_mo_availability_summary(raise_if_empty=True, log=True) def action_confirm_all_mo_if_available(self): self.ensure_one() summary = self.action_check_all_mo_availability_summary() if not summary.get('all_available'): raise UserError(_('Tidak semua kebutuhan material MO terpenuhi. Confirm MO dibatalkan.')) confirmed = [] for mo in self.manufacturing_order_ids.filtered(lambda m: m.state in ('draft', 'confirmed')): if mo.state == 'draft': mo.action_confirm() confirmed.append(mo) if not confirmed: raise UserError(_('Tidak ada MO pada status yang dapat dikonfirmasi.')) self._log_history('manufacturing_order', _('Confirm seluruh MO berhasil dijalankan.')) return confirmed def action_run_mo_pipeline(self, auto_confirm=False): self.ensure_one() estimated_availability = self.action_check_inventory_availability() if not estimated_availability.get('all_available'): raise UserError( _('Estimated BoM belum available sepenuhnya. Selesaikan shortage sebelum jalankan pipeline MO.') ) permanent_bom = self.action_create_permanent_bom_if_available() manufacturing_order = self.action_create_mo_if_available() mo_summary = self.action_check_all_mo_availability_summary() confirmed_mos = [] if auto_confirm: confirmed_mos = self.action_confirm_all_mo_if_available() self._log_history('manufacturing_order', _('Pipeline MO dari Estimated BoM berhasil dijalankan.')) return { 'estimated_availability': estimated_availability, 'permanent_bom': permanent_bom, 'manufacturing_order': manufacturing_order, 'mo_summary': mo_summary, 'confirmed_mos': confirmed_mos, 'auto_confirm': bool(auto_confirm), } def action_preview_mo_pipeline(self): self.ensure_one() if not self.estimated_bom_id: raise UserError(_('Belum ada Estimated BoM. Jalankan generate BoM terlebih dahulu.')) estimated_availability = self._compute_inventory_availability() can_create_permanent_bom = bool(estimated_availability.get('all_available')) permanent_bom_exists = bool(self.permanent_bom_id) can_create_mo = can_create_permanent_bom or permanent_bom_exists existing_mo_summary = self._build_all_mo_availability_summary(raise_if_empty=False, log=False) can_confirm_all_existing_mos = bool(existing_mo_summary.get('mo_count')) and bool(existing_mo_summary.get('all_available')) return { 'document_id': self.id, 'document_ref': self.document_ref, 'estimated_bom_id': self.estimated_bom_id.id, 'permanent_bom_id': self.permanent_bom_id.id if self.permanent_bom_id else False, 'estimated_availability': estimated_availability, 'can_create_permanent_bom': can_create_permanent_bom, 'can_create_mo': can_create_mo, 'existing_mo_summary': existing_mo_summary, 'can_confirm_all_existing_mos': can_confirm_all_existing_mos, 'pipeline_steps': [ { 'step': 'check_estimated_availability', 'ready': True, }, { 'step': 'create_permanent_bom', 'ready': can_create_permanent_bom, }, { 'step': 'create_mo_from_permanent_bom', 'ready': can_create_mo, }, { 'step': 'check_all_mo_availability', 'ready': True, }, { 'step': 'confirm_all_mo_if_available', 'ready': can_confirm_all_existing_mos, }, ], } def _get_or_create_main_product_template(self, unit_uom): self.ensure_one() default_code = self.document_ref template = self.env['product.template'].search([('default_code', '=', default_code)], limit=1) if template: return template return self.env['product.template'].create( { 'name': self.name, 'default_code': default_code, 'detailed_type': 'product', 'uom_id': unit_uom.id, 'uom_po_id': unit_uom.id, } ) def _get_or_create_component_product(self, bom_line, unit_uom): default_code = bom_line.global_id or bom_line.ifc_key if not default_code: return False product = self.env['product.product'].search([('default_code', '=', default_code)], limit=1) if product: return product template = self.env['product.template'].create( { 'name': bom_line.name, 'default_code': default_code, 'detailed_type': 'consu', 'uom_id': unit_uom.id, 'uom_po_id': unit_uom.id, } ) return template.product_variant_id def _create_revision_entry(self, note): self.ensure_one() self.revision_ids.filtered(lambda rev: rev.is_current).write({'is_current': False}) next_revision = (max(self.revision_ids.mapped('revision_number')) + 1) if self.revision_ids else 1 self.env['grt.ifc.document.revision'].create( { 'project_id': self.id, 'revision_number': next_revision, 'revision_date': fields.Datetime.now(), 'revised_by_id': self.env.user.id, 'revision_note': note, 'file_data': self.file_data, 'file_name': self.file_name, 'is_current': True, } ) def _log_history(self, activity_type, note): self.ensure_one() self.env['grt.ifc.activity.history'].create( { 'project_id': self.id, 'activity_type': activity_type, 'user_id': self.env.user.id, 'note': note, 'state_snapshot': self.approval_state, } ) def _schedule_activity(self, user, summary, note): self.ensure_one() if not user: return self.activity_schedule( 'mail.mail_activity_data_todo', user_id=user.id, summary=summary, note=note, ) def _mark_user_activities_done(self, feedback): self.ensure_one() open_activities = self.activity_ids.filtered(lambda act: act.user_id == self.env.user) for activity in open_activities: activity.action_feedback(feedback=feedback) def _get_first_user_of_group(self, group_xmlid): group = self.env.ref(group_xmlid, raise_if_not_found=False) if not group or not group.users: return False return group.users[0] def _ensure_group(self, group_xmlid, error_message): if not self.env.user.has_group(group_xmlid): raise AccessError(error_message) def _ensure_group_any(self, group_xmlids, error_message): if not any(self.env.user.has_group(group_xmlid) for group_xmlid in group_xmlids): raise AccessError(error_message)