Odoo-Modules/fusion_plating/fusion_plating/models/fp_migration.py

# -*- coding: utf-8 -*-
"""Phase H — dry-run + Owner-approval migration workflow."""
import json
import logging
from datetime import timedelta

from markupsafe import Markup

from odoo import _, api, fields, models
from odoo.exceptions import UserError

from .fp_role_constants import (
    _FP_OLD_GROUP_XMLIDS,
    _NEW_ROLE_XMLID,
    fp_resolve_target_role,
)


_logger = logging.getLogger(__name__)


_ROLE_SELECTION = [
    ('no',              'No'),
    ('technician',      'Technician'),
    ('sales_rep',       'Sales Representative'),
    ('shop_manager',    'Shop Manager'),
    ('sales_manager',   'Sales Manager'),
    ('manager',         'Manager'),
    ('quality_manager', 'Quality Manager'),
    ('owner',           'Owner'),
]


class FpMigrationPreview(models.Model):
    _name = 'fp.migration.preview'
    _description = 'Fusion Plating Role Migration Preview'
    _inherit = ['mail.thread']
    _order = 'create_date desc'

    name = fields.Char(
        default=lambda s: _('Migration %s') % fields.Datetime.now(),
        tracking=True,
    )
    state = fields.Selection(
        [
            ('pending',     'Pending Review'),
            ('approved',    'Approved & Applied'),
            ('cancelled',   'Cancelled'),
            ('rolled_back', 'Rolled Back'),
        ],
        default='pending',
        required=True,
        tracking=True,
    )
    line_ids = fields.One2many('fp.migration.preview.line', 'preview_id')
    user_count = fields.Integer(compute='_compute_counts', store=True)
    warning_count = fields.Integer(compute='_compute_counts', store=True)
    approved_by_id = fields.Many2one('res.users', readonly=True)
    approved_at = fields.Datetime(readonly=True)
    rollback_deadline = fields.Datetime(compute='_compute_rollback_deadline')

    @api.depends('line_ids', 'line_ids.warning')
    def _compute_counts(self):
        for rec in self:
            rec.user_count = len(rec.line_ids)
            rec.warning_count = sum(1 for ln in rec.line_ids if ln.warning)

    @api.depends('approved_at')
    def _compute_rollback_deadline(self):
        for rec in self:
            rec.rollback_deadline = (
                rec.approved_at + timedelta(days=30) if rec.approved_at else False
            )

    def _fp_build_lines(self):
        """Walk all active internal users; one line per user with the
        proposed role + capability_delta."""
        self.ensure_one()
        Line = self.env['fp.migration.preview.line']
        users = self.env['res.users'].search([
            ('share', '=', False),
            ('active', '=', True),
        ])
        vals_list = []
        for user in users:
            role, delta = fp_resolve_target_role(user)
            vals_list.append({
                'preview_id': self.id,
                'user_id': user.id,
                'proposed_role': role,
                'capability_delta': delta or '',
                'warning': bool(delta),
            })
        if vals_list:
            Line.create(vals_list)

    def _fp_notify_owners(self):
        """Schedule a 'Review Fusion Plating role migration' activity on
        every Owner user. Idempotent — won't double-schedule."""
        self.ensure_one()
        owner_grp = self.env.ref('fusion_plating.group_fp_owner', raise_if_not_found=False)
        if not owner_grp:
            return
        owners = owner_grp.user_ids.filtered(lambda u: u.active and not u.share)
        if not owners:
            _logger.warning('Fusion Plating migration preview %s: no Owner users to notify', self.id)
            return
        activity_type = self.env.ref('mail.mail_activity_data_todo')
        for owner in owners:
            existing = self.env['mail.activity'].search([
                ('res_model_id', '=', self.env.ref('fusion_plating.model_fp_migration_preview').id),
                ('res_id', '=', self.id),
                ('user_id', '=', owner.id),
            ], limit=1)
            if existing:
                continue
            self.env['mail.activity'].create({
                'res_model_id': self.env.ref('fusion_plating.model_fp_migration_preview').id,
                'res_id': self.id,
                'activity_type_id': activity_type.id,
                'summary': _('Review Fusion Plating role migration'),
                'note': _('%(n)d users affected, %(w)d with capability changes.') % {
                    'n': self.user_count,
                    'w': self.warning_count,
                },
                'user_id': owner.id,
                'date_deadline': fields.Date.today(),
            })

    def action_approve_and_run(self):
        self.ensure_one()
        if not self.env.user.has_group('fusion_plating.group_fp_owner'):
            raise UserError(_('Only Owners can approve role migrations.'))
        if self.state != 'pending':
            raise UserError(_(
                'Migration is no longer pending - current state: %s'
            ) % self.state)

        # Resolve old group ids once
        old_group_ids = []
        for xmlid in _FP_OLD_GROUP_XMLIDS:
            g = self.env.ref(xmlid, raise_if_not_found=False)
            if g:
                old_group_ids.append(g.id)

        for line in self.line_ids:
            user = line.user_id
            # Snapshot current group_ids for rollback
            line.applied_groups_snapshot = json.dumps(user.group_ids.ids)

            # Remove old plating-role groups
            if old_group_ids:
                user.sudo().write({
                    'group_ids': [(3, gid) for gid in old_group_ids]
                })

            # Add the new role group (no-op for 'no')
            target_xmlid = _NEW_ROLE_XMLID.get(line.proposed_role)
            if target_xmlid:
                target = self.env.ref(target_xmlid, raise_if_not_found=False)
                if target:
                    user.sudo().write({'group_ids': [(4, target.id)]})

            # Audit chatter on the user
            user.partner_id.message_post(
                body=Markup(_(
                    'Plating role assigned by migration: <b>%s</b>'
                )) % line.proposed_role,
                message_type='notification',
            )

            # Special: CGP DO becomes a res.company field, not a role
            if line.capability_delta and 'CGP DO' in line.capability_delta:
                user.company_id.x_fc_cgp_designated_official_id = user.id

        self.write({
            'state': 'approved',
            'approved_by_id': self.env.user.id,
            'approved_at': fields.Datetime.now(),
        })

    def action_cancel(self):
        self.ensure_one()
        if self.state != 'pending':
            raise UserError(_('Only pending migrations can be cancelled.'))
        self.state = 'cancelled'

    def action_rollback(self):
        self.ensure_one()
        if self.state != 'approved':
            raise UserError(_('Only approved migrations can be rolled back.'))
        if self.rollback_deadline and fields.Datetime.now() > self.rollback_deadline:
            raise UserError(_(
                'Rollback window has expired (30 days after approval). '
                'Restore from pg_dump backup instead.'
            ))
        for line in self.line_ids:
            if line.applied_groups_snapshot:
                old_ids = json.loads(line.applied_groups_snapshot)
                line.user_id.sudo().write({'group_ids': [(6, 0, old_ids)]})
        self.state = 'rolled_back'

    @api.model
    def _cron_purge_expired_migrations(self):
        """After 30 days, clear snapshots + unlink old plating groups.
        Runs daily via fp_migration_cron.xml."""
        deadline = fields.Datetime.now() - timedelta(days=30)
        expired = self.search([
            ('state', '=', 'approved'),
            ('approved_at', '<', deadline),
        ])
        if not expired:
            return
        # Clear snapshots (no more rollback possible)
        for preview in expired:
            preview.line_ids.write({'applied_groups_snapshot': False})
        # Unlink old plating groups (now confirmed unused — every user is
        # on the new groups; backward-compat implied_ids chains can drop)
        old_group_ids = []
        for xmlid in _FP_OLD_GROUP_XMLIDS:
            g = self.env.ref(xmlid, raise_if_not_found=False)
            if g:
                old_group_ids.append(g.id)
        if old_group_ids:
            # I6 safety check — never unlink a group that still has active
            # internal users on it. If anyone still references the group
            # we'd cascade-strip them silently from their permissions.
            safe_to_unlink = []
            skipped = []
            for old_group in self.env['res.groups'].browse(old_group_ids).exists():
                active_users = old_group.user_ids.filtered(lambda u: u.active and not u.share)
                if active_users:
                    skipped.append((old_group.name, active_users.mapped('login')))
                else:
                    safe_to_unlink.append(old_group.id)
            if skipped:
                _logger.warning(
                    'Fusion Plating migration purge: skipped %d old groups with active users: %s',
                    len(skipped), skipped)
            if safe_to_unlink:
                self.env['res.groups'].browse(safe_to_unlink).unlink()
                _logger.info('Fusion Plating migration: purged %d expired old plating groups',
                             len(safe_to_unlink))


class FpMigrationPreviewLine(models.Model):
    _name = 'fp.migration.preview.line'
    _description = 'Migration Preview Line'

    preview_id = fields.Many2one('fp.migration.preview', required=True, ondelete='cascade')
    user_id = fields.Many2one('res.users', required=True, ondelete='cascade')
    current_groups = fields.Char(compute='_compute_current_groups')
    proposed_role = fields.Selection(_ROLE_SELECTION)
    capability_delta = fields.Char()
    warning = fields.Boolean()
    notes = fields.Text(help='Owner may annotate before approving')
    applied_groups_snapshot = fields.Text(help='JSON of pre-migration group_ids for rollback')

    @api.depends('user_id', 'user_id.group_ids')
    def _compute_current_groups(self):
        for line in self:
            if line.user_id:
                line.current_groups = ', '.join(line.user_id.group_ids.mapped('name'))
            else:
                line.current_groups = ''