diff --git a/addons/link_tracker/models/link_tracker.py b/addons/link_tracker/models/link_tracker.py
index 9938a8ee7da..8abed891105 100644
--- a/addons/link_tracker/models/link_tracker.py
+++ b/addons/link_tracker/models/link_tracker.py
@@ -11,8 +11,7 @@ from werkzeug import urls
from odoo import tools, models, fields, api, _
from odoo.exceptions import UserError
from odoo.osv import expression
-
-URL_MAX_SIZE = 10 * 1024 * 1024
+from odoo.addons.mail.tools import link_preview
class LinkTracker(models.Model):
@@ -112,22 +111,10 @@ class LinkTracker(models.Model):
@api.model
@api.depends('url')
def _get_title_from_url(self, url):
- try:
- head = requests.head(url, allow_redirects=True, timeout=5)
- if (
- int(head.headers.get('Content-Length', 0)) > URL_MAX_SIZE
- or
- 'text/html' not in head.headers.get('Content-Type', 'text/html')
- ):
- return url
- # HTML parser can work with a part of page, so ask server to limit downloading to 50 KB
- page = requests.get(url, timeout=5, headers={"range": "bytes=0-50000"})
- p = html.fromstring(page.text.encode('utf-8'), parser=html.HTMLParser(encoding='utf-8'))
- title = p.find('.//title').text
- except:
- title = url
-
- return title
+ preview = link_preview.get_link_preview_from_url(url)
+ if preview and preview.get('og_title'):
+ return preview['og_title']
+ return url
@api.constrains('url', 'campaign_id', 'medium_id', 'source_id')
def _check_unicity(self):
diff --git a/addons/mail/models/mail_link_preview.py b/addons/mail/models/mail_link_preview.py
index 7a6b1eee905..1166383d884 100644
--- a/addons/mail/models/mail_link_preview.py
+++ b/addons/mail/models/mail_link_preview.py
@@ -3,11 +3,12 @@
from datetime import datetime
from dateutil.relativedelta import relativedelta
-from lxml import html, etree
+from lxml import html
from urllib.parse import urlparse
import requests
from odoo import api, models, fields
+from odoo.addons.mail.tools import link_preview
class LinkPreview(models.Model):
@@ -54,7 +55,7 @@ class LinkPreview(models.Model):
def _create_link_preview(self, url, message_id, request_session):
if self._is_domain_throttled(url):
return self.env['mail.link.preview']
- link_preview_data = self._get_link_preview_from_url(url, request_session)
+ link_preview_data = link_preview.get_link_preview_from_url(url, request_session)
if link_preview_data:
link_preview_data['message_id'] = message_id
return self.create(link_preview_data)
@@ -86,54 +87,6 @@ class LinkPreview(models.Model):
link_preview_throttle = int(self.env['ir.config_parameter'].get_param('mail.link_preview_throttle', 99))
return call_counter > link_preview_throttle
- @api.model
- def _get_link_preview_from_url(self, url, request_session):
- try:
- response = request_session.head(url, timeout=3, allow_redirects=True)
- except requests.exceptions.RequestException:
- return False
- if response.status_code != requests.codes.ok:
- return False
- image_mimetype = (
- 'image/bmp',
- 'image/gif',
- 'image/jpeg',
- 'image/png',
- 'image/tiff',
- 'image/x-icon',
- )
- # Content-Type header can return a charset, but we just need the
- # mimetype (eg: image/jpeg;charset=ISO-8859-1)
- content_type = response.headers['Content-Type'].split(';')
- if response.headers['Content-Type'].startswith(image_mimetype):
- return {
- 'image_mimetype': content_type[0],
- 'source_url': url,
- }
- if response.headers['Content-Type'].startswith('text/html'):
- return self._get_link_preview_from_html(url, request_session)
- return False
-
- def _get_link_preview_from_html(self, url, request_session):
- response = request_session.get(url, timeout=3)
- parser = etree.HTMLParser(encoding=response.encoding)
- tree = html.fromstring(response.content, parser=parser)
- og_title = tree.xpath('//meta[@property="og:title"]/@content')
- if not og_title:
- return False
- og_description = tree.xpath('//meta[@property="og:description"]/@content')
- og_type = tree.xpath('//meta[@property="og:type"]/@content')
- og_image = tree.xpath('//meta[@property="og:image"]/@content')
- og_mimetype = tree.xpath('//meta[@property="og:image:type"]/@content')
- return {
- 'og_description': og_description[0] if og_description else None,
- 'og_image': og_image[0] if og_image else None,
- 'og_mimetype': og_mimetype[0] if og_mimetype else None,
- 'og_title': og_title[0],
- 'og_type': og_type[0] if og_type else None,
- 'source_url': url,
- }
-
def _link_preview_format(self):
return [{
'id': preview.id,
diff --git a/addons/mail/tests/test_link_preview.py b/addons/mail/tests/test_link_preview.py
index 330def1a668..ee4108f1aa6 100644
--- a/addons/mail/tests/test_link_preview.py
+++ b/addons/mail/tests/test_link_preview.py
@@ -1,36 +1,17 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from functools import partial
+import io
from odoo.addons.mail.tests.common import mail_new_test_user
from odoo.addons.mail.tests.common import MailCommon
+from odoo.addons.mail.tools import link_preview
from unittest.mock import patch
import requests
discuss_channel_new_test_user = partial(mail_new_test_user, context={'discuss_channel_nosubscribe': False})
-def _patched_get_html(*args, **kwargs):
- response = requests.Response()
- response.status_code = 200
- response._content = b"""
-
-
-
-
-
-
- """
- response.headers["Content-Type"] = 'text/html'
- return response
-
-def _patch_head_html(*args, **kwargs):
- response = requests.Response()
- response.status_code = 200
- response.headers["Content-Type"] = 'text/html'
- return response
-
-
class TestLinkPreview(MailCommon):
@classmethod
@@ -42,29 +23,82 @@ class TestLinkPreview(MailCommon):
groups='base.group_user')
cls.thread = cls.env['res.partner'].create({'name': 'a partner'})
+ cls.title = 'Test title'
+ cls.og_title = 'Test OG title'
+ cls.og_description = 'Test OG description'
+ cls.og_image = 'https://dummy-image-url.nothing'
+ cls.source_url = 'https://thisdomainedoentexist.nothing'
+
+ def _patch_head_html(self, *args, **kwargs):
+ response = requests.Response()
+ response.status_code = 200
+ response.headers["Content-Type"] = 'text/html'
+ return response
+
+ def _patched_get_html(self, content_type, content):
+ response = requests.Response()
+ response.status_code = 200
+ response._content = content
+ # To handle chunks read on stream requests
+ response.raw = io.BytesIO(response._content)
+ response.headers["Content-Type"] = content_type
+ return response
+
+ def _patch_with_og_properties(self, *args, **kwargs):
+ content = b"""
+
+
+ Test title
+
+
+
+
+
+ """
+ return self._patched_get_html('text/html', content)
+
+ def _patch_without_og_properties(self, *args, **kwargs):
+ content = b"""
+
+
+ Test title
+
+
+ """
+ return self._patched_get_html('text/html', content)
+
+ def _patch_with_image_mimetype(self, *args, **kwargs):
+ content = b"""
+
+
+ 
+
+
+ """
+ return self._patched_get_html('image/png', content)
def test_01_link_preview_throttle(self):
- with patch.object(requests.Session, 'get', _patched_get_html), patch.object(requests.Session, 'head', _patch_head_html):
+ with patch.object(requests.Session, 'get', self._patch_with_og_properties), patch.object(requests.Session, 'head', self._patch_head_html):
throttle = int(self.env['ir.config_parameter'].sudo().get_param('mail.link_preview_throttle', 99))
link_previews = []
for _ in range(throttle):
- link_previews.append({'source_url': 'https://thisdomainedoentexist.nothing', 'message_id': 1})
+ link_previews.append({'source_url': self.source_url, 'message_id': 1})
self.env['mail.link.preview'].create(link_previews)
message = self.env['mail.message'].create({
'model': self.thread._name,
'res_id': self.thread.id,
- 'body': 'Nothing link',
+ 'body': f'Nothing link',
})
self.env['mail.link.preview']._create_link_previews(message)
- link_preview_count = self.env['mail.link.preview'].search_count([('source_url', '=', 'https://thisdomainedoentexist.nothing')])
+ link_preview_count = self.env['mail.link.preview'].search_count([('source_url', '=', self.source_url)])
self.assertEqual(link_preview_count, throttle + 1)
def test_02_link_preview_create(self):
- with patch.object(requests.Session, 'get', _patched_get_html), patch.object(requests.Session, 'head', _patch_head_html):
+ with patch.object(requests.Session, 'get', self._patch_with_og_properties), patch.object(requests.Session, 'head', self._patch_head_html):
message = self.env['mail.message'].create({
'model': self.thread._name,
'res_id': self.thread.id,
- 'body': 'Nothing link',
+ 'body': f'Nothing link',
})
self.env['mail.link.preview']._create_link_previews(message)
self.assertBusNotifications(
@@ -76,13 +110,48 @@ class TestLinkPreview(MailCommon):
'id': link_preview.id,
'message': {'id': message.id},
'image_mimetype': False,
- 'og_description': 'Test description',
- 'og_image': False,
+ 'og_description': self.og_description,
+ 'og_image': self.og_image,
'og_mimetype': False,
- 'og_title': 'Test title',
+ 'og_title': self.og_title,
'og_type': False,
- 'source_url': 'https://thisdomainedoentexist.nothing',
+ 'source_url': self.source_url,
}] for link_preview in message.link_preview_ids
}
}]
)
+
+ def test_get_link_preview_from_url(self):
+ test_cases = [
+ (self._patch_with_og_properties, self.source_url),
+ (self._patch_without_og_properties, self.source_url),
+ (self._patch_with_image_mimetype, self.og_image),
+ ]
+ test_asserts = [
+ {
+ 'og_description': self.og_description,
+ 'og_image': self.og_image,
+ 'og_mimetype': None,
+ 'og_title': self.og_title,
+ 'og_type': None,
+ 'source_url': self.source_url,
+ },
+ {
+ 'og_description': None,
+ 'og_image': None,
+ 'og_mimetype': None,
+ 'og_title': self.title,
+ 'og_type': None,
+ 'source_url': self.source_url,
+ },
+ {
+ 'image_mimetype': 'image/png',
+ 'og_image': self.og_image,
+ 'source_url': self.og_image,
+ },
+ ]
+ session = requests.Session()
+ for (get_patch, url), expected in zip(test_cases, test_asserts):
+ with self.subTest(get_patch=get_patch, url=url, expected=expected), patch.object(requests.Session, 'get', get_patch):
+ preview = link_preview.get_link_preview_from_url(url, session)
+ self.assertEqual(preview, expected)
diff --git a/addons/mail/tools/link_preview.py b/addons/mail/tools/link_preview.py
new file mode 100644
index 00000000000..7fe54479463
--- /dev/null
+++ b/addons/mail/tools/link_preview.py
@@ -0,0 +1,83 @@
+# -*- coding: utf-8 -*-
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+
+from lxml import html
+import requests
+
+
+def get_link_preview_from_url(url, request_session=None):
+ """
+ Get the Open Graph properties of an url. (https://ogp.me/)
+ If the url leads directly to an image mimetype, return
+ the url as preview image else retrieve the properties from
+ the html page.
+
+ Using a stream request to prevent loading the whole page
+ as those properties are declared in the tag.
+
+ The request session is optional as in some cases using
+ a session could be beneficial performance wise
+ (e.g. a lot of url could have the same domain).
+ """
+ # Some websites are blocking non browser user agent.
+ user_agent = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0'}
+ try:
+ if request_session:
+ response = request_session.get(url, timeout=3, headers=user_agent, allow_redirects=True, stream=True)
+ else:
+ response = requests.get(url, timeout=3, headers=user_agent, allow_redirects=True, stream=True)
+ except requests.exceptions.RequestException:
+ return False
+ if not response.ok or not response.headers.get('Content-Type'):
+ return False
+ # Content-Type header can return a charset, but we just need the
+ # mimetype (eg: image/jpeg;charset=ISO-8859-1)
+ content_type = response.headers['Content-Type'].split(';')
+ if response.headers['Content-Type'].startswith('image/'):
+ return {
+ 'image_mimetype': content_type[0],
+ 'og_image': url, # If the url mimetype is already an image type, set url as preview image
+ 'source_url': url,
+ }
+ elif response.headers['Content-Type'].startswith('text/html'):
+ return get_link_preview_from_html(url, response)
+ return False
+
+def get_link_preview_from_html(url, response):
+ """
+ Retrieve the Open Graph properties from the html page. (https://ogp.me/)
+ Load the page with chunks of 8kb to prevent loading the whole
+ html when we only need the tag content.
+ Fallback on the tag if the html doesn't have
+ any Open Graph title property.
+ """
+ content = b""
+ for chunk in response.iter_content(chunk_size=8192):
+ content += chunk
+ # Stop reading once all the <head> data is found
+ if b"</head>" in content:
+ break
+ content = content.decode()
+ if not content:
+ return False
+ tree = html.fromstring(content)
+ og_title = tree.xpath('//meta[@property="og:title"]/@content')
+ if og_title:
+ og_title = og_title[0]
+ elif tree.find('.//title') is not None:
+ # Fallback on the <title> tag if it exists
+ og_title = tree.find('.//title').text
+ else:
+ return False
+ og_description = tree.xpath('//meta[@property="og:description"]/@content')
+ og_type = tree.xpath('//meta[@property="og:type"]/@content')
+ og_image = tree.xpath('//meta[@property="og:image"]/@content')
+ og_mimetype = tree.xpath('//meta[@property="og:image:type"]/@content')
+ return {
+ 'og_description': og_description[0] if og_description else None,
+ 'og_image': og_image[0] if og_image else None,
+ 'og_mimetype': og_mimetype[0] if og_mimetype else None,
+ 'og_title': og_title,
+ 'og_type': og_type[0] if og_type else None,
+ 'source_url': url,
+ }