import csv import xlsxwriter import logging import os import tempfile from smtplib import SMTPException from django.conf import settings from django.core.mail import EmailMultiAlternatives import abc logging.basicConfig(format='[%(levelname)s] %(message)s', level=logging.INFO) logger = logging.getLogger(__name__) class SendExportBase: """Base class of export and sending data.""" def __init__(self): self.success = False self.email_from = settings.EMAIL_HOST_USER self.email_subject = f'Export panel: {self.get_file_name()}' self.email_body = 'Exported panel data' self.file_path = os.path.join( settings.STATIC_ROOT, 'email', tempfile.gettempdir(), self.get_file_name() ) self.type_mapper = { "csv": self.make_csv_file, "xls": self.make_xls_file } @abc.abstractmethod def get_headers(self): pass @abc.abstractmethod def get_emails_to(self): return [ 'kuzmenko.da@gmail.com', 'sinapsit@yandex.ru' ] @abc.abstractmethod def get_data(self): pass @abc.abstractmethod def get_file_name(self): return '' def make_csv_file(self): file_header = self.get_headers() if not self.success: return with open(self.file_path, 'w') as f: file_writer = csv.writer(f, quotechar='"', quoting=csv.QUOTE_MINIMAL) # Write headers to CSV file file_writer.writerow(file_header) for row in self.get_data(): file_writer.writerow(row) def make_xls_file(self): headings = self.get_headers() if not self.success: return with xlsxwriter.Workbook(self.file_path) as workbook: worksheet = workbook.add_worksheet() # Add a bold format to use to highlight cells. bold = workbook.add_format({'bold': True}) # Add the worksheet data that the charts will refer to. data = self.get_data() worksheet.write_row('A1', headings, bold) for n, row in enumerate(data): worksheet.write_row(f'A{n+2}', [str(i) for i in row]) workbook.close() def make_xml_file(self): pass def send_email(self): msg = EmailMultiAlternatives( subject=self.email_subject, body=self.email_body, from_email=self.email_from, to=self.get_emails_to() ) # Create an inline attachment if self.file_path and self.success: msg.attach_file(self.file_path) else: msg.body = 'An error occurred while executing the request.' try: msg.send() logger.debug(f"COMMUTATOR:Email successfully sent") except SMTPException as e: logger.error(f"COMMUTATOR:Email connector: {e}") @abc.abstractmethod def send(self): pass class SendExport(SendExportBase): def __init__(self, panel, user, file_type='csv', **kwargs): super().__init__() self.panel = panel self.user = user self.file_type = file_type self.get_file_method = self.type_mapper[file_type] def get_emails_to(self): return [self.user.email] + super().get_emails_to() def get_file_name(self): name = '_'.join(self.panel.name.split(' ')) return f'export_{name.lower()}.{self.file_type}' def get_data(self): return self.panel.get_data() def get_headers(self): try: header = self.panel.get_headers() self.success = True return header except Exception as err: logger.info(f'HEADER:{err}') def get_file(self): if os.path.exists(self.file_path) and os.path.isfile(self.file_path): with open(self.file_path, 'rb') as export_file: return export_file else: logger.info('COMMUTATOR:image file not found dir: {path}') def send(self): self.get_file_method() print(f'ok: {self.file_path}') self.send_email() class SendGuideExport(SendExportBase): """Send guid export.""" def __init__(self, data, guide, user, file_type='csv', **kwargs): self.type_mapper = { "csv": self.make_csv_file, "xml": self.make_xml_file, "doc": self.make_doc_file } self.guide = guide self.data = data self.user = user self.file_type = file_type self.get_file_method = self.type_mapper[file_type] super().__init__() def get_file_name(self): name = self.guide.slug return f'export_{name}.{self.file_type}' def make_doc_file(self): pass def get_headers(self): headers = list(self.data[0].keys()) headers.pop(headers.index('node_name')) self.success = True return headers def get_data(self): return self.data def send(self): self.get_file_method() print(f'ok: {self.file_path}') self.send_email() def make_csv_file(self): file_header = self.get_headers() if not self.success: return with open(self.file_path, 'w') as f: file_writer = csv.writer(f, quotechar='"', quoting=csv.QUOTE_MINIMAL) # Write headers to CSV file file_writer.writerow(file_header) city = None for row in self.get_data(): row_city = row.get('city_name') if row_city: city = row_city else: row['city_name'] = city if row.pop("node_name") == "EstablishmentNode": file_writer.writerow(row.values())