8c76a16366
Replace em-dashes and en-dashes with hyphens across 789 shipped source files (py/xml/js/scss) so the delivered module reads as human-written; em-dashes had become a recognizable AI-generated tell. Internal .md dev notes are excluded. The WO-sticker mojibake strippers keep their dash search targets (now written — / –). No logic changes: comments and display strings only; validated with py_compile + lxml parse. Rewrite the 7 customer notification emails to be intake-neutral (ship-in / drop-off / pickup) and repair-aware, and fix the Shipped email documents line (packing slip vs bill of lading; certificate only when issued). Subjects use a hyphen separator. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
605 lines
22 KiB
Python
605 lines
22 KiB
Python
# -*- 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='Work Order',
|
|
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(
|
|
'<b>QC PASSED</b> - inspector %s.'
|
|
) % self.env.user.name)
|
|
|
|
def action_fail(self):
|
|
"""Mark QC failed AND auto-spawn a fusion.plating.quality.hold
|
|
so the parts have an AS9100 disposition record. Without this
|
|
spawning the parts are in limbo - operator can't ship and
|
|
nothing tracks scrap/rework/use-as-is decisions.
|
|
|
|
v19.0.4.x - Hold auto-spawn added per the tablet usability pass.
|
|
"""
|
|
Hold = self.env.get('fusion.plating.quality.hold')
|
|
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(
|
|
'<b>QC FAILED</b> - inspector %s.'
|
|
) % self.env.user.name)
|
|
# Auto-spawn the Hold (best-effort; QC failure stands even
|
|
# if Hold creation fails for some odd reason).
|
|
if Hold is not None and rec.job_id:
|
|
# Avoid dupes if the manager calls fail twice
|
|
existing = Hold.sudo().search([
|
|
('x_fc_job_id', '=', rec.job_id.id),
|
|
('hold_reason', '=', 'qc_failure'),
|
|
('state', 'in', ('on_hold', 'under_review')),
|
|
], limit=1) if 'x_fc_job_id' in Hold._fields else False
|
|
if not existing:
|
|
fail_lines = ', '.join(
|
|
l.name for l in rec.line_ids.filtered(
|
|
lambda l: l.result == 'fail'
|
|
)
|
|
) or '(see checklist)'
|
|
vals = {
|
|
'part_ref': rec.job_id.name or '',
|
|
'qty_on_hold': int(rec.job_id.qty or 1),
|
|
'qty_original': int(rec.job_id.qty or 1),
|
|
'hold_reason': (
|
|
'qc_failure'
|
|
if 'qc_failure' in dict(
|
|
Hold._fields['hold_reason'].selection
|
|
)
|
|
else 'other'
|
|
),
|
|
'description': (
|
|
'QC %s failed. Failed checks: %s. '
|
|
'Inspector: %s. Manager: review and decide '
|
|
'rework / scrap / use-as-is.'
|
|
) % (rec.name, fail_lines, self.env.user.name),
|
|
}
|
|
if 'x_fc_job_id' in Hold._fields:
|
|
vals['x_fc_job_id'] = rec.job_id.id
|
|
if 'partner_id' in Hold._fields and rec.partner_id:
|
|
vals['partner_id'] = rec.partner_id.id
|
|
try:
|
|
hold = Hold.sudo().create(vals)
|
|
rec.message_post(body=Markup(
|
|
'Hold <b>%s</b> auto-created for failed QC. '
|
|
'Manager must dispose.'
|
|
) % hold.name)
|
|
except Exception as e:
|
|
_logger.warning(
|
|
'QC %s: failed to auto-spawn hold: %s',
|
|
rec.name, e,
|
|
)
|
|
|
|
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(),
|
|
})
|