142 lines
4.7 KiB
Python
142 lines
4.7 KiB
Python
""" Report """
|
|
|
|
import atexit
|
|
import logging
|
|
|
|
from mylib.config import ConfigurableObject, StringOption
|
|
from mylib.email import EmailClient
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class Report(ConfigurableObject): # pylint: disable=useless-object-inheritance
|
|
"""Logging report"""
|
|
|
|
_config_name = "report"
|
|
_config_comment = "Email report"
|
|
_defaults = {
|
|
"recipient": None,
|
|
"subject": "Report",
|
|
"loglevel": "WARNING",
|
|
"logformat": "%(asctime)s - %(levelname)s - %(message)s",
|
|
"just_try": False,
|
|
}
|
|
|
|
content = []
|
|
handler = None
|
|
formatter = None
|
|
email_client = None
|
|
|
|
def __init__(self, email_client=None, initialize=True, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self.email_client = email_client
|
|
self._attachment_files = []
|
|
self._attachment_payloads = []
|
|
|
|
if initialize:
|
|
self.initialize()
|
|
|
|
def configure(self, **kwargs): # pylint: disable=arguments-differ
|
|
"""Configure options on registered mylib.Config object"""
|
|
section = super().configure(
|
|
just_try_help=kwargs.pop("just_try_help", "Just-try mode: do not really send report"),
|
|
loaded_callback=self.initialize,
|
|
**kwargs,
|
|
)
|
|
|
|
section.add_option(StringOption, "recipient", comment="Report recipient email address")
|
|
section.add_option(
|
|
StringOption,
|
|
"subject",
|
|
default=self._defaults["subject"],
|
|
comment="Report email subject",
|
|
)
|
|
section.add_option(
|
|
StringOption,
|
|
"loglevel",
|
|
default=self._defaults["loglevel"],
|
|
comment='Report log level (as accept by python logging, for instance "INFO")',
|
|
)
|
|
section.add_option(
|
|
StringOption,
|
|
"logformat",
|
|
default=self._defaults["logformat"],
|
|
comment='Report log level (as accept by python logging, for instance "INFO")',
|
|
)
|
|
|
|
if not self.email_client:
|
|
self.email_client = EmailClient(config=self._config)
|
|
self.email_client.configure()
|
|
|
|
return section
|
|
|
|
def initialize(self, loaded_config=None):
|
|
"""Configuration initialized hook"""
|
|
super().initialize(loaded_config=loaded_config)
|
|
self.handler = logging.StreamHandler(self)
|
|
|
|
loglevel = self._get_option("loglevel").upper()
|
|
assert hasattr(logging, loglevel), f"Invalid report loglevel {loglevel}"
|
|
self.handler.setLevel(getattr(logging, loglevel))
|
|
|
|
self.formatter = logging.Formatter(self._get_option("logformat"))
|
|
self.handler.setFormatter(self.formatter)
|
|
|
|
def get_handler(self):
|
|
"""Retreive logging handler"""
|
|
return self.handler
|
|
|
|
def write(self, msg):
|
|
"""Write a message"""
|
|
self.content.append(msg)
|
|
|
|
def get_content(self):
|
|
"""Read the report content"""
|
|
return "".join(self.content)
|
|
|
|
def add_attachment_file(self, filepath):
|
|
"""Add attachment file"""
|
|
self._attachment_files.append(filepath)
|
|
|
|
def add_attachment_payload(self, payload):
|
|
"""Add attachment payload"""
|
|
self._attachment_payloads.append(payload)
|
|
|
|
def send(self, subject=None, rcpt_to=None, email_client=None, just_try=None):
|
|
"""Send report using an EmailClient"""
|
|
if rcpt_to is None:
|
|
rcpt_to = self._get_option("recipient")
|
|
if not rcpt_to:
|
|
log.debug("No report recipient, do not send report")
|
|
return True
|
|
if subject is None:
|
|
subject = self._get_option("subject")
|
|
assert subject, "You must provide report subject using Report.__init__ or Report.send"
|
|
if email_client is None:
|
|
email_client = self.email_client
|
|
assert email_client, (
|
|
"You must provide an email client __init__(), send() or send_at_exit() methods argument"
|
|
" email_client"
|
|
)
|
|
content = self.get_content()
|
|
if not content:
|
|
log.debug("Report is empty, do not send it")
|
|
return True
|
|
msg = email_client.forge_message(
|
|
rcpt_to,
|
|
subject=subject,
|
|
text_body=content,
|
|
attachment_files=self._attachment_files,
|
|
attachment_payloads=self._attachment_payloads,
|
|
)
|
|
if email_client.send(
|
|
rcpt_to, msg=msg, just_try=just_try if just_try is not None else self._just_try
|
|
):
|
|
log.debug("Report sent to %s", rcpt_to)
|
|
return True
|
|
log.error("Fail to send report to %s", rcpt_to)
|
|
return False
|
|
|
|
def send_at_exit(self, **kwargs):
|
|
"""Send report at exit"""
|
|
atexit.register(self.send, **kwargs)
|