Activity feed

hypha.apply.activity.adapters.activity_feed

ActivityAdapter

Bases: AdapterBase

adapter_type class-attribute instance-attribute

adapter_type = 'Activity Feed'

always_send class-attribute instance-attribute

always_send = True

messages class-attribute instance-attribute

messages = {MESSAGES.TRANSITION: 'handle_transition', MESSAGES.BATCH_TRANSITION: 'handle_batch_transition', MESSAGES.NEW_SUBMISSION: _('Submitted {source.title_text_display} for {source.page.title}'), MESSAGES.EDIT_SUBMISSION: _('edited the submission'), MESSAGES.APPLICANT_EDIT: _('edited the submission'), MESSAGES.UPDATE_LEAD: _('updated Lead from {old_lead} to {source.lead}'), MESSAGES.BATCH_UPDATE_LEAD: _('batch updated Lead to {new_lead}'), MESSAGES.DETERMINATION_OUTCOME: _('sent a determination. Outcome: {determination.clean_outcome}'), MESSAGES.BATCH_DETERMINATION_OUTCOME: 'batch_determination', MESSAGES.INVITED_TO_PROPOSAL: _('Invited to submit a proposal'), MESSAGES.REVIEWERS_UPDATED: 'reviewers_updated', MESSAGES.BATCH_REVIEWERS_UPDATED: 'batch_reviewers_updated', MESSAGES.NEW_REVIEW: _('Submitted a review'), MESSAGES.OPENED_SEALED: _('Opened the submission while still sealed'), MESSAGES.SCREENING: 'handle_screening_statuses', MESSAGES.REVIEW_OPINION: _('{opinion.opinion_display}s with {opinion.review.author}s review of {source}'), MESSAGES.DELETE_REVIEW_OPINION: _('deleted the opinion for review: {review_opinion.review}'), MESSAGES.CREATED_PROJECT: _('Created project with initial status of "{status}"'), MESSAGES.CREATED_SOW: _('Created SOW for project "{related.project}"'), MESSAGES.EDITED_SOW: _('Edited SOW for project "{related.project}"'), MESSAGES.CREATED_PF: _('Created project form for project "{related.project}"'), MESSAGES.EDITED_PF: _('Edited project form for project "{related.project}"'), MESSAGES.PROJECT_TRANSITION: 'handle_project_transition', MESSAGES.UPDATE_PROJECT_TITLE: _('updated the project title from {old_title} to {source.title}'), MESSAGES.UPDATE_PROJECT_CONTRACT_NUMBER: _('updated contract number to {source.contract_number}'), MESSAGES.UPDATE_PROJECT_LEAD: _('update Lead from {old_lead} to {source.lead}'), MESSAGES.SEND_FOR_APPROVAL: _('Requested approval'), MESSAGES.APPROVE_PAF: 'handle_paf_assignment', MESSAGES.APPROVE_PROJECT: _('Approved'), MESSAGES.REQUEST_PROJECT_CHANGE: _('requested changes for acceptance'), MESSAGES.SUBMIT_CONTRACT_DOCUMENTS: _('Submitted Contract Documents'), MESSAGES.UPLOAD_CONTRACT: _('Uploaded a {contract.state} contract'), MESSAGES.APPROVE_CONTRACT: _('Approved contract'), MESSAGES.UPDATE_INVOICE_STATUS: 'handle_update_invoice_status', MESSAGES.CREATE_INVOICE: _('Invoice added'), MESSAGES.SUBMIT_REPORT: _('Submitted a report'), MESSAGES.DELETE_REPORT: _('deleted a report'), MESSAGES.SKIPPED_REPORT: 'handle_skipped_report', MESSAGES.REPORT_FREQUENCY_CHANGED: 'handle_report_frequency', MESSAGES.DISABLED_REPORTING: _('disabled reporting'), MESSAGES.BATCH_DELETE_SUBMISSION: 'handle_batch_delete_submission', MESSAGES.BATCH_ARCHIVE_SUBMISSION: 'handle_batch_archive_submission', MESSAGES.BATCH_UPDATE_INVOICE_STATUS: 'handle_batch_update_invoice_status', MESSAGES.ARCHIVE_SUBMISSION: _('archived this submission'), MESSAGES.UNARCHIVE_SUBMISSION: _('un-archived this submission'), MESSAGES.DELETE_INVOICE: _('deleted an invoice'), MESSAGES.REMOVE_TASK: 'handle_task_removal', MESSAGES.UPDATE_AUTHOR: _('updated author from {old_author} to {source.user}')}

recipients

recipients(message_type, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def recipients(self, message_type, **kwargs):
    return [None]

extra_kwargs

extra_kwargs(message_type, source, sources, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def extra_kwargs(self, message_type, source, sources, **kwargs):
    if message_type in [
        MESSAGES.OPENED_SEALED,
        MESSAGES.REVIEWERS_UPDATED,
        MESSAGES.SCREENING,
        MESSAGES.REVIEW_OPINION,
        MESSAGES.DELETE_REVIEW_OPINION,
        MESSAGES.BATCH_REVIEWERS_UPDATED,
        MESSAGES.APPROVE_PROJECT,
        MESSAGES.CREATED_SOW,
        MESSAGES.EDITED_SOW,
        MESSAGES.CREATED_PF,
        MESSAGES.EDITED_PF,
        MESSAGES.REQUEST_PROJECT_CHANGE,
        MESSAGES.SEND_FOR_APPROVAL,
        MESSAGES.APPROVE_PAF,
        MESSAGES.NEW_REVIEW,
        MESSAGES.UPDATE_PROJECT_LEAD,
        MESSAGES.UPDATE_LEAD,
        MESSAGES.BATCH_UPDATE_LEAD,
        MESSAGES.ARCHIVE_SUBMISSION,
        MESSAGES.UNARCHIVE_SUBMISSION,
        MESSAGES.BATCH_ARCHIVE_SUBMISSION,
        MESSAGES.REMOVE_TASK,
        MESSAGES.UPDATE_AUTHOR,
    ]:
        return {"visibility": TEAM}

    if message_type in [
        MESSAGES.CREATED_PROJECT,
        MESSAGES.APPROVE_CONTRACT,
        MESSAGES.UPLOAD_CONTRACT,
        MESSAGES.SUBMIT_CONTRACT_DOCUMENTS,
        MESSAGES.DELETE_INVOICE,
        MESSAGES.CREATE_INVOICE,
        MESSAGES.DELETE_REPORT,
    ]:
        return {"visibility": APPLICANT}

    source = source or sources[0]
    if is_transition(message_type) and not source.phase.permissions.can_view(
        source.user
    ):
        # User's shouldn't see status activity changes for stages that aren't visible to the them
        return {"visibility": TEAM}

    if message_type == MESSAGES.UPDATE_INVOICE_STATUS:
        invoice = kwargs.get("invoice", None)
        if invoice and not is_invoice_public_transition(invoice):
            return {"visibility": TEAM}
        return {"visibility": APPLICANT}
    return {}

reviewers_updated

reviewers_updated(added=None, removed=None, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def reviewers_updated(self, added=None, removed=None, **kwargs):
    message = [_("Reviewers updated.")]
    if added:
        message.append(_("Added:"))
        message.extend(reviewers_message(added))

    if removed:
        message.append(_("Removed:"))
        message.extend(reviewers_message(removed))

    return " ".join(message)

batch_reviewers_updated

batch_reviewers_updated(added, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def batch_reviewers_updated(self, added, **kwargs):
    base = [_("Batch reviewers updated:")]
    base.extend(
        [
            _("{user} as {name}.").format(user=str(user), name=role.name)
            for role, user in added
            if user
        ]
    )
    return " ".join(base)

batch_determination

batch_determination(sources, determinations, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def batch_determination(self, sources, determinations, **kwargs):
    submission = sources[0]
    determination = determinations[submission.id]
    return self.messages[MESSAGES.DETERMINATION_OUTCOME].format(
        determination=determination,
    )

handle_batch_delete_submission

handle_batch_delete_submission(sources, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_batch_delete_submission(self, sources, **kwargs):
    submissions = sources
    submissions_text = ", ".join(
        [submission.title_text_display for submission in submissions]
    )
    return _("Successfully deleted submissions: {title}").format(
        title=submissions_text
    )

handle_batch_archive_submission

handle_batch_archive_submission(sources, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_batch_archive_submission(self, sources, **kwargs):
    submissions = sources
    submissions_text = ", ".join(
        [submission.title_text_display for submission in submissions]
    )
    return _("Successfully archived submissions: {title}").format(
        title=submissions_text
    )

handle_batch_update_invoice_status

handle_batch_update_invoice_status(sources, invoices, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_batch_update_invoice_status(self, sources, invoices, **kwargs):
    invoice_numbers = ", ".join(
        [
            invoice.invoice_number if invoice.invoice_number else ""
            for invoice in invoices
        ]
    )
    invoice_status = invoices[0].status if invoices else ""
    return _(
        "Successfully updated status to {invoice_status} for invoices: {invoice_numbers}"
    ).format(
        invoice_status=get_invoice_status_display_value(invoice_status),
        invoice_numbers=invoice_numbers,
    )

handle_paf_assignment

handle_paf_assignment(source, paf_approvals, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_paf_assignment(self, source, paf_approvals, **kwargs):
    if hasattr(paf_approvals, "__iter__"):  # paf_approvals has to be iterable
        users = ", ".join(
            [
                paf_approval.user.full_name
                if paf_approval.user.full_name
                else paf_approval.user.username
                for paf_approval in paf_approvals
            ]
        )
        users_sentence = " and".join(users.rsplit(",", 1))
        return _("assigned Project form to {}").format(users_sentence)
    return None

handle_task_removal

handle_task_removal(source, task, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_task_removal(self, source, task, **kwargs):
    if task.user:
        return _(
            "removed the task {task.code} for {source} from the task list".format(
                task=task, source=source
            )
        )
    return _(
        "removed the task {task.code} for {source} from whole team's{user_groups} task list.".format(
            task=task,
            source=source,
            user_groups=list(task.user_group.all().values_list("name", flat=True)),
        )
    )

handle_transition

handle_transition(old_phase, source, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_transition(self, old_phase, source, **kwargs):
    def wrap_in_color_class(phase, text):
        return f'<span class="rounded-full inline-block px-2 py-0.5 font-medium text-gray-800 {phase.bg_color}">{text}</span>'

    submission = source
    base_message = _("Progressed from {old_display} to {new_display}")

    new_phase = submission.phase

    staff_message = base_message.format(
        old_display=wrap_in_color_class(old_phase, old_phase.display_name),
        new_display=wrap_in_color_class(new_phase, new_phase.display_name),
    )

    if new_phase.permissions.can_view(submission.user):
        # we need to provide a different message to the applicant
        if not old_phase.permissions.can_view(submission.user):
            old_phase = submission.workflow.previous_visible(
                old_phase, submission.user
            )

        applicant_message = base_message.format(
            old_display=wrap_in_color_class(old_phase, old_phase.public_name),
            new_display=wrap_in_color_class(new_phase, new_phase.public_name),
        )

        return json.dumps(
            {
                TEAM: staff_message,
                ALL: applicant_message,
            }
        )

    return staff_message

handle_project_transition

handle_project_transition(old_stage, source, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_project_transition(self, old_stage, source, **kwargs):
    project = source
    base_message = _("Progressed from {old_display} to {new_display}")

    staff_message = base_message.format(
        old_display=get_project_status_display_value(old_stage),
        new_display=project.status_display,
    )

    applicant_message = base_message.format(
        old_display=get_project_public_status(project_status=old_stage),
        new_display=get_project_public_status(project_status=project.status),
    )

    return json.dumps(
        {
            TEAM: staff_message,
            ALL: applicant_message,
        }
    )

handle_batch_transition

handle_batch_transition(transitions, sources, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_batch_transition(self, transitions, sources, **kwargs):
    submissions = sources
    kwargs.pop("source")
    for submission in submissions:
        old_phase = transitions[submission.id]
        return self.handle_transition(
            old_phase=old_phase, source=submission, **kwargs
        )

handle_report_frequency

handle_report_frequency(config, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_report_frequency(self, config, **kwargs):
    new_schedule = config.get_frequency_display()
    return _(
        "Updated reporting frequency. New schedule is: {new_schedule} starting on {schedule_start}"
    ).format(new_schedule=new_schedule, schedule_start=config.schedule_start)

handle_skipped_report

handle_skipped_report(report, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_skipped_report(self, report, **kwargs):
    if report.skipped:
        return _("Skipped a Report")
    else:
        return _("Marked a Report as required")

handle_update_invoice_status

handle_update_invoice_status(invoice, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_update_invoice_status(self, invoice, **kwargs):
    base_message = _("Updated Invoice status to: {invoice_status}.")
    staff_message = base_message.format(invoice_status=invoice.get_status_display())

    if is_invoice_public_transition(invoice):
        public_status = get_invoice_public_status(invoice_status=invoice.status)
        applicant_message = base_message.format(invoice_status=public_status)
        return json.dumps({TEAM: staff_message, ALL: applicant_message})

    return staff_message

send_message

send_message(message, user, source, sources, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def send_message(self, message, user, source, sources, **kwargs):
    from ..models import Activity

    visibility = kwargs.get("visibility", ALL)

    related = kwargs["related"]
    if isinstance(related, dict):
        try:
            related = related[source.id]
        except KeyError:
            pass

    has_correct_fields = all(
        hasattr(related, attr) for attr in ["get_absolute_url"]
    )
    isnt_source = source != related
    is_model = isinstance(related, DjangoModel)
    if has_correct_fields and isnt_source and is_model:
        related_object = related
    else:
        related_object = None

    Activity.actions.create(
        user=user,
        source=source,
        timestamp=timezone.now(),
        message=message,
        visibility=visibility,
        related_object=related_object,
    )

handle_screening_statuses

handle_screening_statuses(source, old_status, **kwargs)

Source code in hypha/apply/activity/adapters/activity_feed.py

def handle_screening_statuses(self, source, old_status, **kwargs):
    new_status = source.get_current_screening_status()

    if str(new_status) == old_status:
        return

    if new_status and old_status != "-":
        return _(
            'Updated screening decision from "{old_status}" to "{new_status}"'
        ).format(old_status=old_status, new_status=new_status)
    elif new_status:
        return _('Added screening decision to "{new_status}"').format(
            new_status=new_status
        )
    elif old_status != "-":
        return _('Removed "{old_status}" screening decision.').format(
            old_status=old_status
        )

message

message(message_type, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def message(self, message_type, **kwargs):
    try:
        message = self.messages[message_type]
    except KeyError:
        # We don't know how to handle that message type
        return

    try:
        # see if its a method on the adapter
        method = getattr(self, message)
    except AttributeError:
        return self.render_message(message, **kwargs)
    else:
        # Delegate all responsibility to the custom method
        return method(**kwargs)

render_message

render_message(message, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def render_message(self, message, **kwargs):
    return message.format(**kwargs)

get_neat_related

get_neat_related(message_type, related)

Source code in hypha/apply/activity/adapters/base.py

def get_neat_related(self, message_type, related):
    # We translate the related kwarg into something we can understand
    try:
        neat_name = neat_related[message_type]
    except KeyError:
        # Message type doesn't expect a related object
        if related:
            raise ValueError(
                f"Unexpected 'related' kwarg provided for {message_type}"
            ) from None
        return {}
    else:
        if not related:
            raise ValueError(f"{message_type} expects a 'related' kwarg")
        return {neat_name: related}

batch_recipients

batch_recipients(message_type, sources, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def batch_recipients(self, message_type, sources, **kwargs):
    # Default batch recipients is to send a message to each of the recipients that would
    # receive a message under normal conditions
    return [
        {
            "recipients": self.recipients(message_type, source=source, **kwargs),
            "sources": [source],
        }
        for source in sources
    ]

process_batch

process_batch(message_type, events, request, user, sources, related=None, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def process_batch(
    self, message_type, events, request, user, sources, related=None, **kwargs
):
    events_by_source = {event.source.id: event for event in events}
    for recipient in self.batch_recipients(
        message_type, sources, user=user, **kwargs
    ):
        recipients = recipient["recipients"]
        sources = recipient["sources"]
        events = [events_by_source[source.id] for source in sources]
        self.process_send(
            message_type,
            recipients,
            events,
            request,
            user,
            sources=sources,
            source=None,
            related=related,
            **kwargs,
        )

process

process(message_type, event, request, user, source, related=None, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def process(
    self, message_type, event, request, user, source, related=None, **kwargs
):
    recipients = self.recipients(
        message_type,
        source=source,
        related=related,
        user=user,
        request=request,
        **kwargs,
    )
    self.process_send(
        message_type,
        recipients,
        [event],
        request,
        user,
        source,
        related=related,
        **kwargs,
    )

process_send

process_send(message_type, recipients, events, request, user, source, sources=None, related=None, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def process_send(
    self,
    message_type,
    recipients,
    events,
    request,
    user,
    source,
    sources=None,
    related=None,
    **kwargs,
):
    if sources is None:
        sources = []
    try:
        # If this was a batch action we want to pull out the submission
        source = sources[0]
    except IndexError:
        pass

    kwargs = {
        "request": request,
        "user": user,
        "source": source,
        "sources": sources,
        "related": related,
        **kwargs,
    }
    kwargs.update(self.get_neat_related(message_type, related))
    kwargs.update(self.extra_kwargs(message_type, **kwargs))

    for recipient in recipients:
        # Allow for customization of message based on recipient string (will vary based on adapter)
        message_kwargs = {**kwargs, "recipient": recipient}
        message = self.message(message_type, **message_kwargs)
        if not message:
            continue

        message_logs = self.create_logs(message, recipient, *events)

        if settings.SEND_MESSAGES or self.always_send:
            status = self.send_message(
                message, recipient=recipient, logs=message_logs, **kwargs
            )
        else:
            status = "Message not sent as SEND_MESSAGES==FALSE"

        message_logs.update_status(status)

        if not settings.SEND_MESSAGES:
            if recipient:
                debug_message = _("{adapter} [to: {recipient}]: {message}").format(
                    adapter=self.adapter_type, recipient=recipient, message=message
                )
            else:
                debug_message = _("{adapter}: {message}").format(
                    adapter=self.adapter_type, message=message
                )
            messages.add_message(request, messages.DEBUG, debug_message)

create_logs

create_logs(message, recipient, *events)

Source code in hypha/apply/activity/adapters/base.py

def create_logs(self, message, recipient, *events):
    from ..models import Message

    messages = Message.objects.bulk_create(
        Message(**self.log_kwargs(message, recipient, event)) for event in events
    )
    return Message.objects.filter(id__in=[message.id for message in messages])

log_kwargs

log_kwargs(message, recipient, event)

Source code in hypha/apply/activity/adapters/base.py

def log_kwargs(self, message, recipient, event):
    return {
        "type": self.adapter_type,
        "content": message,
        "recipient": recipient or "",
        "event": event,
    }