mirror of
https://github.com/GSA/notifications-api.git
synced 2026-01-27 04:52:07 -05:00
2463cd8fb5
it now uses the ft_notification_status table - nice and quick. if you ask for the current tax year it'll top up the data with numbers from the notification table. the data is in exactly the same shape as the old endpoint - no changes to the admin app are necessary
109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
from collections import defaultdict
|
|
from datetime import datetime
|
|
|
|
from app.models import NOTIFICATION_STATUS_TYPES, TEMPLATE_TYPES
|
|
from app.utils import convert_utc_to_bst
|
|
from app.dao.date_util import get_months_for_financial_year
|
|
|
|
|
|
def format_statistics(statistics):
|
|
# statistics come in a named tuple with uniqueness from 'notification_type', 'status' - however missing
|
|
# statuses/notification types won't be represented and the status types need to be simplified/summed up
|
|
# so we can return emails/sms * created, sent, and failed
|
|
counts = create_zeroed_stats_dicts()
|
|
for row in statistics:
|
|
_update_statuses_from_row(counts[row.notification_type], row)
|
|
|
|
return counts
|
|
|
|
|
|
def format_admin_stats(statistics):
|
|
counts = create_stats_dict()
|
|
|
|
for row in statistics:
|
|
if row.key_type == 'test':
|
|
counts[row.notification_type]['test-key'] += row.count
|
|
else:
|
|
counts[row.notification_type]['total'] += row.count
|
|
if row.status in ('technical-failure', 'permanent-failure', 'temporary-failure', 'virus-scan-failed'):
|
|
counts[row.notification_type]['failures'][row.status] += row.count
|
|
|
|
return counts
|
|
|
|
|
|
def create_stats_dict():
|
|
stats_dict = {}
|
|
for template in TEMPLATE_TYPES:
|
|
stats_dict[template] = {}
|
|
|
|
for status in ('total', 'test-key'):
|
|
stats_dict[template][status] = 0
|
|
|
|
stats_dict[template]['failures'] = {
|
|
'technical-failure': 0,
|
|
'permanent-failure': 0,
|
|
'temporary-failure': 0,
|
|
'virus-scan-failed': 0,
|
|
}
|
|
return stats_dict
|
|
|
|
|
|
def format_monthly_template_notification_stats(year, rows):
|
|
stats = {
|
|
datetime.strftime(date, '%Y-%m'): {}
|
|
for date in [
|
|
datetime(year, month, 1) for month in range(4, 13)
|
|
] + [
|
|
datetime(year + 1, month, 1) for month in range(1, 4)
|
|
]
|
|
}
|
|
|
|
for row in rows:
|
|
formatted_month = row.month.strftime('%Y-%m')
|
|
if str(row.template_id) not in stats[formatted_month]:
|
|
stats[formatted_month][str(row.template_id)] = {
|
|
"name": row.name,
|
|
"type": row.template_type,
|
|
"counts": dict.fromkeys(NOTIFICATION_STATUS_TYPES, 0)
|
|
}
|
|
stats[formatted_month][str(row.template_id)]["counts"][row.status] += row.count
|
|
|
|
return stats
|
|
|
|
|
|
def create_zeroed_stats_dicts():
|
|
return {
|
|
template_type: {
|
|
status: 0 for status in ('requested', 'delivered', 'failed')
|
|
} for template_type in TEMPLATE_TYPES
|
|
}
|
|
|
|
|
|
def _update_statuses_from_row(update_dict, row):
|
|
update_dict['requested'] += row.count
|
|
if row.status in ('delivered', 'sent'):
|
|
update_dict['delivered'] += row.count
|
|
elif row.status in ('failed', 'technical-failure', 'temporary-failure', 'permanent-failure'):
|
|
update_dict['failed'] += row.count
|
|
|
|
|
|
def create_empty_monthly_notification_status_stats_dict(year):
|
|
utc_month_starts = get_months_for_financial_year(year)
|
|
# nested dicts - data[month][template type][status] = count
|
|
return {
|
|
convert_utc_to_bst(start).strftime('%Y-%m'): {
|
|
template_type: defaultdict(int)
|
|
for template_type in TEMPLATE_TYPES
|
|
}
|
|
for start in utc_month_starts
|
|
}
|
|
|
|
|
|
def add_monthly_notification_status_stats(data, stats):
|
|
for row in stats:
|
|
month = row.month.strftime('%Y-%m')
|
|
|
|
data[month][row.notification_type][row.notification_status] = row.count
|
|
|
|
return data
|