mirror of
https://github.com/apache/superset.git
synced 2026-04-22 17:45:21 +00:00
afb8bd5fe6
* feat: improve HTML table in text reports * Remove unused import * Update tests * Fix test
155 lines
5.0 KiB
Python
155 lines
5.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Licensed to the Apache Software Foundation (ASF) under one
|
|
# or more contributor license agreements. See the NOTICE file
|
|
# distributed with this work for additional information
|
|
# regarding copyright ownership. The ASF licenses this file
|
|
# to you under the Apache License, Version 2.0 (the
|
|
# "License"); you may not use this file except in compliance
|
|
# with the License. You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing,
|
|
# software distributed under the License is distributed on an
|
|
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
# KIND, either express or implied. See the License for the
|
|
# specific language governing permissions and limitations
|
|
# under the License.
|
|
import json
|
|
import logging
|
|
import textwrap
|
|
from dataclasses import dataclass
|
|
from email.utils import make_msgid, parseaddr
|
|
from typing import Any, Dict, Optional
|
|
|
|
import bleach
|
|
from flask_babel import gettext as __
|
|
|
|
from superset import app
|
|
from superset.models.reports import ReportRecipientType
|
|
from superset.reports.notifications.base import BaseNotification
|
|
from superset.reports.notifications.exceptions import NotificationError
|
|
from superset.utils.core import send_email_smtp
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TABLE_TAGS = ["table", "th", "tr", "td", "thead", "tbody", "tfoot"]
|
|
TABLE_ATTRIBUTES = ["colspan", "rowspan", "halign", "border", "class"]
|
|
|
|
|
|
@dataclass
|
|
class EmailContent:
|
|
body: str
|
|
data: Optional[Dict[str, Any]] = None
|
|
images: Optional[Dict[str, bytes]] = None
|
|
|
|
|
|
class EmailNotification(BaseNotification): # pylint: disable=too-few-public-methods
|
|
"""
|
|
Sends an email notification for a report recipient
|
|
"""
|
|
|
|
type = ReportRecipientType.EMAIL
|
|
|
|
@staticmethod
|
|
def _get_smtp_domain() -> str:
|
|
return parseaddr(app.config["SMTP_MAIL_FROM"])[1].split("@")[1]
|
|
|
|
@staticmethod
|
|
def _error_template(text: str) -> str:
|
|
return __(
|
|
"""
|
|
Error: %(text)s
|
|
""",
|
|
text=text,
|
|
)
|
|
|
|
def _get_content(self) -> EmailContent:
|
|
if self._content.text:
|
|
return EmailContent(body=self._error_template(self._content.text))
|
|
# Get the domain from the 'From' address ..
|
|
# and make a message id without the < > in the end
|
|
image = None
|
|
csv_data = None
|
|
domain = self._get_smtp_domain()
|
|
msgid = make_msgid(domain)[1:-1]
|
|
|
|
# Strip any malicious HTML from the description
|
|
description = bleach.clean(self._content.description or "")
|
|
|
|
# Strip malicious HTML from embedded data, allowing only table elements
|
|
if self._content.embedded_data is not None:
|
|
df = self._content.embedded_data
|
|
html_table = bleach.clean(
|
|
df.to_html(na_rep="", index=True),
|
|
tags=TABLE_TAGS,
|
|
attributes=TABLE_ATTRIBUTES,
|
|
)
|
|
else:
|
|
html_table = ""
|
|
|
|
call_to_action = __("Explore in Superset")
|
|
img_tag = (
|
|
f'<img width="1000px" src="cid:{msgid}">'
|
|
if self._content.screenshot
|
|
else ""
|
|
)
|
|
body = textwrap.dedent(
|
|
f"""
|
|
<html>
|
|
<head>
|
|
<style type="text/css">
|
|
table, th, td {{
|
|
border-collapse: collapse;
|
|
border-color: rgb(200, 212, 227);
|
|
color: rgb(42, 63, 95);
|
|
padding: 4px 8px;
|
|
}}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<p>{description}</p>
|
|
<b><a href="{self._content.url}">{call_to_action}</a></b><p></p>
|
|
{html_table}
|
|
{img_tag}
|
|
</body>
|
|
</html>
|
|
"""
|
|
)
|
|
if self._content.screenshot:
|
|
image = {msgid: self._content.screenshot}
|
|
if self._content.csv:
|
|
csv_data = {__("%(name)s.csv", name=self._content.name): self._content.csv}
|
|
return EmailContent(body=body, images=image, data=csv_data)
|
|
|
|
def _get_subject(self) -> str:
|
|
return __(
|
|
"%(prefix)s %(title)s",
|
|
prefix=app.config["EMAIL_REPORTS_SUBJECT_PREFIX"],
|
|
title=self._content.name,
|
|
)
|
|
|
|
def _get_to(self) -> str:
|
|
return json.loads(self._recipient.recipient_config_json)["target"]
|
|
|
|
def send(self) -> None:
|
|
subject = self._get_subject()
|
|
content = self._get_content()
|
|
to = self._get_to()
|
|
try:
|
|
send_email_smtp(
|
|
to,
|
|
subject,
|
|
content.body,
|
|
app.config,
|
|
files=[],
|
|
data=content.data,
|
|
images=content.images,
|
|
bcc="",
|
|
mime_subtype="related",
|
|
dryrun=False,
|
|
)
|
|
logger.info("Report sent to email")
|
|
except Exception as ex:
|
|
raise NotificationError(ex) from ex
|