Создать и отправить отчет в формате Excel данные, полученные из SQL


Это мой первый питон (3.6) скрипт, который я ценю критику.

Я генерировать ежедневный отчет о сигнала тревоги. Сбора данных от SQL-сервера баз данных, представленных в файле XLSX с помощью XlsxWriter и по электронной почте. Параметры конфигурации извлекаются из пару файлов JSON, так что люди, которые не знают Python может обновить его как оборудование и идет.

send_email.py

import smtplib
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText


def send_email(from_addr, to_addrs, subject, body, attachments, server, port=25):

    msg = MIMEMultipart()
    msg['From'] = from_addr
    if to_addrs is list:
        msg['To'] = ', '.join(to_addrs)
    else:
        msg['To'] = to_addrs
    msg['Subject'] = subject

    msg.attach(MIMEText(body))

    for a in attachments or []:
        with open(a, 'rb') as f:
            attachment = MIMEApplication(f.read(), Name=a.name)
            attachment['Content-Disposition'] = f'attachment; filename="{a.name}"'
            msg.attach(attachment)

    with smtplib.SMTP(server, port) as smtp:
        smtp.sendmail(from_addr=from_addr, to_addrs=to_addrs, msg=msg.as_string())

alarm_report.py

# _*_ coding: utf-8

import json
import os
import pyodbc
import sys
from datetime import datetime, time, date, timedelta
from pathlib import Path

import xlsxwriter

from send_email import send_email



DEFAULT_DAY_START_HOUR = 5
DEFAULT_ALM_CATEGORIES = {'All': '[0-5]', 'Critical': '[12]', 'Major': '3', 'Minor': '4', 'Event': '5'}
DEFAULT_ALM_TYPES = ('DIG_ALM', 'STATUS', 'EVENT')
DEFAULT_REPORT_TITLE = 'Daily Alarm Report'
DEFAULT_SAVE_PATH = r"//spam/Reports/Alarm Report/"


def reporting_dates(day_offset=1, start_of_reporting_day=DEFAULT_DAY_START_HOUR):
    report_date = date.today() - timedelta(day_offset)
    report_start = datetime.combine(report_date, time(hour=start_of_reporting_day, minute=0, second=0, microsecond=0))
    report_end = report_start + timedelta(days=1)
    return report_date, report_start, report_end


class Equipment:

    def __init__(self, name, equip_numbers=None):
        self.name = name
        self.alm_cat_totals = {}
        if equip_numbers is None:
            self.equip_numbers = ['%']
        else:
            self.equip_numbers = equip_numbers
        self.alm_counts()

    @property
    def alm_total(self):
        """
        Returns the total number of alarms for the equipment.
        :return: total number of alarms
        :rtype: int
        """
        return self.alm_cat_totals['All']

    def alm_count(self, category=None):
        if category is None:
            category = 'All'
        equipment_clause = " OR ".join(["[Equipment] LIKE '" + x + "'" for x in self.equip_numbers])
        alm_type_clause = " OR ".join(["[Type] LIKE '" + a + "'" for a in almmgr_alm_types])
        select = f"""\
        SELECT COUNT(ID) AS [Total] 
        FROM (
            SELECT [ID], [On_Time] AS [On Time], [Equipment] AS [Equipment] 
            FROM [Database].[dbo].[AlarmTable] 
            WHERE ({alm_type_clause}) AND ({equipment_clause}) AND [Category] LIKE '{alm_categories.get(category)}'
            ) q1
        WHERE [On Time] BETWEEN '{report_start_datetime}' AND '{report_end_datetime}'"""

        self.alm_cat_totals[category] = cursor.execute(select).fetchone()[0]
        return self.alm_cat_totals[category]

    def alm_counts(self):
        for category, category_number in alm_categories.items():
            self.alm_count(category=category)
        return self.alm_cat_totals

    def top_alarms(self):
        equipment_clause = " OR ".join(["L.[Equipment] LIKE '" + x + "'" for x in self.equip_numbers])
        alm_type_clause = " OR ".join(["L.[Type] LIKE '" + a + "'" for a in almmgr_alm_types])
        select = f"""\
        SELECT  TOP ({top_alm_count})
        COUNT([ID]) AS [Occurrences],
        IIF(SUM([Seconds On])/86400 > 0,
            CONVERT(VARCHAR(4), SUM([Seconds On])/86400)+ 'd ' +
                CONVERT(VARCHAR(8), CONVERT(TIME(0), DATEADD(SECOND, SUM([Seconds On])%86400, 0))
            ), 
            CONVERT(VARCHAR(8), CONVERT(TIME(0), DATEADD(SECOND, SUM([Seconds On])%86400, 0))
            )
        ) AS [Duration], 
        [Equipment], 
        [Description], 
        [Tag], 
        (SELECT CASE [Category] 
            WHEN '1' THEN 'Audible' 
            WHEN '2' THEN 'Critical' 
            WHEN '3' THEN 'Major' 
            WHEN '4' THEN 'Minor' 
            WHEN '5' THEN 'Event' 
            ELSE '' 
        END ) AS [Category] 
        FROM (
              SELECT 
                L.[ID], 
                L.[On_Time], 
                L.[Off_Time], 
                IIF(LEFT(L.[Off_Time], 10) = '9999-12-31',
                    '',
                    DATEDIFF(SECOND,
                            L.[On_Time],
                            IIF(L.[Off_Time] > '{report_end_datetime}', '{report_end_datetime}', L.[Off_Time])
                    )
                ) AS [Seconds On], 
                L.[Equipment] AS [Equipment], 
                T.[TagDescription] AS [Description], 
                L.[Category] AS [Category], 
                T.[TagName] AS [Tag] 
              FROM  [Database].[dbo].[AlarmTable] L 
              JOIN  [Database].[dbo].[TagTable] T ON L.Tag = T.Tag 
              WHERE L.On_Time BETWEEN '{report_start_datetime}' AND '{report_end_datetime}' 
                    AND ({alm_type_clause}) 
                    AND ({equipment_clause}) 
               )q1 
        GROUP BY [Equipment], [Tag], [Description], [Category] 
        ORDER BY [Occurrences] DESC, [Duration] DESC"""

        return cursor.execute(select).fetchall()

    def alm_log(self):
        equipment_clause = " OR ".join(["L.[Equipment] LIKE '" + x + "'" for x in self.equip_numbers])
        alm_type_clause = " OR ".join(["L.[Type] LIKE '" + a + "'" for a in almmgr_alm_types])
        select = f"""\
        SELECT
            CONVERT(VARCHAR(8), [On_Time], 108) AS [On Time], 
            IIF(LEFT([Off_Time], 10) = '9999-12-31', '', CONVERT(VARCHAR(8), [Off_Time], 108)) AS [Off Time], 
            [Equipment], 
            [Description], 
            [Tag], 
            (SELECT CASE [Category] 
            WHEN '1' THEN 'Audible' 
            WHEN '2' THEN 'Critical' 
            WHEN '3' THEN 'Major' 
            WHEN '4' THEN 'Minor' 
            WHEN '5' THEN 'Event' 
            ELSE '' END) 
            AS [Category], 
            IIF([Seconds On] / 86400 > 0, 
            CONVERT(VARCHAR(4), [Seconds On] / 86400) + 'd ' + 
            CONVERT(VARCHAR(8), CONVERT(TIME(0), DATEADD(SECOND, [Seconds On] % 86400, 0))), 
            CONVERT(VARCHAR(8), CONVERT(TIME(0), DATEADD(SECOND, [Seconds On] % 86400, 0))) ) AS [Duration] 
        FROM ( 
            SELECT
                L.[On_Time], 
                L.[Off_Time], 
                IIF(LEFT(L.[Off_Time], 10) = '9999-12-31', '', DATEDIFF(SECOND, L.[On_Time], L.[Off_Time]) )
                AS [Seconds On], 
                L.[Equipment]        AS [Equipment], 
                T.[TagDescription] AS [Description], 
                T.[TagName]        AS [Tag], 
                L.[Category]        AS [Category] 
            FROM [Database].[dbo].[AlarmTable] L 
            JOIN [Datanase].[dbo].[TagTable] T ON L.Tag = T.Tag 
            WHERE L.On_Time BETWEEN '{report_start_datetime}' AND '{report_end_datetime}' 
                AND ({alm_type_clause}) 
                AND ({equipment_clause}) 
        ) q1 
        ORDER BY [On_Time] ASC"""

        return cursor.execute(select).fetchall()

    def create_worksheet(self, workbook):
        top_alarms_first_row = 12
        top_alarms_last_row = top_alarms_first_row + top_alm_count

        worksheet = workbook.add_worksheet(self.name)
        # format the worksheet
        worksheet.set_column(0, 0, 5, general_format)  # left padding
        worksheet.set_column(1, 3, 10)  # on time, off time, equipment
        worksheet.set_column(4, 4, 60)  # description
        worksheet.set_column(5, 5, 40)  # tag
        worksheet.set_column(6, 6, 10)  # priority
        worksheet.set_column(7, 7, 10)  # duration
        for col in range(8, 50):
            worksheet.set_column(col, col, None, general_format)
        for row in range(2, 12):
            worksheet.set_row(row, None, general_format)
        for row in range(top_alarms_last_row + 1, top_alarms_last_row + 3):
            worksheet.set_row(row, None, general_format)

        # home button
        worksheet.write_url('A1',
                            url='internal:Summary!A1',
                            string="\u2302",
                            cell_format=title_format
                            )

        worksheet.merge_range('B1:H1', report_title, title_format)
        worksheet.merge_range('B4:H4', self.name, equipment_format)
        worksheet.merge_range('B2:H2', str(report_date), title_format)
        worksheet.write('B5', 'Alarm Count', heading_format)

        if self.alm_total:
            # alarm category counts
            row, col = 5, 1
            for category, count in self.alm_cat_totals.items():
                if category != 'All':
                    worksheet.write(row, col, category, text_format)
                    worksheet.write(row, col + 1, count, text_format)
                    row += 1
            worksheet.write(row, col + 1, self.alm_total, total_format)

            # top 5 alarms
            for row in range(top_alarms_first_row, top_alarms_last_row + 1):
                worksheet.write_blank(row, 7, None, general_format)
            worksheet.write('B12', 'Top ' + str(top_alm_count) + ' Alarms', heading_format)
            worksheet.add_table(first_row=top_alarms_first_row,
                                first_col=1,
                                last_row=top_alarms_last_row,
                                last_col=6,
                                options={'data': self.top_alarms(),
                                         'autofilter': False,
                                         'columns': [{'header': 'Count'},
                                                     {'header': 'Duration'},
                                                     {'header': 'Equipment'},
                                                     {'header': 'Description'},
                                                     {'header': 'Tag'},
                                                     {'header': 'Category'}, ],
                                         'name': self.name.replace(' ', '_').replace('.', '') + 'TopAlarms',
                                         'banded_rows': False,
                                         'style': 'Table Style Light 9',
                                         })

            # alarm log
            alm_log_first_row = top_alarms_first_row + top_alm_count + 3
            worksheet.write(alm_log_first_row - 1, 1, 'Alarm Log', heading_format)
            alm_log_last_row = self.alm_total + alm_log_first_row
            worksheet.add_table(first_row=alm_log_first_row,
                                first_col=1,
                                last_row=alm_log_last_row,
                                last_col=7,
                                options={'data': self.alm_log(),
                                         'columns': [{'header': 'On Time'},
                                                     {'header': 'Off Time'},
                                                     {'header': 'Equipment'},
                                                     {'header': 'Description'},
                                                     {'header': 'Tag'},
                                                     {'header': 'Category'},
                                                     {'header': 'Duration'}, ],
                                         'name': self.name.replace(' ', '_').replace('.', '') + 'AlarmLog',
                                         'banded_rows': False,
                                         'style': 'Table Style Light 9',
                                         })
            for row in range(alm_log_last_row + 1, alm_log_last_row + 200):
                worksheet.set_row(row, None, general_format)
        else:
            worksheet.write('C5', 0, text_format)
            for row in range(top_alarms_first_row, 200):
                worksheet.set_row(row, None, general_format)


if __name__ == '__main__':

    # use yesterday's date for the report
    report_date, report_start_datetime, report_end_datetime = reporting_dates(day_offset=1)

    # get configuration paramters from config files
    path = Path(os.path.dirname(sys.argv[0]))
    with open(path / 'equipment.json', 'r') as infile:
        equipment_list = json.load(infile)
    with open(path / 'config.json', 'r') as infile:
        config = json.load(infile)
    almmgr_alm_types = config.get('AlarmManager', {}).get('AlarmTypes', DEFAULT_ALM_TYPES)
    top_alm_count = config.get('Report', {}).get('TopAlarmCount', 10)
    report_title = config.get('Report', {}).get('Title', DEFAULT_REPORT_TITLE)
    root_path = config.get('Report', {}).get('SavePath', DEFAULT_SAVE_PATH)
    reporting_day_start_hour = config.get('Report', {}).get('StartOfDayHour', DEFAULT_DAY_START_HOUR)
    alm_categories = config.get('AlarmCategories', DEFAULT_ALM_CATEGORIES)
    alm_name_field_length = config.get('AlarmManager', {}).get('AlarmNameFieldLength', 5)

    # excel report
    save_path = Path(fr"{root_path}/archive/{report_date.strftime('%Y')}/{report_date.strftime('%m')}/")
    if not save_path.exists():
        save_path.mkdir(parents=True)
    file_path = save_path / f'{report_date}.xlsx'
    workbook = xlsxwriter.Workbook(file_path)

    # cell formats
    general_format = workbook.add_format({
        'bg_color': 'white',
    })
    title_format = workbook.add_format({
        'font_size': 10,
        'bg_color': 'white',
        'align': 'center',
        'valign': 'vcenter',
    })
    equipment_format = workbook.add_format({
        'font_size': 20,
        'bg_color': 'white',
        'align': 'center',
        'valign': 'vcenter',
    })
    box_heading_format = workbook.add_format({
        'bold': True,
        'font_size': 10,
        'font_color': 'white',
        'bg_color': '#4f81bd',
    })
    text_format = workbook.add_format({
        'font_size': 10,
        'bg_color': 'white',
    })
    link_format = workbook.add_format({
        'font_color': 'blue',
        'underline': True,
        'bg_color': 'white',
    })
    heading_format = workbook.add_format({
        'bold': True,
        'font_size': 10,
        'bg_color': 'white',
    })
    total_format = workbook.add_format({
        'font_color': 'blue',
        'underline': True,
        'align': 'right',
        'bold': True,
        'font_size': 10,
        'bg_color': 'white',
        'top': 1
    })

    # SQL ODBC connection
    sql_driver = config.get('SQL', {}).get('Driver', '{ODBC Driver 13 for SQL Server}')
    sql_server = config.get('SQL', {}).get('Server', r'SERVER\SOMEPLACE')
    sql_uid = config.get('SQL', {}).get('UID', 'monty')
    sql_pwd = config.get('SQL', {}).get('PWD', 'python')
    sql_database = config.get('SQL', {}).get('Database', 'SpamDatabase')
    sql_connection_string = f'Driver={sql_driver};' \
                            f'Server={sql_server};' \
                            f'Database={sql_database};' \
                            f'uid={sql_uid};' \
                            f'pwd={sql_pwd}'
    connection = pyodbc.connect(sql_connection_string)
    cursor = connection.cursor()

    # create worksheets for each bit of equipment
    equipment = {'All': Equipment(name='All')}
    equipment['All'].create_worksheet(workbook=workbook)
    for key, values in equipment_list.items():
        # Alarm Manager seems to trim the alarm name field of length to 5
        # so we trim anything with an alarm name > 5 characters.
        # TODO: update this after fixing the Alarm Manager rules
        values = [val[:alm_name_field_length + 1] for val in values]
        equipment[key] = Equipment(name=key, equip_numbers=values)
        equipment[key].create_worksheet(workbook=workbook)

    # =============================
    # excel report statistics page
    # =============================
    worksheet = workbook.add_worksheet('Statistics')
    for column in range(50):
        worksheet.set_column(column, column, 6.5, general_format)
    worksheet.merge_range('B2:U2', 'Spam Industries', title_format)
    worksheet.merge_range('B3:U3', 'Alarm History', equipment_format)
    worksheet.merge_range('B4:U4', str(report_date), title_format)

    # get the daily alarm counts for the last four weeks
    days_headings = []
    days_alm_total = []
    for days in range(28, 0, -1):
        rep_date, report_start_datetime, report_end_datetime = reporting_dates(day_offset=days)
        e = Equipment(name='All')
        days_headings.append(rep_date.strftime('%d/%m'))
        days_alm_total.append(e.alm_count())
    worksheet.write_row('B21', days_headings, heading_format)
    worksheet.write_row('B22', days_alm_total, text_format)
    worksheet.set_row(20, None, None, {'hidden': True})
    worksheet.set_row(21, None, None, {'hidden': True})

    # chart the alarm count for the last four weeks
    chart_days = workbook.add_chart({'type': 'column'})
    chart_days.add_series({
        'name': 'Date',
        'categories': '=Statistics!$B$21:$AC$21',
        'values': '=Statistics!$B$22:$AC$22',
        'data_labels': {
            'value': True,
            'font': {'size': 8}
        }
    })
    chart_days.set_style(2)
    chart_days.set_title({
        'name': 'Daily Alarm Totals',
        'name_font': {
            'name': 'Calibri',
            'bold': False,
            'size': 14
        }
    })
    chart_days.set_x_axis({'major_tick_mark': 'none'})
    chart_days.set_y_axis({
        'visible': False,
        'major_gridlines': {'visible': False}
    })
    chart_days.set_legend({'none': True})
    chart_days.set_size({'width': 1000, 'height': 250})
    chart_days.set_chartarea({'border': {'none': True}})
    chart_days.show_hidden_data()
    worksheet.insert_chart('B7', chart_days)

    # ===============================
    # excel report summary/home page
    # ===============================
    first_row = 5
    worksheet = workbook.add_worksheet('Summary')
    worksheet.set_column(0, 0, 10, general_format)  # left padding
    worksheet.set_column(1, 1, 18, general_format)  # equipment names
    worksheet.set_column(2, 2, 10, general_format)  # alarm counts
    worksheet.set_column(3, 3, 50, general_format)  # data bars
    for column in range(4, 50):
        worksheet.set_column(column, column, None, general_format)
    worksheet.merge_range('A2:C2', 'Spam Industries', title_format)
    worksheet.merge_range('A3:C3', 'Daily Alarm Report', equipment_format)
    worksheet.merge_range('A4:C4', str(report_date), title_format)

    row, col = first_row, 1
    for equip_name, equip in equipment.items():
        if equip_name != 'All':
            worksheet.write_url(row=row,
                                col=col,
                                url="internal:'" + equip_name + "'!A1",
                                string=equip_name,
                                cell_format=link_format
                                )
            worksheet.write(row, col + 1, equip.alm_total)
            row += 1
    worksheet.write_url(row=row,
                        col=col + 1,
                        url="internal:'All'!A1",
                        string=str(equipment['All'].alm_total),
                        cell_format=total_format
                        )
    worksheet.set_first_sheet()
    worksheet.activate()

    # close the SQL connection and excel workbook
    connection.close()
    workbook.close()

    # email
    send_email(from_addr='Reports <no-reply@example.com>',
               to_addrs='everybody@example.com',
               subject='Spam Industries Daily Alarm Report',
               body=f"{equipment['All'].alm_total} alarms yesterday.",
               attachments=[file_path])

оборудование.в JSON выглядит {"SpamArea": ["Squasher","Canner"]}

Я все правильно делаю?



193
2
задан 31 марта 2018 в 07:03 Источник Поделиться
Комментарии