# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
# Part of the Fusion Plating product family.
#
# Phase 1 (Sub 11) — relocated from fusion_plating_bridge_mrp.
# Now binds to fp.job (native) instead of mrp.production.
"""Per-job QC instance.
When a plating job confirms and the customer requires QC, we clone
the active checklist template into a `fusion.plating.quality.check`
with one line per template line. The inspector picks it up on the
tablet, walks the checks, and signs off — which unblocks the job's
`button_mark_done`.
The QC also owns the Fischerscope / XDAL 600 thickness report PDF.
When the operator uploads one, we extract per-reading data server-side
and auto-create `fp.thickness.reading` rows so the CoC PDF picks them up.
"""
import base64
import logging
import re
import subprocess
import tempfile
from markupsafe import Markup
from odoo import api, fields, models, _
from odoo.exceptions import UserError
_logger = logging.getLogger(__name__)
class FpQualityCheck(models.Model):
_name = 'fusion.plating.quality.check'
_description = 'Fusion Plating — Quality Check'
_inherit = ['mail.thread', 'mail.activity.mixin']
_order = 'create_date desc'
name = fields.Char(
string='Reference', required=True, copy=False, readonly=True,
default=lambda self: self._default_name(), tracking=True,
)
job_id = fields.Many2one(
'fp.job', string='Plating Job',
required=True, ondelete='cascade', tracking=True,
index=True,
)
sale_order_id = fields.Many2one(
'sale.order', related='job_id.sale_order_id',
store=True, readonly=True,
)
partner_id = fields.Many2one(
'res.partner', related='job_id.partner_id',
store=True, readonly=True,
)
template_id = fields.Many2one(
'fp.qc.checklist.template', string='Template',
help='The checklist template these lines were cloned from.',
)
state = fields.Selection(
[
('draft', 'Draft'),
('in_progress', 'In Progress'),
('passed', 'Passed'),
('failed', 'Failed'),
('rework', 'Rework Required'),
],
string='Status', default='draft', required=True, tracking=True,
)
overall_result = fields.Selection(
[('pass', 'Pass'), ('fail', 'Fail'), ('partial', 'Partial Pass')],
string='Result', tracking=True,
help='Summary outcome — set when inspector signs off.',
)
line_ids = fields.One2many(
'fusion.plating.quality.check.line', 'check_id',
string='Check Items',
)
line_count = fields.Integer(compute='_compute_line_stats', store=True)
lines_passed = fields.Integer(compute='_compute_line_stats', store=True)
lines_failed = fields.Integer(compute='_compute_line_stats', store=True)
lines_pending = fields.Integer(compute='_compute_line_stats', store=True)
inspector_id = fields.Many2one(
'res.users', string='Inspector',
help='Whoever signed the QC off. Filled when state moves to '
'passed/failed.',
tracking=True,
)
started_at = fields.Datetime(
string='Started', help='First time inspector opened this check.',
)
completed_at = fields.Datetime(
string='Completed', help='When the check was signed off.',
tracking=True,
)
notes = fields.Html(string='Inspector Notes')
thickness_report_pdf_id = fields.Many2one(
'ir.attachment', string='Thickness Report PDF',
help='Upload the Fischerscope / XDAL 600 export. On upload we '
'parse the PDF and auto-create fp.thickness.reading rows.',
)
thickness_reading_ids = fields.One2many(
'fp.thickness.reading', 'quality_check_id',
string='Thickness Readings',
)
thickness_reading_count = fields.Integer(
compute='_compute_thickness_count',
)
require_thickness_readings = fields.Boolean(
related='template_id.require_thickness_readings',
store=True, readonly=True,
)
require_thickness_report_pdf = fields.Boolean(
related='template_id.require_thickness_report_pdf',
store=True, readonly=True,
)
require_inspector_signoff = fields.Boolean(
related='template_id.require_inspector_signoff',
store=True, readonly=True,
)
company_id = fields.Many2one(
'res.company', related='job_id.company_id',
store=True, readonly=True,
)
@api.depends('line_ids.result')
def _compute_line_stats(self):
for rec in self:
rec.line_count = len(rec.line_ids)
rec.lines_passed = len(rec.line_ids.filtered(
lambda l: l.result == 'pass'
))
rec.lines_failed = len(rec.line_ids.filtered(
lambda l: l.result == 'fail'
))
rec.lines_pending = len(rec.line_ids.filtered(
lambda l: l.result in (False, 'pending')
))
@api.depends('thickness_reading_ids')
def _compute_thickness_count(self):
for rec in self:
rec.thickness_reading_count = len(rec.thickness_reading_ids)
@api.model
def _default_name(self):
seq = self.env['ir.sequence'].next_by_code(
'fusion.plating.quality.check',
)
return seq or 'QC/NEW'
@api.model_create_multi
def create(self, vals_list):
for vals in vals_list:
if not vals.get('name') or vals.get('name') == '/':
vals['name'] = self._default_name()
return super().create(vals_list)
@api.model
def create_for_job(self, job, template=None):
"""Spin up a QC record for a plating job, cloning lines from the
template.
If no template is passed, we resolve one from the job's customer.
Returns the created check, or an empty recordset if no template
matches (=> no QC required for this customer).
"""
self = self.sudo()
if template is None:
template = self.env['fp.qc.checklist.template'].resolve_for_partner(
job.partner_id,
)
if not template:
return self.browse()
existing = self.search([
('job_id', '=', job.id),
('state', '!=', 'failed'),
], limit=1)
if existing:
return existing
check = self.create({
'job_id': job.id,
'template_id': template.id,
'state': 'draft',
})
Line = self.env['fusion.plating.quality.check.line']
for tline in template.line_ids.sorted('sequence'):
Line.create({
'check_id': check.id,
'sequence': tline.sequence,
'name': tline.name,
'description': tline.description,
'check_type': tline.check_type,
'required': tline.required,
'requires_value': tline.requires_value,
'value_min': tline.value_min,
'value_max': tline.value_max,
'value_uom': tline.value_uom,
'requires_photo': tline.requires_photo,
'result': 'pending',
})
job.message_post(
body=_('QC checklist "%s" created — %d items to inspect.') % (
template.name, len(template.line_ids),
),
)
return check
def action_start(self):
for rec in self:
if rec.state == 'draft':
rec.write({
'state': 'in_progress',
'started_at': fields.Datetime.now(),
'inspector_id': self.env.user.id,
})
rec.message_post(body=_('QC started.'))
def action_pass(self):
for rec in self:
rec._ensure_all_required_complete()
rec.write({
'state': 'passed',
'overall_result': 'pass',
'completed_at': fields.Datetime.now(),
'inspector_id': self.env.user.id,
})
rec.message_post(body=Markup(
'QC PASSED — inspector %s.'
) % self.env.user.name)
def action_fail(self):
for rec in self:
rec.write({
'state': 'failed',
'overall_result': 'fail',
'completed_at': fields.Datetime.now(),
'inspector_id': self.env.user.id,
})
rec.message_post(body=Markup(
'QC FAILED — inspector %s.'
) % self.env.user.name)
def action_rework(self):
for rec in self:
rec.write({
'state': 'rework',
'overall_result': 'partial',
'completed_at': fields.Datetime.now(),
'inspector_id': self.env.user.id,
})
rec.message_post(body=_('QC flagged for rework.'))
def action_reset_to_draft(self):
for rec in self:
rec.write({
'state': 'draft',
'overall_result': False,
'completed_at': False,
})
def action_spawn_retry(self):
"""Spin up a fresh QC instance for the same job after a failure."""
self.ensure_one()
if self.state != 'failed':
return
new_check = self.sudo().create_for_job(
self.job_id, template=self.template_id,
)
if not new_check:
return False
self.message_post(body=_('Retry QC created: %s') % new_check.name)
new_check.message_post(body=_('Retry of failed QC %s') % self.name)
return new_check.action_open_tablet()
def _ensure_all_required_complete(self):
for rec in self:
pending = rec.line_ids.filtered(
lambda l: l.required and l.result in (False, 'pending')
)
if pending:
raise UserError(_(
'Cannot pass QC "%(name)s" — %(n)d required check '
'item(s) still pending:\n • %(items)s'
) % {
'name': rec.name,
'n': len(pending),
'items': '\n • '.join(pending.mapped('name')),
})
failed = rec.line_ids.filtered(lambda l: l.result == 'fail')
if failed:
raise UserError(_(
'Cannot pass QC "%(name)s" — %(n)d check item(s) '
'failed. Fail the QC instead, or reset those '
'items to pass.'
) % {'name': rec.name, 'n': len(failed)})
def _on_thickness_pdf_uploaded(self):
"""Parse the attached PDF and create fp.thickness.reading rows."""
ThicknessReading = self.env['fp.thickness.reading']
for rec in self:
if not rec.thickness_report_pdf_id:
continue
try:
text = rec._extract_pdf_text(rec.thickness_report_pdf_id)
except Exception:
_logger.exception(
'QC %s: pdftotext extraction failed', rec.name,
)
continue
readings = rec._parse_fischerscope_text(text)
if not readings:
rec.message_post(body=_(
'Thickness report PDF attached but no readings '
'could be extracted automatically. Please enter '
'readings manually.'
))
continue
auto = rec.thickness_reading_ids.filtered(
lambda r: r.auto_extracted
)
auto.unlink()
for idx, row in enumerate(readings, start=1):
vals = {
'quality_check_id': rec.id,
'reading_number': idx,
'nip_mils': row.get('nip_mils', 0.0),
'ni_percent': row.get('ni_percent', 0.0),
'p_percent': row.get('p_percent', 0.0),
'position_label': row.get('position', ''),
'auto_extracted': True,
}
ThicknessReading.create(vals)
rec.message_post(body=_(
'Extracted %d thickness reading(s) from "%s".'
) % (len(readings), rec.thickness_report_pdf_id.name))
@staticmethod
def _extract_pdf_text(attachment):
raw = base64.b64decode(attachment.datas or b'')
if not raw:
return ''
with tempfile.NamedTemporaryFile(
suffix='.pdf', delete=True,
) as tmp:
tmp.write(raw)
tmp.flush()
try:
result = subprocess.run(
['pdftotext', '-layout', tmp.name, '-'],
capture_output=True, text=True, timeout=30,
)
return result.stdout or ''
except FileNotFoundError:
_logger.warning(
'pdftotext not installed — cannot auto-extract '
'Fischerscope PDF data. Install poppler-utils on '
'the Odoo host.',
)
return ''
@staticmethod
def _parse_fischerscope_text(text):
readings = []
row_re = re.compile(
r'^\s*(?:n\s*=\s*|N\s*)?(\d{1,3})[\s.:]+'
r'([0-9]*\.[0-9]+|\d+)'
r'(?:\s*(?:mils|microns|µm|um))?'
r'[\s|]+'
r'([0-9]*\.?[0-9]+)'
r'[\s|%]+'
r'([0-9]*\.?[0-9]+)'
r'[\s|%]*'
r'(.*)$',
re.IGNORECASE,
)
for raw_line in text.splitlines():
line = raw_line.strip()
if not line:
continue
m = row_re.match(line)
if not m:
continue
try:
idx = int(m.group(1))
nip = float(m.group(2))
ni = float(m.group(3))
p = float(m.group(4))
except (TypeError, ValueError):
continue
if not (0 < nip < 1) and not (0 < nip < 30):
continue
if not (0 < ni < 100):
continue
if not (0 < p < 30):
continue
if idx < 1 or idx > 500:
continue
position = (m.group(5) or '').strip()[:60]
readings.append({
'index': idx,
'nip_mils': nip,
'ni_percent': ni,
'p_percent': p,
'position': position,
})
seen = set()
dedup = []
for r in readings:
if r['index'] in seen:
continue
seen.add(r['index'])
dedup.append(r)
return dedup
def write(self, vals):
trigger = 'thickness_report_pdf_id' in vals and vals.get(
'thickness_report_pdf_id'
)
res = super().write(vals)
if trigger:
self._on_thickness_pdf_uploaded()
return res
def action_open_tablet(self):
"""Launch the mobile QC checklist OWL client action."""
self.ensure_one()
return {
'type': 'ir.actions.client',
'tag': 'fp_qc_checklist',
'name': _('QC — %s') % (self.job_id.name or ''),
'params': {'check_id': self.id},
'target': 'current',
}
class FpQualityCheckLine(models.Model):
_name = 'fusion.plating.quality.check.line'
_description = 'Fusion Plating — Quality Check Line'
_order = 'sequence, id'
check_id = fields.Many2one(
'fusion.plating.quality.check', string='Check',
required=True, ondelete='cascade', index=True,
)
sequence = fields.Integer(default=10)
name = fields.Char(string='Check Item', required=True)
description = fields.Text(string='Guidance')
check_type = fields.Selection(
selection=lambda self: self.env[
'fp.qc.checklist.template.line'
]._fields['check_type'].selection,
string='Type', default='visual',
)
required = fields.Boolean(default=True)
requires_value = fields.Boolean()
value = fields.Float(digits=(12, 4))
value_min = fields.Float(digits=(12, 4))
value_max = fields.Float(digits=(12, 4))
value_uom = fields.Char(string='Unit')
requires_photo = fields.Boolean()
photo_attachment_id = fields.Many2one(
'ir.attachment', string='Photo',
)
result = fields.Selection(
[
('pending', 'Pending'),
('pass', 'Pass'),
('fail', 'Fail'),
('na', 'N/A'),
],
string='Result', default='pending', required=True,
)
notes = fields.Text(string='Note')
inspector_id = fields.Many2one('res.users', string='Inspector')
completed_at = fields.Datetime(string='Completed At')
value_in_range = fields.Boolean(
compute='_compute_value_in_range', store=True,
)
@api.depends('value', 'value_min', 'value_max', 'requires_value')
def _compute_value_in_range(self):
for rec in self:
if not rec.requires_value:
rec.value_in_range = True
continue
vmin = rec.value_min
vmax = rec.value_max
if vmin and rec.value < vmin:
rec.value_in_range = False
elif vmax and rec.value > vmax:
rec.value_in_range = False
else:
rec.value_in_range = True
def action_mark_pass(self):
for rec in self:
if rec.requires_value and not rec.value_in_range:
raise UserError(_(
'Cannot pass "%(item)s" — value %(val)s is outside '
'the acceptance range (%(min)s – %(max)s %(uom)s).'
) % {
'item': rec.name,
'val': rec.value,
'min': rec.value_min,
'max': rec.value_max,
'uom': rec.value_uom or '',
})
if rec.requires_photo and not rec.photo_attachment_id:
raise UserError(_(
'Cannot pass "%(item)s" — a photo is required.'
) % {'item': rec.name})
rec.write({
'result': 'pass',
'inspector_id': self.env.user.id,
'completed_at': fields.Datetime.now(),
})
def action_mark_fail(self):
for rec in self:
rec.write({
'result': 'fail',
'inspector_id': self.env.user.id,
'completed_at': fields.Datetime.now(),
})
def action_mark_na(self):
for rec in self:
if rec.required:
raise UserError(_(
'"%(item)s" is a required check and cannot be '
'marked N/A.'
) % {'item': rec.name})
rec.write({
'result': 'na',
'inspector_id': self.env.user.id,
'completed_at': fields.Datetime.now(),
})