750 lines
32 KiB
Python
750 lines
32 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.
|
||
|
||
import logging
|
||
|
||
from odoo import api, fields, models, _
|
||
from odoo.exceptions import UserError
|
||
|
||
_logger = logging.getLogger(__name__)
|
||
|
||
|
||
class FpCertificate(models.Model):
|
||
"""Unified certificate registry.
|
||
|
||
Logs every quality document issued to customers: CoC, thickness
|
||
reports, mill test reports, Nadcap certs, and customer-specific
|
||
formats. Auto-created when reports are generated.
|
||
"""
|
||
_name = 'fp.certificate'
|
||
_description = 'Fusion Plating — Certificate'
|
||
_inherit = ['mail.thread', 'mail.activity.mixin', 'fp.parent.numbered.mixin']
|
||
_order = 'issue_date desc, id desc'
|
||
|
||
name = fields.Char(string='Reference', readonly=True, copy=False, default='New')
|
||
certificate_type = fields.Selection(
|
||
[
|
||
('coc', 'Certificate of Conformance'),
|
||
('thickness_report', 'Thickness Report'),
|
||
('mill_test', 'Mill Test Report'),
|
||
('nadcap_cert', 'Nadcap Certificate'),
|
||
('customer_specific', 'Customer-Specific'),
|
||
],
|
||
string='Type', required=True, default='coc', tracking=True,
|
||
)
|
||
partner_id = fields.Many2one(
|
||
'res.partner', string='Customer', required=True, tracking=True,
|
||
domain="[('customer_rank', '>', 0)]",
|
||
)
|
||
sale_order_id = fields.Many2one('sale.order', string='Sale Order')
|
||
# Phase 6 (Sub 11) — production_id retired (MRP module gone).
|
||
# Certificates link via sale_order_id + portal_job_id natively.
|
||
portal_job_id = fields.Many2one('fusion.plating.portal.job', string='Portal Job')
|
||
part_number = fields.Char(string='Part Number', help='Denormalized for fast search.')
|
||
process_description = fields.Char(
|
||
string='Process', help='e.g. "ELECTROLESS NICKEL PLATING PER AMS 2404"',
|
||
)
|
||
spec_reference = fields.Char(string='Spec Reference')
|
||
po_number = fields.Char(string='Customer PO #')
|
||
entech_wo_number = fields.Char(string='Entech WO #')
|
||
quantity_shipped = fields.Integer(string='Qty Shipped')
|
||
nc_quantity = fields.Integer(
|
||
string='NC Qty',
|
||
help='Non-conforming quantity — parts that failed inspection / rework.',
|
||
)
|
||
customer_job_no = fields.Char(
|
||
string='Customer Job No.',
|
||
help="Customer's internal job / traveler reference.",
|
||
)
|
||
contact_partner_id = fields.Many2one(
|
||
'res.partner', string='Customer Contact',
|
||
domain="[('parent_id', '=', partner_id)]",
|
||
help="Specific contact person at the customer for this certificate. "
|
||
'Their name, email, and phone are printed on the CoC.',
|
||
)
|
||
issued_by_id = fields.Many2one(
|
||
'res.users', string='Issued By', default=lambda self: self.env.user,
|
||
)
|
||
certified_by_id = fields.Many2one(
|
||
'res.users', string='Certified By', help='Signing authority (e.g. Quality Manager).',
|
||
)
|
||
|
||
# ===== Sub 12c — chronological CoC opt-in ===============================
|
||
body_style = fields.Selection(
|
||
[
|
||
('classic', 'Classic (recipe-order)'),
|
||
('chronological', 'Chronological (chain-of-custody)'),
|
||
],
|
||
string='CoC Body Style', default='classic',
|
||
help='Chronological walks fp.job.step.move records in time order '
|
||
'with measurement sub-tables per move, matching Steelhead\'s '
|
||
'CoC PDF layout. Classic uses the existing recipe-order body.',
|
||
)
|
||
issue_date = fields.Date(string='Issue Date', default=fields.Date.today, tracking=True)
|
||
attachment_id = fields.Many2one('ir.attachment', string='Certificate PDF')
|
||
thickness_reading_ids = fields.One2many(
|
||
'fp.thickness.reading', 'certificate_id', string='Thickness Readings',
|
||
)
|
||
|
||
# ----- Inline Fischerscope PDF upload (cert-local) ----------------------
|
||
# The merge pipeline normally pulls the Fischerscope/XDAL PDF from the
|
||
# linked QC check. That works when the operator uploaded it via the
|
||
# tablet, but managers issuing certs after the fact don't want to
|
||
# navigate to the QC. This pair of fields gives them a direct upload
|
||
# path on the cert form. When set, _fp_merge_thickness_into_pdf uses
|
||
# this in preference to the QC-side upload.
|
||
x_fc_local_thickness_pdf = fields.Binary(
|
||
string='Fischerscope PDF (Upload Here)',
|
||
attachment=True,
|
||
help='Drop the Fischerscope / XDAL 600 XRF export PDF here. '
|
||
'When the cert is issued it will be appended as page 2 of '
|
||
'the CoC. Overrides any PDF on the linked QC check.',
|
||
)
|
||
x_fc_local_thickness_pdf_filename = fields.Char(
|
||
string='Fischerscope PDF filename',
|
||
)
|
||
|
||
# ---- Material traceability (T2.3) ----
|
||
batch_ids = fields.Many2many(
|
||
'fusion.plating.batch', compute='_compute_batch_ids',
|
||
string='Batches', help='All batches used for this MO.',
|
||
)
|
||
batch_count = fields.Integer(
|
||
string='Batches', compute='_compute_batch_ids',
|
||
)
|
||
bath_ids = fields.Many2many(
|
||
'fusion.plating.bath', compute='_compute_batch_ids',
|
||
string='Baths Used',
|
||
)
|
||
|
||
@api.depends('sale_order_id')
|
||
def _compute_batch_ids(self):
|
||
# Phase 6 (Sub 11) — walks fp.job via SO instead of mrp.production.
|
||
Batch = self.env.get('fusion.plating.batch')
|
||
Bath = self.env['fusion.plating.bath']
|
||
Job = self.env.get('fp.job')
|
||
empty_batch = self.env['fusion.plating.batch']
|
||
for rec in self:
|
||
if Batch is None or Job is None or not rec.sale_order_id:
|
||
rec.batch_ids = empty_batch
|
||
rec.batch_count = 0
|
||
rec.bath_ids = Bath
|
||
continue
|
||
jobs = Job.search([('sale_order_id', '=', rec.sale_order_id.id)])
|
||
if not jobs:
|
||
rec.batch_ids = empty_batch
|
||
rec.batch_count = 0
|
||
rec.bath_ids = Bath
|
||
continue
|
||
if 'x_fc_job_id' in Batch._fields:
|
||
batches = Batch.search([('x_fc_job_id', 'in', jobs.ids)])
|
||
else:
|
||
batches = empty_batch
|
||
rec.batch_ids = batches
|
||
rec.batch_count = len(batches)
|
||
rec.bath_ids = batches.mapped('bath_id')
|
||
state = fields.Selection(
|
||
[('draft', 'Draft'), ('issued', 'Issued'), ('voided', 'Voided')],
|
||
string='Status', default='draft', tracking=True, required=True,
|
||
)
|
||
void_reason = fields.Text(string='Void Reason')
|
||
notes = fields.Html(string='Notes')
|
||
|
||
# ----- Computed stats from readings -------------------------------------
|
||
reading_count = fields.Integer(
|
||
string='Readings', compute='_compute_reading_stats', store=True,
|
||
)
|
||
mean_nip_mils = fields.Float(
|
||
string='Mean NiP (mils)', compute='_compute_reading_stats',
|
||
store=True, digits=(10, 4),
|
||
)
|
||
|
||
# ---- SPC (T2.1) ----
|
||
spec_min_mils = fields.Float(
|
||
string='Spec Min (mils)', digits=(10, 4),
|
||
help='Lower specification limit pulled from the coating config. '
|
||
'Override per certificate if needed.',
|
||
)
|
||
spec_max_mils = fields.Float(
|
||
string='Spec Max (mils)', digits=(10, 4),
|
||
help='Upper specification limit pulled from the coating config.',
|
||
)
|
||
std_dev_mils = fields.Float(
|
||
string='Std Dev (mils)', compute='_compute_reading_stats',
|
||
store=True, digits=(10, 4),
|
||
)
|
||
min_reading_mils = fields.Float(
|
||
string='Min Reading', compute='_compute_reading_stats',
|
||
store=True, digits=(10, 4),
|
||
)
|
||
max_reading_mils = fields.Float(
|
||
string='Max Reading', compute='_compute_reading_stats',
|
||
store=True, digits=(10, 4),
|
||
)
|
||
cpk = fields.Float(
|
||
string='Cpk', compute='_compute_reading_stats', store=True, digits=(6, 3),
|
||
help='Process capability index. <1.0 = incapable · 1.0-1.33 = marginal · '
|
||
'≥1.33 = capable · ≥1.67 = excellent.',
|
||
)
|
||
cpk_status = fields.Selection(
|
||
[('incapable', 'Incapable'),
|
||
('marginal', 'Marginal'),
|
||
('capable', 'Capable'),
|
||
('excellent', 'Excellent'),
|
||
('insufficient', 'Insufficient Data')],
|
||
string='Cpk Status', compute='_compute_reading_stats', store=True,
|
||
)
|
||
trend_alert = fields.Selection(
|
||
[('ok', 'OK'),
|
||
('warning', 'Trend Detected'),
|
||
('alert', 'Out of Control')],
|
||
string='Trend Alert', compute='_compute_reading_stats', store=True,
|
||
help='Western Electric rule 1 (any point beyond 3σ) or rule 4 '
|
||
'(8 consecutive points on one side of centre).',
|
||
)
|
||
trend_explanation = fields.Char(
|
||
string='Trend Note', compute='_compute_reading_stats', store=True,
|
||
)
|
||
|
||
@api.depends('thickness_reading_ids', 'thickness_reading_ids.nip_mils',
|
||
'spec_min_mils', 'spec_max_mils')
|
||
def _compute_reading_stats(self):
|
||
for rec in self:
|
||
readings = rec.thickness_reading_ids
|
||
rec.reading_count = len(readings)
|
||
values = [r.nip_mils for r in readings if r.nip_mils]
|
||
n = len(values)
|
||
if n == 0:
|
||
rec.mean_nip_mils = 0.0
|
||
rec.std_dev_mils = 0.0
|
||
rec.min_reading_mils = 0.0
|
||
rec.max_reading_mils = 0.0
|
||
rec.cpk = 0.0
|
||
rec.cpk_status = 'insufficient'
|
||
rec.trend_alert = 'ok'
|
||
rec.trend_explanation = ''
|
||
continue
|
||
|
||
mean = sum(values) / n
|
||
rec.mean_nip_mils = round(mean, 4)
|
||
rec.min_reading_mils = round(min(values), 4)
|
||
rec.max_reading_mils = round(max(values), 4)
|
||
|
||
# Sample standard deviation (Bessel's correction)
|
||
if n >= 2:
|
||
sq_diff = sum((v - mean) ** 2 for v in values)
|
||
sigma = (sq_diff / (n - 1)) ** 0.5
|
||
else:
|
||
sigma = 0.0
|
||
rec.std_dev_mils = round(sigma, 4)
|
||
|
||
# Cpk — requires spec limits + non-zero sigma
|
||
usl = rec.spec_max_mils or 0.0
|
||
lsl = rec.spec_min_mils or 0.0
|
||
if n < 5 or sigma == 0 or (usl == 0 and lsl == 0):
|
||
rec.cpk = 0.0
|
||
rec.cpk_status = 'insufficient'
|
||
else:
|
||
if usl and lsl:
|
||
cpu = (usl - mean) / (3.0 * sigma)
|
||
cpl = (mean - lsl) / (3.0 * sigma)
|
||
cpk = min(cpu, cpl)
|
||
elif usl:
|
||
cpk = (usl - mean) / (3.0 * sigma)
|
||
else:
|
||
cpk = (mean - lsl) / (3.0 * sigma)
|
||
rec.cpk = round(cpk, 3)
|
||
if cpk < 1.0:
|
||
rec.cpk_status = 'incapable'
|
||
elif cpk < 1.33:
|
||
rec.cpk_status = 'marginal'
|
||
elif cpk < 1.67:
|
||
rec.cpk_status = 'capable'
|
||
else:
|
||
rec.cpk_status = 'excellent'
|
||
|
||
# Trend detection (Western Electric rules)
|
||
# Rule 1: any point outside 3σ from mean
|
||
# Rule 4: 8+ consecutive on one side of mean
|
||
alert = 'ok'
|
||
explanation = ''
|
||
if sigma > 0:
|
||
three_sigma = 3.0 * sigma
|
||
for v in values:
|
||
if abs(v - mean) > three_sigma:
|
||
alert = 'alert'
|
||
explanation = 'Rule 1: reading beyond 3σ from mean'
|
||
break
|
||
if alert == 'ok' and n >= 8:
|
||
# Check last 8 readings for all-above or all-below
|
||
last_eight = values[-8:]
|
||
if all(v > mean for v in last_eight):
|
||
alert = 'warning'
|
||
explanation = 'Rule 4: 8 consecutive readings above mean'
|
||
elif all(v < mean for v in last_eight):
|
||
alert = 'warning'
|
||
explanation = 'Rule 4: 8 consecutive readings below mean'
|
||
rec.trend_alert = alert
|
||
rec.trend_explanation = explanation
|
||
|
||
# ----- Parent-numbered mixin hooks -------------------------------------
|
||
def _fp_parent_sale_order(self):
|
||
return self.sale_order_id
|
||
|
||
def _fp_name_prefix(self):
|
||
return 'CoC'
|
||
|
||
def _fp_parent_counter_field(self):
|
||
return 'x_fc_pn_cert_count'
|
||
|
||
# ----- Create: parent-derived name (fallback to legacy sequence) -------
|
||
@api.model_create_multi
|
||
def create(self, vals_list):
|
||
SaleOrder = self.env['sale.order']
|
||
for vals in vals_list:
|
||
# Spec-limit auto-fill — sources thickness range from the
|
||
# recipe (Phase A moved the thickness fields onto the
|
||
# recipe root). Falls back gracefully when the SO has no
|
||
# recipe-bearing line.
|
||
already_set = vals.get('spec_min_mils') or vals.get('spec_max_mils')
|
||
if not already_set and vals.get('sale_order_id'):
|
||
so = SaleOrder.browse(vals['sale_order_id'])
|
||
# Look across order_line for the first recipe with a
|
||
# populated thickness range.
|
||
first_line = so.order_line[:1] if so.order_line else False
|
||
recipe = (
|
||
first_line.x_fc_process_variant_id
|
||
if (first_line
|
||
and 'x_fc_process_variant_id' in first_line._fields)
|
||
else False
|
||
)
|
||
if (recipe
|
||
and 'thickness_uom' in recipe._fields
|
||
and recipe.thickness_uom == 'mils'):
|
||
vals.setdefault('spec_min_mils', recipe.thickness_min or 0.0)
|
||
vals.setdefault('spec_max_mils', recipe.thickness_max or 0.0)
|
||
# Defer naming: let the record exist so the mixin can write
|
||
# name via raw SQL, then fall back to the legacy sequence if
|
||
# no parent SO is reachable.
|
||
if not vals.get('name'):
|
||
vals['name'] = 'New'
|
||
records = super().create(vals_list)
|
||
for rec in records:
|
||
if rec.name and rec.name != 'New':
|
||
continue
|
||
if not rec._fp_assign_parent_name():
|
||
seq = self.env['ir.sequence'].next_by_code('fp.certificate') or 'New'
|
||
self.env.cr.execute(
|
||
"UPDATE fp_certificate SET name = %s WHERE id = %s",
|
||
(seq, rec.id),
|
||
)
|
||
rec.invalidate_recordset(['name'])
|
||
return records
|
||
|
||
# ----- State actions ----------------------------------------------------
|
||
def action_issue(self):
|
||
for rec in self:
|
||
if rec.state != 'draft':
|
||
raise UserError(_('Only draft certificates can be issued.'))
|
||
# Lazy-fill from partner defaults BEFORE running the gates.
|
||
# Without this, a cert created before partner.x_fc_default_*
|
||
# was configured would still trip the gate even after sales
|
||
# set the default. Robust-by-construction: the defaults take
|
||
# effect retroactively at issue time.
|
||
if (not rec.contact_partner_id
|
||
and rec.partner_id
|
||
and 'x_fc_default_coc_contact_id' in rec.partner_id._fields
|
||
and rec.partner_id.x_fc_default_coc_contact_id):
|
||
rec.contact_partner_id = (
|
||
rec.partner_id.x_fc_default_coc_contact_id
|
||
)
|
||
if (not rec.certified_by_id
|
||
and rec.company_id
|
||
and 'x_fc_owner_user_id' in rec.company_id._fields
|
||
and rec.company_id.x_fc_owner_user_id):
|
||
rec.certified_by_id = rec.company_id.x_fc_owner_user_id
|
||
# Spec reference is what the cert ATTESTS — without it the
|
||
# cert is just a piece of paper. AS9100 / Nadcap require
|
||
# naming the spec the work was performed to.
|
||
if not rec.spec_reference:
|
||
raise UserError(_(
|
||
'Cannot issue certificate "%(name)s" — no Spec '
|
||
'Reference set.\n\nFill the Spec Reference field '
|
||
'(e.g. "AMS 2404", "MIL-C-26074") so the cert '
|
||
'states which standard the work meets.'
|
||
) % {'name': rec.name or rec.display_name})
|
||
# Process description (what was done to the parts). Without
|
||
# it the cert PDF just shows blank process text — customer
|
||
# has no idea what they paid for. Auto-filled from the
|
||
# recipe at create time; manager can override before issuing.
|
||
if not rec.process_description:
|
||
raise UserError(_(
|
||
'Cannot issue certificate "%(name)s" — Process '
|
||
'Description is blank.\n\nFill it manually (e.g. '
|
||
'"ELECTROLESS NICKEL PLATING PER AMS 2404") or '
|
||
'assign a recipe to the job so it auto-fills.'
|
||
) % {'name': rec.name or rec.display_name})
|
||
# Signing authority — the human who attests the work. Auto-
|
||
# filled from per-spec signer_user_id, falling back to
|
||
# company.x_fc_owner_user_id. If neither is configured, the
|
||
# manager must pick before issuing.
|
||
if not rec.certified_by_id:
|
||
raise UserError(_(
|
||
'Cannot issue certificate "%(name)s" — Certified By '
|
||
'is not set.\n\nPick the signing authority, or have '
|
||
'an admin configure the company\'s Certificate Owner '
|
||
'(Settings > Fusion Plating).'
|
||
) % {'name': rec.name or rec.display_name})
|
||
# Customer contact — the named recipient printed on the
|
||
# cert and emailed when it ships. Auto-filled from
|
||
# partner.x_fc_default_coc_contact_id when set.
|
||
if not rec.contact_partner_id:
|
||
raise UserError(_(
|
||
'Cannot issue certificate "%(name)s" — Customer '
|
||
'Contact is not set.\n\nPick the recipient contact, '
|
||
'or configure a Default CoC Contact on customer '
|
||
'"%(cust)s".'
|
||
) % {
|
||
'name': rec.name or rec.display_name,
|
||
'cust': rec.partner_id.name if rec.partner_id else '?',
|
||
})
|
||
if not (rec.contact_partner_id.email or '').strip():
|
||
raise UserError(_(
|
||
'Cannot issue certificate "%(name)s" — contact '
|
||
'"%(c)s" has no email address.\n\nAdd an email '
|
||
'to the contact before issuing (the cert is sent '
|
||
'by email post-issue).'
|
||
) % {
|
||
'name': rec.name or rec.display_name,
|
||
'c': rec.contact_partner_id.name,
|
||
})
|
||
# Thickness data requirement — unified gate covering both
|
||
# cert types. A customer needs thickness data on the cert
|
||
# when ANY of these is true:
|
||
# 1. cert type is thickness_report (the cert IS the data)
|
||
# 2. partner.x_fc_strict_thickness_required (aerospace /
|
||
# Nadcap — always strict)
|
||
# 3. partner.x_fc_send_thickness_report (the bundling
|
||
# rule — CoC carries thickness as page 2 by default
|
||
# for these customers; see CLAUDE.md "CoC + thickness
|
||
# = ONE cert (page 2 merge)")
|
||
# Acceptable data: logged readings on the cert OR a
|
||
# Fischerscope PDF on the linked QC OR a cert-local
|
||
# Fischerscope upload. Any one is enough.
|
||
partner = rec.partner_id
|
||
needs_thickness = (
|
||
rec.certificate_type == 'thickness_report'
|
||
or (rec.certificate_type == 'coc' and partner and (
|
||
('x_fc_strict_thickness_required' in partner._fields
|
||
and partner.x_fc_strict_thickness_required)
|
||
or ('x_fc_send_thickness_report' in partner._fields
|
||
and partner.x_fc_send_thickness_report)
|
||
))
|
||
)
|
||
if needs_thickness:
|
||
has_readings = bool(rec.thickness_reading_ids)
|
||
has_qc_fischer_pdf = bool(
|
||
rec.x_fc_thickness_pdf_id
|
||
if 'x_fc_thickness_pdf_id' in rec._fields else False
|
||
)
|
||
has_local_pdf = bool(rec.x_fc_local_thickness_pdf)
|
||
if not (has_readings or has_qc_fischer_pdf or has_local_pdf):
|
||
type_label = (
|
||
_('Thickness Report')
|
||
if rec.certificate_type == 'thickness_report'
|
||
else _('CoC')
|
||
)
|
||
raise UserError(_(
|
||
'Cannot issue %(type)s "%(name)s" — customer '
|
||
'"%(cust)s" requires thickness data on every '
|
||
'%(type)s. No readings, no Fischerscope PDF on '
|
||
'the linked QC, and no local Fischerscope upload '
|
||
'on this cert.\n\nUse the Issue Certs wizard '
|
||
'from the work order to upload the Fischerscope '
|
||
'report, or log readings against the job for '
|
||
'SO %(so)s via the Tablet Station.'
|
||
) % {
|
||
'type': type_label,
|
||
'name': rec.name or rec.display_name,
|
||
'cust': partner.name if partner else '?',
|
||
'so': rec.sale_order_id.name if rec.sale_order_id else '?',
|
||
})
|
||
# Defensive qty reconciliation — should already be guaranteed
|
||
# by fp.job.button_mark_done's gate, but re-checked here so
|
||
# certs created outside the job flow (manual, scripts) still
|
||
# can't issue with a mismatched job. No bypass — qty integrity
|
||
# is non-negotiable at issue.
|
||
job = (rec.x_fc_job_id
|
||
if 'x_fc_job_id' in rec._fields else False)
|
||
if job and job.qty_received:
|
||
rejects = job.qty_visual_inspection_rejects or 0
|
||
accounted = (
|
||
(job.qty_done or 0)
|
||
+ (job.qty_scrapped or 0)
|
||
+ rejects
|
||
)
|
||
if abs(job.qty_received - accounted) > 0.0001:
|
||
raise UserError(_(
|
||
'Cannot issue certificate "%(name)s" — job '
|
||
'%(job)s qty mismatch (received %(r)g vs '
|
||
'accounted-out %(a)g). Reconcile job '
|
||
'quantities before issuing.'
|
||
) % {
|
||
'name': rec.name or rec.display_name,
|
||
'job': job.name,
|
||
'r': job.qty_received,
|
||
'a': accounted,
|
||
})
|
||
rec.state = 'issued'
|
||
# Generate the CoC PDF and attach it so action_send_to_customer
|
||
# has something to email. Without this the workflow goes:
|
||
# Issue → Send → opens composer with no attachment → operator
|
||
# closes confused. Best-effort: if the report renders, attach;
|
||
# if it fails, log + continue (cert is still issued).
|
||
try:
|
||
rec._fp_render_and_attach_pdf()
|
||
except Exception as e:
|
||
_logger.warning(
|
||
'Cert %s: PDF render failed: %s', rec.name, e,
|
||
)
|
||
rec.message_post(body=_('Certificate issued.'))
|
||
|
||
def _fp_render_and_attach_pdf(self):
|
||
"""Render the CoC PDF via the bound report action, OPTIONALLY
|
||
merge the Fischerscope thickness report PDF (uploaded by the
|
||
QC tablet operator) as page 2, and attach the result.
|
||
|
||
Without the merge, a customer who specs "CoC must include the
|
||
XRF report" gets two separate PDFs to chase down. AS9100 wants
|
||
the supporting evidence inline with the cert.
|
||
|
||
Tries the EN-language CoC report first, falls back to the
|
||
generic action_report_coc. Idempotent — skips if attachment_id
|
||
is already set. PDF merge is best-effort: corrupt Fischerscope
|
||
upload or missing pypdf falls back to CoC-only with a warning.
|
||
"""
|
||
import base64
|
||
import io
|
||
self.ensure_one()
|
||
if self.attachment_id:
|
||
return self.attachment_id
|
||
report = (
|
||
self.env.ref(
|
||
'fusion_plating_reports.action_report_coc_en',
|
||
raise_if_not_found=False,
|
||
)
|
||
or self.env.ref(
|
||
'fusion_plating_reports.action_report_coc',
|
||
raise_if_not_found=False,
|
||
)
|
||
)
|
||
if not report:
|
||
_logger.warning(
|
||
'Cert %s: no CoC report action found, cannot render PDF',
|
||
self.name,
|
||
)
|
||
return False
|
||
coc_pdf_bytes, _content_type = report._render_qweb_pdf(
|
||
report.report_name, res_ids=self.ids,
|
||
)
|
||
# Try to append the Fischerscope thickness-report PDF as page 2.
|
||
merged_bytes = self._fp_merge_thickness_into_pdf(coc_pdf_bytes)
|
||
final_pdf = merged_bytes or coc_pdf_bytes
|
||
|
||
att = self.env['ir.attachment'].sudo().create({
|
||
'name': '%s.pdf' % (self.name or 'certificate'),
|
||
'type': 'binary',
|
||
'datas': base64.b64encode(final_pdf),
|
||
'mimetype': 'application/pdf',
|
||
'res_model': self._name,
|
||
'res_id': self.id,
|
||
})
|
||
self.attachment_id = att.id
|
||
return att
|
||
|
||
def _fp_merge_thickness_into_pdf(self, coc_pdf_bytes):
|
||
"""Look up the linked QC check, find its thickness_report_pdf_id
|
||
(Fischerscope / XDAL 600 XRF export), and return a merged PDF
|
||
with the CoC first + Fischerscope appended as page 2+.
|
||
|
||
Returns None when:
|
||
- cert isn't a CoC, or
|
||
- no fp.job linked, or
|
||
- no fp.quality.check on the job has a PDF uploaded, or
|
||
- pypdf / PyPDF2 not installed, or
|
||
- either PDF fails to parse.
|
||
|
||
Caller falls back to CoC-only when None is returned.
|
||
"""
|
||
import io
|
||
import base64 as _b64
|
||
self.ensure_one()
|
||
if self.certificate_type != 'coc':
|
||
return None
|
||
# Resolution order for the source of the Fischerscope bytes:
|
||
# 1. Cert-local upload (x_fc_local_thickness_pdf) — manager
|
||
# dropped it directly on the cert form
|
||
# 2. Linked QC's thickness_report_pdf_id — operator uploaded
|
||
# via the tablet during inspection
|
||
# Either path yields the same merged-PDF outcome.
|
||
fischer_bytes = b''
|
||
qc = False
|
||
if self.x_fc_local_thickness_pdf:
|
||
try:
|
||
fischer_bytes = _b64.b64decode(
|
||
self.x_fc_local_thickness_pdf or b''
|
||
)
|
||
except Exception:
|
||
fischer_bytes = b''
|
||
if not fischer_bytes:
|
||
# Fall through to the QC-side PDF.
|
||
job = False
|
||
if 'x_fc_job_id' in self._fields:
|
||
job = self.x_fc_job_id
|
||
if not job and 'job_id' in self._fields:
|
||
job = self.job_id
|
||
if not job:
|
||
return None
|
||
QC = self.env.get('fusion.plating.quality.check')
|
||
if QC is None:
|
||
return None
|
||
qc = QC.sudo().search([
|
||
('job_id', '=', job.id),
|
||
('state', '=', 'passed'),
|
||
('thickness_report_pdf_id', '!=', False),
|
||
], order='completed_at desc', limit=1)
|
||
if not qc:
|
||
qc = QC.sudo().search([
|
||
('job_id', '=', job.id),
|
||
('thickness_report_pdf_id', '!=', False),
|
||
], order='create_date desc', limit=1)
|
||
if not qc or not qc.thickness_report_pdf_id:
|
||
return None
|
||
fischer_bytes = _b64.b64decode(
|
||
qc.thickness_report_pdf_id.datas or b''
|
||
)
|
||
if not fischer_bytes:
|
||
return None
|
||
# Merge — pypdf is the modern name; PyPDF2 still works on older
|
||
# Odoo bundles. Either is fine.
|
||
try:
|
||
from pypdf import PdfWriter
|
||
writer_cls = PdfWriter
|
||
use_append = True
|
||
except ImportError:
|
||
try:
|
||
from PyPDF2 import PdfMerger
|
||
writer_cls = PdfMerger
|
||
use_append = False
|
||
except ImportError:
|
||
_logger.warning(
|
||
'Cert %s: neither pypdf nor PyPDF2 installed, '
|
||
'cannot append Fischerscope PDF to CoC.',
|
||
self.name,
|
||
)
|
||
return None
|
||
try:
|
||
if use_append:
|
||
# pypdf 3.x — PdfWriter.append() handles bytes/streams
|
||
writer = writer_cls()
|
||
writer.append(io.BytesIO(coc_pdf_bytes))
|
||
writer.append(io.BytesIO(fischer_bytes))
|
||
out = io.BytesIO()
|
||
writer.write(out)
|
||
merged = out.getvalue()
|
||
else:
|
||
# PyPDF2 — PdfMerger.append + write
|
||
merger = writer_cls()
|
||
merger.append(io.BytesIO(coc_pdf_bytes))
|
||
merger.append(io.BytesIO(fischer_bytes))
|
||
out = io.BytesIO()
|
||
merger.write(out)
|
||
merger.close()
|
||
merged = out.getvalue()
|
||
except Exception:
|
||
_logger.exception(
|
||
'PDF merge failed for cert %s — Fischerscope PDF may '
|
||
'be corrupt / encrypted / malformed. Falling back to '
|
||
'CoC-only.', self.name,
|
||
)
|
||
return None
|
||
source = (
|
||
_('cert upload') if self.x_fc_local_thickness_pdf
|
||
else _('QC %s') % (qc.name if qc else '?')
|
||
)
|
||
self.message_post(body=_(
|
||
'Fischerscope thickness report (%s) appended to CoC PDF.'
|
||
) % source)
|
||
return merged
|
||
|
||
def action_void(self):
|
||
for rec in self:
|
||
if rec.state != 'issued':
|
||
raise UserError(_('Only issued certificates can be voided.'))
|
||
if not rec.void_reason:
|
||
raise UserError(_('Please enter a void reason before voiding.'))
|
||
rec.state = 'voided'
|
||
rec.message_post(body=_('Certificate voided. Reason: %s') % rec.void_reason)
|
||
|
||
def action_open_void_wizard(self):
|
||
"""Open the void-reason wizard. Bound to the Void header button
|
||
instead of action_void directly so the manager always supplies a
|
||
written reason (the underlying action_void still blocks on a
|
||
blank reason as a defensive last-line check)."""
|
||
self.ensure_one()
|
||
if self.state != 'issued':
|
||
raise UserError(_(
|
||
'Only issued certificates can be voided '
|
||
'(current state: %s).'
|
||
) % self.state)
|
||
Wizard = self.env.get('fp.cert.void.wizard')
|
||
if Wizard is None:
|
||
raise UserError(_(
|
||
'Void wizard not available. Reinstall '
|
||
'fusion_plating_certificates.'
|
||
))
|
||
wiz = Wizard.create({'cert_id': self.id})
|
||
return {
|
||
'type': 'ir.actions.act_window',
|
||
'name': _('Void %s') % self.name,
|
||
'res_model': Wizard._name,
|
||
'res_id': wiz.id,
|
||
'view_mode': 'form',
|
||
'target': 'new',
|
||
}
|
||
|
||
def action_view_traceability(self):
|
||
"""Show the batches (and their chemistry logs) that produced
|
||
these parts — auditor's dream, customer's RMA friend."""
|
||
self.ensure_one()
|
||
return {
|
||
'type': 'ir.actions.act_window',
|
||
'name': _('Traceability — %s') % self.name,
|
||
'res_model': 'fusion.plating.batch',
|
||
'view_mode': 'list,form',
|
||
'domain': [('id', 'in', self.batch_ids.ids)],
|
||
'target': 'current',
|
||
}
|
||
|
||
def action_send_to_customer(self):
|
||
"""Open email composer with the certificate PDF attached."""
|
||
self.ensure_one()
|
||
template = self.env.ref('mail.email_compose_message_wizard_form', raise_if_not_found=False)
|
||
ctx = {
|
||
'default_model': 'fp.certificate',
|
||
'default_res_ids': self.ids,
|
||
'default_composition_mode': 'comment',
|
||
'default_partner_ids': [self.partner_id.id] if self.partner_id else [],
|
||
}
|
||
if self.attachment_id:
|
||
ctx['default_attachment_ids'] = [self.attachment_id.id]
|
||
return {
|
||
'type': 'ir.actions.act_window',
|
||
'res_model': 'mail.compose.message',
|
||
'view_mode': 'form',
|
||
'target': 'new',
|
||
'context': ctx,
|
||
}
|