349 lines
11 KiB
Python
349 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Mapping
|
|
|
|
from govoplan_core.core.idm import (
|
|
IdmFunctionAssignmentDirectory,
|
|
OrganizationFunctionAssignmentRef,
|
|
)
|
|
from govoplan_core.core.organizations import (
|
|
OrganizationFunctionRef,
|
|
OrganizationHierarchyDirectory,
|
|
OrganizationHierarchyEdgeRef,
|
|
OrganizationUnitRef,
|
|
)
|
|
from govoplan_postbox.backend.schemas import PostboxRoutingPolicyPayload
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class HierarchyRouteCandidate:
|
|
depth: int
|
|
unit: OrganizationUnitRef
|
|
function: OrganizationFunctionRef | None
|
|
holders: tuple[OrganizationFunctionAssignmentRef, ...]
|
|
status: str
|
|
path: tuple[Mapping[str, object], ...]
|
|
diagnostics: tuple[str, ...] = ()
|
|
|
|
@property
|
|
def holder_count(self) -> int:
|
|
return len(
|
|
{
|
|
holder.identity_id or holder.account_id or holder.id
|
|
for holder in self.holders
|
|
}
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class HierarchyRoutePlan:
|
|
status: str
|
|
policy: Mapping[str, object]
|
|
candidates: tuple[HierarchyRouteCandidate, ...] = ()
|
|
diagnostics: tuple[str, ...] = ()
|
|
|
|
|
|
def normalized_routing_policy(
|
|
value: Mapping[str, object] | None,
|
|
) -> dict[str, object]:
|
|
return PostboxRoutingPolicyPayload.model_validate(value or {}).model_dump(
|
|
mode="json"
|
|
)
|
|
|
|
|
|
def plan_hierarchy_routes(
|
|
*,
|
|
hierarchy: OrganizationHierarchyDirectory | None,
|
|
incumbencies: IdmFunctionAssignmentDirectory,
|
|
tenant_id: str,
|
|
source_unit_id: str | None,
|
|
routing_policy: Mapping[str, object] | None,
|
|
producer_module: str,
|
|
classification: str,
|
|
expires_at: datetime | None,
|
|
now: datetime,
|
|
) -> HierarchyRoutePlan:
|
|
policy = normalized_routing_policy(routing_policy)
|
|
linked_copy = _mapping(policy.get("linked_copy"))
|
|
if not linked_copy.get("enabled"):
|
|
return HierarchyRoutePlan(
|
|
status="disabled",
|
|
policy=policy,
|
|
diagnostics=("hierarchy_routing_disabled",),
|
|
)
|
|
if hierarchy is None:
|
|
return HierarchyRoutePlan(
|
|
status="blocked",
|
|
policy=policy,
|
|
diagnostics=("organization_hierarchy_unavailable",),
|
|
)
|
|
if not source_unit_id:
|
|
return HierarchyRoutePlan(
|
|
status="blocked",
|
|
policy=policy,
|
|
diagnostics=("source_organization_unit_missing",),
|
|
)
|
|
|
|
diagnostics = list(
|
|
_policy_gate_diagnostics(
|
|
linked_copy,
|
|
producer_module=producer_module,
|
|
classification=classification,
|
|
expires_at=expires_at,
|
|
now=now,
|
|
)
|
|
)
|
|
if diagnostics:
|
|
return HierarchyRoutePlan(
|
|
status="blocked",
|
|
policy=policy,
|
|
diagnostics=tuple(diagnostics),
|
|
)
|
|
|
|
structure_id = str(linked_copy["structure_id"])
|
|
relation_type_ids = tuple(
|
|
str(value) for value in linked_copy.get("relation_type_ids", ())
|
|
)
|
|
max_depth = int(linked_copy["max_depth"])
|
|
try:
|
|
resolutions = hierarchy.resolve_hierarchy_relatives(
|
|
tenant_id,
|
|
(source_unit_id,),
|
|
structure_id=structure_id,
|
|
relation_type_ids=relation_type_ids,
|
|
direction="ancestors",
|
|
max_depth=max_depth,
|
|
)
|
|
except ValueError as exc:
|
|
return HierarchyRoutePlan(
|
|
status="blocked",
|
|
policy=policy,
|
|
diagnostics=(f"hierarchy_request_invalid:{exc}",),
|
|
)
|
|
if not resolutions:
|
|
return HierarchyRoutePlan(
|
|
status="blocked",
|
|
policy=policy,
|
|
diagnostics=("hierarchy_resolution_missing",),
|
|
)
|
|
resolution = resolutions[0]
|
|
diagnostics.extend(resolution.diagnostics)
|
|
if resolution.cycle_detected:
|
|
diagnostics.append("hierarchy_cycle_bounded")
|
|
if resolution.depth_limited:
|
|
diagnostics.append("hierarchy_depth_limited")
|
|
if resolution.status != "active":
|
|
diagnostics.append(f"hierarchy_{resolution.status}")
|
|
return HierarchyRoutePlan(
|
|
status="blocked",
|
|
policy=policy,
|
|
diagnostics=tuple(dict.fromkeys(diagnostics)),
|
|
)
|
|
|
|
matches = []
|
|
stop_unit_id = linked_copy.get("stop_unit_id")
|
|
stop_unit_type_id = linked_copy.get("stop_unit_type_id")
|
|
for match in sorted(
|
|
resolution.matches,
|
|
key=lambda item: (item.depth, item.unit.name, item.unit.id),
|
|
):
|
|
matches.append(match)
|
|
if (
|
|
stop_unit_id
|
|
and match.unit.id == stop_unit_id
|
|
or stop_unit_type_id
|
|
and match.unit.unit_type_id == stop_unit_type_id
|
|
):
|
|
diagnostics.append("hierarchy_stop_reached")
|
|
break
|
|
if not matches:
|
|
return HierarchyRoutePlan(
|
|
status="no_route",
|
|
policy=policy,
|
|
diagnostics=tuple(
|
|
dict.fromkeys((*diagnostics, "no_hierarchy_ancestor"))
|
|
),
|
|
)
|
|
|
|
unit_ids = tuple(dict.fromkeys(match.unit.id for match in matches))
|
|
target_function_type_id = str(linked_copy["target_function_type_id"])
|
|
try:
|
|
function_resolution = hierarchy.resolve_functions_by_type(
|
|
tenant_id,
|
|
target_function_type_id,
|
|
organization_unit_ids=unit_ids,
|
|
)
|
|
except ValueError as exc:
|
|
return HierarchyRoutePlan(
|
|
status="blocked",
|
|
policy=policy,
|
|
diagnostics=tuple(
|
|
dict.fromkeys(
|
|
(*diagnostics, f"function_resolution_invalid:{exc}")
|
|
)
|
|
),
|
|
)
|
|
diagnostics.extend(function_resolution.diagnostics)
|
|
if function_resolution.status != "active":
|
|
diagnostics.append(
|
|
f"target_function_type_{function_resolution.status}"
|
|
)
|
|
|
|
functions_by_unit: dict[str, list[OrganizationFunctionRef]] = {}
|
|
for function in function_resolution.matches:
|
|
if function.status != "active":
|
|
continue
|
|
functions_by_unit.setdefault(
|
|
function.organization_unit_id,
|
|
[],
|
|
).append(function)
|
|
function_ids = tuple(
|
|
function.id
|
|
for functions in functions_by_unit.values()
|
|
if len(functions) == 1
|
|
for function in functions
|
|
)
|
|
try:
|
|
holder_map = (
|
|
incumbencies.organization_function_incumbencies(
|
|
function_ids,
|
|
tenant_id=tenant_id,
|
|
)
|
|
if function_ids
|
|
else {}
|
|
)
|
|
except ValueError:
|
|
holder_map = {}
|
|
diagnostics.append("target_incumbency_unavailable")
|
|
|
|
candidates: list[HierarchyRouteCandidate] = []
|
|
seen_targets: set[tuple[str, str]] = set()
|
|
for match in matches:
|
|
functions = functions_by_unit.get(match.unit.id, [])
|
|
candidate_diagnostics: list[str] = []
|
|
function = functions[0] if len(functions) == 1 else None
|
|
if match.unit.status != "active":
|
|
status = "unit_inactive"
|
|
candidate_diagnostics.append("target_unit_inactive")
|
|
elif not functions:
|
|
status = "function_missing"
|
|
candidate_diagnostics.append("target_function_missing")
|
|
elif len(functions) > 1:
|
|
status = "function_ambiguous"
|
|
candidate_diagnostics.append("target_function_ambiguous")
|
|
elif (match.unit.id, function.id) in seen_targets:
|
|
status = "duplicate"
|
|
candidate_diagnostics.append("duplicate_target_suppressed")
|
|
else:
|
|
seen_targets.add((match.unit.id, function.id))
|
|
holders = tuple(
|
|
holder_map.get(function.id).assignments
|
|
if function.id in holder_map
|
|
else ()
|
|
)
|
|
status = "available" if holders else "vacant"
|
|
candidates.append(
|
|
HierarchyRouteCandidate(
|
|
depth=match.depth,
|
|
unit=match.unit,
|
|
function=function,
|
|
holders=holders,
|
|
status=status,
|
|
path=tuple(_edge_snapshot(edge) for edge in match.path),
|
|
)
|
|
)
|
|
continue
|
|
candidates.append(
|
|
HierarchyRouteCandidate(
|
|
depth=match.depth,
|
|
unit=match.unit,
|
|
function=function,
|
|
holders=(),
|
|
status=status,
|
|
path=tuple(_edge_snapshot(edge) for edge in match.path),
|
|
diagnostics=tuple(candidate_diagnostics),
|
|
)
|
|
)
|
|
|
|
routable = [
|
|
candidate
|
|
for candidate in candidates
|
|
if candidate.status in {"available", "vacant"}
|
|
]
|
|
status = "planned" if routable else "no_route"
|
|
if not routable:
|
|
diagnostics.append("no_available_hierarchy_target")
|
|
return HierarchyRoutePlan(
|
|
status=status,
|
|
policy=policy,
|
|
candidates=tuple(candidates),
|
|
diagnostics=tuple(dict.fromkeys(diagnostics)),
|
|
)
|
|
|
|
|
|
def _policy_gate_diagnostics(
|
|
linked_copy: Mapping[str, object],
|
|
*,
|
|
producer_module: str,
|
|
classification: str,
|
|
expires_at: datetime | None,
|
|
now: datetime,
|
|
) -> tuple[str, ...]:
|
|
diagnostics: list[str] = []
|
|
classifications = {
|
|
str(value) for value in linked_copy.get("allowed_classifications", ())
|
|
}
|
|
if classification not in classifications:
|
|
diagnostics.append("classification_not_allowed")
|
|
producers = {
|
|
str(value) for value in linked_copy.get("allowed_producer_modules", ())
|
|
}
|
|
if producer_module not in producers and "*" not in producers:
|
|
diagnostics.append("producer_not_authorized")
|
|
normalized_expiry = _as_utc(expires_at) if expires_at else None
|
|
if linked_copy.get("require_expiry") and normalized_expiry is None:
|
|
diagnostics.append("expiry_required")
|
|
max_retention_days = linked_copy.get("max_retention_days")
|
|
if (
|
|
normalized_expiry is not None
|
|
and max_retention_days is not None
|
|
and normalized_expiry
|
|
> _as_utc(now) + timedelta(days=int(max_retention_days))
|
|
):
|
|
diagnostics.append("retention_limit_exceeded")
|
|
if normalized_expiry is not None and normalized_expiry <= _as_utc(now):
|
|
diagnostics.append("message_already_expired")
|
|
return tuple(diagnostics)
|
|
|
|
|
|
def _edge_snapshot(edge: OrganizationHierarchyEdgeRef) -> dict[str, object]:
|
|
return {
|
|
"edge_id": edge.id,
|
|
"structure_id": edge.structure.id,
|
|
"structure_slug": edge.structure.slug,
|
|
"relation_type_id": edge.relation_type.id,
|
|
"relation_type_slug": edge.relation_type.slug,
|
|
"source_unit_id": edge.source_unit_id,
|
|
"target_unit_id": edge.target_unit_id,
|
|
}
|
|
|
|
|
|
def _mapping(value: object) -> dict[str, Any]:
|
|
return dict(value) if isinstance(value, Mapping) else {}
|
|
|
|
|
|
def _as_utc(value: datetime) -> datetime:
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value.astimezone(timezone.utc)
|
|
|
|
|
|
__all__ = [
|
|
"HierarchyRouteCandidate",
|
|
"HierarchyRoutePlan",
|
|
"normalized_routing_policy",
|
|
"plan_hierarchy_routes",
|
|
]
|