# coding: utf8 """ Report """ import atexit import logging from mylib.config import ConfigurableObject from mylib.config import 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', } 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(**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=False): """ 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): 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)