diff --git a/fusion_plating/fusion_plating_quality/controllers/fp_quality_dashboard.py b/fusion_plating/fusion_plating_quality/controllers/fp_quality_dashboard.py index 695d7708..1b3549fb 100644 --- a/fusion_plating/fusion_plating_quality/controllers/fp_quality_dashboard.py +++ b/fusion_plating/fusion_plating_quality/controllers/fp_quality_dashboard.py @@ -90,13 +90,23 @@ class FpQualityDashboardSnapshot: def build(self): sections = [] + all_banner_candidates = [] for type_code in SECTION_ORDER: + cfg = TYPE_CONFIG[type_code] + if cfg['model'] not in self.env: + continue + Model = self.env[cfg['model']].sudo() section = self._build_section(type_code) if section is None: - continue # model not installed — omit + continue sections.append(section) - # Banner placeholder for Task 2 — fully implemented in Task 4. - banner = {'items': [], 'all_clear': True, 'total_matching': 0} + # Collect banner candidates with type_code attached + for rec, urgency, badge in self._fetch_banner_candidates( + type_code, Model): + all_banner_candidates.append( + (rec, urgency, badge, type_code), + ) + banner = self._build_banner(all_banner_candidates) return { 'banner': banner, 'sections': sections, @@ -217,6 +227,132 @@ class FpQualityDashboardSnapshot: return f'{age_str} overdue' if age_str else 'overdue' return age_str + # ===== Banner ===================================================== + + def _fetch_banner_candidates(self, type_code, Model): + """Per-type pull of records that qualify for the banner: + (overdue) OR (critical-customer AND state-is-open). Returns + list of (rec, urgency, critical_badge) tuples — deduped. + """ + overdue_ids = set(self._overdue_ids(type_code, Model)) + critical_ids = set(self._critical_customer_ids(type_code, Model)) + # Open state filter (anything in the type's state_domain) + state_dom = list(OVERDUE_THRESHOLDS[type_code]['state_domain']) + open_recs = Model.search(state_dom) + results = [] + for rec in open_recs: + is_overdue = rec.id in overdue_ids + is_critical = rec.id in critical_ids + if not (is_overdue or is_critical): + continue + urgency = 'overdue' if is_overdue else 'critical_customer' + badge = None if is_overdue else self._critical_badge(rec) + results.append((rec, urgency, badge)) + return results + + def _critical_customer_ids(self, type_code, Model): + """IDs of records belonging to critical customers (rush / vip / + aerospace). Defensive on missing fields per Rule 24.""" + partner_field_map = { + 'cert': 'partner_id', + 'hold': 'partner_id', + 'ncr': 'partner_id', + 'rma': 'partner_id', + 'capa': 'partner_id', # may be False if linked via ncr_id + 'check': 'job_id.partner_id', + } + # Find partners with critical flags + Partner = self.env['res.partner'].sudo() + critical_partner_clauses = [] + if 'x_fc_rush' in Partner._fields: + critical_partner_clauses.append(('x_fc_rush', '=', True)) + if 'x_fc_vip' in Partner._fields: + critical_partner_clauses.append(('x_fc_vip', '=', True)) + if not critical_partner_clauses: + return [] # no flags defined in this codebase + # OR the conditions + critical_partner_domain = list(critical_partner_clauses) + if len(critical_partner_clauses) > 1: + critical_partner_domain = ( + ['|'] * (len(critical_partner_clauses) - 1) + + critical_partner_clauses + ) + critical_partner_ids = Partner.search(critical_partner_domain).ids + if not critical_partner_ids: + return [] + partner_path = partner_field_map.get(type_code, 'partner_id') + # Compose the record-side filter — direct or via dotted path + dom = list(OVERDUE_THRESHOLDS[type_code]['state_domain']) + dom.append((partner_path, 'in', critical_partner_ids)) + try: + return Model.search(dom).ids + except Exception as e: + # Field doesn't exist on the model (e.g. fusion.plating.capa + # may not have partner_id directly). Best-effort: skip. + _logger.debug( + "critical_customer probe failed for %s: %s", type_code, e, + ) + return [] + + def _critical_badge(self, rec): + """Return 'RUSH' / 'VIP' / 'AEROSPACE' / 'AS9100' / None based + on which signal qualified the record. Defensive on missing fields.""" + partner = self._resolve_partner(rec) + if partner: + if 'x_fc_rush' in partner._fields and getattr(partner, 'x_fc_rush', False): + return 'RUSH' + if 'x_fc_vip' in partner._fields and getattr(partner, 'x_fc_vip', False): + return 'VIP' + # Aerospace — check part name OR spec code if reachable + for path in ('part_catalog_id.name', 'customer_spec_id.code'): + head, _, attr = path.partition('.') + if head in rec._fields and rec[head] \ + and attr in rec[head]._fields: + val = (rec[head][attr] or '').upper() + if 'AEROSPACE' in val: + return 'AEROSPACE' + if val.startswith('AS9100') or val.startswith('NADCAP'): + return 'AS9100' + return None + + def _build_banner(self, all_candidates): + """Rank: overdue first by oldest create_date, then critical- + customer by oldest create_date. Take top 6. + all_candidates: list of (rec, urgency, badge, type_code) tuples. + """ + def rank_key(tup): + rec, urgency, _badge, _t = tup + # overdue=0 sorts before critical_customer=1; then oldest first. + return (0 if urgency == 'overdue' else 1, + rec.create_date or self.now) + ranked = sorted(all_candidates, key=rank_key) + top = ranked[:self.BANNER_MAX] + return { + 'items': [self._build_banner_item(t) for t in top], + 'all_clear': len(all_candidates) == 0, + 'total_matching': len(all_candidates), + } + + def _build_banner_item(self, tup): + """Shape one banner item from (rec, urgency, badge, type_code).""" + rec, urgency, badge, type_code = tup + cfg = TYPE_CONFIG[type_code] + partner = self._resolve_partner(rec) + is_overdue = (urgency == 'overdue') + return { + 'type': type_code, + 'id': rec.id, + 'name': rec.display_name or rec.name or f'#{rec.id}', + 'customer': partner.name if partner else '', + 'subtitle': self._build_subtitle(type_code, rec, is_overdue), + 'urgency': urgency, + 'critical_badge': badge, + 'open_action': { + 'res_model': cfg['model'], + 'res_id': rec.id, + }, + } + class FpQualityDashboardController(http.Controller): diff --git a/fusion_plating/fusion_plating_quality/tests/test_dashboard_snapshot.py b/fusion_plating/fusion_plating_quality/tests/test_dashboard_snapshot.py index 7930d3a2..41ea9855 100644 --- a/fusion_plating/fusion_plating_quality/tests/test_dashboard_snapshot.py +++ b/fusion_plating/fusion_plating_quality/tests/test_dashboard_snapshot.py @@ -109,3 +109,64 @@ class TestDashboardSnapshotItems(TransactionCase): self.assertEqual(item['open_action']['res_model'], 'fusion.plating.quality.hold') self.assertEqual(item['open_action']['res_id'], hold.id) + + +class TestDashboardSnapshotBanner(TransactionCase): + """Banner population + ranking.""" + + def _build(self): + from odoo.addons.fusion_plating_quality.controllers.fp_quality_dashboard \ + import FpQualityDashboardSnapshot + return FpQualityDashboardSnapshot(self.env).build() + + def test_banner_picks_up_overdue_hold(self): + from datetime import datetime, timedelta + if 'fusion.plating.quality.hold' not in self.env: + self.skipTest('fusion.plating.quality.hold not installed') + partner = self.env['res.partner'].create({'name': 'Cust'}) + hold = self.env['fusion.plating.quality.hold'].create({ + 'partner_id': partner.id, 'state': 'on_hold', 'reason': 'old', + }) + # Backdate create_date past the 3-day overdue threshold + old = datetime.now() - timedelta(days=5) + self.env.cr.execute( + "UPDATE fusion_plating_quality_hold SET create_date = %s WHERE id = %s", + (old, hold.id), + ) + hold.invalidate_recordset(['create_date']) + + snap = self._build() + self.assertFalse(snap['banner']['all_clear']) + self.assertEqual(snap['banner']['total_matching'], 1) + self.assertEqual(len(snap['banner']['items']), 1) + item = snap['banner']['items'][0] + self.assertEqual(item['type'], 'hold') + self.assertEqual(item['urgency'], 'overdue') + + def test_banner_caps_at_6_with_overflow_count(self): + from datetime import datetime, timedelta + if 'fusion.plating.quality.hold' not in self.env: + self.skipTest('fusion.plating.quality.hold not installed') + partner = self.env['res.partner'].create({'name': 'Cust'}) + Hold = self.env['fusion.plating.quality.hold'] + # 8 overdue holds + old = datetime.now() - timedelta(days=5) + for i in range(8): + h = Hold.create({ + 'partner_id': partner.id, 'state': 'on_hold', + 'reason': f'overdue {i}', + }) + self.env.cr.execute( + "UPDATE fusion_plating_quality_hold SET create_date = %s WHERE id = %s", + (old, h.id), + ) + snap = self._build() + self.assertEqual(len(snap['banner']['items']), 6) + self.assertEqual(snap['banner']['total_matching'], 8) + self.assertFalse(snap['banner']['all_clear']) + + def test_banner_all_clear_when_zero(self): + snap = self._build() + # Empty DB — no overdue, no critical + self.assertTrue(snap['banner']['all_clear']) + self.assertEqual(snap['banner']['items'], [])