Odoo-Modules/fusion_plating/scripts/fp_per_step_audit.py

# -*- coding: utf-8 -*-
"""Per-step compliance audit - walks every WO of the most recent MO
and reports which compliance data points are captured vs missing,
broken down by WO kind.

Output is the diagnostic the user asked for: "check and report if
all the data needed for compliance is being enforced for every step."
"""
env = env  # noqa


def banner(t):
    print(f'\n{"="*78}\n  {t}\n{"="*78}')


# Per-kind required data points. Each tuple is (field_or_check, severity, why)
KIND_RULES = {
    'wet': [
        ('x_fc_assigned_user_id',  'CRITICAL', 'Operator (audit trail)'),
        ('x_fc_bath_id',           'CRITICAL', 'Which bath ran (chemistry traceability)'),
        ('x_fc_tank_id',           'CRITICAL', 'Which physical tank'),
        ('duration',               'CRITICAL', 'Actual run time'),
        ('x_fc_thickness_target',  'IMPORTANT','Spec target (QC accept criterion)'),
        ('x_fc_dwell_time_minutes','IMPORTANT','Recipe dwell vs actual'),
        ('x_fc_rack_id',           'IMPORTANT','Which rack/fixture used'),
        ('bath_log_during_window', 'IMPORTANT','Chemistry reading recorded during WO time window'),
        ('x_fc_started_by_user_id','IMPORTANT','Who actually clicked Start'),
        ('x_fc_finished_by_user_id','IMPORTANT','Who clicked Finish'),
    ],
    'bake': [
        ('x_fc_assigned_user_id',  'CRITICAL', 'Operator'),
        ('x_fc_oven_id',           'CRITICAL', 'Which oven'),
        ('x_fc_bake_temp',         'CRITICAL', 'Setpoint temp (Nadcap req)'),
        ('x_fc_bake_duration_hours','CRITICAL','Actual bake duration'),
        ('chart_recorder_ref',     'CRITICAL', 'Chart-recorder ref on the OVEN - auditor demands the chart for the run'),
        ('duration',               'CRITICAL', 'WO timer duration'),
        ('x_fc_started_by_user_id','IMPORTANT','Who started'),
        ('x_fc_finished_by_user_id','IMPORTANT','Who finished'),
    ],
    'mask': [
        ('x_fc_assigned_user_id',  'CRITICAL', 'Operator'),
        ('duration',               'CRITICAL', 'Run time'),
        ('masking_material',       'IMPORTANT','Which material - needed for stripping later'),
        ('x_fc_started_by_user_id','IMPORTANT','Who started'),
        ('x_fc_finished_by_user_id','IMPORTANT','Who finished'),
    ],
    'rack': [
        ('x_fc_assigned_user_id',  'CRITICAL', 'Operator'),
        ('x_fc_rack_id',           'CRITICAL', 'Which rack/fixture (per-rack MTO life tracking)'),
        ('duration',               'CRITICAL', 'Run time'),
        ('x_fc_started_by_user_id','IMPORTANT','Who started'),
        ('x_fc_finished_by_user_id','IMPORTANT','Who finished'),
    ],
    'inspect': [
        ('x_fc_assigned_user_id',  'CRITICAL', 'Inspector'),
        ('duration',               'CRITICAL', 'Run time'),
        ('thickness_readings',     'CRITICAL', 'Fischerscope readings logged for this MO'),
        ('cal_std_on_readings',    'CRITICAL', 'Every reading has calibration std (Nadcap)'),
        ('gauge_serial',           'IMPORTANT','Which gauge (links to calibration record)'),
        ('x_fc_started_by_user_id','IMPORTANT','Who started'),
        ('x_fc_finished_by_user_id','IMPORTANT','Who finished'),
    ],
    'other': [
        ('x_fc_assigned_user_id',  'IMPORTANT','Operator'),
        ('duration',               'IMPORTANT','Run time'),
    ],
}


def check_field(wo, field):
    """Return (value, is_filled, label_for_display)."""
    if field == 'bath_log_during_window':
        # Look for any bath log on this WO's bath, between start+finish
        if not wo.x_fc_bath_id or not wo.x_fc_started_at or not wo.x_fc_finished_at:
            return ('-', False, 'no log searchable')
        Log = env['fusion.plating.bath.log']
        n = Log.search_count([
            ('bath_id', '=', wo.x_fc_bath_id.id),
            ('log_date', '>=', wo.x_fc_started_at),
            ('log_date', '<=', wo.x_fc_finished_at),
        ])
        return (f'{n} log(s)', n > 0, '')
    if field == 'chart_recorder_ref':
        ref = wo.x_fc_oven_id.chart_recorder_ref if wo.x_fc_oven_id else False
        return (ref or '-', bool(ref), 'on oven')
    if field == 'masking_material':
        val = wo.x_fc_masking_material if 'x_fc_masking_material' in wo._fields else False
        if not val:
            return ('-', False, '')
        label = dict(wo._fields['x_fc_masking_material'].selection).get(val, val)
        return (label, True, '')
    if field == 'thickness_readings':
        n = env['fp.thickness.reading'].search_count([
            ('production_id', '=', wo.production_id.id),
        ])
        return (f'{n} reading(s)', n > 0, '')
    if field == 'cal_std_on_readings':
        rs = env['fp.thickness.reading'].search([
            ('production_id', '=', wo.production_id.id),
        ])
        if not rs:
            return ('-', False, 'no readings')
        n_with = sum(1 for r in rs if r.calibration_std_ref)
        return (f'{n_with}/{len(rs)} have cal std', n_with == len(rs), '')
    if field == 'gauge_serial':
        # Pull from any reading on this MO
        r = env['fp.thickness.reading'].search(
            [('production_id', '=', wo.production_id.id)], limit=1)
        if not r:
            return ('-', False, 'no readings')
        return (r.equipment_model or '-', bool(r.equipment_model), 'from reading.equipment_model')
    # Direct field on WO
    val = getattr(wo, field, False) if field in wo._fields else None
    if val is None:
        return ('(field n/a)', False, '')
    if hasattr(val, '_name'):
        label = val.display_name if val else '-'
        return (label, bool(val.ids), '')
    if isinstance(val, (int, float)):
        return (str(val), val > 0, '')
    return (str(val), bool(val), '')


# Pull the most recent MO with all its WOs (sudo to bypass any
# multi-company / record-rule filter so we always pick the truly latest).
mo = env['mrp.production'].sudo().search(
    [('state', '=', 'done')], order='id desc', limit=1)
print(f'\nAuditing MO: {mo.name} (state={mo.state}, recipe={mo.x_fc_recipe_id.name})')
print(f'{len(mo.workorder_ids)} work orders\n')

GAP_TOTALS = {'CRITICAL': 0, 'IMPORTANT': 0}
PER_KIND = {}

for wo in mo.workorder_ids.sorted('sequence'):
    kind = wo._fp_classify_kind() if hasattr(wo, '_fp_classify_kind') else 'other'
    rules = KIND_RULES.get(kind, KIND_RULES['other'])
    banner(f'WO {wo.id}: "{wo.name}"   kind={kind}')
    show_gaps = []
    show_ok = []
    for field, severity, why in rules:
        val_str, is_filled, note = check_field(wo, field)
        sym = '✓' if is_filled else '✗'
        line = f'  {sym} {severity:<9} {field:<30} → {val_str:<35} {why}'
        if note:
            line += f'  [{note}]'
        if is_filled:
            show_ok.append(line)
        else:
            show_gaps.append(line)
            if severity in GAP_TOTALS:
                GAP_TOTALS[severity] += 1
            PER_KIND.setdefault(kind, []).append(field)
    for ln in show_ok:
        print(ln)
    if show_gaps:
        print('  ── GAPS ──')
        for ln in show_gaps:
            print(ln)

# =====================================================================
banner('SUMMARY - gaps per WO kind across this MO')
# =====================================================================

for kind, gaps in PER_KIND.items():
    from collections import Counter
    c = Counter(gaps)
    print(f'\n  {kind} WOs ({sum(1 for w in mo.workorder_ids if (w._fp_classify_kind() if hasattr(w,"_fp_classify_kind") else "other") == kind)} of them):')
    for field, n in c.most_common():
        print(f'    × {field:<30} missing in {n} WO(s)')

print(f'\n  Totals: {GAP_TOTALS["CRITICAL"]} CRITICAL gaps, {GAP_TOTALS["IMPORTANT"]} IMPORTANT gaps')
print('\n  Note: "missing" doesn\'t always mean "broken" - some fields')
print('        are optional today but should be required for stricter')
print('        AS9100 / Nadcap compliance. See the per-kind list to')
print('        decide which are real bugs vs roadmap items.')