1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from collections import defaultdict
from dateutil.relativedelta import relativedelta
from odoo import api, fields, models, _
from odoo.exceptions import ValidationError
from odoo.tools.float_utils import float_round
class UtmCampaign(models.Model):
_inherit = 'utm.campaign'
mailing_mail_ids = fields.One2many(
'mailing.mailing', 'campaign_id',
domain=[('mailing_type', '=', 'mail')],
string='Mass Mailings',
groups="mass_mailing.group_mass_mailing_user")
mailing_mail_count = fields.Integer('Number of Mass Mailing',
compute="_compute_mailing_mail_count",
groups="mass_mailing.group_mass_mailing_user")
is_mailing_campaign_activated = fields.Boolean(compute="_compute_is_mailing_campaign_activated")
# A/B Testing
ab_testing_mailings_count = fields.Integer("A/B Test Mailings #", compute="_compute_mailing_mail_count")
ab_testing_completed = fields.Boolean("A/B Testing Campaign Finished", compute="_compute_ab_testing_completed",
copy=False, readonly=True, store=True)
ab_testing_winner_mailing_id = fields.Many2one("mailing.mailing", "A/B Campaign Winner Mailing", copy=False)
ab_testing_schedule_datetime = fields.Datetime('Send Final On',
default=lambda self: fields.Datetime.now() + relativedelta(days=1),
help="Date that will be used to know when to determine and send the winner mailing")
ab_testing_winner_selection = fields.Selection([
('manual', 'Manual'),
('opened_ratio', 'Highest Open Rate'),
('clicks_ratio', 'Highest Click Rate'),
('replied_ratio', 'Highest Reply Rate')], string="Winner Selection", default="opened_ratio",
help="Selection to determine the winner mailing that will be sent.")
# stat fields
received_ratio = fields.Float(compute="_compute_statistics", string='Received Ratio')
opened_ratio = fields.Float(compute="_compute_statistics", string='Opened Ratio')
replied_ratio = fields.Float(compute="_compute_statistics", string='Replied Ratio')
bounced_ratio = fields.Float(compute="_compute_statistics", string='Bounced Ratio')
@api.depends('ab_testing_winner_mailing_id')
def _compute_ab_testing_completed(self):
for campaign in self:
campaign.ab_testing_completed = bool(self.ab_testing_winner_mailing_id)
@api.depends('mailing_mail_ids')
def _compute_mailing_mail_count(self):
mailing_data = self.env['mailing.mailing']._read_group(
[('campaign_id', 'in', self.ids), ('mailing_type', '=', 'mail')],
['campaign_id', 'ab_testing_enabled'],
['__count'],
)
ab_testing_mapped_data = defaultdict(list)
mapped_data = defaultdict(list)
for campaign, ab_testing_enabled, count in mailing_data:
if ab_testing_enabled:
ab_testing_mapped_data[campaign.id].append(count)
mapped_data[campaign.id].append(count)
for campaign in self:
campaign.mailing_mail_count = sum(mapped_data[campaign._origin.id or campaign.id])
campaign.ab_testing_mailings_count = sum(ab_testing_mapped_data[campaign._origin.id or campaign.id])
def _compute_statistics(self):
""" Compute statistics of the mass mailing campaign """
default_vals = {
'received_ratio': 0,
'opened_ratio': 0,
'replied_ratio': 0,
'bounced_ratio': 0
}
if not self.ids:
self.update(default_vals)
return
self.env.cr.execute("""
SELECT
c.id as campaign_id,
COUNT(s.id) AS expected,
COUNT(s.sent_datetime) AS sent,
COUNT(s.trace_status) FILTER (WHERE s.trace_status in ('sent', 'open', 'reply')) AS delivered,
COUNT(s.trace_status) FILTER (WHERE s.trace_status in ('open', 'reply')) AS open,
COUNT(s.trace_status) FILTER (WHERE s.trace_status = 'reply') AS reply,
COUNT(s.trace_status) FILTER (WHERE s.trace_status = 'bounce') AS bounce,
COUNT(s.trace_status) FILTER (WHERE s.trace_status = 'cancel') AS cancel
FROM
mailing_trace s
RIGHT JOIN
utm_campaign c
ON (c.id = s.campaign_id)
WHERE
c.id IN %s
GROUP BY
c.id
""", (tuple(self.ids), ))
all_stats = self.env.cr.dictfetchall()
stats_per_campaign = {
stats['campaign_id']: stats
for stats in all_stats
}
for campaign in self:
stats = stats_per_campaign.get(campaign.id)
if not stats:
vals = default_vals
else:
total = (stats['expected'] - stats['cancel']) or 1
delivered = stats['sent'] - stats['bounce']
vals = {
'received_ratio': float_round(100.0 * delivered / total, precision_digits=2),
'opened_ratio': float_round(100.0 * stats['open'] / total, precision_digits=2),
'replied_ratio': float_round(100.0 * stats['reply'] / total, precision_digits=2),
'bounced_ratio': float_round(100.0 * stats['bounce'] / total, precision_digits=2)
}
campaign.update(vals)
def _compute_is_mailing_campaign_activated(self):
self.is_mailing_campaign_activated = self.env.user.has_group('mass_mailing.group_mass_mailing_campaign')
def _get_mailing_recipients(self, model=None):
"""Return the recipients of a mailing campaign. This is based on the statistics
build for each mailing. """
res = dict.fromkeys(self.ids, {})
for campaign in self:
domain = [('campaign_id', '=', campaign.id)]
if model:
domain += [('model', '=', model)]
res[campaign.id] = set(self.env['mailing.trace'].search(domain).mapped('res_id'))
return res
@api.model
def _cron_process_mass_mailing_ab_testing(self):
""" Cron that manages A/B testing and sends a winner mailing computed based on
the value set on the A/B testing campaign.
In case there is no mailing sent for an A/B testing campaign we ignore this campaign
"""
ab_testing_campaign = self.search([
('ab_testing_schedule_datetime', '<=', fields.Datetime.now()),
('ab_testing_winner_selection', '!=', 'manual'),
('ab_testing_completed', '=', False),
])
for campaign in ab_testing_campaign:
ab_testing_mailings = campaign.mailing_mail_ids.filtered(lambda m: m.ab_testing_enabled)
if not ab_testing_mailings.filtered(lambda m: m.state == 'done'):
continue
ab_testing_mailings.action_send_winner_mailing()
return ab_testing_campaign
|
