"""Private batch adapter for deterministic transfer resource forecasts.""" from __future__ import annotations from collections import defaultdict from dataclasses import asdict, dataclass from datetime import UTC, datetime, timedelta from typing import Any from uuid import UUID from zoneinfo import ZoneInfo from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.coordinators.role_change_coordinator import ( REVIEW_STATUS_PENDING, ROLE_CHANGE_EVENT_TYPE, ) from app.core.provision_match import policy_matches_values from app.domains.asset.models import AssetItem, SoftwareSeatAllocation from app.domains.cmdb.models import ( GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING, CmdbCloudAccessGrant, ) from app.domains.hr.models import HrLifecycleEvent from app.domains.permission.models import ( PERM_ACTIVE, PERM_PENDING, UserPermission, ) from app.domains.provisioning.models import ProvisionPolicy from app.identity.person_models import IdentityPerson from app.intelligence.demand_forecast import ResourceKey, ScopeMode from app.intelligence.privacy import EvidenceObject, assert_llm_safe, build_reference_map from app.intelligence.transfer_forecast import ( CurrentHolding, TransferCandidate, TransferForecast, build_transfer_forecast, ) _APP_TZ = ZoneInfo("Asia/Shanghai") _WINDOW_DAYS = 90 @dataclass(frozen=True) class TransferForecastFactPackage: forecast: TransferForecast reference_map: dict[str, Any] excluded_counts: dict[str, int] data_as_of: datetime | None def aggregate_facts(self) -> dict[str, Any]: payload = { "forecast": asdict(self.forecast), "excluded_counts": dict(sorted(self.excluded_counts.items())), } assert_llm_safe(payload) return payload def _utc(value: datetime) -> datetime: return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) def _reverse_reference_map( objects: list[EvidenceObject], ) -> tuple[dict[tuple[str, str], str], dict[str, Any]]: _, mapping = build_reference_map(objects) return { (str(value["kind"]), str(value["object_id"])): ref for ref, value in mapping.items() }, mapping def _policy_matches(policy: ProvisionPolicy, person: IdentityPerson) -> bool: """Python projection of the canonical lenient person matcher.""" return policy_matches_values( rule_department=policy.department_name, rule_job_title=policy.match_job_title, rule_location=policy.match_location, department=person.primary_department, job_title=person.job_title, location=person.work_location_code, ) async def collect_transfer_forecast_facts( db: AsyncSession, *, now: datetime | None = None, ) -> TransferForecastFactPackage: observed_at = _utc(now or datetime.now(UTC)) today = observed_at.astimezone(_APP_TZ).date() window_start = today - timedelta(days=_WINDOW_DAYS) window_end = today + timedelta(days=_WINDOW_DAYS) excluded: defaultdict[str, int] = defaultdict(int) events = list((await db.execute(select(HrLifecycleEvent).where( HrLifecycleEvent.event_type == ROLE_CHANGE_EVENT_TYPE, HrLifecycleEvent.effective_date >= window_start, HrLifecycleEvent.effective_date <= window_end, ).order_by( HrLifecycleEvent.person_id, HrLifecycleEvent.effective_date.desc(), HrLifecycleEvent.created_at.desc(), HrLifecycleEvent.id.desc(), ))).scalars()) latest_event_by_person: dict[UUID, HrLifecycleEvent] = {} for event in events: latest_event_by_person.setdefault(event.person_id, event) excluded["superseded_transfer_event"] = len(events) - len(latest_event_by_person) person_ids = set(latest_event_by_person) people = list((await db.execute(select(IdentityPerson).where( IdentityPerson.id.in_(person_ids), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ))).scalars()) if person_ids else [] people_by_id = {person.id: person for person in people} excluded["transfer_without_active_person"] = len(person_ids - people_by_id.keys()) person_ids = set(people_by_id) policies = list((await db.execute(select(ProvisionPolicy).where( ProvisionPolicy.is_active.is_(True), ))).scalars()) seats = list((await db.execute(select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.user_id.in_(person_ids), SoftwareSeatAllocation.status == "active", ))).scalars()) if person_ids else [] permissions = list((await db.execute(select(UserPermission).where( UserPermission.user_id.in_(person_ids), UserPermission.status.in_((PERM_ACTIVE, PERM_PENDING)), ))).scalars()) if person_ids else [] assets = list((await db.execute(select(AssetItem).where( AssetItem.assigned_to.in_(person_ids), AssetItem.deleted_at.is_(None), AssetItem.disposed_at.is_(None), AssetItem.status != "disposed", ))).scalars()) if person_ids else [] grants = list((await db.execute(select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.person_id.in_(tuple(str(value) for value in person_ids)), CmdbCloudAccessGrant.status.in_((GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING)), CmdbCloudAccessGrant.review_status == REVIEW_STATUS_PENDING, ))).scalars()) if person_ids else [] objects: list[EvidenceObject] = [] objects.extend( EvidenceObject("transfer", str(event.id)) for event in latest_event_by_person.values() if event.person_id in person_ids ) objects.extend(EvidenceObject("software", str(row.plan_id)) for row in seats) objects.extend(EvidenceObject("access", f"permission:{row.permission_id}") for row in permissions) objects.extend(EvidenceObject("hardware", row.category) for row in assets) objects.extend(EvidenceObject("access", f"cloud_grant:{row.id}") for row in grants) for policy in policies: if policy.target_type == "seat": objects.append(EvidenceObject("software", str(policy.target_id))) elif policy.target_type == "permission": objects.append(EvidenceObject("access", f"permission:{policy.target_id}")) refs, reference_map = _reverse_reference_map(objects) def resource( kind: str, object_id: object, scope: ScopeMode = "global", ) -> ResourceKey: return ResourceKey(kind, refs[(kind, str(object_id))], scope) current_by_person: defaultdict[UUID, dict[ResourceKey, CurrentHolding]] = defaultdict(dict) for seat in seats: item = resource("software", seat.plan_id) current_by_person[seat.user_id][item] = CurrentHolding(item) for permission in permissions: item = resource("access", f"permission:{permission.permission_id}") current_by_person[permission.user_id][item] = CurrentHolding(item) for asset in assets: if asset.assigned_to is None: # excluded by the query; keep the type honest continue item = resource("hardware", asset.category, "location") current_by_person[asset.assigned_to][item] = CurrentHolding( item, retain_on_transfer=True, ) for grant in grants: try: person_id = UUID(str(grant.person_id)) except (TypeError, ValueError): excluded["invalid_cloud_person_reference"] += 1 continue item = resource("access", f"cloud_grant:{grant.id}") current_by_person[person_id][item] = CurrentHolding(item, review_on_transfer=True) target_by_person: defaultdict[UUID, set[ResourceKey]] = defaultdict(set) matched_unsupported_policies: set[UUID] = set() for person in people: for policy in policies: if not _policy_matches(policy, person): continue if policy.target_type == "seat": target_by_person[person.id].add(resource("software", policy.target_id)) elif policy.target_type == "permission": target_by_person[person.id].add(resource( "access", f"permission:{policy.target_id}", )) else: matched_unsupported_policies.add(policy.id) excluded["unsupported_target_policy"] = len(matched_unsupported_policies) transfers: list[TransferCandidate] = [] for person_id in sorted(person_ids, key=str): person = people_by_id[person_id] event = latest_event_by_person[person_id] complete = bool( person.primary_department and person.job_title and person.work_location_code ) if not complete: excluded["transfer_incomplete_target_attributes"] += 1 current = tuple(current_by_person[person_id].values()) target = frozenset(target_by_person[person_id]) actionable = bool( target - current_by_person[person_id].keys() or any( holding.review_on_transfer or ( holding.resource not in target and not holding.retain_on_transfer and holding.resource.kind != "hardware" ) for holding in current ) ) # A closed historical move must leave the current forecast. Future and # same-day moves remain visible even when already prepared. if event.effective_date < today and not actionable and complete: excluded["resolved_historical_transfer"] += 1 continue transfers.append(TransferCandidate( ref=refs[("transfer", str(event.id))], effective_date=event.effective_date, current_resources=current, target_resources=target, target_complete=complete, )) data_times = [_utc(row.created_at) for row in events] data_times.extend(_utc(row.updated_at) for row in people if row.updated_at is not None) data_times.extend(_utc(row.updated_at) for row in policies if row.updated_at is not None) data_times.extend(_utc(row.allocated_at) for row in seats) data_times.extend(_utc(row.updated_at) for row in permissions) data_times.extend(_utc(row.updated_at) for row in assets if row.updated_at is not None) data_times.extend(_utc(row.updated_at) for row in grants) package = TransferForecastFactPackage( forecast=build_transfer_forecast(as_of_date=today, transfers=tuple(transfers)), reference_map=reference_map, excluded_counts=dict(excluded), data_as_of=max(data_times, default=None), ) package.aggregate_facts() return package __all__ = ["TransferForecastFactPackage", "collect_transfer_forecast_facts"]