Techsystech
/
server-tools
mirror of https://github.com/OCA/server-tools.git
3
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[FIX] fetchmail_*: adapt to Odoo 16.0 and python 3.x

pull/2800/head
Ronald Portier 2024-01-10 22:56:47 +01:00
parent 7a7b1a6c9d
commit 6a24dec96f
13 changed files with 194 additions and 224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
fetchmail_attach_from_folder/match_algorithm/__init__.py
View File

@ -2,4 +2,3 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
from . import email_exact
from . import email_domain
from . import odoo_standard

14
fetchmail_attach_from_folder/match_algorithm/base.py
View File

@ -1,14 +0,0 @@
# Copyright - 2013-2024 Therp BV <https://therp.nl>.
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
class Base(object):
def search_matches(self, folder, mail_message):
"""Returns recordset found for model with mail_message."""
return []
def handle_match(
self, connection, match_object, folder, mail_message, mail_message_org, msgid
):
"""Do whatever it takes to handle a match"""
folder.attach_mail(match_object, mail_message)

10
fetchmail_attach_from_folder/match_algorithm/email_domain.py
View File

@ -9,21 +9,21 @@ class EmailDomain(EmailExact):
Beware of match_first here, this is most likely to get it wrong (gmail).
"""
def search_matches(self, folder, mail_message):
def search_matches(self, folder, message_dict):
"""Returns recordset of matching objects."""
matches = super().search_matches(folder, mail_message)
matches = super().search_matches(folder, message_dict)
if not matches:
object_model = folder.env[folder.model_id.model]
domains = []
for addr in self._get_mailaddresses(folder, mail_message):
for addr in self._get_mailaddresses(folder, message_dict):
domains.append(addr.split("@")[-1])
matches = object_model.search(
self._get_mailaddress_search_domain(
folder,
mail_message,
message_dict,
operator="like",
values=["%@" + domain for domain in set(domains)],
),
order=folder.model_order,
)
return matches
return matches.ids

21
fetchmail_attach_from_folder/match_algorithm/email_exact.py
View File

@ -3,24 +3,22 @@
from odoo.tools.mail import email_split
from odoo.tools.safe_eval import safe_eval
from .base import Base
class EmailExact(Base):
class EmailExact:
"""Search for exactly the mailadress as noted in the email"""
def _get_mailaddresses(self, folder, mail_message):
def _get_mailaddresses(self, folder, message_dict):
mailaddresses = []
fields = folder.mail_field.split(",")
for field in fields:
if field in mail_message:
mailaddresses += email_split(mail_message[field])
if field in message_dict:
mailaddresses += email_split(message_dict[field])
return [addr.lower() for addr in mailaddresses]
def _get_mailaddress_search_domain(
self, folder, mail_message, operator="=", values=None
self, folder, message_dict, operator="=", values=None
):
mailaddresses = values or self._get_mailaddresses(folder, mail_message)
mailaddresses = values or self._get_mailaddresses(folder, message_dict)
if not mailaddresses:
return [(0, "=", 1)]
search_domain = (
@ -30,8 +28,9 @@ class EmailExact(Base):
)
return search_domain
def search_matches(self, folder, mail_message):
def search_matches(self, folder, message_dict):
"""Returns recordset of matching objects."""
object_model = folder.env[folder.model_id.model]
search_domain = self._get_mailaddress_search_domain(folder, mail_message)
return object_model.search(search_domain, order=folder.model_order)
search_domain = self._get_mailaddress_search_domain(folder, message_dict)
matches = object_model.search(search_domain, order=folder.model_order)
return matches.ids

23
fetchmail_attach_from_folder/match_algorithm/odoo_standard.py
View File

@ -1,23 +0,0 @@
# Copyright - 2013-2024 Therp BV <https://therp.nl>.
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
from .base import Base
class OdooStandard(Base):
"""No search at all. Use Odoo's standard mechanism to attach mails to
mail.thread objects. Note that this algorithm always matches."""
def search_matches(self, folder, mail_message):
"""Always match. Duplicates will be fished out by message_id"""
return [True]
def handle_match(
self, connection, match_object, folder, mail_message, mail_message_org, msgid
):
thread_model = folder.env["mail.thread"]
thread_model.message_process(
folder.model_id.model,
mail_message_org,
save_original=folder.server_id.original,
strip_attachments=(not folder.server_id.attach),
)

1
fetchmail_attach_from_folder/models/__init__.py
View File

@ -2,3 +2,4 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
from . import fetchmail_server
from . import fetchmail_server_folder
from . import mail_thread

12
fetchmail_attach_from_folder/models/fetchmail_server.py
View File

@ -19,7 +19,10 @@ class FetchmailServer(models.Model):
"""Retrieve available folders from IMAP server."""
def parse_list_response(line):
flags, delimiter, mailbox_name = list_response_pattern.match(line).groups()
string_line = line.decode("utf-8")
flags, delimiter, mailbox_name = list_response_pattern.match(
string_line
).groups()
mailbox_name = mailbox_name.strip('"')
return (flags, delimiter, mailbox_name)
@ -34,7 +37,7 @@ class FetchmailServer(models.Model):
continue
folders_available = []
for folder_entry in list_result[1]:
folders_available.append(parse_list_response(str(folder_entry))[2])
folders_available.append(parse_list_response(folder_entry)[2])
this.folders_available = "\n".join(folders_available)
connection.logout()
@ -47,13 +50,14 @@ class FetchmailServer(models.Model):
string="Folders",
context={"active_test": False},
)
object_id = fields.Many2one(required=False) # comodel_name='ir.model'
server_type = fields.Selection(default="imap")
folders_only = fields.Boolean(
string="Only folders, not inbox",
help="Check this field to leave imap inbox alone"
" and only retrieve mail from configured folders.",
)
# Below existing fields, that are modified by this module.
object_id = fields.Many2one(required=False) # comodel_name='ir.model'
server_type = fields.Selection(default="imap")
@api.onchange("server_type", "is_ssl", "object_id")
def onchange_server_type(self):

214
fetchmail_attach_from_folder/models/fetchmail_server_folder.py
View File

@ -1,18 +1,18 @@
# Copyright - 2013-2024 Therp BV <https://therp.nl>.
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import base64
import logging
from odoo import _, fields, models
from odoo.exceptions import UserError, ValidationError
from .. import match_algorithm
_logger = logging.getLogger(__name__)
class FetchmailServerFolder(models.Model):
"""Define folders (IMAP mailboxes) from which to fetch mail."""
_name = "fetchmail.server.folder"
_description = __doc__
_rec_name = "path"
_order = "sequence"
@ -84,22 +84,6 @@ class FetchmailServerFolder(models.Model):
)
active = fields.Boolean(default=True)
def get_algorithm(self):
"""Translate algorithm code to implementation class.
We used to load this dynamically, but having it more or less hardcoded
allows to adapt the UI to the selected algorithm, withouth needing
the (deprecated) fields_view_get trickery we used in the past.
"""
self.ensure_one()
if self.match_algorithm == "odoo_standard":
return match_algorithm.odoo_standard.OdooStandard
if self.match_algorithm == "email_domain":
return match_algorithm.email_domain.EmailDomain
if self.match_algorithm == "email_exact":
return match_algorithm.email_exact.EmailExact
return None
def button_confirm_folder(self):
self.write({"state": "draft"})
for this in self:
@ -127,64 +111,6 @@ class FetchmailServerFolder(models.Model):
self.write({"state": "draft"})
return True
def get_msgids(self, connection, criteria):
"""Return imap ids of messages to process"""
self.ensure_one()
server = self.server_id
_logger.info(
"start checking for emails in folder %(folder)s on server %(server)s",
{"folder": self.path, "server": server.name},
)
if connection.select(self.path)[0] != "OK":
raise UserError(
_("Could not open folder %(folder)s on server %(server)s")
% {"folder": self.path, "server": server.name}
)
result, msgids = connection.search(None, criteria)
if result != "OK":
raise UserError(
_("Could not search folder %(folder)s on server %(server)s")
% {"folder": self.path, "server": server.name}
)
_logger.info(
"finished checking for emails in folder %(folder)s on server %(server)s",
{"folder": self.path, "server": server.name},
)
return msgids
def fetch_msg(self, connection, msgid):
"""Select a single message from a folder."""
self.ensure_one()
result, msgdata = connection.fetch(msgid, "(RFC822)")
if result != "OK":
raise UserError(
_("Could not fetch %(msgid)s in folder %(folder)s on server %(server)s")
% {"msgid": msgid, "folder": self.path, "server": self.server_id.name}
)
message_org = msgdata[0][1] # rfc822 message source
mail_message = self.env["mail.thread"].message_parse(
message_org, save_original=self.server_id.original
)
return (mail_message, message_org)
def retrieve_imap_folder(self, connection):
"""Retrieve all mails for one IMAP folder."""
self.ensure_one()
msgids = self.get_msgids(connection, "UNDELETED")
match_algorithm = self.get_algorithm()
for msgid in msgids[0].split():
# We will accept exceptions for single messages
try:
self.env.cr.execute("savepoint apply_matching")
self.apply_matching(connection, msgid, match_algorithm)
self.env.cr.execute("release savepoint apply_matching")
except Exception:
self.env.cr.execute("rollback to savepoint apply_matching")
_logger.exception(
"Failed to fetch mail %(msgid)s from server %(server)s",
{"msgid": msgid, "server": self.server_id.name},
)
def fetch_mail(self):
"""Retrieve all mails for IMAP folders.
@ -215,6 +141,82 @@ class FetchmailServerFolder(models.Model):
if connection:
connection.logout()
def retrieve_imap_folder(self, connection):
"""Retrieve all mails for one IMAP folder."""
self.ensure_one()
msgids = self.get_msgids(connection, "UNDELETED")
for msgid in msgids[0].split():
# We will accept exceptions for single messages
try:
self.env.cr.execute("savepoint apply_matching")
self.apply_matching(connection, msgid)
self.env.cr.execute("release savepoint apply_matching")
except Exception:
self.env.cr.execute("rollback to savepoint apply_matching")
_logger.exception(
"Failed to fetch mail %(msgid)s from server %(server)s",
{"msgid": msgid, "server": self.server_id.name},
)
def get_msgids(self, connection, criteria):
"""Return imap ids of messages to process"""
self.ensure_one()
server = self.server_id
_logger.info(
"start checking for emails in folder %(folder)s on server %(server)s",
{"folder": self.path, "server": server.name},
)
if connection.select(self.path)[0] != "OK":
raise UserError(
_("Could not open folder %(folder)s on server %(server)s")
% {"folder": self.path, "server": server.name}
)
result, msgids = connection.search(None, criteria)
if result != "OK":
raise UserError(
_("Could not search folder %(folder)s on server %(server)s")
% {"folder": self.path, "server": server.name}
)
_logger.info(
"finished checking for emails in folder %(folder)s on server %(server)s",
{"folder": self.path, "server": server.name},
)
return msgids
def apply_matching(self, connection, msgid):
"""Return ids of objects matched"""
self.ensure_one()
thread_model = self.env["mail.thread"]
message_org = self.fetch_msg(connection, msgid)
custom_values = (
None
if self.match_algorithm == "odoo_standard"
else {
"folder": self,
}
)
thread_id = thread_model.message_process(
self.model_id.model,
message_org,
custom_values=custom_values,
save_original=self.server_id.original,
strip_attachments=(not self.server_id.attach),
)
matched = True if thread_id else False
self.update_msg(connection, msgid, matched=matched)
def fetch_msg(self, connection, msgid):
"""Select a single message from a folder."""
self.ensure_one()
result, msgdata = connection.fetch(msgid, "(RFC822)")
if result != "OK":
raise UserError(
_("Could not fetch %(msgid)s in folder %(folder)s on server %(server)s")
% {"msgid": msgid, "folder": self.path, "server": self.server_id.name}
)
message_org = msgdata[0][1] # rfc822 message source
return message_org
def update_msg(self, connection, msgid, matched=True, flagged=False):
"""Update msg in imap folder depending on match and settings."""
if matched:
@ -225,61 +227,3 @@ class FetchmailServerFolder(models.Model):
else:
if self.flag_nonmatching:
connection.store(msgid, "+FLAGS", "\\FLAGGED")
def apply_matching(self, connection, msgid, match_algorithm):
"""Return ids of objects matched"""
self.ensure_one()
mail_message, message_org = self.fetch_msg(connection, msgid)
if self.env["mail.message"].search(
[("message_id", "=", mail_message["message_id"])]
):
# Ignore mails that have been handled already
return
matches = match_algorithm.search_matches(self, mail_message)
matched = matches and (len(matches) == 1 or self.match_first)
if matched:
match_algorithm.handle_match(
connection, matches[0], self, mail_message, message_org, msgid
)
self.update_msg(connection, msgid, matched=matched)
def attach_mail(self, match_object, mail_message):
"""Attach mail to match_object."""
self.ensure_one()
partner = False
model_name = self.model_id.model
if model_name == "res.partner":
partner = match_object
elif "partner_id" in self.env[model_name]._fields:
partner = match_object.partner_id
attachments = []
if self.server_id.attach and mail_message.get("attachments"):
for attachment in mail_message["attachments"]:
# Attachment should at least have filename and data, but
# might have some extra element(s)
if len(attachment) < 2:
continue
fname, fcontent = attachment[:2]
data_attach = {
"name": fname,
"datas": base64.b64encode(fcontent),
"datas_fname": fname,
"description": _("Mail attachment"),
"res_model": model_name,
"res_id": match_object.id,
}
attachments.append(self.env["ir.attachment"].create(data_attach))
self.env["mail.message"].create(
{
"author_id": partner and partner.id or False,
"model": model_name,
"res_id": match_object.id,
"message_type": "email",
"body": mail_message.get("body"),
"subject": mail_message.get("subject"),
"email_from": mail_message.get("from"),
"date": mail_message.get("date"),
"message_id": mail_message.get("message_id"),
"attachment_ids": [(6, 0, [a.id for a in attachments])],
}
)

70
fetchmail_attach_from_folder/models/mail_thread.py 100644
View File

@ -0,0 +1,70 @@
# Copyright - 2024 Therp BV <https://therp.nl>.
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import logging
from odoo import api, models
from .. import match_algorithm
_logger = logging.getLogger(__name__)
class MailThread(models.AbstractModel):
_inherit = "mail.thread"
@api.model
def message_route(
self,
message,
message_dict,
model=None,
thread_id=None,
custom_values=None,
):
"""Override to apply matching algorithm to determine thread_id if requested."""
if not thread_id and custom_values and "folder" in custom_values:
thread_id = self._find_match(custom_values, message_dict)
if not thread_id:
return [] # This will ultimately return thread_id = False
return super().message_route(
message,
message_dict,
model=model,
thread_id=thread_id,
custom_values=custom_values,
)
@api.model
def _find_match(self, custom_values, message_dict):
"""Try to find existing object to link mail to."""
folder = custom_values.pop("folder")
matcher = self._get_algorithm(folder.match_algorithm)
if not matcher:
return None
matches = matcher.search_matches(folder, message_dict)
if not matches:
_logger.info(
"No match found for message %(subject)s with msgid %(msgid)s",
{
"subject": message_dict.get("subject", "no subject"),
"msgid": message_dict.get("message_id", "no msgid"),
},
)
return None
matched = len(matches) == 1 or folder.match_first
return matched and matches[0] or None
@api.model
def _get_algorithm(self, algorithm):
"""Translate algorithm code to implementation class.
We used to load this dynamically, but having it more or less hardcoded
allows to adapt the UI to the selected algorithm, withouth needing
the (deprecated) fields_view_get trickery we used in the past.
"""
if algorithm == "email_domain":
return match_algorithm.email_domain.EmailDomain()
if algorithm == "email_exact":
return match_algorithm.email_exact.EmailExact()
_logger.error("Unknown algorithm %(algorithm)s", {"algorithm": algorithm})
return None

2
fetchmail_attach_from_folder/security/ir.model.access.csv
View File

@ -1,2 +1,4 @@
id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
access_model_fetchmail_server_folder,fetchmail.server.folder,model_fetchmail_server_folder,base.group_system,1,1,1,1
access_fetchmail_attach_mail_manually,access_fetchmail_attach_mail_manually,model_fetchmail_attach_mail_manually,base.group_system,1,1,1,1
access_fetchmail_attach_mail_manually_mail,access_fetchmail_attach_mail_manually_mail,model_fetchmail_attach_mail_manually_mail,base.group_system,1,1,1,1

1 id name model_id:id group_id:id perm_read perm_write perm_create perm_unlink
2 access_model_fetchmail_server_folder fetchmail.server.folder model_fetchmail_server_folder base.group_system 1 1 1 1
3 access_fetchmail_attach_mail_manually access_fetchmail_attach_mail_manually model_fetchmail_attach_mail_manually base.group_system 1 1 1 1
4 access_fetchmail_attach_mail_manually_mail access_fetchmail_attach_mail_manually_mail model_fetchmail_attach_mail_manually_mail base.group_system 1 1 1 1

28
fetchmail_attach_from_folder/tests/test_match_algorithms.py
View File

@ -2,7 +2,7 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
from odoo.tests.common import TransactionCase
from ..match_algorithm import email_domain, email_exact, odoo_standard
from ..match_algorithm import email_domain, email_exact
MSG_BODY = [
(
@ -116,7 +116,7 @@ class TestMatchAlgorithms(TransactionCase):
}
folder = self.folder
folder.match_algorithm = "email_domain"
folder.use_first_match = True
folder.match_first = True
self.do_matching(
email_domain.EmailDomain,
"base.res_partner_address_31",
@ -128,32 +128,12 @@ class TestMatchAlgorithms(TransactionCase):
mail_message["subject"],
)
def test_odoo_standard(self):
mail_message_org = (
"To: demo@yourcompany.example.com\n"
"From: someone@else.com\n"
"Subject: testsubject\n"
"Message-Id: 42\n"
"Hello world"
)
folder = self.folder
folder.match_algorithm = "odoo_standard"
matcher = odoo_standard.OdooStandard()
matches = matcher.search_matches(folder, None)
self.assertEqual(len(matches), 1)
matcher.handle_match(None, matches[0], folder, None, mail_message_org, None)
self.assertIn(
"Hello world",
self.env["mail.message"].search([("subject", "=", "testsubject")]).body,
)
def test_apply_matching_exact(self):
folder = self.folder
folder.match_algorithm = "email_domain"
folder.match_algorithm = "email_exact"
connection = MockConnection()
msgid = "<485a8041-d560-a981-5afc-d31c1f136748@acme.com>"
matcher = email_exact.EmailExact()
folder.apply_matching(connection, msgid, matcher)
folder.apply_matching(connection, msgid)
def test_retrieve_imap_folder_domain(self):
folder = self.folder

16
fetchmail_attach_from_folder/views/fetchmail_server.xml
View File

@ -11,14 +11,14 @@
name="attrs"
>{'required': [('server_type', '!=', 'imap')]}</attribute>
</field>
<field name="server_type" position="after">
<field name="folders_only" />
</field>
<xpath expr="//notebook" position="inside">
<page
string="Folders to monitor"
attrs="{'invisible': [('server_type','!=','imap')]}"
>
<group>
<field name="folders_only" />
</group>
<field name="folder_ids" nolabel="1">
<tree decoration-muted="active == False">
<field name="active" invisible="True" />
@ -58,10 +58,12 @@
states="done"
/>
</header>
<group>
<field name="path" placeholder="INBOX.subfolder1" />
<field name="model_id" />
<field name="match_algorithm" />
<group colspan="4" col="2">
<group>
<field name="path" placeholder="INBOX.subfolder1" />
<field name="model_id" />
<field name="match_algorithm" />
</group>
<group
name="group_email_match"
attrs="{'invisible':

6
fetchmail_attach_from_folder/wizard/attach_mail_manually.py
View File

@ -8,7 +8,10 @@ _logger = logging.getLogger(__name__)
class AttachMailManually(models.TransientModel):
"""Attach mail to selected documents."""
_name = "fetchmail.attach.mail.manually"
_description = __doc__
name = fields.Char()
folder_id = fields.Many2one(comodel_name="fetchmail.server.folder", readonly=True)
@ -88,7 +91,10 @@ class AttachMailManually(models.TransientModel):
class AttachMailManuallyMail(models.TransientModel):
"""Attach single mail to selected documents."""
_name = "fetchmail.attach.mail.manually.mail"
_description = __doc__
wizard_id = fields.Many2one("fetchmail.attach.mail.manually", readonly=True)
msgid = fields.Char("Message id", readonly=True)