mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-06-23 17:02:01 -04:00
83156bd16e
Different emergencies will need broadcasts to last for a variable amount of time. We give users some control over this by letting them stop a broadcast early. But we should also let them set a maximum broadcast time, for: - when the duration of the danger is known - when the broadcast has been live long enough to alert everyone who needs to know about it This code re-uses the pattern for scheduling jobs, which has some constraints that are probably OK for now: - end time is limited to an hour - longest duration is 3 whole days (eg if you start broadcasting Friday you have the choice of Saturday, Sunday and all of Monday, up to midnight)
162 lines
4.2 KiB
Python
162 lines
4.2 KiB
Python
from datetime import datetime
|
|
|
|
from notifications_utils.broadcast_areas import broadcast_area_libraries
|
|
from notifications_utils.template import BroadcastPreviewTemplate
|
|
from orderedset import OrderedSet
|
|
|
|
from app.models import JSONModel, ModelList
|
|
from app.models.user import User
|
|
from app.notify_client.broadcast_message_api_client import (
|
|
broadcast_message_api_client,
|
|
)
|
|
from app.notify_client.service_api_client import service_api_client
|
|
|
|
|
|
class BroadcastMessage(JSONModel):
|
|
|
|
ALLOWED_PROPERTIES = {
|
|
'id',
|
|
'service_id',
|
|
'template_id',
|
|
'template_name',
|
|
'template_version',
|
|
'service_id',
|
|
'created_by',
|
|
'personalisation',
|
|
'starts_at',
|
|
'finishes_at',
|
|
'created_at',
|
|
'approved_at',
|
|
'cancelled_at',
|
|
'updated_at',
|
|
'created_by_id',
|
|
'approved_by_id',
|
|
'cancelled_by_id',
|
|
}
|
|
|
|
libraries = broadcast_area_libraries
|
|
|
|
def __lt__(self, other):
|
|
return (
|
|
self.cancelled_at or self.finishes_at
|
|
) < (
|
|
other.cancelled_at or other.finishes_at
|
|
)
|
|
|
|
@classmethod
|
|
def create(cls, *, service_id, template_id):
|
|
return cls(broadcast_message_api_client.create_broadcast_message(
|
|
service_id=service_id,
|
|
template_id=template_id,
|
|
))
|
|
|
|
@classmethod
|
|
def from_id(cls, broadcast_message_id, *, service_id):
|
|
return cls(broadcast_message_api_client.get_broadcast_message(
|
|
service_id=service_id,
|
|
broadcast_message_id=broadcast_message_id,
|
|
))
|
|
|
|
@property
|
|
def areas(self):
|
|
return broadcast_area_libraries.get_areas(
|
|
*self._dict['areas']
|
|
)
|
|
|
|
@property
|
|
def initial_area_names(self):
|
|
return [
|
|
area.name for area in self.areas
|
|
][:10]
|
|
|
|
@property
|
|
def polygons(self):
|
|
return broadcast_area_libraries.get_polygons_for_areas_lat_long(
|
|
*self._dict['areas']
|
|
)
|
|
|
|
@property
|
|
def template(self):
|
|
response = service_api_client.get_service_template(
|
|
self.service_id,
|
|
self.template_id,
|
|
version=self.template_version,
|
|
)
|
|
return BroadcastPreviewTemplate(response['data'])
|
|
|
|
@property
|
|
def status(self):
|
|
if (
|
|
self._dict['status']
|
|
and self._dict['status'] == 'broadcasting'
|
|
and self.finishes_at < datetime.utcnow().isoformat()
|
|
):
|
|
return 'completed'
|
|
return self._dict['status']
|
|
|
|
@property
|
|
def created_by(self):
|
|
return User.from_id(self.created_by_id)
|
|
|
|
@property
|
|
def approved_by(self):
|
|
return User.from_id(self.approved_by_id)
|
|
|
|
@property
|
|
def cancelled_by(self):
|
|
return User.from_id(self.cancelled_by_id)
|
|
|
|
def add_areas(self, *new_areas):
|
|
self._update(areas=list(OrderedSet(
|
|
self._dict['areas'] + list(new_areas)
|
|
)))
|
|
|
|
def remove_area(self, area_to_remove):
|
|
self._update(areas=[
|
|
area for area in self._dict['areas']
|
|
if area != area_to_remove
|
|
])
|
|
|
|
def _set_status_to(self, status):
|
|
broadcast_message_api_client.update_broadcast_message_status(
|
|
status,
|
|
broadcast_message_id=self.id,
|
|
service_id=self.service_id,
|
|
)
|
|
|
|
def _update(self, **kwargs):
|
|
broadcast_message_api_client.update_broadcast_message(
|
|
broadcast_message_id=self.id,
|
|
service_id=self.service_id,
|
|
data=kwargs,
|
|
)
|
|
|
|
def request_approval(self, until):
|
|
self._update(
|
|
finishes_at=until,
|
|
)
|
|
self._set_status_to('pending-approval')
|
|
|
|
def approve_broadcast(self):
|
|
self._update(
|
|
starts_at=datetime.utcnow().isoformat(),
|
|
)
|
|
self._set_status_to('broadcasting')
|
|
|
|
def reject_broadcast(self):
|
|
self._set_status_to('rejected')
|
|
|
|
def cancel_broadcast(self):
|
|
self._set_status_to('cancelled')
|
|
|
|
|
|
class BroadcastMessages(ModelList):
|
|
|
|
model = BroadcastMessage
|
|
client_method = broadcast_message_api_client.get_broadcast_messages
|
|
|
|
def with_status(self, *statuses):
|
|
return [
|
|
broadcast for broadcast in self if broadcast.status in statuses
|
|
]
|