Merge branch 'main' into 2966-create-organization-dashboard-pageroutetemplate-page

This commit is contained in:
Beverly Nguyen
2025-10-14 12:33:46 -07:00
101 changed files with 1939 additions and 6429 deletions

View File

@@ -1564,20 +1564,20 @@ class SearchUsersForm(StripWhitespaceForm):
search = UsaSearchField("Search by name or email address")
class SearchNotificationsForm(StripWhitespaceForm):
to = UsaSearchField()
labels = {
"email": "Search by email address",
"sms": "Search by phone number",
}
def __init__(self, message_type, *args, **kwargs):
super().__init__(*args, **kwargs)
self.to.label.text = self.labels.get(
message_type,
"Search by phone number or email address",
)
# class SearchNotificationsForm(StripWhitespaceForm):
# to = UsaSearchField()
#
# labels = {
# "email": "Search by email address",
# "sms": "Search by phone number",
# }
#
# def __init__(self, message_type, *args, **kwargs):
# super().__init__(*args, **kwargs)
# self.to.label.text = self.labels.get(
# message_type,
# "Search by phone number or email address",
# )
class SearchTemplatesForm(StripWhitespaceForm):

View File

@@ -194,6 +194,22 @@ def template_usage(service_id):
get_monthly_template_stats(month, stats)
for month in get_months_for_financial_year(year, time_format="%B")
]
months.reverse()
# Get the year from stats
# If the year is this year, remove months in the future
if len(stats) > 0:
stat_year = stats[0]["year"]
current_year = datetime.now().year
current_month_num = datetime.now().month
new_months = []
if stat_year == current_year:
for m in months:
month_num_full = datetime.strptime(m["name"], "%B").month
if month_num_full <= current_month_num:
new_months.append(m)
months = new_months
return render_template(
"views/dashboard/all-template-statistics.html",

View File

@@ -1,6 +1,4 @@
# -*- coding: utf-8 -*-
import os
from functools import partial
from flask import (
Response,
@@ -13,27 +11,15 @@ from flask import (
stream_with_context,
url_for,
)
from flask_login import current_user
from markupsafe import Markup
from app import (
current_service,
format_datetime_table,
notification_api_client,
service_api_client,
)
from app.enums import NotificationStatus, NotificationType, ServicePermission
from app import current_service, format_datetime_table
from app.enums import ServicePermission
from app.formatters import get_time_left, message_count_noun
from app.main import main
from app.main.forms import SearchNotificationsForm
from app.models.job import Job
from app.utils import parse_filter_args, set_status_filters
from app.utils.csv import generate_notifications_csv
from app.utils.pagination import (
generate_next_dict,
generate_previous_dict,
get_page_from_request,
)
from app.utils.user import user_has_permissions
from notifications_python_client.errors import HTTPError
from notifications_utils.template import EmailPreviewTemplate, SMSBodyPreviewTemplate
@@ -59,7 +45,6 @@ def view_job(service_id, job_id):
filter_args = parse_filter_args(request.args)
filter_args["status"] = set_status_filters(filter_args)
api_public_url = os.environ.get("API_PUBLIC_URL")
notifications = None
more_than_one_page = False
@@ -74,12 +59,16 @@ def view_job(service_id, job_id):
return render_template(
"views/jobs/job.html",
api_public_url=api_public_url,
job=job,
status=request.args.get("status", ""),
counts=_get_job_counts(job),
notifications=notifications,
more_than_one_page=more_than_one_page,
uploaded_file_name=job.original_file_name,
time_left=get_time_left(job.created_at),
service_data_retention_days=current_service.get_days_of_retention(
job.template_type, number_of_days="seven_day"
),
download_link=(
url_for(
".view_job_csv",
@@ -195,244 +184,6 @@ def view_job_notifications_table(service_id, job_id):
)
@main.route("/services/<uuid:service_id>/notifications", methods=["GET", "POST"])
@main.route(
"/services/<uuid:service_id>/notifications/<template_type:message_type>",
methods=["GET", "POST"],
)
@user_has_permissions()
def view_notifications(service_id, message_type=None):
return render_template(
"views/notifications.html",
partials=get_notifications(service_id, message_type),
message_type=message_type,
status=request.args.get("status") or "sending,delivered,failed",
page=request.args.get("page", 1),
search_form=SearchNotificationsForm(
message_type=message_type,
to=request.form.get("to"),
),
things_you_can_search_by={
"email": ["email address"],
"sms": ["phone number"],
None: ["recipient"],
}.get(message_type)
+ {
True: ["reference"],
False: [],
}.get(bool(current_service.api_keys)),
download_link_one_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="one_day",
),
download_link_today=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="today",
),
download_link_three_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="three_day",
),
download_link_five_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="five_day",
),
download_link_seven_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="seven_day",
),
)
@main.route("/services/<uuid:service_id>/notifications.json", methods=["GET", "POST"])
@main.route(
"/services/<uuid:service_id>/notifications/<template_type:message_type>.json",
methods=["GET", "POST"],
)
@user_has_permissions()
def get_notifications_as_json(service_id, message_type=None):
return jsonify(
get_notifications(
service_id, message_type, status_override=request.args.get("status")
)
)
@main.route(
"/services/<uuid:service_id>/notifications.csv", endpoint="view_notifications_csv"
)
@main.route(
"/services/<uuid:service_id>/notifications/<template_type:message_type>.csv",
endpoint="view_notifications_csv",
)
@user_has_permissions()
def get_notifications(service_id, message_type, status_override=None): # noqa
# TODO get the api to return count of pages as well.
page = get_page_from_request()
if page is None:
abort(404, "Invalid page argument ({}).".format(request.args.get("page")))
filter_args = parse_filter_args(request.args)
filter_args["status"] = set_status_filters(filter_args)
service_data_retention_days = None
search_term = request.form.get("to", "")
if message_type is not None:
service_data_retention_days = current_service.get_days_of_retention(
message_type, number_of_days="seven_day"
)
if request.path.endswith("csv") and current_user.has_permissions(
ServicePermission.VIEW_ACTIVITY
):
return Response(
generate_notifications_csv(
service_id=service_id,
page=page,
page_size=5000,
template_type=[message_type],
status=filter_args.get("status"),
limit_days=service_data_retention_days,
),
mimetype="text/csv",
headers={"Content-Disposition": 'inline; filename="notifications.csv"'},
)
notifications = notification_api_client.get_notifications_for_service(
service_id=service_id,
page=page,
template_type=[message_type] if message_type else [],
status=filter_args.get("status"),
limit_days=service_data_retention_days,
to=search_term,
)
notifications_list = notifications.get("notifications", [])
url_args = {"message_type": message_type, "status": request.args.get("status")}
prev_page = None
if "links" in notifications and notifications["links"].get("prev", None):
prev_page = generate_previous_dict(
"main.view_notifications", service_id, page, url_args=url_args
)
next_page = None
total_items = notifications.get("total", 0)
page_size = notifications.get("page_size", 50)
total_pages = (total_items + page_size - 1) // page_size
if (
"links" in notifications
and notifications["links"].get("next", None)
and total_items > 50
and page < total_pages
):
next_page = generate_next_dict(
"main.view_notifications", service_id, page, url_args
)
if message_type:
download_link = url_for(
".view_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
)
else:
download_link = None
return {
"service_data_retention_days": service_data_retention_days,
"counts": render_template(
"views/activity/counts.html",
status=request.args.get("status"),
status_filters=get_status_filters(
current_service,
message_type,
service_api_client.get_service_statistics(
service_id, limit_days=service_data_retention_days
),
),
),
"notifications": render_template(
"views/activity/notifications.html",
notifications=list(
add_preview_of_content_to_notifications(notifications_list)
),
page=page,
limit_days=service_data_retention_days,
prev_page=prev_page,
next_page=next_page,
show_pagination=(not search_term),
status=request.args.get("status"),
message_type=message_type,
download_link=download_link,
single_notification_url=partial(
url_for,
".view_notification",
service_id=current_service.id,
),
),
}
def get_status_filters(service, message_type, statistics):
message_types = (
[message_type]
if message_type
else [NotificationType.EMAIL, NotificationType.SMS]
)
stats = {
NotificationStatus.REQUESTED: sum(
statistics[mt].get(NotificationStatus.REQUESTED, 0) for mt in message_types
),
NotificationStatus.DELIVERED: sum(
statistics[mt].get(NotificationStatus.DELIVERED, 0) for mt in message_types
),
NotificationStatus.FAILED: sum(
statistics[mt].get(NotificationStatus.FAILED, 0) for mt in message_types
),
}
stats[NotificationStatus.PENDING] = (
stats[NotificationStatus.REQUESTED]
- stats[NotificationStatus.DELIVERED]
- stats[NotificationStatus.FAILED]
)
filters = [
(NotificationStatus.REQUESTED, "total", "sending,delivered,failed"),
(NotificationStatus.PENDING, "pending", "sending,pending"),
(NotificationStatus.DELIVERED, "delivered", "delivered"),
(NotificationStatus.FAILED, "failed", "failed"),
]
return [
(
label,
option,
url_for(
".view_notifications",
service_id=service.id,
message_type=message_type,
status=option,
),
stats.get(key),
)
for key, label, option in filters
]
def _get_job_counts(job):
job_type = job.template_type
return [

View File

@@ -62,9 +62,9 @@ def view_notification(service_id, notification_id, error_message=None):
job = None
if get_help_argument() or request.args.get("help") == "0":
# help=0 is set when youve just sent a notification. We
# only want to show the back link when youve navigated to a
# notification, not when youve just sent it.
# help=0 is set when you've just sent a notification. We
# only want to show the back link when you've navigated to a
# notification, not when you've just sent it.
back_link = None
elif request.args.get("from_job"):
back_link = url_for(
@@ -74,10 +74,8 @@ def view_notification(service_id, notification_id, error_message=None):
)
else:
back_link = url_for(
"main.view_notifications",
"main.all_jobs_activity",
service_id=current_service.id,
message_type=template.template_type,
status="sending,delivered,failed",
)
return render_template(
@@ -198,10 +196,8 @@ def download_notifications_csv(service_id):
)
return redirect(
url_for(
"main.view_notifications",
"main.all_jobs_activity",
service_id=service_id,
message_type=filter_args["message_type"][0],
status="sending,delivered,failed",
)
)
return Response(

View File

@@ -25,12 +25,3 @@ def pricing():
search_form=SearchByNameForm(),
navigation_links=using_notify_nav(),
)
@main.route("/pricing/how-to-pay")
@user_is_logged_in
def how_to_pay():
return render_template(
"views/pricing/how-to-pay.html",
navigation_links=using_notify_nav(),
)