import difflib import logging from odoo import _, api, fields, models from odoo.exceptions import UserError _logger = logging.getLogger(__name__) # Fuzzy name-match threshold (difflib.SequenceMatcher ratio). Per spec. FUZZY_THRESHOLD = 0.85 class FlConflictCheck(models.Model): _name = 'fl.conflict.check' _description = 'Conflict of Interest Check' _inherit = ['mail.thread'] _order = 'create_date desc' _rec_name = 'name' name = fields.Char( string='Reference', compute='_compute_name', store=True ) case_id = fields.Many2one( 'fl.case', string='Case', required=True, ondelete='cascade', index=True ) check_date = fields.Datetime( string='Check Date', default=fields.Datetime.now, readonly=True ) status = fields.Selection([ ('clear', 'Clear — No Conflict'), ('conflict', 'Conflict Detected'), ('override', 'Conflict Overridden'), ], string='Status', default='clear', required=True, tracking=True) matched_case_ids = fields.Many2many( 'fl.case', 'fl_conflict_check_matched_case_rel', 'conflict_id', 'matched_case_id', string='Conflicting Cases' ) match_detail = fields.Text(string='Match Detail', readonly=True) # ── Override (group_admin only) ──────────────────────────────────────── override_justification = fields.Text( string='Override Justification', help='Written justification required to override a detected conflict.' ) override_user_id = fields.Many2one( 'res.users', string='Overridden By', readonly=True ) override_date = fields.Datetime(string='Override Date', readonly=True) @api.depends('case_id', 'case_id.name') def _compute_name(self): for rec in self: rec.name = _('Conflict Check — %s') % (rec.case_id.name or _('New')) # ────────────────────────────────────────────────────────────────────── # Engine # ────────────────────────────────────────────────────────────────────── @api.model def run_check(self, case): """ Run a conflict-of-interest screen for the given case. Creates and returns an fl.conflict.check record, sets case.conflict_check_passed and links the latest check. Never silently passes a conflict — always surfaces it to chatter. """ names = self._collect_party_names(case) matches = self._find_conflicts(case, names) check = self.create({ 'case_id': case.id, 'status': 'conflict' if matches else 'clear', 'matched_case_ids': ( [(6, 0, list({m['case_id'] for m in matches}))] if matches else False ), 'match_detail': ( self._format_matches(matches) if matches else 'No conflicts found.' ), }) case.conflict_check_id = check.id if matches: case.conflict_check_passed = False case.message_post( body=( '
' '

' '🚩 CONFLICT OF INTEREST DETECTED

' '

One or more parties on this case match parties on other ' 'open cases. This case cannot advance past Intake until ' 'the conflict is resolved or overridden by an administrator.

' '
'
                    f'{self._format_matches(matches)}
' '
' ), subtype_xmlid='mail.mt_comment', ) else: case.conflict_check_passed = True case.message_post( body='✅ Conflict-of-interest check clear — no matching parties on other open cases.', subtype_xmlid='mail.mt_note', ) _logger.info( "FL Conflict Check: case %s → %s (%d match(es))", case.id, check.status, len(matches) ) return check def _collect_party_names(self, case): """ Collect (partner_id, name, role_label) tuples for every party on a case. Pulls from petitioner_id, respondent_id, and party_ids — deduplicated by partner. Drops entries with no name. """ entries = [] seen = set() def _add(partner, role_label): if partner and partner.id not in seen and partner.name: entries.append((partner.id, partner.name, role_label)) seen.add(partner.id) _add(case.petitioner_id, 'Petitioner') _add(case.respondent_id, 'Respondent') for party in case.party_ids: role_label = dict(party._fields['role'].selection).get(party.role, '') _add(party.partner_id, role_label) return entries def _find_conflicts(self, case, names): """ Compare this case's party names against parties on other open cases. Exact partner-id match OR fuzzy name match (>= FUZZY_THRESHOLD) flags a conflict. Folded stages (Closed, Referred to Attorney) are treated as not-open and skipped. Returns a list of match dicts. """ if not names: return [] other_cases = self.env['fl.case'].search([ ('id', '!=', case.id), ('active', '=', True), ]) matches = [] for other in other_cases: if other.stage_id and other.stage_id.fold: continue other_names = self._collect_party_names(other) for pid, pname, prole in names: for opid, opname, oprole in other_names: if pid == opid: ratio = 1.0 else: ratio = difflib.SequenceMatcher( None, (pname or '').lower().strip(), (opname or '').lower().strip(), ).ratio() if ratio >= FUZZY_THRESHOLD: matches.append({ 'case_id': other.id, 'case_ref': other.name, 'new_party': f'{pname} ({prole})', 'existing_party': f'{opname} ({oprole})', 'score': round(ratio, 2), }) return matches def _format_matches(self, matches): lines = [] for m in matches: lines.append( f"• {m['new_party']} ↔ {m['existing_party']} " f"on case {m['case_ref']} (similarity {m['score']:.0%})" ) return '\n'.join(lines) # ────────────────────────────────────────────────────────────────────── # Actions # ────────────────────────────────────────────────────────────────────── def action_override(self): """Manager-only override of a detected conflict. Requires justification.""" self.ensure_one() if not self.env.user.has_group('activeblue_familylaw.group_admin'): raise UserError(_( 'Only a Family Law Administrator can override a conflict of interest.' )) if not (self.override_justification or '').strip(): raise UserError(_( 'A written justification is required to override a conflict.' )) self.write({ 'status': 'override', 'override_user_id': self.env.user.id, 'override_date': fields.Datetime.now(), }) self.case_id.conflict_check_passed = True self.case_id.message_post( body=( f'⚠ CONFLICT OVERRIDE authorized by ' f'{self.env.user.name}.
' f'Justification: {self.override_justification}' ), subtype_xmlid='mail.mt_comment', ) return True