# -*- coding: utf-8 -*- # Copyright 2026 Nexa Systems Inc. # License OPL-1 (Odoo Proprietary License v1.0) # # sale.order.action_confirm hook — creates fp.job records on confirm. # Sub 11 (2026-04-26) removed MRP entirely; fp.job is the only fulfilment # path. The former x_fc_use_native_jobs migration toggle was dropped in # 19.0.8.19.0 once the legacy bridge_mrp fallback became unreachable. import logging from markupsafe import Markup from odoo import _, api, fields, models _logger = logging.getLogger(__name__) class SaleOrder(models.Model): _inherit = 'sale.order' x_fc_fp_job_count = fields.Integer( string='Plating Jobs', compute='_compute_fp_job_count', ) x_fc_fp_certificate_count = fields.Integer( string='Certificates', compute='_compute_fp_certificate_count', help='Number of fp.certificate records issued (or draft) against ' 'this sale order. Surfaced as a smart button so Sarah/Tom ' 'can jump straight from the SO to the cert without having ' 'to drill through the linked Plating Job first.', ) # ------------------------------------------------------------------ # Phase 4 (Sub 11) — workflow-stage field + assigned-manager field # relocated from fusion_plating_bridge_mrp. Field re-declared with # the same selection + compute pointer; jobs is now the source of # truth so Phase 5 can delete bridge_mrp without losing the field. # ------------------------------------------------------------------ x_fc_workflow_stage = fields.Selection( [ ('draft', 'Quote'), ('awaiting_parts', 'Parts'), ('inspecting', 'Inspecting'), ('accept_parts', 'Accept'), ('assign_work', 'Assign'), ('in_production', 'Production'), ('ready_to_ship', 'Ready'), ('shipped', 'Shipped'), ('invoicing', 'Invoicing'), ('paid', 'Paid'), ('complete', 'Done'), ('cancelled', 'Cancelled'), ], compute='_compute_workflow_stage', string='Workflow Stage', help='Current position in the SO → Ship → Invoice workflow. ' 'Drives which next-step button is shown on the SO header.', ) x_fc_assigned_manager_id = fields.Many2one( 'res.users', string='Assigned Manager', help='The manager responsible for this job. Set when the job ' 'is confirmed (falls back to the salesperson).', tracking=True, ) def _compute_fp_job_count(self): Job = self.env['fp.job'].sudo() for so in self: so.x_fc_fp_job_count = Job.search_count( [('sale_order_id', '=', so.id)] ) def _compute_fp_certificate_count(self): Cert = self.env['fp.certificate'].sudo() for so in self: so.x_fc_fp_certificate_count = Cert.search_count( [('sale_order_id', '=', so.id)] ) def _compute_workflow_stage(self): """Walk fp.job state to derive the SO workflow banner.""" Job = self.env['fp.job'] Delivery = self.env.get('fusion.plating.delivery') for so in self: if so.state == 'cancel': so.x_fc_workflow_stage = 'cancelled' continue if so.state in ('draft', 'sent'): so.x_fc_workflow_stage = 'draft' continue jobs = Job.search([('sale_order_id', '=', so.id)]) all_jobs_done = bool(jobs) and all( j.state == 'done' for j in jobs ) shipped = False if Delivery is not None and jobs: if 'x_fc_job_id' in Delivery._fields: shipped = bool(Delivery.search_count([ ('x_fc_job_id', 'in', jobs.ids), ('state', '=', 'delivered'), ])) posted_invoices = so.invoice_ids.filtered( lambda i: i.state == 'posted' ) has_posted_invoice = bool(posted_invoices) all_paid = has_posted_invoice and all( i.payment_state in ('paid', 'in_payment') for i in posted_invoices ) if shipped and all_paid: so.x_fc_workflow_stage = 'complete' continue if all_paid and not shipped: so.x_fc_workflow_stage = 'paid' continue if shipped and has_posted_invoice: so.x_fc_workflow_stage = 'invoicing' continue if shipped: so.x_fc_workflow_stage = 'shipped' continue if all_jobs_done: so.x_fc_workflow_stage = 'ready_to_ship' continue recv_status = so.x_fc_receiving_status or 'not_received' if recv_status == 'not_received': so.x_fc_workflow_stage = 'awaiting_parts' continue if recv_status in ('partial', 'received'): so.x_fc_workflow_stage = 'inspecting' continue if recv_status == 'inspected': if not so.x_fc_assigned_manager_id and not jobs: so.x_fc_workflow_stage = 'assign_work' continue so.x_fc_workflow_stage = 'in_production' continue so.x_fc_workflow_stage = ( 'in_production' if jobs else 'awaiting_parts' ) def action_view_fp_jobs(self): self.ensure_one() jobs = self.env['fp.job'].search([('sale_order_id', '=', self.id)]) action = { 'type': 'ir.actions.act_window', 'name': _('Plating Jobs'), 'res_model': 'fp.job', 'view_mode': 'list,form', 'domain': [('sale_order_id', '=', self.id)], 'context': {'default_sale_order_id': self.id}, } if len(jobs) == 1: action.update({ 'view_mode': 'form', 'res_id': jobs.id, }) return action def action_view_fp_certificates(self): """Smart-button target — open the certificate(s) linked to this SO. One cert → form view; many → list view filtered to this SO.""" self.ensure_one() certs = self.env['fp.certificate'].search([ ('sale_order_id', '=', self.id), ]) action = { 'type': 'ir.actions.act_window', 'name': _('Certificates'), 'res_model': 'fp.certificate', 'view_mode': 'list,form', 'domain': [('sale_order_id', '=', self.id)], 'context': { 'default_sale_order_id': self.id, 'default_partner_id': self.partner_id.id, }, } if len(certs) == 1: action.update({'view_mode': 'form', 'res_id': certs.id}) return action def action_confirm(self): result = super().action_confirm() for so in self: so._fp_auto_create_job() # Auto-confirm any draft jobs we just created so steps # generate immediately (no manager click required). # Best-effort: an exception in side-effects shouldn't # block the SO confirm itself. draft_jobs = self.env['fp.job'].sudo().search([ ('sale_order_id', '=', so.id), ('state', '=', 'draft'), ]) for job in draft_jobs: try: job.action_confirm() except Exception as exc: so.message_post(body=_( 'Auto-confirm of fp.job %(job)s failed: %(err)s. ' 'Confirm manually from the job form.' ) % {'job': job.name, 'err': exc}) return result def _fp_auto_create_job(self): """Create fp.job(s) from the SO's plating lines. Lines that share a `x_fc_wo_group_tag` collapse into one job; untagged lines get one job per line. Mirrors bridge_mrp's _fp_auto_create_mo grouping logic. """ self.ensure_one() Job = self.env['fp.job'].sudo() # Idempotency: skip if a job already references this SO existing = Job.search([('sale_order_id', '=', self.id)], limit=1) if existing: return # Find plating lines (those with a part_catalog_id or coating_config_id) plating_lines = self.order_line.filtered( lambda l: ( ('x_fc_part_catalog_id' in l._fields and l.x_fc_part_catalog_id) or ('x_fc_coating_config_id' in l._fields and l.x_fc_coating_config_id) ) ) # Fallback: legacy/configurator SOs that carry part+coating on the # header but not on the line. Treat the entire order as one # plating line so the planner gets an fp.job to work against. if not plating_lines and self.order_line and ( ('x_fc_part_catalog_id' in self._fields and self.x_fc_part_catalog_id) or ('x_fc_coating_config_id' in self._fields and self.x_fc_coating_config_id) ): _logger.info( 'SO %s: no line-level part/coating but header carries one — ' 'treating all lines as a single plating job.', self.name, ) plating_lines = self.order_line if not plating_lines: _logger.info('SO %s: no plating lines, skipping job creation.', self.name) return # Group by x_fc_wo_group_tag (untagged → distinct group per line) groups = {} # tag → recordset of lines untagged_idx = 0 for line in plating_lines: tag = ( 'x_fc_wo_group_tag' in line._fields and line.x_fc_wo_group_tag ) or False if not tag: untagged_idx += 1 tag = '__untagged_%d' % untagged_idx groups[tag] = groups.get(tag, self.env['sale.order.line']) | line # Create a job per group for tag, lines in groups.items(): first_line = lines[0] qty = sum(lines.mapped('product_uom_qty')) part = ( 'x_fc_part_catalog_id' in first_line._fields and first_line.x_fc_part_catalog_id or False ) coating = ( 'x_fc_coating_config_id' in first_line._fields and first_line.x_fc_coating_config_id or False ) # Header fallback for legacy/configurator SOs that put part + # coating on the SO header instead of the line. if not part and 'x_fc_part_catalog_id' in self._fields: part = self.x_fc_part_catalog_id or False if not coating and 'x_fc_coating_config_id' in self._fields: coating = self.x_fc_coating_config_id or False # Recipe lookup priority: # 1. line.x_fc_process_variant_id — Sarah explicitly picked # a part-scoped variant on this order line. Always wins. # 2. coating.recipe_id — coating-config recipe. # 3. part.default_process_id — part's flagged default. # 4. part.recipe_id — legacy fallback. # # If multiple lines in the same WO group have different # variants we use the FIRST line's variant (consistent with # everything else in this loop using `first_line`). recipe = False picked_variant = ( 'x_fc_process_variant_id' in first_line._fields and first_line.x_fc_process_variant_id or False ) if picked_variant: recipe = picked_variant if not recipe and coating and 'recipe_id' in coating._fields \ and coating.recipe_id: recipe = coating.recipe_id if not recipe and part and 'default_process_id' in part._fields \ and part.default_process_id: recipe = part.default_process_id if not recipe and part and 'recipe_id' in part._fields \ and part.recipe_id: recipe = part.recipe_id vals = { 'partner_id': self.partner_id.id, 'product_id': first_line.product_id.id if first_line.product_id else False, 'qty': qty, 'origin': self.name, 'sale_order_id': self.id, 'sale_order_line_ids': [(6, 0, lines.ids)], 'date_deadline': self.commitment_date or self.date_order, } if part: vals['part_catalog_id'] = part.id if coating: vals['coating_config_id'] = coating.id if recipe: vals['recipe_id'] = recipe.id # Customer spec / facility / manager — copy from SO if present if 'x_fc_customer_spec_id' in self._fields and self.x_fc_customer_spec_id: vals['customer_spec_id'] = self.x_fc_customer_spec_id.id if 'x_fc_facility_id' in self._fields and self.x_fc_facility_id: vals['facility_id'] = self.x_fc_facility_id.id if 'x_fc_manager_id' in self._fields and self.x_fc_manager_id: vals['manager_id'] = self.x_fc_manager_id.id # Quoted revenue: sum line totals vals['quoted_revenue'] = sum(lines.mapped('price_subtotal')) job = Job.create(vals) _logger.info( 'SO %s: created fp.job %s (qty=%s, recipe=%s)', self.name, job.name, qty, (recipe.name if recipe else '-'), ) return True # ------------------------------------------------------------------ # Phase 4 (Sub 11) — workflow stage action buttons. # Native versions of bridge_mrp's action_fp_* methods. Drop the # mrp.production lookups; talk to fp.job and fp.receiving directly. # ------------------------------------------------------------------ def action_fp_mark_inspected(self): """Flip open receivings from draft → inspecting.""" self.ensure_one() Recv = self.env.get('fp.receiving') if Recv is None: return False for rec in Recv.search([('sale_order_id', '=', self.id)]): if rec.state == 'draft': rec.state = 'inspecting' self.message_post(body=_('Parts marked as inspecting.')) return True def action_fp_accept_parts(self): """Mark receiving accepted; flip SO receiving status to inspected.""" self.ensure_one() Recv = self.env.get('fp.receiving') if Recv is None: return False for rec in Recv.search([('sale_order_id', '=', self.id)]): if rec.state in ('draft', 'inspecting'): rec.state = 'accepted' if 'x_fc_receiving_status' in self._fields: self.x_fc_receiving_status = 'inspected' self.message_post(body=_('Parts accepted — ready to assign manager.')) return True def action_fp_assign_to_me(self): """Manager claims the SO and confirms its draft fp.jobs.""" self.ensure_one() user = self.env.user self.x_fc_assigned_manager_id = user.id Job = self.env['fp.job'] jobs = Job.search([ ('sale_order_id', '=', self.id), ('state', '=', 'draft'), ]) for job in jobs: try: job.action_confirm() except Exception as exc: self.message_post(body=_( 'Auto-confirm of fp.job %s failed: %s' ) % (job.name, exc)) if 'manager_id' in job._fields and not job.manager_id: job.manager_id = user.id self.message_post( body=Markup(_( 'Job assigned to %s. %d plating job(s) released to the floor.' )) % (user.name, len(jobs)), ) return True def action_fp_mark_shipped(self): """Mark linked deliveries delivered (triggers auto-invoice).""" self.ensure_one() Delivery = self.env.get('fusion.plating.delivery') if Delivery is None: return False Job = self.env['fp.job'] jobs = Job.search([('sale_order_id', '=', self.id)]) deliveries = Delivery.browse([]) if 'x_fc_job_id' in Delivery._fields: deliveries = Delivery.search([ ('x_fc_job_id', 'in', jobs.ids), ('state', '!=', 'delivered'), ]) for dlv in deliveries: dlv.action_mark_delivered() self.message_post( body=_( '%d delivery record(s) marked delivered. ' 'Invoice flow triggered per invoice strategy.' ) % len(deliveries), ) return True def action_fp_open_shop_floor(self): """Jump to the Plant Overview filtered to this SO's jobs.""" self.ensure_one() return { 'type': 'ir.actions.client', 'tag': 'fp_plant_overview', 'name': _('Shop Floor — %s') % self.name, 'target': 'current', }