<filename>addons/mail/models/mail_blacklist.py
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.

import logging

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

_logger = logging.getLogger(__name__)


class MailBlackList(models.Model):
    """ Model of blacklisted email addresses to stop sending emails."""
    _name = 'mail.blacklist'
    _inherit = ['mail.thread']
    _description = 'Mail Blacklist'
    _rec_name = 'email'

    email = fields.Char(string='Email Address', required=True, index=True, help='This field is case insensitive.',
                        track_visibility=True)
    active = fields.Boolean(default=True, track_visibility=True)

    _sql_constraints = [
        ('unique_email', 'unique (email)', 'Email address already exists!')
    ]

    @api.model_create_multi
    def create(self, values):
        # First of all, extract values to ensure emails are really unique (and don't modify values in place)
        new_values = []
        all_emails = []
        for value in values:
            email = self._sanitize_email(value.get('email'))
            if not email:
                raise UserError(_('Invalid email address %r') % value['email'])
            if email in all_emails:
                continue
            all_emails.append(email)
            new_value = dict(value, email=email)
            new_values.append(new_value)

        """ To avoid crash during import due to unique email, return the existing records if any """
        sql = '''SELECT email, id FROM mail_blacklist WHERE email = ANY(%s)'''
        emails = [v['email'] for v in new_values]
        self._cr.execute(sql, (emails,))
        bl_entries = dict(self._cr.fetchall())
        to_create = [v for v in new_values
                     if v['email'] not in bl_entries]

        # TODO DBE Fixme : reorder ids according to incoming ids.
        results = super(MailBlackList, self).create(to_create)
        return self.env['mail.blacklist'].browse(bl_entries.values()) | results

    @api.multi
    def write(self, values):
        if 'email' in values:
            values['email'] = self._sanitize_email(values['email'])
        return super(MailBlackList, self).write(values)

    def _search(self, args, offset=0, limit=None, order=None, count=False, access_rights_uid=None):
        """ Override _search in order to grep search on email field and make it
        lower-case and sanitized """
        if args:
            new_args = []
            for arg in args:
                if isinstance(arg, (list, tuple)) and arg[0] == 'email' and isinstance(arg[2], tools.pycompat.text_type):
                    sanitized = self.env['mail.blacklist']._sanitize_email(arg[2])
                    if sanitized:
                        new_args.append([arg[0], arg[1], sanitized])
                    else:
                        new_args.append(arg)
                else:
                    new_args.append(arg)
        else:
            new_args = args
        return super(MailBlackList, self)._search(new_args, offset=offset, limit=limit, order=order, count=count, access_rights_uid=access_rights_uid)

    def _add(self, email):
        sanitized = self._sanitize_email(email)
        record = self.env["mail.blacklist"].with_context(active_test=False).search([('email', '=', sanitized)])
        if len(record) > 0:
            record.write({'active': True})
        else:
            record = self.create({'email': email})
        return record

    def _remove(self, email):
        sanitized = self._sanitize_email(email)
        record = self.env["mail.blacklist"].with_context(active_test=False).search([('email', '=', sanitized)])
        if len(record) > 0:
            record.write({'active': False})
        else:
            record = record.create({'email': email, 'active': False})
        return record

    def _sanitize_email(self, email):
        """ Sanitize and standardize blacklist entries: all emails should be
        only real email extracted from strings (A <a@a> -> a@a)  and should be
        lower case. """
        emails = tools.email_split(email)
        if not emails or len(emails) != 1:
            return False
        return emails[0].lower()


class MailBlackListMixin(models.AbstractModel):
    """ Mixin that is inherited by all model with opt out.
        USAGE : the field '_primary_email' must be overridden in the model that inherit the mixin
        and must contain the email field of the model.
        """
    _name = 'mail.blacklist.mixin'
    _description = 'Mail Blacklist mixin'
    _primary_email = ['email']

    # Note : is_blacklisted sould only be used for display. As the compute is not depending on the blacklist,
    # once read, it won't be re-computed again if the blacklist is modified in the same request.
    is_blacklisted = fields.Boolean(string='Blacklist', compute="_compute_is_blacklisted", compute_sudo=True,
        store=False, search="_search_is_blacklisted", groups="base.group_user",
        help="If the email address is on the blacklist, the contact won't receive mass mailing anymore, from any list")

    def _assert_primary_email(self):
        if not hasattr(self, "_primary_email") or \
                not isinstance(self._primary_email, (list, tuple)) or \
                not len(self._primary_email) == 1:
            raise UserError(_('Invalid primary email field on model %s') % self._name)
        field_name = self._primary_email[0]
        if field_name not in self._fields or self._fields[field_name].type != 'char':
            raise UserError(_('Invalid primary email field on model %s') % self._name)

    @api.model
    def _search_is_blacklisted(self, operator, value):
        # Assumes operator is '=' or '!=' and value is True or False
        self._assert_primary_email()
        if operator != '=':
            if operator == '!=' and isinstance(value, bool):
                value = not value
            else:
                raise NotImplementedError()

        [email_field] = self._primary_email
        if value:
            query = """
                SELECT m.id
                    FROM mail_blacklist bl
                    JOIN %s m
                    ON (LOWER(substring(m.%s, '([^ ,;<@]+@[^> ,;]+)')) = bl.email AND bl.active)
            """
        else:
            query = """
                SELECT m.id
                    FROM %s m
                    LEFT JOIN mail_blacklist bl
                    ON (LOWER(substring(m.%s, '([^ ,;<@]+@[^> ,;]+)')) = bl.email AND bl.active)
                    WHERE bl.id IS NULL
            """
        self._cr.execute(query % (self._table, email_field))
        res = self._cr.fetchall()
        if not res:
            return [(0, '=', 1)]
        return [('id', 'in', [r[0] for r in res])]

    @api.depends(lambda self: self._primary_email)
    def _compute_is_blacklisted(self):
        self._assert_primary_email()
        [email_field] = self._primary_email
        # TODO : Should remove the sudo as compute_sudo defined on methods.
        # But if user doesn't have access to mail.blacklist, doen't work without sudo().
        sanitized = [self.env['mail.blacklist']._sanitize_email(email) for email in self.mapped(email_field)]
        blacklist = set(self.env['mail.blacklist'].sudo().search([('email', 'in', sanitized)]).mapped('email'))
        for record in self:
            record.is_blacklisted = self.env['mail.blacklist']._sanitize_email(record[email_field]) in blacklist

<filename>addons/mail/models/mail_blacklist.py
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.

import logging

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

_logger = logging.getLogger(__name__)


class MailBlackList(models.Model):
    """ Model of blacklisted email addresses to stop sending emails."""
    _name = 'mail.blacklist'
    _inherit = ['mail.thread']
    _description = 'Mail Blacklist'
    _rec_name = 'email'

    email = fields.Char(string='Email Address', required=True, index=True, help='This field is case insensitive.',
                        track_visibility=True)
    active = fields.Boolean(default=True, track_visibility=True)

    _sql_constraints = [
        ('unique_email', 'unique (email)', 'Email address already exists!')
    ]

    @api.model_create_multi
    def create(self, values):
        # First of all, extract values to ensure emails are really unique (and don't modify values in place)
        new_values = []
        all_emails = []
        for value in values:
            email = self._sanitize_email(value.get('email'))
            if not email:
                raise UserError(_('Invalid email address %r') % value['email'])
            if email in all_emails:
                continue
            all_emails.append(email)
            new_value = dict(value, email=email)
            new_values.append(new_value)

        """ To avoid crash during import due to unique email, return the existing records if any """
        sql = '''SELECT email, id FROM mail_blacklist WHERE email = ANY(%s)'''
        emails = [v['email'] for v in new_values]
        self._cr.execute(sql, (emails,))
        bl_entries = dict(self._cr.fetchall())
        to_create = [v for v in new_values
                     if v['email'] not in bl_entries]

        # TODO DBE Fixme : reorder ids according to incoming ids.
        results = super(MailBlackList, self).create(to_create)
        return self.env['mail.blacklist'].browse(bl_entries.values()) | results

    @api.multi
    def write(self, values):
        if 'email' in values:
            values['email'] = self._sanitize_email(values['email'])
        return super(MailBlackList, self).write(values)

    def _search(self, args, offset=0, limit=None, order=None, count=False, access_rights_uid=None):
        """ Override _search in order to grep search on email field and make it
        lower-case and sanitized """
        if args:
            new_args = []
            for arg in args:
                if isinstance(arg, (list, tuple)) and arg[0] == 'email' and isinstance(arg[2], tools.pycompat.text_type):
                    sanitized = self.env['mail.blacklist']._sanitize_email(arg[2])
                    if sanitized:
                        new_args.append([arg[0], arg[1], sanitized])
                    else:
                        new_args.append(arg)
                else:
                    new_args.append(arg)
        else:
            new_args = args
        return super(MailBlackList, self)._search(new_args, offset=offset, limit=limit, order=order, count=count, access_rights_uid=access_rights_uid)

    def _add(self, email):
        sanitized = self._sanitize_email(email)
        record = self.env["mail.blacklist"].with_context(active_test=False).search([('email', '=', sanitized)])
        if len(record) > 0:
            record.write({'active': True})
        else:
            record = self.create({'email': email})
        return record

    def _remove(self, email):
        sanitized = self._sanitize_email(email)
        record = self.env["mail.blacklist"].with_context(active_test=False).search([('email', '=', sanitized)])
        if len(record) > 0:
            record.write({'active': False})
        else:
            record = record.create({'email': email, 'active': False})
        return record

    def _sanitize_email(self, email):
        """ Sanitize and standardize blacklist entries: all emails should be
        only real email extracted from strings (A <a@a> -> a@a)  and should be
        lower case. """
        emails = tools.email_split(email)
        if not emails or len(emails) != 1:
            return False
        return emails[0].lower()


class MailBlackListMixin(models.AbstractModel):
    """ Mixin that is inherited by all model with opt out.
        USAGE : the field '_primary_email' must be overridden in the model that inherit the mixin
        and must contain the email field of the model.
        """
    _name = 'mail.blacklist.mixin'
    _description = 'Mail Blacklist mixin'
    _primary_email = ['email']

    # Note : is_blacklisted sould only be used for display. As the compute is not depending on the blacklist,
    # once read, it won't be re-computed again if the blacklist is modified in the same request.
    is_blacklisted = fields.Boolean(string='Blacklist', compute="_compute_is_blacklisted", compute_sudo=True,
        store=False, search="_search_is_blacklisted", groups="base.group_user",
        help="If the email address is on the blacklist, the contact won't receive mass mailing anymore, from any list")

    def _assert_primary_email(self):
        if not hasattr(self, "_primary_email") or \
                not isinstance(self._primary_email, (list, tuple)) or \
                not len(self._primary_email) == 1:
            raise UserError(_('Invalid primary email field on model %s') % self._name)
        field_name = self._primary_email[0]
        if field_name not in self._fields or self._fields[field_name].type != 'char':
            raise UserError(_('Invalid primary email field on model %s') % self._name)

    @api.model
    def _search_is_blacklisted(self, operator, value):
        # Assumes operator is '=' or '!=' and value is True or False
        self._assert_primary_email()
        if operator != '=':
            if operator == '!=' and isinstance(value, bool):
                value = not value
            else:
                raise NotImplementedError()

        [email_field] = self._primary_email
        if value:
            query = """
                SELECT m.id
                    FROM mail_blacklist bl
                    JOIN %s m
                    ON (LOWER(substring(m.%s, '([^ ,;<@]+@[^> ,;]+)')) = bl.email AND bl.active)
            """
        else:
            query = """
                SELECT m.id
                    FROM %s m
                    LEFT JOIN mail_blacklist bl
                    ON (LOWER(substring(m.%s, '([^ ,;<@]+@[^> ,;]+)')) = bl.email AND bl.active)
                    WHERE bl.id IS NULL
            """
        self._cr.execute(query % (self._table, email_field))
        res = self._cr.fetchall()
        if not res:
            return [(0, '=', 1)]
        return [('id', 'in', [r[0] for r in res])]

    @api.depends(lambda self: self._primary_email)
    def _compute_is_blacklisted(self):
        self._assert_primary_email()
        [email_field] = self._primary_email
        # TODO : Should remove the sudo as compute_sudo defined on methods.
        # But if user doesn't have access to mail.blacklist, doen't work without sudo().
        sanitized = [self.env['mail.blacklist']._sanitize_email(email) for email in self.mapped(email_field)]
        blacklist = set(self.env['mail.blacklist'].sudo().search([('email', 'in', sanitized)]).mapped('email'))
        for record in self:
            record.is_blacklisted = self.env['mail.blacklist']._sanitize_email(record[email_field]) in blacklist

# -*- coding: utf-8 -*- # Part of Odoo. See LICENSE file for full copyright and licensing details. Model of blacklisted email addresses to stop sending emails. # First of all, extract values to ensure emails are really unique (and don't modify values in place) To avoid crash during import due to unique email, return the existing records if any SELECT email, id FROM mail_blacklist WHERE email = ANY(%s) # TODO DBE Fixme : reorder ids according to incoming ids. Override _search in order to grep search on email field and make it         lower-case and sanitized Sanitize and standardize blacklist entries: all emails should be         only real email extracted from strings (A <a@a> -> a@a)  and should be         lower case. Mixin that is inherited by all model with opt out.         USAGE : the field '_primary_email' must be overridden in the model that inherit the mixin         and must contain the email field of the model. # Note : is_blacklisted sould only be used for display. As the compute is not depending on the blacklist, # once read, it won't be re-computed again if the blacklist is modified in the same request. # Assumes operator is '=' or '!=' and value is True or False SELECT m.id                     FROM mail_blacklist bl                     JOIN %s m                     ON (LOWER(substring(m.%s, '([^ ,;<@]+@[^> ,;]+)')) = bl.email AND bl.active) SELECT m.id                     FROM %s m                     LEFT JOIN mail_blacklist bl                     ON (LOWER(substring(m.%s, '([^ ,;<@]+@[^> ,;]+)')) = bl.email AND bl.active)                     WHERE bl.id IS NULL # TODO : Should remove the sudo as compute_sudo defined on methods. # But if user doesn't have access to mail.blacklist, doen't work without sudo().