eligibility-signposting-api/src/eligibility_signposting_api/audit/audit_context.py at 68daff8e1cd1bcbb5e11bfa5b908dc8d24ee3877 · NHSDigital/eligibility-signposting-api · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
import logging
from datetime import UTC, datetime
from uuid import UUID

from flask import Request, g

from eligibility_signposting_api.audit.audit_models import (
    AuditAction,
    AuditCondition,
    AuditEligibilityCohortGroups,
    AuditEligibilityCohorts,
    AuditEvent,
    AuditFilterRule,
    AuditRedirectRule,
    AuditSuitabilityRule,
    RequestAuditData,
    RequestAuditHeader,
    RequestAuditQueryParams,
)
from eligibility_signposting_api.audit.audit_service import AuditService
from eligibility_signposting_api.model.eligibility import (
    CohortGroupResult,
    ConditionName,
    IterationResult,
    Status,
    SuggestedAction,
)
from eligibility_signposting_api.model.rules import CampaignID, CampaignVersion, Iteration, RuleName, RulePriority

logger = logging.getLogger(__name__)


class AuditContext:
    @staticmethod
    def add_request_details(request: Request) -> None:
        g.audit_log = AuditEvent()
        resource_id = None
        if request.view_args and request.view_args["nhs_number"]:
            resource_id = request.view_args["nhs_number"]
        g.audit_log.request = RequestAuditData(
            nhs_number=resource_id,
            request_timestamp=datetime.now(tz=UTC),
            headers=(
                RequestAuditHeader(
                    x_request_id=request.headers.get("X-Request-ID"),
                    x_correlation_id=request.headers.get("X-Correlation-ID"),
                    nhsd_end_user_organisation_ods=request.headers.get("NHSD-End-User-Organisation-ODS"),
                    nhsd_application_id=request.headers.get("nhsd-application-id"),
                )
            ),
            query_params=(
                RequestAuditQueryParams(
                    category=request.args.get("category"),
                    conditions=request.args.get("conditions"),
                    include_actions=request.args.get("includeActions"),
                )
            ),
        )

    @staticmethod
    def append_audit_condition(
        suggested_actions: list[SuggestedAction] | None,
        condition_name: ConditionName,
        best_results: tuple[Iteration | None, IterationResult | None, dict[str, CohortGroupResult] | None],
        campaign_details: tuple[CampaignID | None, CampaignVersion | None],
        redirect_rule_details: tuple[RulePriority | None, RuleName | None],
    ) -> None:
        audit_eligibility_cohorts, audit_eligibility_cohort_groups, audit_actions = [], [], []
        audit_filter_rule, audit_suitability_rule, audit_redirect_rule = None, None, None
        best_active_iteration = best_results[0]
        best_candidate = best_results[1]
        best_cohort_results = best_results[2]

        if best_cohort_results:
            for cohort_label, result in sorted(best_cohort_results.items(), key=lambda item: item[1].cohort_code):
                cohort_status_name = result.status.name if result.status else None
                audit_eligibility_cohorts.append(
                    AuditEligibilityCohorts(cohort_code=cohort_label, cohort_status=cohort_status_name)
                )

                audit_eligibility_cohort_groups.append(
                    AuditEligibilityCohortGroups(
                        cohort_code=result.cohort_code, cohort_status=cohort_status_name, cohort_text=result.description
                    )
                )

                if result.audit_rules and best_candidate:
                    if best_candidate.status and best_candidate.status.name == Status.not_eligible.name:
                        audit_filter_rule = AuditFilterRule(
                            rule_priority=result.audit_rules[0].rule_priority,
                            rule_name=result.audit_rules[0].rule_name,
                        )
                    if best_candidate.status and best_candidate.status.name == Status.not_actionable.name:
                        audit_suitability_rule = AuditSuitabilityRule(
                            rule_priority=result.audit_rules[0].rule_priority,
                            rule_name=result.audit_rules[0].rule_name,
                            rule_message=result.audit_rules[0].rule_description,
                        )

        if best_candidate and best_candidate.status and best_candidate.status.name == Status.actionable.name:
            audit_redirect_rule = AuditRedirectRule(
                rule_priority=str(redirect_rule_details[0]), rule_name=redirect_rule_details[1]
            )

        if suggested_actions is None:
            audit_actions = None
        elif len(suggested_actions) > 0:
            for action in suggested_actions:
                audit_actions.append(
                    AuditAction(
                        internal_action_code=action.internal_action_code,
                        action_code=action.action_code,
                        action_type=action.action_type,
                        action_description=action.action_description,
                        action_url=str(action.url_link) if action.url_link else None,
                        action_url_label=action.url_label,
                    )
                )

        audit_condition = AuditCondition(
            campaign_id=campaign_details[0],
            campaign_version=campaign_details[1],
            iteration_id=best_active_iteration.id if best_active_iteration else None,
            iteration_version=best_active_iteration.version if best_active_iteration else None,
            condition_name=condition_name,
            status=best_candidate.status.name if best_candidate and best_candidate.status else None,
            status_text=best_candidate.status.name if best_candidate and best_candidate.status else None,
            eligibility_cohorts=audit_eligibility_cohorts,
            eligibility_cohort_groups=audit_eligibility_cohort_groups,
            filter_rules=audit_filter_rule,
            suitability_rules=audit_suitability_rule,
            action_rule=audit_redirect_rule,
            actions=audit_actions,
        )

        g.audit_log.response.condition.append(audit_condition)

    @staticmethod
    def add_response_details(response_id: UUID, last_updated: datetime) -> None:
        g.audit_log.response.response_id = response_id
        g.audit_log.response.last_updated = last_updated

    @staticmethod
    def write_to_firehose(service: AuditService) -> None:
        service.audit(g.audit_log.model_dump(by_alias=True))