# -*- 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 from odoo.exceptions import UserError _logger = logging.getLogger(__name__) class SaleOrder(models.Model): _inherit = 'sale.order' x_fc_fp_job_count = fields.Integer( string='Work Orders', 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.', ) # ------------------------------------------------------------------ # Parent-number hierarchy (2026-05-12 design) # See docs/superpowers/specs/2026-05-12-parent-number-hierarchy-design.md # ------------------------------------------------------------------ x_fc_parent_number = fields.Integer( string='Parent Number', readonly=True, copy=False, index=True, help='Set on confirm. Drives every linked document\'s name ' '(WO-NNN, IN-NNN, CoC-NNN, ...). Immutable post-assignment.', ) x_fc_quote_ref = fields.Char( string='Originally Quoted As', readonly=True, copy=False, help='The quote-stage name (e.g. Q202605-200). Preserved when ' 'the SO is renamed on confirm.', ) # Per-model counters — monotonic, never decrement. Source of truth # for the next sibling's x_fc_doc_index. Updated via row-locked SQL # in fp.parent.numbered.mixin so concurrent creates can't drift. # # Naming: `x_fc_pn_*_count` — the `pn_` infix distinguishes our # storage counters from pre-existing compute fields (e.g. the # `x_fc_delivery_count` compute in bridge_mrp, `x_fc_ncr_count` # in configurator, `x_fc_receiving_count` in fp_receiving) which # are surface counters for smart buttons. Distinct names avoid # the silent compute-override that made Tasks 3+9 fail until 9.5. x_fc_pn_wo_count = fields.Integer(string='Parent: WO Count', readonly=True, copy=False, default=0) x_fc_pn_invoice_count = fields.Integer(string='Parent: Invoice Count', readonly=True, copy=False, default=0) x_fc_pn_cn_count = fields.Integer(string='Parent: Credit Note Count', readonly=True, copy=False, default=0) x_fc_pn_cert_count = fields.Integer(string='Parent: Certificate Count', readonly=True, copy=False, default=0) x_fc_pn_delivery_count = fields.Integer(string='Parent: Delivery Count', readonly=True, copy=False, default=0) x_fc_pn_receiving_count = fields.Integer(string='Parent: Receiving Count', readonly=True, copy=False, default=0) x_fc_pn_pickup_count = fields.Integer(string='Parent: Pickup Count', readonly=True, copy=False, default=0) x_fc_pn_ncr_count = fields.Integer(string='Parent: NCR Count', readonly=True, copy=False, default=0) x_fc_pn_capa_count = fields.Integer(string='Parent: CAPA Count', readonly=True, copy=False, default=0) x_fc_pn_hold_count = fields.Integer(string='Parent: Hold Count', readonly=True, copy=False, default=0) x_fc_pn_rma_count = fields.Integer(string='Parent: RMA Count', readonly=True, copy=False, default=0) # ------------------------------------------------------------------ # 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 # ------------------------------------------------------------------ # Parent-number hierarchy — quote naming on create # ------------------------------------------------------------------ @api.model_create_multi def create(self, vals_list): """Draw Q-YYYYMM-N from fp.quote.number when no explicit name. The drawn name is also stashed in x_fc_quote_ref so it survives the confirm-time rename to SO-. If the caller passed an explicit name we preserve that AND mirror it into x_fc_quote_ref (covers data migration, restore, etc.). """ Seq = self.env['ir.sequence'] for vals in vals_list: existing = vals.get('name') if not existing or existing == _('New') or existing == 'New': quote_name = Seq.next_by_code('fp.quote.number') if quote_name: vals['name'] = quote_name vals.setdefault('x_fc_quote_ref', quote_name) elif not vals.get('x_fc_quote_ref'): vals['x_fc_quote_ref'] = existing return super().create(vals_list) def action_confirm(self): """Assign parent number + rename Q-…-N to SO-, then run the standard confirm (which kicks off WO creation). Parent number is drawn from fp.parent.number; the quote name was already saved to x_fc_quote_ref on create() so it survives the rename. Idempotent — if x_fc_parent_number is already set, the rename is skipped (re-confirm scenarios).""" Seq = self.env['ir.sequence'] for so in self: if so.x_fc_parent_number: continue parent = Seq.next_by_code('fp.parent.number') if not parent: raise UserError(_( 'Sequence fp.parent.number is missing. Reinstall ' 'fusion_plating to restore it.' )) parent_int = int(parent) old_name = so.name # fp_allow_name_rename whitelists this single legitimate # rename path through the immutability write() guard # (added in Task 11). so.with_context(fp_allow_name_rename=True).write({ 'name': f'SO-{parent_int}', 'x_fc_parent_number': parent_int, }) so.message_post(body=Markup(_( 'Confirmed quote %s as %s.' )) % (old_name, so.name)) 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 _create_invoices(self, grouped=False, final=False, date=None): """Set fp_from_so_invoice=True so account.move.create() allows the customer-invoice creation (the direct-creation block is bypassed via this context flag). Also lets the parent-numbered mixin find the originating SO without depending on invoice_origin. """ return super(SaleOrder, self.with_context( fp_from_so_invoice=True, fp_invoice_source_so_id=self.id if len(self) == 1 else False, ))._create_invoices(grouped=grouped, final=final, date=date) def unlink(self): """Spec §6.2 — confirmed SOs are part of the compliance audit trail and cannot be deleted. Cancellation must go through the state machine instead. Draft SOs (no parent_number assigned yet) remain freely deletable per Odoo standard. Applies to all users including administrators.""" for so in self: if so.x_fc_parent_number: raise UserError(_( 'Sale Order "%(name)s" cannot be deleted — it has ' 'been confirmed (parent number %(parent)s issued) ' 'and is part of the compliance audit trail. Cancel ' 'it instead. This rule applies to all users ' 'including administrators.' ) % {'name': so.display_name, 'parent': so.x_fc_parent_number}) return super().unlink() def _fp_resolve_recipe_for_line(self, line): """4-tier recipe resolution. Used BOTH for grouping (Task 6 recipe-driven WO splits) AND for the per-job vals construction. Priority (most-specific first): 1. line.x_fc_process_variant_id — Sarah explicitly picked a part-scoped variant on this order line. Always wins. 2. part.default_process_id — part's flagged default variant. Customer-and-part-tuned recipe; must beat any generic coating template. 3. coating.recipe_id — coating-config recipe (generic template fallback). 4. part.recipe_id — legacy fallback. Returns the recipe record or an empty recordset. """ Node = self.env['fusion.plating.process.node'] part = ( 'x_fc_part_catalog_id' in line._fields and line.x_fc_part_catalog_id ) or False if not part and 'x_fc_part_catalog_id' in self._fields: part = self.x_fc_part_catalog_id or False coating = ( 'x_fc_coating_config_id' in line._fields and line.x_fc_coating_config_id ) or False if not coating and 'x_fc_coating_config_id' in self._fields: coating = self.x_fc_coating_config_id or False picked = ( 'x_fc_process_variant_id' in line._fields and line.x_fc_process_variant_id ) or False if picked: return picked if part and 'default_process_id' in part._fields and part.default_process_id: return part.default_process_id if coating and 'recipe_id' in coating._fields and coating.recipe_id: return coating.recipe_id if part and 'recipe_id' in part._fields and part.recipe_id: return part.recipe_id return Node def _fp_auto_create_job(self): """Create fp.job(s) from the SO's plating lines. 2026-05-12 parent-number rewrite: lines are grouped by resolved recipe id (NOT by x_fc_wo_group_tag). If 1 group → one WO named WO- (bare). If N>1 groups → N WOs named WO--01, WO--02, ..., ordered by min line sequence so suffixes mirror SO display order. WO names are then immutable; later manual additions to the SO get the next index via the mixin. """ 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 (recipe, part, coating, thickness, serial). Lines that # share ALL FIVE collapse into one WO. Same compliance reasoning # as part_id + coating_id: bundling lines with different thicknesses # or different serials under one WO would carry the first line's # values onto the cert + sticker — silent mis-attestation. Sub 5 # added thickness_id + serial_id; this extends the grouping logic # to honour them. No-recipe lines still get their own group each. groups = {} unrecipe_idx = 0 for line in plating_lines: recipe = self._fp_resolve_recipe_for_line(line) part_id = ( 'x_fc_part_catalog_id' in line._fields and line.x_fc_part_catalog_id.id ) or False coating_id = ( 'x_fc_coating_config_id' in line._fields and line.x_fc_coating_config_id.id ) or False thickness_id = ( 'x_fc_thickness_id' in line._fields and line.x_fc_thickness_id.id ) or False serial_id = ( 'x_fc_serial_id' in line._fields and line.x_fc_serial_id.id ) or False if recipe: key = (recipe.id, part_id, coating_id, thickness_id, serial_id) else: unrecipe_idx += 1 key = ('no_recipe', unrecipe_idx) groups[key] = groups.get(key, self.env['sale.order.line']) | line # Order groups by min line sequence so dash-suffixes mirror SO # display order. Deterministic regardless of dict iteration order. ordered_keys = sorted( groups.keys(), key=lambda k: min(groups[k].mapped('sequence') or [0]), ) n_groups = len(ordered_keys) parent = self.x_fc_parent_number # set by action_confirm earlier # Create a job per group for idx, key in enumerate(ordered_keys, start=1): lines = groups[key] 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 ) 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 = self._fp_resolve_recipe_for_line(first_line) 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 references — mirror onto the job so the shop floor # has them without round-tripping to the SO. if 'x_fc_customer_job_number' in self._fields \ and self.x_fc_customer_job_number: vals['x_fc_customer_job_number'] = self.x_fc_customer_job_number if 'x_fc_po_number' in self._fields and self.x_fc_po_number: vals['x_fc_po_number'] = self.x_fc_po_number if 'x_fc_rush_order' in self._fields: vals['x_fc_rush_order'] = bool(self.x_fc_rush_order) # Scheduling targets — mirror the SO's customer-facing dates. if 'x_fc_internal_deadline' in self._fields \ and self.x_fc_internal_deadline: vals['x_fc_internal_deadline'] = self.x_fc_internal_deadline if 'x_fc_planned_start_date' in self._fields \ and self.x_fc_planned_start_date: vals['x_fc_planned_start_date'] = self.x_fc_planned_start_date # Operational notes — mirror so the shop has them on the WO. if 'x_fc_internal_note' in self._fields \ and self.x_fc_internal_note: vals['x_fc_internal_note'] = self.x_fc_internal_note if 'x_fc_external_note' in self._fields \ and self.x_fc_external_note: vals['x_fc_external_note'] = self.x_fc_external_note # 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')) # Parent-number naming (2026-05-12). Bare for the single-group # case; zero-padded -NN suffix when multiple recipes split the # SO into multiple WOs. Set explicitly so fp.job.create() skips # its own naming fallback. if parent: if n_groups == 1: vals['name'] = f'WO-{parent}' vals['x_fc_doc_index'] = 1 else: vals['name'] = f'WO-{parent}-{idx:02d}' if idx <= 99 else f'WO-{parent}-{idx}' vals['x_fc_doc_index'] = idx 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 '-'), ) # Bump SO counter to reflect the bulk creation. Future manual # WO additions pick up from here via the mixin standard path. if parent and n_groups: self.env.cr.execute( "UPDATE sale_order SET x_fc_pn_wo_count = %s WHERE id = %s", (n_groups, self.id), ) self.invalidate_recordset(['x_fc_pn_wo_count']) 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', }