GovOPlaN/govoplan-dist-lists
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Integrate governed address contact resolution

Browse Source
This commit is contained in:
zemion
2026-08-02 07:24:29 +02:00
parent 7b0c9cfdde
commit 8099e34c6c
9 changed files with 774 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_dist_lists/backend/expansion.py
+396 -2
View File
@@ -16,6 +16,14 @@ from govoplan_core.core.dataflows import (
DataflowDatasetRequest,
dataflow_dataset_output,
)
from govoplan_core.core.contact_points import (
CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION,
ContactPointCandidate,
ContactPointResolution,
ContactPointResolutionProvider,
ContactPointResolutionRequest,
ContactPointSourceRequest,
)
from govoplan_core.core.distribution_lists import (
CAPABILITY_POLICY_DISTRIBUTION_CHANNELS,
CAPABILITY_RECIPIENT_CHANNEL_FACTS,
@@ -566,6 +574,199 @@ def _nested_list_recipients(
def _address_recipients(
context: _ExpansionContext,
entry: DistributionListEntryRef,
) -> list[DistributionRecipientRef]:
provider = _typed_capability(
context.registry,
CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION,
ContactPointResolutionProvider,
)
if provider is not None:
return _contact_point_recipients(context, entry, provider)
return _legacy_address_recipients(context, entry)
def _contact_point_recipients(
context: _ExpansionContext,
entry: DistributionListEntryRef,
provider: ContactPointResolutionProvider,
) -> list[DistributionRecipientRef]:
requested_channels, constrained = _requested_channel_filter(context, entry)
if constrained and not requested_channels:
return [
_unresolved(
entry,
"channel.no_common_selection",
"The entry and expansion request do not allow a common delivery channel.",
status="suppressed",
)
]
if entry.kind == "address_list":
return _contact_point_source_recipients(
context,
entry,
provider,
requested_channels=requested_channels,
)
try:
resolution = provider.resolve_contact_points(
context.session,
context.principal,
request=ContactPointResolutionRequest(
tenant_id=context.principal.tenant_id,
subject=entry.source,
effective_at=context.effective_at,
purpose=entry.purpose or context.request.purpose,
requested_channels=requested_channels,
address_purpose=_text(entry.configuration.get("address_purpose")),
fallback_rule=_contact_point_fallback_rule(entry),
locale=_text(entry.configuration.get("locale")),
postal_format=_postal_format(entry),
context={
"distribution_list_id": context.request.list_id,
"distribution_list_entry_id": entry.id,
},
),
)
except (LookupError, PermissionError, ValueError) as exc:
return [_unresolved(entry, "addresses.contact_point_failed", str(exc))]
contact_revision = _contact_record_revision(resolution)
stale = _is_stale(
entry.source,
actual_revision=contact_revision,
actual_fingerprint=resolution.source_fingerprint,
)
context.evidence.append(
_contact_point_evidence(
entry,
actual_revision=contact_revision or resolution.source_revision,
actual_fingerprint=resolution.source_fingerprint,
stale=stale,
details={
"contract_version": resolution.contract_version,
"contact_id": resolution.contact_id,
"contact_point_revision": resolution.source_revision,
"contact_point_fingerprint": resolution.source_fingerprint,
"provenance": dict(resolution.provenance),
},
)
)
return [
_contact_point_recipient(
entry,
resolution,
stale=stale,
expected_email=(
_text(entry.configuration.get("email"))
or _text(entry.source.metadata.get("email"))
if entry.kind == "address_email"
else None
),
)
]
def _contact_point_source_recipients(
context: _ExpansionContext,
entry: DistributionListEntryRef,
provider: ContactPointResolutionProvider,
*,
requested_channels: tuple[str, ...],
) -> list[DistributionRecipientRef]:
source_id = _contact_point_source_id(entry)
source_request = ContactPointSourceRequest(
tenant_id=context.principal.tenant_id,
source_id=source_id,
effective_at=context.effective_at,
purpose=entry.purpose or context.request.purpose,
requested_channels=requested_channels, # type: ignore[arg-type]
address_purpose=_text(entry.configuration.get("address_purpose")),
fallback_rule=_contact_point_fallback_rule(entry),
locale=_text(entry.configuration.get("locale")),
postal_format=_postal_format(entry),
max_items=20_000,
context={
"distribution_list_id": context.request.list_id,
"distribution_list_entry_id": entry.id,
},
)
resolutions: list[ContactPointResolution] = []
preview = None
source_revision: str | None = None
source_fingerprint: str | None = None
offset = 0
provider_limit = context.request.limits.max_provider_results
try:
while len(resolutions) < provider_limit:
page_limit = min(500, provider_limit - len(resolutions))
preview = provider.preview_source(
context.session,
context.principal,
request=source_request,
offset=offset,
limit=page_limit,
)
if source_revision is None:
source_revision = preview.source_revision
source_fingerprint = preview.source_fingerprint
elif (
preview.source_revision != source_revision
or preview.source_fingerprint != source_fingerprint
):
return [
_unresolved(
entry,
"addresses.source_changed",
"The Addresses source changed while it was being expanded; retry the expansion.",
status="stale",
)
]
resolutions.extend(preview.resolutions)
offset += len(preview.resolutions)
if not preview.has_more or not preview.resolutions:
break
except (LookupError, PermissionError, ValueError) as exc:
return [_unresolved(entry, "addresses.source_failed", str(exc))]
if preview is None:
return []
stale = _is_stale(
entry.source,
actual_revision=source_revision,
actual_fingerprint=source_fingerprint,
)
context.evidence.append(
_contact_point_evidence(
entry,
actual_revision=source_revision,
actual_fingerprint=source_fingerprint,
stale=stale,
generated_at=preview.generated_at,
details={
"contract_version": preview.contract_version,
"source_id": source_id,
"total_count": preview.total_count,
"provenance": dict(preview.provenance),
},
)
)
if preview.total_count > len(resolutions):
context.candidate_truncated = True
context.diagnostics.append(
_explanation(
"provider.result_limit",
"Addresses results were truncated by the provider limit.",
provider="addresses",
source=entry.source,
)
)
return [
_contact_point_recipient(entry, resolution, stale=stale)
for resolution in resolutions
]
def _legacy_address_recipients(
context: _ExpansionContext,
entry: DistributionListEntryRef,
) -> list[DistributionRecipientRef]:
if entry.kind == "address_list":
provider = _capability(context.registry, ADDRESSES_RECIPIENT_SOURCE)
@@ -631,6 +832,194 @@ def _address_recipients(
return [_address_lookup_recipient(entry, item) for item in exact]
def _contact_point_recipient(
entry: DistributionListEntryRef,
resolution: ContactPointResolution,
*,
stale: bool,
expected_email: str | None = None,
) -> DistributionRecipientRef:
accepted = list(resolution.candidates)
rejected = list(resolution.excluded)
if expected_email is not None:
matching: list[ContactPointCandidate] = []
for candidate in accepted:
if candidate.channel == "email" and candidate.target.casefold() == expected_email.casefold():
matching.append(candidate)
else:
rejected.append(
replace(
candidate,
status="suppressed",
reason_code="addresses.email.not_selected",
explanation="This contact point is not the email address selected by the list entry.",
)
)
accepted = matching
channels = tuple(
_distribution_channel_candidate(item, entry=entry, stale=stale)
for item in (*accepted, *rejected)
)
selected_channels = channels[: len(accepted)]
if any(item.status == "usable" for item in selected_channels):
status = "usable"
elif any(item.status == "stale" for item in selected_channels):
status = "stale"
else:
status = _recipient_outcome(channels, resolution.status)
explanations = [*resolution.explanations]
explanations.extend(
DistributionExplanation(
code=item.reason_code,
message=item.explanation,
severity="warning",
provider="addresses",
source=item.source or entry.source,
provenance=dict(item.provenance),
)
for item in rejected
if item.reason_code and item.explanation
)
contact_id = resolution.contact_id
return DistributionRecipientRef(
recipient_key=f"contact:{contact_id}" if contact_id else f"unresolved:{entry.id}",
display_name=(
resolution.display_name
or entry.label
or entry.source.label
or contact_id
or entry.source.resource_id
),
status=status, # type: ignore[arg-type]
channels=channels,
contact_id=contact_id,
source_entry_ids=(entry.id,),
explanations=tuple(_unique_explanations(explanations)),
attributes={
"contact_points": [
{
"channel": item.channel,
"contact_point_id": item.contact_point_id,
"address_purpose": item.address_purpose,
"value": dict(item.value),
}
for item in (*accepted, *rejected)
],
},
provenance={
**_entry_provenance(entry),
"contact_point_resolution": {
"contract_version": resolution.contract_version,
"source_revision": resolution.source_revision,
"source_fingerprint": resolution.source_fingerprint,
"provenance": dict(resolution.provenance),
},
"channel_facts_resolved": True,
},
)
def _distribution_channel_candidate(
candidate: ContactPointCandidate,
*,
entry: DistributionListEntryRef,
stale: bool,
) -> DistributionChannelCandidate:
status = candidate.status
reason_code = candidate.reason_code
explanation = candidate.explanation
if stale and status == "usable":
status = "stale"
reason_code = "source.stale"
explanation = "The source revision changed."
return DistributionChannelCandidate(
channel=candidate.channel,
target=candidate.target,
target_key=candidate.target_key,
status=status,
contact_point_id=candidate.contact_point_id,
locale=candidate.locale,
preferred=candidate.preferred,
reason_code=reason_code,
explanation=explanation,
source=candidate.source or entry.source,
decision_provenance={
**dict(candidate.provenance),
"address_purpose": candidate.address_purpose,
"preference_rank": candidate.preference_rank,
"source_revision": candidate.source_revision,
"preference_revision": candidate.preference_revision,
"consent_revision": candidate.consent_revision,
"value": dict(candidate.value),
},
)
def _contact_point_evidence(
entry: DistributionListEntryRef,
*,
actual_revision: str | None,
actual_fingerprint: str | None,
stale: bool,
details: Mapping[str, object],
generated_at: datetime | None = None,
) -> DistributionProviderEvidence:
return DistributionProviderEvidence(
provider="addresses",
source=entry.source,
actual_revision=actual_revision,
actual_fingerprint=actual_fingerprint,
stale=stale,
generated_at=generated_at,
details=details,
)
def _contact_record_revision(resolution: ContactPointResolution) -> str | None:
revisions = {
str(revision).strip()
for candidate in (*resolution.candidates, *resolution.excluded)
if (revision := candidate.provenance.get("source_revision")) is not None
and str(revision).strip()
}
return next(iter(revisions)) if len(revisions) == 1 else None
def _contact_point_source_id(entry: DistributionListEntryRef) -> str:
resource_id = entry.source.resource_id
if resource_id.startswith("addresses:"):
return resource_id
resource_type = entry.source.resource_type
if resource_type in {"address_book", "book"}:
return f"addresses:address_book:{resource_id}"
return f"addresses:address_list:{resource_id}"
def _contact_point_fallback_rule(entry: DistributionListEntryRef):
value = _text(entry.configuration.get("fallback_rule")) or "primary"
return value if value in {"none", "primary", "any"} else "primary"
def _postal_format(entry: DistributionListEntryRef):
value = _text(entry.configuration.get("postal_format")) or "domestic"
return value if value in {"domestic", "international"} else "domestic"
def _requested_channel_filter(
context: _ExpansionContext,
entry: DistributionListEntryRef,
) -> tuple[tuple[str, ...], bool]:
entry_channels = set(entry.requested_channels)
if entry.kind == "address_email":
entry_channels = {"email"}
caller_channels = set(context.request.requested_channels)
if entry_channels and caller_channels:
selected = entry_channels.intersection(caller_channels)
else:
selected = entry_channels or caller_channels
return tuple(sorted(selected)), bool(entry_channels or caller_channels)
def _address_snapshot_recipient(
entry: DistributionListEntryRef,
item: object,
@@ -1063,7 +1452,11 @@ def _apply_channel_decisions(
CAPABILITY_RECIPIENT_CHANNEL_FACTS,
RecipientChannelFactsProvider,
)
if facts_provider is not None and recipient.source_entry_ids:
if (
facts_provider is not None
and recipient.source_entry_ids
and not recipient.provenance.get("channel_facts_resolved")
):
source = channels[0].source if channels else None
if source is not None:
facts = facts_provider.resolve_channel_facts(
@@ -1097,12 +1490,13 @@ def _apply_channel_decisions(
else set()
)
requested_by_caller = set(context.request.requested_channels)
has_channel_constraint = bool(requested_by_entry or requested_by_caller)
requested_channels = (
requested_by_entry.intersection(requested_by_caller)
if requested_by_entry and requested_by_caller
else requested_by_entry or requested_by_caller
)
if requested_channels:
if has_channel_constraint:
channels = [
item
if item.channel in requested_channels