115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
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
|
|
|
|
logging.basicConfig(format='[%(levelname)s] %(message)s', level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SendExport:
|
|
|
|
def __init__(self, user, panel, file_type='csv'):
|
|
self.type_mapper = {
|
|
"csv": self.make_csv_file,
|
|
"xls": self.make_xls_file
|
|
}
|
|
self.file_type = file_type
|
|
self.user = user
|
|
self.panel = panel
|
|
self.email_from = settings.EMAIL_HOST_USER
|
|
self.email_subject = f'Export panel: {self.get_file_name()}'
|
|
self.email_body = 'Exported panel data'
|
|
self.get_file_method = self.type_mapper[file_type]
|
|
self.file_path = os.path.join(
|
|
settings.STATIC_ROOT,
|
|
'email', tempfile.gettempdir(),
|
|
self.get_file_name()
|
|
)
|
|
self.success = False
|
|
|
|
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 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 send(self):
|
|
self.get_file_method()
|
|
print(f'ok: {self.file_path}')
|
|
self.send_email()
|
|
|
|
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_email(self):
|
|
|
|
msg = EmailMultiAlternatives(
|
|
subject=self.email_subject,
|
|
body=self.email_body,
|
|
from_email=self.email_from,
|
|
to=[
|
|
self.user.email,
|
|
'kuzmenko.da@gmail.com',
|
|
'sinapsit@yandex.ru'
|
|
]
|
|
)
|
|
|
|
# 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}") |