Merge pull request #13307 from leela/incoming-mail-refactoring

refactor: Incoming mail refactoring

frappe/core/doctype/communication/communication.py

@@ -21,9 +21,11 @@ from frappe.automation.doctype.assignment_rule.assignment_rule import apply as a
 exclude_from_linked_with = True
 
 class Communication(Document):
+	"""Communication represents an external communication like Email.
+	"""
 	no_feed_on_delete = True
+	DOCTYPE = 'Communication'
 
-	"""Communication represents an external communication like Email."""
 	def onload(self):
 		"""create email flag queue"""
 		if self.communication_type == "Communication" and self.communication_medium == "Email" \
@@ -149,6 +151,23 @@ class Communication(Document):
 
 			self.email_status = "Spam"
 
+	@classmethod
+	def find(cls, name, ignore_error=False):
+		try:
+			return frappe.get_doc(cls.DOCTYPE, name)
+		except frappe.DoesNotExistError:
+			if ignore_error:
+				return
+			raise
+
+	@classmethod
+	def find_one_by_filters(cls, *, order_by=None, **kwargs):
+		name = frappe.db.get_value(cls.DOCTYPE, kwargs, order_by=order_by)
+		return cls.find(name) if name else None
+
+	def update_db(self, **kwargs):
+		frappe.db.set_value(self.DOCTYPE, self.name, kwargs)
+
 	def set_sender_full_name(self):
 		if not self.sender_full_name and self.sender:
 			if self.sender == "Administrator":
@@ -485,4 +504,4 @@ def set_avg_response_time(parent, communication):
 						response_times.append(response_time)
 			if response_times:
 				avg_response_time = sum(response_times) / len(response_times)
-				parent.db_set("avg_response_time", avg_response_time)
+				parent.db_set("avg_response_time", avg_response_time)

frappe/email/doctype/email_account/email_account.py

@@ -19,7 +19,7 @@ from frappe.utils import (validate_email_address, cint, cstr, get_datetime,
 from frappe.utils.user import is_system_user
 from frappe.utils.jinja import render_template
 from frappe.email.smtp import SMTPServer
-from frappe.email.receive import EmailServer, Email
+from frappe.email.receive import EmailServer, InboundMail, SentEmailInInboxError
 from poplib import error_proto
 from dateutil.relativedelta import relativedelta
 from datetime import datetime, timedelta
@@ -430,89 +430,76 @@ class EmailAccount(Document):
 
 	def receive(self, test_mails=None):
 		"""Called by scheduler to receive emails from this EMail account using POP3/IMAP."""
-		def get_seen(status):
-			if not status:
-				return None
-			seen = 1 if status == "SEEN" else 0
-			return seen
-
-		if self.enable_incoming:
-			uid_list = []
-			exceptions = []
-			seen_status = []
-			uid_reindexed = False
-			email_server = None
-
-			if frappe.local.flags.in_test:
-				incoming_mails = test_mails or []
-			else:
-				email_sync_rule = self.build_email_sync_rule()
-
-				try:
-					email_server = self.get_incoming_server(in_receive=True, email_sync_rule=email_sync_rule)
-				except Exception:
-					frappe.log_error(title=_("Error while connecting to email account {0}").format(self.name))
-
-				if not email_server:
-					return
-
-				emails = email_server.get_messages()
-				if not emails:
-					return
-
-				incoming_mails = emails.get("latest_messages", [])
-				uid_list = emails.get("uid_list", [])
-				seen_status = emails.get("seen_status", [])
-				uid_reindexed = emails.get("uid_reindexed", False)
-
-			for idx, msg in enumerate(incoming_mails):
-				uid = None if not uid_list else uid_list[idx]
-				self.flags.notify = True
-
-				try:
-					args = {
-						"uid": uid,
-						"seen": None if not seen_status else get_seen(seen_status.get(uid, None)),
-						"uid_reindexed": uid_reindexed
-					}
-					communication = self.insert_communication(msg, args=args)
-
-				except SentEmailInInbox:
-					frappe.db.rollback()
-
-				except Exception:
-					frappe.db.rollback()
-					frappe.log_error('email_account.receive')
-					if self.use_imap:
-						self.handle_bad_emails(email_server, uid, msg, frappe.get_traceback())
-					exceptions.append(frappe.get_traceback())
+		exceptions = []
+		inbound_mails = self.get_inbound_mails(test_mails=test_mails)
+		for mail in inbound_mails:
+			try:
+				communication = mail.process()
+				frappe.db.commit()
+				# If email already exists in the system
+				# then do not send notifications for the same email.
+				if communication and mail.flags.is_new_communication:
+					# notify all participants of this thread
+					if self.enable_auto_reply:
+						self.send_auto_reply(communication, mail)
+
+					attachments = []
+					if hasattr(communication, '_attachments'):
+						attachments = [d.file_name for d in communication._attachments]
+					communication.notify(attachments=attachments, fetched_from_email_account=True)
+			except SentEmailInInboxError:
+				frappe.db.rollback()
+			except Exception:
+				frappe.db.rollback()
+				frappe.log_error('email_account.receive')
+				if self.use_imap:
+					self.handle_bad_emails(mail.uid, mail.raw_message, frappe.get_traceback())
+				exceptions.append(frappe.get_traceback())
+
+		#notify if user is linked to account
+		if len(inbound_mails)>0 and not frappe.local.flags.in_test:
+			frappe.publish_realtime('new_email',
+				{"account":self.email_account_name, "number":len(inbound_mails)}
+			)
 
-				else:
-					frappe.db.commit()
-					if communication and self.flags.notify:
+		if exceptions:
+			raise Exception(frappe.as_json(exceptions))
 
-						# If email already exists in the system
-						# then do not send notifications for the same email.
+	def get_inbound_mails(self, test_mails=None):
+		"""retrive and return inbound mails.
 
-						attachments = []
+		"""
+		if frappe.local.flags.in_test:
+			return [InboundMail(msg, self) for msg in test_mails or []]
 
-						if hasattr(communication, '_attachments'):
-							attachments = [d.file_name for d in communication._attachments]
+		if not self.enable_incoming:
+			return []
 
-						communication.notify(attachments=attachments, fetched_from_email_account=True)
+		email_sync_rule = self.build_email_sync_rule()
+		try:
+			email_server = self.get_incoming_server(in_receive=True, email_sync_rule=email_sync_rule)
+			messages = email_server.get_messages() or {}
+		except Exception:
+			raise
+			frappe.log_error(title=_("Error while connecting to email account {0}").format(self.name))
+			return []
 
-			#notify if user is linked to account
-			if len(incoming_mails)>0 and not frappe.local.flags.in_test:
-				frappe.publish_realtime('new_email', {"account":self.email_account_name, "number":len(incoming_mails)})
+		mails = []
+		for index, message in enumerate(messages.get("latest_messages", [])):
+			uid = messages['uid_list'][index]
+			seen_status = 1 if messages['seen_status'][uid]=='SEEN' else 0
+			mails.append(InboundMail(message, self, uid, seen_status))
 
-			if exceptions:
-				raise Exception(frappe.as_json(exceptions))
+		return mails
 
-	def handle_bad_emails(self, email_server, uid, raw, reason):
-		if email_server and cint(email_server.settings.use_imap):
+	def handle_bad_emails(self, uid, raw, reason):
+		if cint(self.use_imap):
 			import email
 			try:
-				mail = email.message_from_string(raw)
+				if isinstance(raw, bytes):
+					mail = email.message_from_bytes(raw)
+				else:
+					mail = email.message_from_string(raw)
 
 				message_id = mail.get('Message-ID')
 			except Exception:
@@ -524,275 +511,18 @@ class EmailAccount(Document):
 				"reason":reason,
 				"message_id": message_id,
 				"doctype": "Unhandled Email",
-				"email_account": email_server.settings.email_account
+				"email_account": self.name
 			})
 			unhandled_email.insert(ignore_permissions=True)
 			frappe.db.commit()
 
-	def insert_communication(self, msg, args=None):
-		if isinstance(msg, list):
-			raw, uid, seen = msg
-		else:
-			raw = msg
-			uid = -1
-			seen = 0
-		if isinstance(args, dict):
-			if args.get("uid", -1): uid = args.get("uid", -1)
-			if args.get("seen", 0): seen = args.get("seen", 0)
-
-		email = Email(raw)
-
-		if email.from_email == self.email_id and not email.mail.get("Reply-To"):
-			# gmail shows sent emails in inbox
-			# and we don't want emails sent by us to be pulled back into the system again
-			# dont count emails sent by the system get those
-			if frappe.flags.in_test:
-				print('WARN: Cannot pull email. Sender sames as recipient inbox')
-			raise SentEmailInInbox
-
-		if email.message_id:
-			# https://stackoverflow.com/a/18367248
-			names = frappe.db.sql("""SELECT DISTINCT `name`, `creation` FROM `tabCommunication`
-				WHERE `message_id`='{message_id}'
-				ORDER BY `creation` DESC LIMIT 1""".format(
-					message_id=email.message_id
-				), as_dict=True)
-
-			if names:
-				name = names[0].get("name")
-				# email is already available update communication uid instead
-				frappe.db.set_value("Communication", name, "uid", uid, update_modified=False)
-
-				self.flags.notify = False
-
-				return frappe.get_doc("Communication", name)
-
-		if email.content_type == 'text/html':
-			email.content = clean_email_html(email.content)
-
-		communication = frappe.get_doc({
-			"doctype": "Communication",
-			"subject": email.subject,
-			"content": email.content,
-			'text_content': email.text_content,
-			"sent_or_received": "Received",
-			"sender_full_name": email.from_real_name,
-			"sender": email.from_email,
-			"recipients": email.mail.get("To"),
-			"cc": email.mail.get("CC"),
-			"email_account": self.name,
-			"communication_medium": "Email",
-			"uid": int(uid or -1),
-			"message_id": email.message_id,
-			"communication_date": email.date,
-			"has_attachment": 1 if email.attachments else 0,
-			"seen": seen or 0
-		})
-
-		self.set_thread(communication, email)
-		if communication.seen:
-			# get email account user and set communication as seen
-			users = frappe.get_all("User Email", filters={ "email_account": self.name },
-				fields=["parent"])
-			users = list(set([ user.get("parent") for user in users ]))
-			communication._seen = json.dumps(users)
-
-		communication.flags.in_receive = True
-		communication.insert(ignore_permissions=True)
-
-		# save attachments
-		communication._attachments = email.save_attachments_in_doc(communication)
-
-		# replace inline images
-		dirty = False
-		for file in communication._attachments:
-			if file.name in email.cid_map and email.cid_map[file.name]:
-				dirty = True
-
-				email.content = email.content.replace("cid:{0}".format(email.cid_map[file.name]),
-					file.file_url)
-
-		if dirty:
-			# not sure if using save() will trigger anything
-			communication.db_set("content", sanitize_html(email.content))
-
-		# notify all participants of this thread
-		if self.enable_auto_reply and getattr(communication, "is_first", False):
-			self.send_auto_reply(communication, email)
-
-		return communication
-
-	def set_thread(self, communication, email):
-		"""Appends communication to parent based on thread ID. Will extract
-		parent communication and will link the communication to the reference of that
-		communication. Also set the status of parent transaction to Open or Replied.
-
-		If no thread id is found and `append_to` is set for the email account,
-		it will create a new parent transaction (e.g. Issue)"""
-		parent = None
-
-		parent = self.find_parent_from_in_reply_to(communication, email)
-
-		if not parent and self.append_to:
-			self.set_sender_field_and_subject_field()
-
-		if not parent and self.append_to:
-			parent = self.find_parent_based_on_subject_and_sender(communication, email)
-
-		if not parent and self.append_to and self.append_to!="Communication":
-			parent = self.create_new_parent(communication, email)
-
-		if parent:
-			communication.reference_doctype = parent.doctype
-			communication.reference_name = parent.name
-
-		# check if message is notification and disable notifications for this message
-		isnotification = email.mail.get("isnotification")
-		if isnotification:
-			if "notification" in isnotification:
-				communication.unread_notification_sent = 1
-
-	def set_sender_field_and_subject_field(self):
-		'''Identify the sender and subject fields from the `append_to` DocType'''
-		# set subject_field and sender_field
-		meta = frappe.get_meta(self.append_to)
-		self.subject_field = None
-		self.sender_field = None
-
-		if hasattr(meta, "subject_field"):
-			self.subject_field = meta.subject_field
-
-		if hasattr(meta, "sender_field"):
-			self.sender_field = meta.sender_field
-
-	def find_parent_based_on_subject_and_sender(self, communication, email):
-		'''Find parent document based on subject and sender match'''
-		parent = None
-
-		if self.append_to and self.sender_field:
-			if self.subject_field:
-				if '#' in email.subject:
-					# try and match if ID is found
-					# document ID is appended to subject
-					# example "Re: Your email (#OPP-2020-2334343)"
-					parent_id = email.subject.rsplit('#', 1)[-1].strip(' ()')
-					if parent_id:
-						parent = frappe.db.get_all(self.append_to, filters = dict(name = parent_id),
-							fields = 'name')
-
-				if not parent:
-					# try and match by subject and sender
-					# if sent by same sender with same subject,
-					# append it to old coversation
-					subject = frappe.as_unicode(strip(re.sub(r"(^\s*(fw|fwd|wg)[^:]*:|\s*(re|aw)[^:]*:\s*)*",
-						"", email.subject, 0, flags=re.IGNORECASE)))
-
-					parent = frappe.db.get_all(self.append_to, filters={
-						self.sender_field: email.from_email,
-						self.subject_field: ("like", "%{0}%".format(subject)),
-						"creation": (">", (get_datetime() - relativedelta(days=60)).strftime(DATE_FORMAT))
-					}, fields = "name", limit = 1)
-
-				if not parent and len(subject) > 10 and is_system_user(email.from_email):
-					# match only subject field
-					# when the from_email is of a user in the system
-					# and subject is atleast 10 chars long
-					parent = frappe.db.get_all(self.append_to, filters={
-						self.subject_field: ("like", "%{0}%".format(subject)),
-						"creation": (">", (get_datetime() - relativedelta(days=60)).strftime(DATE_FORMAT))
-					}, fields = "name", limit = 1)
-
-
-
-			if parent:
-				parent = frappe._dict(doctype=self.append_to, name=parent[0].name)
-				return parent
-
-	def create_new_parent(self, communication, email):
-		'''If no parent found, create a new reference document'''
-
-		# no parent found, but must be tagged
-		# insert parent type doc
-		parent = frappe.new_doc(self.append_to)
-
-		if self.subject_field:
-			parent.set(self.subject_field, frappe.as_unicode(email.subject)[:140])
-
-		if self.sender_field:
-			parent.set(self.sender_field, frappe.as_unicode(email.from_email))
-
-		if parent.meta.has_field("email_account"):
-			parent.email_account = self.name
-
-		parent.flags.ignore_mandatory = True
-
-		try:
-			parent.insert(ignore_permissions=True)
-		except frappe.DuplicateEntryError:
-			# try and find matching parent
-			parent_name = frappe.db.get_value(self.append_to, {self.sender_field: email.from_email})
-			if parent_name:
-				parent.name = parent_name
-			else:
-				parent = None
-
-		# NOTE if parent isn't found and there's no subject match, it is likely that it is a new conversation thread and hence is_first = True
-		communication.is_first = True
-
-		return parent
-
-	def find_parent_from_in_reply_to(self, communication, email):
-		'''Returns parent reference if embedded in In-Reply-To header
-
-		Message-ID is formatted as `{message_id}@{site}`'''
-		parent = None
-		in_reply_to = (email.mail.get("In-Reply-To") or "").strip(" <>")
-
-		if in_reply_to:
-			if "@{0}".format(frappe.local.site) in in_reply_to:
-				# reply to a communication sent from the system
-				email_queue = frappe.db.get_value('Email Queue', dict(message_id=in_reply_to), ['communication','reference_doctype', 'reference_name'])
-				if email_queue:
-					parent_communication, parent_doctype, parent_name = email_queue
-					if parent_communication:
-						communication.in_reply_to = parent_communication
-				else:
-					reference, domain = in_reply_to.split("@", 1)
-					parent_doctype, parent_name = 'Communication', reference
-
-				if frappe.db.exists(parent_doctype, parent_name):
-					parent = frappe._dict(doctype=parent_doctype, name=parent_name)
-
-					# set in_reply_to of current communication
-					if parent_doctype=='Communication':
-						# communication.in_reply_to = email_queue.communication
-
-						if parent.reference_name:
-							# the true parent is the communication parent
-							parent = frappe.get_doc(parent.reference_doctype,
-								parent.reference_name)
-			else:
-				comm = frappe.db.get_value('Communication',
-						dict(
-							message_id=in_reply_to,
-							creation=['>=', add_days(get_datetime(), -30)]),
-						['reference_doctype', 'reference_name'], as_dict=1)
-				if comm:
-					parent = frappe._dict(doctype=comm.reference_doctype, name=comm.reference_name)
-
-		return parent
-
 	def send_auto_reply(self, communication, email):
 		"""Send auto reply if set."""
 		from frappe.core.doctype.communication.email import set_incoming_outgoing_accounts
-
 		if self.enable_auto_reply:
 			set_incoming_outgoing_accounts(communication)
 
-			if self.send_unsubscribe_message:
-				unsubscribe_message = _("Leave this conversation")
-			else:
-				unsubscribe_message = ""
+			unsubscribe_message = (self.send_unsubscribe_message and _("Leave this conversation")) or ""
 
 			frappe.sendmail(recipients = [email.from_email],
 				sender = self.email_id,

frappe/email/doctype/email_account/test_email_account.py

@@ -1,45 +1,56 @@
 # Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
 # See license.txt
 
-from __future__ import unicode_literals
-import frappe, os
-import unittest, email
+import os
+import email
+import unittest
+from datetime import datetime, timedelta
 
+from frappe.email.receive import InboundMail, SentEmailInInboxError, Email
+from frappe.email.email_body import get_message_id
+import frappe
 from frappe.test_runner import make_test_records
+from frappe.core.doctype.communication.email import make
+from frappe.desk.form.load import get_attachments
+from frappe.email.doctype.email_account.email_account import notify_unreplied
 
 make_test_records("User")
 make_test_records("Email Account")
 
-from frappe.core.doctype.communication.email import make
-from frappe.desk.form.load import get_attachments
-from frappe.email.doctype.email_account.email_account import notify_unreplied
-from datetime import datetime, timedelta
 
-class TestEmailAccount(unittest.TestCase):
-	def setUp(self):
-		frappe.flags.mute_emails = False
-		frappe.flags.sent_mail = None
 
+class TestEmailAccount(unittest.TestCase):
+	@classmethod
+	def setUpClass(cls):
 		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
 		email_account.db_set("enable_incoming", 1)
-		frappe.db.sql('delete from `tabEmail Queue`')
+		email_account.db_set("enable_auto_reply", 1)
 
-	def tearDown(self):
+	@classmethod
+	def tearDownClass(cls):
 		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
 		email_account.db_set("enable_incoming", 0)
 
+	def setUp(self):
+		frappe.flags.mute_emails = False
+		frappe.flags.sent_mail = None
+		frappe.db.sql('delete from `tabEmail Queue`')
+		frappe.db.sql('delete from `tabUnhandled Email`')
+
+	def get_test_mail(self, fname):
+		with open(os.path.join(os.path.dirname(__file__), "test_mails", fname), "r") as f:
+			return f.read()
+
 	def test_incoming(self):
 		cleanup("test_sender@example.com")
 
-		with open(os.path.join(os.path.dirname(__file__), "test_mails", "incoming-1.raw"), "r") as f:
-			test_mails = [f.read()]
+		test_mails = [self.get_test_mail('incoming-1.raw')]
 
 		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
 		email_account.receive(test_mails=test_mails)
 
 		comm = frappe.get_doc("Communication", {"sender": "test_sender@example.com"})
 		self.assertTrue("test_receiver@example.com" in comm.recipients)
-
 		# check if todo is created
 		self.assertTrue(frappe.db.get_value(comm.reference_doctype, comm.reference_name, "name"))
 
@@ -88,7 +99,7 @@ class TestEmailAccount(unittest.TestCase):
 		email_account.receive(test_mails=test_mails)
 
 		comm = frappe.get_doc("Communication", {"sender": "test_sender@example.com"})
-		self.assertTrue("From: \"Microsoft Outlook\" <test_sender@example.com>" in comm.content)
+		self.assertTrue("From: "Microsoft Outlook" <test_sender@example.com>" in comm.content)
 		self.assertTrue("This is an e-mail message sent automatically by Microsoft Outlook while" in comm.content)
 
 	def test_incoming_attached_email_from_outlook_layers(self):
@@ -101,7 +112,7 @@ class TestEmailAccount(unittest.TestCase):
 		email_account.receive(test_mails=test_mails)
 
 		comm = frappe.get_doc("Communication", {"sender": "test_sender@example.com"})
-		self.assertTrue("From: \"Microsoft Outlook\" <test_sender@example.com>" in comm.content)
+		self.assertTrue("From: "Microsoft Outlook" <test_sender@example.com>" in comm.content)
 		self.assertTrue("This is an e-mail message sent automatically by Microsoft Outlook while" in comm.content)
 
 	def test_outgoing(self):
@@ -166,7 +177,6 @@ class TestEmailAccount(unittest.TestCase):
 
 		comm_list = frappe.get_all("Communication", filters={"sender":"test_sender@example.com"},
 			fields=["name", "reference_doctype", "reference_name"])
-
 		# both communications attached to the same reference
 		self.assertEqual(comm_list[0].reference_doctype, comm_list[1].reference_doctype)
 		self.assertEqual(comm_list[0].reference_name, comm_list[1].reference_name)
@@ -199,6 +209,215 @@ class TestEmailAccount(unittest.TestCase):
 		self.assertEqual(comm_list[0].reference_doctype, event.doctype)
 		self.assertEqual(comm_list[0].reference_name, event.name)
 
+	def test_auto_reply(self):
+		cleanup("test_sender@example.com")
+
+		test_mails = [self.get_test_mail('incoming-1.raw')]
+
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		email_account.receive(test_mails=test_mails)
+
+		comm = frappe.get_doc("Communication", {"sender": "test_sender@example.com"})
+		self.assertTrue(frappe.db.get_value("Email Queue", {"reference_doctype": comm.reference_doctype,
+			"reference_name": comm.reference_name}))
+
+	def test_handle_bad_emails(self):
+		mail_content = self.get_test_mail(fname="incoming-1.raw")
+		message_id = Email(mail_content).mail.get('Message-ID')
+
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		email_account.handle_bad_emails(uid=-1, raw=mail_content, reason="Testing")
+		self.assertTrue(frappe.db.get_value("Unhandled Email", {'message_id': message_id}))
+
+class TestInboundMail(unittest.TestCase):
+	@classmethod
+	def setUpClass(cls):
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		email_account.db_set("enable_incoming", 1)
+
+	@classmethod
+	def tearDownClass(cls):
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		email_account.db_set("enable_incoming", 0)
+
+	def setUp(self):
+		cleanup()
+		frappe.db.sql('delete from `tabEmail Queue`')
+		frappe.db.sql('delete from `tabToDo`')
+
+	def get_test_mail(self, fname):
+		with open(os.path.join(os.path.dirname(__file__), "test_mails", fname), "r") as f:
+			return f.read()
+
+	def new_doc(self, doctype, **data):
+		doc = frappe.new_doc(doctype)
+		for field, value in data.items():
+			setattr(doc, field, value)
+		doc.insert()
+		return doc
+
+	def new_communication(self, **kwargs):
+		defaults = {
+			'subject': "Test Subject"
+		}
+		d = {**defaults, **kwargs}
+		return self.new_doc('Communication', **d)
+
+	def new_email_queue(self, **kwargs):
+		defaults = {
+			'message_id': get_message_id().strip(" <>")
+		}
+		d = {**defaults, **kwargs}
+		return self.new_doc('Email Queue', **d)
+
+	def new_todo(self, **kwargs):
+		defaults = {
+			'description': "Description"
+		}
+		d = {**defaults, **kwargs}
+		return self.new_doc('ToDo', **d)
+
+	def test_self_sent_mail(self):
+		"""Check that we raise SentEmailInInboxError if the inbound mail is self sent mail.
+		"""
+		mail_content = self.get_test_mail(fname="incoming-self-sent.raw")
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 1, 1)
+		with self.assertRaises(SentEmailInInboxError):
+			inbound_mail.process()
+
+	def test_mail_exist_validation(self):
+		"""Do not create communication record if the mail is already downloaded into the system.
+		"""
+		mail_content = self.get_test_mail(fname="incoming-1.raw")
+		message_id = Email(mail_content).message_id
+		# Create new communication record in DB
+		communication = self.new_communication(message_id=message_id)
+
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		new_communiction = inbound_mail.process()
+
+		# Make sure that uid is changed to new uid
+		self.assertEqual(new_communiction.uid, 12345)
+		self.assertEqual(communication.name, new_communiction.name)
+
+	def test_find_parent_email_queue(self):
+		"""If the mail is reply to the already sent mail, there will be a email queue record.
+		"""
+		# Create email queue record
+		queue_record = self.new_email_queue()
+
+		mail_content = self.get_test_mail(fname="reply-4.raw").replace(
+			"{{ message_id }}", queue_record.message_id
+		)
+
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		parent_queue = inbound_mail.parent_email_queue()
+		self.assertEqual(queue_record.name, parent_queue.name)
+
+	def test_find_parent_communication_through_queue(self):
+		"""Find parent communication of an inbound mail.
+		Cases where parent communication does exist:
+		1. No parent communication is the mail is not a reply.
+
+		Cases where parent communication does not exist:
+		2. If mail is not a reply to system sent mail, then there can exist co
+		"""
+		# Create email queue record
+		communication = self.new_communication()
+		queue_record = self.new_email_queue(communication=communication.name)
+		mail_content = self.get_test_mail(fname="reply-4.raw").replace(
+			"{{ message_id }}", queue_record.message_id
+		)
+
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		parent_communication = inbound_mail.parent_communication()
+		self.assertEqual(parent_communication.name, communication.name)
+
+	def test_find_parent_communication_for_self_reply(self):
+		"""If the inbound email is a reply but not reply to system sent mail.
+
+		Ex: User replied to his/her mail.
+		"""
+		message_id = "new-message-id"
+		mail_content = self.get_test_mail(fname="reply-4.raw").replace(
+			"{{ message_id }}", message_id
+		)
+
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		parent_communication = inbound_mail.parent_communication()
+		self.assertFalse(parent_communication)
+
+		communication = self.new_communication(message_id=message_id)
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		parent_communication = inbound_mail.parent_communication()
+		self.assertEqual(parent_communication.name, communication.name)
+
+	def test_find_parent_communication_from_header(self):
+		"""Incase of header contains parent communication name
+		"""
+		communication = self.new_communication()
+		mail_content = self.get_test_mail(fname="reply-4.raw").replace(
+			"{{ message_id }}", f"<{communication.name}@{frappe.local.site}>"
+		)
+
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		parent_communication = inbound_mail.parent_communication()
+		self.assertEqual(parent_communication.name, communication.name)
+
+	def test_reference_document(self):
+		# Create email queue record
+		todo = self.new_todo()
+		# communication = self.new_communication(reference_doctype='ToDo', reference_name=todo.name)
+		queue_record = self.new_email_queue(reference_doctype='ToDo', reference_name=todo.name)
+		mail_content = self.get_test_mail(fname="reply-4.raw").replace(
+			"{{ message_id }}", queue_record.message_id
+		)
+
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		reference_doc = inbound_mail.reference_document()
+		self.assertEqual(todo.name, reference_doc.name)
+
+	def test_reference_document_by_record_name_in_subject(self):
+		# Create email queue record
+		todo = self.new_todo()
+
+		mail_content = self.get_test_mail(fname="incoming-subject-placeholder.raw").replace(
+			"{{ subject }}", f"RE: (#{todo.name})"
+		)
+
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		reference_doc = inbound_mail.reference_document()
+		self.assertEqual(todo.name, reference_doc.name)
+
+	def test_reference_document_by_subject_match(self):
+		subject = "New todo"
+		todo = self.new_todo(sender='test_sender@example.com', description=subject)
+
+		mail_content = self.get_test_mail(fname="incoming-subject-placeholder.raw").replace(
+			"{{ subject }}", f"RE: {subject}"
+		)
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		reference_doc = inbound_mail.reference_document()
+		self.assertEqual(todo.name, reference_doc.name)
+
+	def test_create_communication_from_mail(self):
+		# Create email queue record
+		mail_content = self.get_test_mail(fname="incoming-2.raw")
+		email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
+		inbound_mail = InboundMail(mail_content, email_account, 12345, 1)
+		communication = inbound_mail.process()
+		self.assertTrue(communication.is_first)
+		self.assertTrue(communication._attachments)
+
 def cleanup(sender=None):
 	filters = {}
 	if sender:
@@ -207,4 +426,4 @@ def cleanup(sender=None):
 	names = frappe.get_list("Communication", filters=filters, fields=["name"])
 	for name in names:
 		frappe.delete_doc_if_exists("Communication", name.name)
-		frappe.delete_doc_if_exists("Communication Link", {"parent": name.name})
+		frappe.delete_doc_if_exists("Communication Link", {"parent": name.name})

frappe/email/doctype/email_account/test_mails/incoming-self-sent.raw

@@ -0,0 +1,91 @@
+Delivered-To: test_receiver@example.com
+Received: by 10.96.153.227 with SMTP id vj3csp416144qdb;
+        Mon, 15 Sep 2014 03:35:07 -0700 (PDT)
+X-Received: by 10.66.119.103 with SMTP id kt7mr36981968pab.95.1410777306321;
+        Mon, 15 Sep 2014 03:35:06 -0700 (PDT)
+Return-Path: <test@example.com>
+Received: from mail-pa0-x230.google.com (mail-pa0-x230.google.com [2607:f8b0:400e:c03::230])
+        by mx.google.com with ESMTPS id dg10si22178346pdb.115.2014.09.15.03.35.06
+        for <test_receiver@example.com>
+        (version=TLSv1 cipher=ECDHE-RSA-RC4-SHA bits=128/128);
+        Mon, 15 Sep 2014 03:35:06 -0700 (PDT)
+Received-SPF: pass (google.com: domain of test@example.com designates 2607:f8b0:400e:c03::230 as permitted sender) client-ip=2607:f8b0:400e:c03::230;
+Authentication-Results: mx.google.com;
+       spf=pass (google.com: domain of test@example.com designates 2607:f8b0:400e:c03::230 as permitted sender) smtp.mail=test@example.com;
+       dkim=pass header.i=@gmail.com;
+       dmarc=pass (p=NONE dis=NONE) header.from=gmail.com
+Received: by mail-pa0-f48.google.com with SMTP id hz1so6118714pad.21
+        for <test_receiver@example.com>; Mon, 15 Sep 2014 03:35:06 -0700 (PDT)
+DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed;
+        d=gmail.com; s=20120113;
+        h=from:content-type:subject:message-id:date:to:mime-version;
+        bh=rwiLijtF3lfy9M6cP/7dv2Hm7NJuBwFZn1OFsN8Tlvs=;
+        b=x7U4Ny3Kz2ULRJ7a04NDBrBTVhP2ImIB9n3LVNGQDnDonPUM5Ro/wZcxPTVnBWZ2L1
+         o1bGfP+lhBrvYUlHsd5r4FYC0Uvpad6hbzLr0DGUQgPTxW4cGKbtDEAq+BR2JWd9f803
+         vdjSWdGk8w2dt2qbngTqIZkm5U2XWjICDOAYuPIseLUgCFwi9lLyOSARFB7mjAa2YL7Q
+         Nswk7mbWU1hbnHP6jaBb0m8QanTc7Up944HpNDRxIrB1ZHgKzYhXtx8nhnOx588ZGIAe
+         E6tyG8IwogR11vLkkrBhtMaOme9PohYx4F1CSTiwspmDCadEzJFGRe//lEXKmZHAYH6g
+         90Zg==
+X-Received: by 10.70.38.135 with SMTP id g7mr22078275pdk.100.1410777305744;
+        Mon, 15 Sep 2014 03:35:05 -0700 (PDT)
+Return-Path: <test@example.com>
+Received: from [192.168.0.100] ([27.106.4.70])
+        by mx.google.com with ESMTPSA id zr6sm11025126pbc.50.2014.09.15.03.35.02
+        for <test_receiver@example.com>
+        (version=TLSv1 cipher=ECDHE-RSA-RC4-SHA bits=128/128);
+        Mon, 15 Sep 2014 03:35:04 -0700 (PDT)
+From: Rushabh Mehta <test@example.com>
+Content-Type: multipart/alternative; boundary="Apple-Mail=_57F71261-5C3A-43F6-918B-4438B96F61AA"
+Subject: test mail 🦄🌈😎
+Message-Id: <9143999C-8456-4399-9CF1-4A2DA9DD7711@gmail.com>
+Date: Mon, 15 Sep 2014 16:04:57 +0530
+To: Rushabh Mehta <test_receiver@example.com>
+Mime-Version: 1.0 (Mac OS X Mail 7.3 \(1878.6\))
+X-Mailer: Apple Mail (2.1878.6)
+
+
+--Apple-Mail=_57F71261-5C3A-43F6-918B-4438B96F61AA
+Content-Transfer-Encoding: 7bit
+Content-Type: text/plain;
+	charset=us-ascii
+
+test mail
+
+
+
+@rushabh_mehta
+https://erpnext.org
+
+
+--Apple-Mail=_57F71261-5C3A-43F6-918B-4438B96F61AA
+Content-Transfer-Encoding: quoted-printable
+Content-Type: text/html;
+	charset=us-ascii
+
+<html><head><meta http-equiv=3D"Content-Type" content=3D"text/html =
+charset=3Dus-ascii"></head><body style=3D"word-wrap: break-word; =
+-webkit-nbsp-mode: space; -webkit-line-break: after-white-space;">test =
+mail<br><div apple-content-edited=3D"true">
+<div style=3D"color: rgb(0, 0, 0); letter-spacing: normal; orphans: =
+auto; text-align: start; text-indent: 0px; text-transform: none; =
+white-space: normal; widows: auto; word-spacing: 0px; =
+-webkit-text-stroke-width: 0px; word-wrap: break-word; =
+-webkit-nbsp-mode: space; -webkit-line-break: after-white-space;"><div =
+style=3D"color: rgb(0, 0, 0); font-family: Helvetica; font-style: =
+normal; font-variant: normal; font-weight: normal; letter-spacing: =
+normal; line-height: normal; orphans: 2; text-align: -webkit-auto; =
+text-indent: 0px; text-transform: none; white-space: normal; widows: 2; =
+word-spacing: 0px; -webkit-text-stroke-width: 0px; word-wrap: =
+break-word; -webkit-nbsp-mode: space; -webkit-line-break: =
+after-white-space;"><br><br><br>@rushabh_mehta</div><div style=3D"color: =
+rgb(0, 0, 0); font-family: Helvetica; font-style: normal; font-variant: =
+normal; font-weight: normal; letter-spacing: normal; line-height: =
+normal; orphans: 2; text-align: -webkit-auto; text-indent: 0px; =
+text-transform: none; white-space: normal; widows: 2; word-spacing: 0px; =
+-webkit-text-stroke-width: 0px; word-wrap: break-word; =
+-webkit-nbsp-mode: space; -webkit-line-break: after-white-space;"><a =
+href=3D"https://erpnext.org">https://erpnext.org</a><br></div></div>
+</div>
+<br></body></html>=
+
+--Apple-Mail=_57F71261-5C3A-43F6-918B-4438B96F61AA--

frappe/email/doctype/email_account/test_mails/incoming-subject-placeholder.raw

@@ -0,0 +1,183 @@
+Return-path: <test_sender@example.com>
+Envelope-to: test_receiver@example.com
+Delivery-date: Wed, 27 Jan 2016 16:24:20 +0800
+Received: from 23-59-23-10.perm.iinet.net.au ([23.59.23.10]:62191 helo=DESKTOP7C66I2M)
+	by webcloud85.au.syrahost.com with esmtp (Exim 4.86)
+	(envelope-from <test_sender@example.com>)
+	id 1aOLOj-002xFL-CP
+	for test_receiver@example.com; Wed, 27 Jan 2016 16:24:20 +0800
+From: <test_sender@example.com>
+To: <test_receiver@example.com>
+References: <COMM-02154@site1.local>
+In-Reply-To: <COMM-02154@site1.local>
+Subject: RE: {{ subject }}
+Date: Wed, 27 Jan 2016 16:24:09 +0800
+Message-ID: <000001d158dc$1b8363a0$528a2ae0$@example.com>
+MIME-Version: 1.0
+Content-Type: multipart/mixed;
+	boundary="----=_NextPart_000_0001_01D1591F.29A7DC20"
+X-Mailer: Microsoft Outlook 14.0
+Thread-Index: AQJZfZxrgcB9KnMqoZ+S4Qq9hcoSeZ3+vGiQ
+Content-Language: en-au
+
+This is a multipart message in MIME format.
+
+------=_NextPart_000_0001_01D1591F.29A7DC20
+Content-Type: multipart/alternative;
+	boundary="----=_NextPart_001_0002_01D1591F.29A7DC20"
+
+
+------=_NextPart_001_0002_01D1591F.29A7DC20
+Content-Type: text/plain;
+	charset="utf-8"
+Content-Transfer-Encoding: quoted-printable
+
+Test purely for testing with the debugger has email attached
+
+=20
+
+From: Notification [mailto:test_receiver@example.com]=20
+Sent: Wednesday, 27 January 2016 9:30 AM
+To: test_receiver@example.com
+Subject: Sales Invoice: SINV-12276
+
+=20
+
+test no 6 sent from bench to outlook to be replied to with messaging
+
+
+
+
+------=_NextPart_001_0002_01D1591F.29A7DC20
+Content-Type: text/html;
+	charset="utf-8"
+Content-Transfer-Encoding: quoted-printable
+
+<html xmlns:v=3D"urn:schemas-microsoft-com:vml" =
+xmlns:o=3D"urn:schemas-microsoft-com:office:office" =
+xmlns:w=3D"urn:schemas-microsoft-com:office:word" =
+xmlns:m=3D"http://schemas.microsoft.com/office/2004/12/omml" =
+xmlns=3D"http://www.w3.org/TR/REC-html40"><head><meta =
+http-equiv=3DContent-Type content=3D"text/html; charset=3Dutf-8"><meta =
+name=3DGenerator content=3D"Microsoft Word 14 (filtered =
+medium)"><title>hi there</title><style><!--
+/* Font Definitions */
+@font-face
+	{font-family:Helvetica;
+	panose-1:2 11 6 4 2 2 2 2 2 4;}
+@font-face
+	{font-family:"Cambria Math";
+	panose-1:0 0 0 0 0 0 0 0 0 0;}
+@font-face
+	{font-family:Calibri;
+	panose-1:2 15 5 2 2 2 4 3 2 4;}
+@font-face
+	{font-family:Tahoma;
+	panose-1:2 11 6 4 3 5 4 4 2 4;}
+/* Style Definitions */
+p.MsoNormal, li.MsoNormal, div.MsoNormal
+	{margin:0cm;
+	margin-bottom:.0001pt;
+	font-size:12.0pt;
+	font-family:"Times New Roman","serif";}
+a:link, span.MsoHyperlink
+	{mso-style-priority:99;
+	color:blue;
+	text-decoration:underline;}
+a:visited, span.MsoHyperlinkFollowed
+	{mso-style-priority:99;
+	color:purple;
+	text-decoration:underline;}
+p
+	{mso-style-priority:99;
+	mso-margin-top-alt:auto;
+	margin-right:0cm;
+	mso-margin-bottom-alt:auto;
+	margin-left:0cm;
+	font-size:12.0pt;
+	font-family:"Times New Roman","serif";}
+span.EmailStyle18
+	{mso-style-type:personal-reply;
+	font-family:"Calibri","sans-serif";
+	color:#1F497D;}
+.MsoChpDefault
+	{mso-style-type:export-only;
+	font-size:10.0pt;}
+@page WordSection1
+	{size:612.0pt 792.0pt;
+	margin:72.0pt 72.0pt 72.0pt 72.0pt;}
+div.WordSection1
+	{page:WordSection1;}
+--></style><!--[if gte mso 9]><xml>
+<o:shapedefaults v:ext=3D"edit" spidmax=3D"1026" />
+</xml><![endif]--><!--[if gte mso 9]><xml>
+<o:shapelayout v:ext=3D"edit">
+<o:idmap v:ext=3D"edit" data=3D"1" />
+</o:shapelayout></xml><![endif]--></head><body lang=3DEN-AU link=3Dblue =
+vlink=3Dpurple><div class=3DWordSection1><p class=3DMsoNormal><span =
+style=3D'font-size:11.0pt;font-family:"Calibri","sans-serif";color:#1F497=
+D'>Test purely for testing with the debugger has email =
+attached<o:p></o:p></span></p><p class=3DMsoNormal><a =
+name=3D"_MailEndCompose"><span =
+style=3D'font-size:11.0pt;font-family:"Calibri","sans-serif";color:#1F497=
+D'><o:p>&nbsp;</o:p></span></a></p><div><div =
+style=3D'border:none;border-top:solid #B5C4DF 1.0pt;padding:3.0pt 0cm =
+0cm 0cm'><p class=3DMsoNormal><b><span lang=3DEN-US =
+style=3D'font-size:10.0pt;font-family:"Tahoma","sans-serif"'>From:</span>=
+</b><span lang=3DEN-US =
+style=3D'font-size:10.0pt;font-family:"Tahoma","sans-serif"'> =
+Notification [mailto:test_receiver@example.com] <br><b>Sent:</b> Wednesday, 27 =
+January 2016 9:30 AM<br><b>To:</b> =
+test_receiver@example.com<br><b>Subject:</b> Sales Invoice: =
+SINV-12276<o:p></o:p></span></p></div></div><p =
+class=3DMsoNormal><o:p>&nbsp;</o:p></p><div><p><span =
+style=3D'font-size:10.5pt;font-family:"Helvetica","sans-serif";color:#364=
+14C'>test no 3 sent from bench to outlook to be replied to with =
+messaging<o:p></o:p></span></p><p><span =
+style=3D'font-size:10.5pt;font-family:"Helvetica","sans-serif";color:#364=
+14C'>fizz buzz <o:p></o:p></span></p></div><div =
+style=3D'border:none;border-top:solid #D1D8DD 1.0pt;padding:0cm 0cm 0cm =
+0cm;margin-top:22.5pt;margin-bottom:11.25pt'><div =
+style=3D'margin-top:11.25pt;margin-bottom:11.25pt'><p class=3DMsoNormal =
+align=3Dcenter style=3D'text-align:center'><span =
+style=3D'font-size:8.5pt;font-family:"Helvetica","sans-serif";color:#8D99=
+A6'>This email was sent to <a =
+href=3D"mailto:test_receiver@example.com">test_receiver@example.=
+com</a> and copied to SuperUser <o:p></o:p></span></p><p =
+align=3Dcenter =
+style=3D'mso-margin-top-alt:11.25pt;margin-right:0cm;margin-bottom:11.25p=
+t;margin-left:0cm;text-align:center'><span =
+style=3D'font-size:8.5pt;font-family:"Helvetica","sans-serif";color:#8D99=
+A6'><span =
+style=3D'color:#8D99A6'>Leave this conversation =
+</span></a><o:p></o:p></span></p></div><div =
+style=3D'margin-top:11.25pt;margin-bottom:11.25pt'><p class=3DMsoNormal =
+align=3Dcenter style=3D'text-align:center'><span =
+style=3D'font-size:8.5pt;font-family:"Helvetica","sans-serif";color:#8D99=
+A6'>hi<o:p></o:p></span></p></div></div></div></body></html>
+------=_NextPart_001_0002_01D1591F.29A7DC20--
+
+------=_NextPart_000_0001_01D1591F.29A7DC20
+Content-Type: message/rfc822
+Content-Transfer-Encoding: 7bit
+Content-Disposition: attachment
+
+Received: from 203-59-223-10.perm.iinet.net.au ([23.59.23.10]:49772 helo=DESKTOP7C66I2M)
+	by webcloud85.au.syrahost.com with esmtpsa (TLSv1.2:DHE-RSA-AES256-GCM-SHA384:256)
+	(Exim 4.86)
+	(envelope-from <test_sender@example.com>)
+	id 1aOEtO-003tI4-Kv
+	for test_receiver@example.com; Wed, 27 Jan 2016 09:27:30 +0800
+Return-Path: <test_sender@example.com>
+From: "Microsoft Outlook" <test_sender@example.com>
+To: <test_receiver@example.com>
+Subject: Microsoft Outlook Test Message
+MIME-Version: 1.0
+Content-Type: text/plain;
+	charset="utf-8"
+Content-Transfer-Encoding: quoted-printable
+X-Mailer: Microsoft Outlook 14.0
+Thread-Index: AdFYoeN8x8wUI/+QSoCJkp33NKPVmw==
+
+This is an e-mail message sent automatically by Microsoft Outlook while =
+testing the settings for your account.

frappe/email/doctype/email_queue/email_queue.py

@@ -45,6 +45,11 @@ class EmailQueue(Document):
 	def find(cls, name):
 		return frappe.get_doc(cls.DOCTYPE, name)
 
+	@classmethod
+	def find_one_by_filters(cls, **kwargs):
+		name = frappe.db.get_value(cls.DOCTYPE, kwargs)
+		return cls.find(name) if name else None
+
 	def update_db(self, commit=False, **kwargs):
 		frappe.db.set_value(self.DOCTYPE, self.name, kwargs)
 		if commit:

frappe/email/email_body.py

@@ -353,9 +353,7 @@ def add_attachment(fname, fcontent, content_type=None,
 
 def get_message_id():
 	'''Returns Message ID created from doctype and name'''
-	return "<{unique}@{site}>".format(
-			site=frappe.local.site,
-			unique=email.utils.make_msgid(random_string(10)).split('@')[0].split('<')[1])
+	return email.utils.make_msgid(domain=frappe.local.site)
 
 def get_signature(email_account):
 	if email_account and email_account.add_signature and email_account.signature:

frappe/email/receive.py

@@ -8,6 +8,7 @@ import imaplib
 import poplib
 import re
 import time
+import json
 from email.header import decode_header
 
 import _socket
@@ -20,13 +21,26 @@ from frappe import _, safe_decode, safe_encode
 from frappe.core.doctype.file.file import (MaxFileSizeReachedError,
 	get_random_filename)
 from frappe.utils import (cint, convert_utc_to_user_timezone, cstr,
-	extract_email_id, markdown, now, parse_addr, strip)
+	extract_email_id, markdown, now, parse_addr, strip, get_datetime,
+	add_days, sanitize_html)
+from frappe.utils.user import is_system_user
+from frappe.utils.html_utils import clean_email_html
+
+# fix due to a python bug in poplib that limits it to 2048
+poplib._MAXLINE = 20480
+imaplib._MAXLINE = 20480
+
+# fix due to a python bug in poplib that limits it to 2048
+poplib._MAXLINE = 20480
+imaplib._MAXLINE = 20480
 
 
 class EmailSizeExceededError(frappe.ValidationError): pass
 class EmailTimeoutError(frappe.ValidationError): pass
 class TotalSizeExceededError(frappe.ValidationError): pass
 class LoginLimitExceeded(frappe.ValidationError): pass
+class SentEmailInInboxError(Exception):
+	pass
 
 class EmailServer:
 	"""Wrapper for POP server to pull emails."""
@@ -100,14 +114,11 @@ class EmailServer:
 
 	def get_messages(self):
 		"""Returns new email messages in a list."""
-		if not self.check_mails():
-			return # nothing to do
+		if not (self.check_mails() or self.connect()):
+			return []
 
 		frappe.db.commit()
 
-		if not self.connect():
-			return
-
 		uid_list = []
 
 		try:
@@ -116,7 +127,6 @@ class EmailServer:
 			self.latest_messages = []
 			self.seen_status = {}
 			self.uid_reindexed = False
-
 			uid_list = email_list = self.get_new_mails()
 
 			if not email_list:
@@ -132,11 +142,7 @@ class EmailServer:
 			self.max_email_size = cint(frappe.local.conf.get("max_email_size"))
 			self.max_total_size = 5 * self.max_email_size
 
-			for i, message_meta in enumerate(email_list):
-				# do not pull more than NUM emails
-				if (i+1) > num:
-					break
-
+			for i, message_meta in enumerate(email_list[:num]):
 				try:
 					self.retrieve_message(message_meta, i+1)
 				except (TotalSizeExceededError, EmailTimeoutError, LoginLimitExceeded):
@@ -152,7 +158,6 @@ class EmailServer:
 		except Exception as e:
 			if self.has_login_limit_exceeded(e):
 				pass
-
 			else:
 				raise
 
@@ -369,6 +374,7 @@ class Email:
 			else:
 				self.mail = email.message_from_string(content)
 
+		self.raw_message = content
 		self.text_content = ''
 		self.html_content = ''
 		self.attachments = []
@@ -391,6 +397,10 @@ class Email:
 		if self.date > now():
 			self.date = now()
 
+	@property
+	def in_reply_to(self):
+		return (self.mail.get("In-Reply-To") or "").strip(" <>")
+
 	def parse(self):
 		"""Walk and process multi-part email."""
 		for part in self.mail.walk():
@@ -558,10 +568,330 @@ class Email:
 		l = re.findall(r'(?<=\[)[\w/-]+', self.subject)
 		return l and l[0] or None
 
+	def is_reply(self):
+		return bool(self.in_reply_to)
+
+class InboundMail(Email):
+	"""Class representation of incoming mail along with mail handlers.
+	"""
+	def __init__(self, content, email_account, uid=None, seen_status=None):
+		super().__init__(content)
+		self.email_account = email_account
+		self.uid = uid or -1
+		self.seen_status = seen_status or 0
+
+		# System documents related to this mail
+		self._parent_email_queue = None
+		self._parent_communication = None
+		self._reference_document = None
+
+		self.flags = frappe._dict()
+
+	def get_content(self):
+		if self.content_type == 'text/html':
+			return clean_email_html(self.content)
+
+	def process(self):
+		"""Create communication record from email.
+		"""
+		if self.is_sender_same_as_receiver() and not self.is_reply():
+			if frappe.flags.in_test:
+				print('WARN: Cannot pull email. Sender same as recipient inbox')
+			raise SentEmailInInboxError
+
+		communication = self.is_exist_in_system()
+		if communication:
+			communication.update_db(uid=self.uid)
+			communication.reload()
+			return communication
+
+		self.flags.is_new_communication = True
+		return self._build_communication_doc()
+
+	def _build_communication_doc(self):
+		data = self.as_dict()
+		data['doctype'] = "Communication"
+
+		if self.parent_communication():
+			data['in_reply_to'] = self.parent_communication().name
+
+		if self.reference_document():
+			data['reference_doctype'] = self.reference_document().doctype
+			data['reference_name'] = self.reference_document().name
+		elif self.email_account.append_to and self.email_account.append_to != 'Communication':
+			reference_doc = self._create_reference_document(self.email_account.append_to)
+			if reference_doc:
+				data['reference_doctype'] = reference_doc.doctype
+				data['reference_name'] = reference_doc.name
+				data['is_first'] = True
+
+		if self.is_notification():
+			# Disable notifications for notification.
+			data['unread_notification_sent'] = 1
+
+		if self.seen_status:
+			data['_seen'] = json.dumps(self.get_users_linked_to_account(self.email_account))
+
+		communication = frappe.get_doc(data)
+		communication.flags.in_receive = True
+		communication.insert(ignore_permissions=True)
+
+		# save attachments
+		communication._attachments = self.save_attachments_in_doc(communication)
+		communication.content = sanitize_html(self.replace_inline_images(communication._attachments))
+		communication.save()
+		return communication
+
+	def replace_inline_images(self, attachments):
+		# replace inline images
+		content = self.content
+		for file in attachments:
+			if file.name in self.cid_map and self.cid_map[file.name]:
+				content = content.replace("cid:{0}".format(self.cid_map[file.name]),
+					file.file_url)
+		return content
+
+	def is_notification(self):
+		isnotification = self.mail.get("isnotification")
+		return isnotification and ("notification" in isnotification)
+
+	def is_exist_in_system(self):
+		"""Check if this email already exists in the system(as communication document).
+		"""
+		from frappe.core.doctype.communication.communication import Communication
+		if not self.message_id:
+			return
 
-# fix due to a python bug in poplib that limits it to 2048
-poplib._MAXLINE = 20480
-imaplib._MAXLINE = 20480
+		return Communication.find_one_by_filters(message_id = self.message_id,
+			order_by = 'creation DESC')
+
+	def is_sender_same_as_receiver(self):
+		return self.from_email == self.email_account.email_id
+
+	def is_reply_to_system_sent_mail(self):
+		"""Is it a reply to already sent mail.
+		"""
+		return self.is_reply() and frappe.local.site in self.in_reply_to
+
+	def parent_email_queue(self):
+		"""Get parent record from `Email Queue`.
+
+		If it is a reply to already sent mail, then there will be a parent record in EMail Queue.
+		"""
+		from frappe.email.doctype.email_queue.email_queue import EmailQueue
+
+		if self._parent_email_queue is not None:
+			return self._parent_email_queue
+
+		parent_email_queue = ''
+		if self.is_reply_to_system_sent_mail():
+			parent_email_queue = EmailQueue.find_one_by_filters(message_id=self.in_reply_to)
+
+		self._parent_email_queue = parent_email_queue or ''
+		return self._parent_email_queue
+
+	def parent_communication(self):
+		"""Find a related communication so that we can prepare a mail thread.
+
+		The way it happens is by using in-reply-to header, and we can't make thread if it does not exist.
+
+		Here are the cases to handle:
+		1. If mail is a reply to already sent mail, then we can get parent communicaion from
+			Email Queue record.
+		2. Sometimes we send communication name in message-ID directly, use that to get parent communication.
+		3. Sender sent a reply but reply is on top of what (s)he sent before,
+			then parent record exists directly in communication.
+		"""
+		from frappe.core.doctype.communication.communication import Communication
+		if self._parent_communication is not None:
+			return self._parent_communication
+
+		if not self.is_reply():
+			return ''
+
+		if not self.is_reply_to_system_sent_mail():
+			communication = Communication.find_one_by_filters(message_id=self.in_reply_to,
+				creation = ['>=', self.get_relative_dt(-30)])
+		elif self.parent_email_queue() and self.parent_email_queue().communication:
+			communication = Communication.find(self.parent_email_queue().communication, ignore_error=True)
+		else:
+			reference = self.in_reply_to
+			if '@' in self.in_reply_to:
+				reference, _ = self.in_reply_to.split("@", 1)
+			communication = Communication.find(reference, ignore_error=True)
+
+		self._parent_communication = communication or ''
+		return self._parent_communication
+
+	def reference_document(self):
+		"""Reference document is a document to which mail relate to.
+
+		We can get reference document from Parent record(EmailQueue | Communication) if exists.
+		Otherwise we do subject match to find reference document if we know the reference(append_to) doctype.
+		"""
+		if self._reference_document is not None:
+			return self._reference_document
+
+		reference_document = ""
+		parent = self.parent_email_queue() or self.parent_communication()
+
+		if parent and parent.reference_doctype:
+			reference_doctype, reference_name = parent.reference_doctype, parent.reference_name
+			reference_document = self.get_doc(reference_doctype, reference_name, ignore_error=True)
+
+		if not reference_document and self.email_account.append_to:
+			reference_document = self.match_record_by_subject_and_sender(self.email_account.append_to)
+
+			# if not reference_document:
+			# 	reference_document = Create_reference_document(self.email_account.append_to)
+
+		self._reference_document = reference_document or ''
+		return self._reference_document
+
+	def get_reference_name_from_subject(self):
+		"""
+		Ex: "Re: Your email (#OPP-2020-2334343)"
+		"""
+		return self.subject.rsplit('#', 1)[-1].strip(' ()')
+
+	def match_record_by_subject_and_sender(self, doctype):
+		"""Find a record in the given doctype that matches with email subject and sender.
+
+		Cases:
+		1. Sometimes record name is part of subject. We can get document by parsing name from subject
+		2. Find by matching sender and subject
+		3. Find by matching subject alone (Special case)
+			Ex: when a System User is using Outlook and replies to an email from their own client,
+			it reaches the Email Account with the threading info lost and the (sender + subject match)
+			doesn't work because the sender in the first communication was someone different to whom
+			the system user is replying to via the common email account in Frappe. This fix bypasses
+			the sender match when the sender is a system user and subject is atleast 10 chars long
+			(for additional safety)
+
+		NOTE: We consider not to match by subject if match record is very old.
+		"""
+		name = self.get_reference_name_from_subject()
+		email_fields = self.get_email_fields(doctype)
+
+		record = self.get_doc(doctype, name, ignore_error=True) if name else None
+
+		if not record:
+			subject = self.clean_subject(self.subject)
+			filters = {
+				email_fields.subject_field: ("like", f"%{subject}%"),
+				"creation": (">", self.get_relative_dt(days=-60))
+			}
+
+			# Sender check is not needed incase mail is from system user.
+			if not (len(subject) > 10 and is_system_user(self.from_email)):
+				filters[email_fields.sender_field] = self.from_email
+
+			name = frappe.db.get_value(self.email_account.append_to, filters = filters)
+			record = self.get_doc(doctype, name, ignore_error=True) if name else None
+		return record
+
+	def _create_reference_document(self, doctype):
+		""" Create reference document if it does not exist in the system.
+		"""
+		parent = frappe.new_doc(doctype)
+		email_fileds = self.get_email_fields(doctype)
+
+		if email_fileds.subject_field:
+			parent.set(email_fileds.subject_field, frappe.as_unicode(self.subject)[:140])
+
+		if email_fileds.sender_field:
+			parent.set(email_fileds.sender_field, frappe.as_unicode(self.from_email))
+
+		parent.flags.ignore_mandatory = True
+
+		try:
+			parent.insert(ignore_permissions=True)
+		except frappe.DuplicateEntryError:
+			# try and find matching parent
+			parent_name = frappe.db.get_value(self.email_account.append_to,
+				{email_fileds.sender_field: email.from_email}
+			)
+			if parent_name:
+				parent.name = parent_name
+			else:
+				parent = None
+		return parent
+
+
+	@staticmethod
+	def get_doc(doctype, docname, ignore_error=False):
+		try:
+			return frappe.get_doc(doctype, docname)
+		except frappe.DoesNotExistError:
+			if ignore_error:
+				return
+			raise
+
+	@staticmethod
+	def get_relative_dt(days):
+		"""Get relative to current datetime. Only relative days are supported.
+		"""
+		return add_days(get_datetime(), days)
+
+	@staticmethod
+	def get_users_linked_to_account(email_account):
+		"""Get list of users who linked to Email account.
+		"""
+		users = frappe.get_all("User Email", filters={"email_account": email_account.name},
+			fields=["parent"])
+		return list(set([user.get("parent") for user in users]))
+
+	@staticmethod
+	def clean_subject(subject):
+		"""Remove Prefixes like 'fw', FWD', 're' etc from subject.
+		"""
+		# Match strings like "fw:", "re	:" etc.
+		regex = r"(^\s*(fw|fwd|wg)[^:]*:|\s*(re|aw)[^:]*:\s*)*"
+		return frappe.as_unicode(strip(re.sub(regex, "", subject, 0, flags=re.IGNORECASE)))
+
+	@staticmethod
+	def get_email_fields(doctype):
+		"""Returns Email related fields of a doctype.
+		"""
+		fields = frappe._dict()
+
+		email_fields = ['subject_field', 'sender_field']
+		meta = frappe.get_meta(doctype)
+
+		for field in email_fields:
+			if hasattr(meta, field):
+				fields[field] = getattr(meta, field)
+		return fields
+
+	@staticmethod
+	def get_document(self, doctype, name):
+		"""Is same as frappe.get_doc but suppresses the DoesNotExist error.
+		"""
+		try:
+			return frappe.get_doc(doctype, name)
+		except frappe.DoesNotExistError:
+			return None
+
+	def as_dict(self):
+		"""
+		"""
+		return {
+			"subject": self.subject,
+			"content": self.get_content(),
+			'text_content': self.text_content,
+			"sent_or_received": "Received",
+			"sender_full_name": self.from_real_name,
+			"sender": self.from_email,
+			"recipients": self.mail.get("To"),
+			"cc": self.mail.get("CC"),
+			"email_account": self.email_account.name,
+			"communication_medium": "Email",
+			"uid": self.uid,
+			"message_id": self.message_id,
+			"communication_date": self.date,
+			"has_attachment": 1 if self.attachments else 0,
+			"seen": self.seen_status or 0
+		}
 
 class TimerMixin(object):
 	def __init__(self, *args, **kwargs):

frappe/tests/test_email.py

@@ -1,8 +1,6 @@
 # Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
 # MIT License. See license.txt
 
-from __future__ import unicode_literals
-
 import unittest, frappe, re, email
 from six import PY3
 
@@ -178,7 +176,8 @@ class TestEmail(unittest.TestCase):
 		frappe.db.sql('''delete from `tabCommunication` where sender = 'sukh@yyy.com' ''')
 
 		with open(frappe.get_app_path('frappe', 'tests', 'data', 'email_with_image.txt'), 'r') as raw:
-			communication = email_account.insert_communication(raw.read())
+			mails = email_account.get_inbound_mails(test_mails=[raw.read()])
+			communication = mails[0].process()
 
 		self.assertTrue(re.search('''<img[^>]*src=["']/private/files/rtco1.png[^>]*>''', communication.content))
 		self.assertTrue(re.search('''<img[^>]*src=["']/private/files/rtco2.png[^>]*>''', communication.content))