"""Leave-health insights — aggregate leave *utilisation* across the org. The counter-intuitive lens: a team whose people barely touch their granted annual leave is not "hard-working" — it is a **team-health / leadership risk signal** (a lead who doesn't make it safe to rest). This service turns the per-person SF-synced leave data (:mod:`app.domains.hr.leave_models`) into utilisation roll-ups by **BU**, **department (team / 组)**, and **lead (full reporting subtree)**, plus the company baseline they're measured against. Design notes that matter for correctness: * **Denominator = ``entitled_days``.** Verified against prod (2026-06): CN annual leave is granted as a single upfront ``ACCRUAL`` at fiscal-year start, so ``entitled_days`` already equals the *full-year* quota (``entitled ≈ used + balance``). No proration / policy-calculator needed, and the same holds for the MY/US annual buckets. Utilisation is therefore ``used_days / entitled_days`` — "of the leave you were granted this year, how much have you actually taken". * **Primary health signal is RELATIVE, not absolute.** Leave is seasonal (summer + Spring Festival + National Day) so at fiscal-month 3 *everyone* looks "behind" on a flat pace line. Classifying a team as healthy/needs- attention is done by comparing its utilisation to the **company utilisation at the same moment** (a fair snapshot), never to a fixed threshold. The fiscal-pace index is carried as a forward-looking *context* column only. * **Annual leave only for the headline.** 法定年假 + 福利年假 (+ MY annual) drive utilisation; 调休 (comp / TOIL) is surfaced *separately* as a "banked overtime" signal (a high comp balance is its own red flag — paid overtime never taken back); 病假 / 陪产假 / sick buckets are excluded from the rate entirely. * **Coverage is shown honestly.** Only ~279 of ~500 staff have SF leave accounts; people without a quota are counted as "no data", never as 0%. * **Small-group privacy guard.** Groups below ``min_group_size`` covered members are flagged ``insufficient_data`` and kept out of the ranking, so the board never singles out an individual under a "team" label. """ from __future__ import annotations import logging from collections import Counter from dataclasses import dataclass, field from datetime import UTC, date, datetime from decimal import Decimal from zoneinfo import ZoneInfo from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.fiscal_year import ( current_fiscal_year, get_fiscal_year_start_month, ) from app.domains.hr.leave_models import HrLeaveAccount from app.identity.bu_helpers import bu_from_org_path from app.identity.person_models import ( IdentityOrgMembership, IdentityPerson, ) logger = logging.getLogger(__name__) # ── Account-type classification ────────────────────────────────────────── # Defaults reflect the live tenant (CN-dominant + a few MY/US). Overridable # via the ``hr.leave_insights.account_type_overrides`` platform setting # (a ``{code: "annual"|"comp"|"exclude"}`` map) so a new tenant / new SF # account-type code never needs a code change. Classification is: # annual → counts toward utilisation (numerator + denominator) # comp → surfaced separately as banked-overtime balance # exclude → sick / paternity / hospitalisation; ignored entirely _DEFAULT_ANNUAL_CODES = frozenset({"CHN_AnnualLeave", "CHN_PL", "MY_AnnuL"}) _DEFAULT_COMP_CODES = frozenset({"CHN_COMPL", "CHN_CompLeave", "CHN_TOIL"}) # The annual buckets that forfeit at period end if untaken — their current # balance is the "use-it-or-lose-it" at-risk figure. # # 福利年假 (``CHN_PL``) sat OUTSIDE this set until 2026-08-18 on the assumption # that only the statutory bucket lapses. HR states the opposite: both the # statutory and the welfare grant run an 18-month window and both are forfeited # on 09-30 — so the employee card, the 临期未休 total and the expiry reminder # were all under-reporting the same real deadline. Recorded in DECISIONS.md. _DEFAULT_FORFEIT_CODES = frozenset({"CHN_AnnualLeave", "CHN_PL", "MY_AnnuL"}) # Heuristic fallbacks for codes not in the explicit sets / overrides, so an # unseen ``XX_AnnualLeave`` still classifies sensibly instead of vanishing. _ANNUAL_HINTS = ("annualleave", "_annul", "annual") _COMP_HINTS = ("compl", "comp_", "toil", "overtime") _EXCLUDE_HINTS = ("sick", "_sl", "hosip", "hospital", "paternit", "matern", "_psl") ANNUAL = "annual" COMP = "comp" EXCLUDE = "exclude" # ── Health classification defaults (relative to company utilisation) ────── SETTING_THRESHOLDS = "hr.leave_insights.thresholds" SETTING_TYPE_OVERRIDES = "hr.leave_insights.account_type_overrides" _DEFAULT_THRESHOLDS = { # group_util / company_util >= this → healthy "healthy_ratio": 1.0, # group_util / company_util >= this (and < healthy) → normal; below → attention "watch_ratio": 0.6, # a member is a "low utiliser" when their utilisation < company_util * this "low_utiliser_ratio": 0.5, # groups with fewer than this many *covered* members are not ranked "min_group_size": 3, } HEALTH_HEALTHY = "healthy" HEALTH_NORMAL = "normal" HEALTH_ATTENTION = "attention" HEALTH_INSUFFICIENT = "insufficient_data" # A group where NObody has a leave account at all (covered == 0) — distinct # from `insufficient_data` (has some data but below the small-group privacy # guard). At a global company most non-CN teams land here because SF EC # Time-Off only carries the CN entity's leave; surfacing them as "uncovered" # (not a low-usage problem) keeps the board honest and uncluttered. HEALTH_UNCOVERED = "uncovered" DIMENSION_BU = "bu" DIMENSION_DEPARTMENT = "department" DIMENSION_LEAD = "lead" DIMENSION_COMPANY = "company" COMPANY_KEY = "__company__" def forfeits_at_period_end(code: str | None, overrides: dict[str, str]) -> bool: """Whether an unused balance in this bucket is LOST at period end. Still narrower than ``classify_account_type(...) == ANNUAL``: the at-risk figure this board reports counts only ``_DEFAULT_FORFEIT_CODES``, which is every ANNUAL code we know forfeits (法定年假 / 福利年假 / MY_AnnuL) but not the class as a whole — a tenant-specific annual code we have never seen must not be declared "use it or lose it" on a guess. The employee leave card calls this same function, so the card and the board can never tell one person two different stories about the same account. Tenant overrides still win: a code reclassified to comp/exclude does not forfeit, and one reclassified to annual is only treated as forfeiting if it is also in the forfeit set — the override vocabulary has no separate "forfeits" axis, so an explicit non-annual override is the reliable signal we can honour here. """ if not code: return False override = overrides.get(code) if override in (COMP, EXCLUDE): return False return code in _DEFAULT_FORFEIT_CODES async def load_type_overrides(db) -> dict[str, str]: """Tenant's ``{code: annual|comp|exclude}`` reclassification map. Module-level so the employee-facing leave cards can classify with the SAME overrides the admin insights board uses — a tenant that reclassifies a code must not get one answer on the board and another on the employee's card. """ from app.integrations.service import get_platform_setting raw = await get_platform_setting(SETTING_TYPE_OVERRIDES, default=None, session=db) if not isinstance(raw, dict): return {} return { str(k): str(v) for k, v in raw.items() if str(v) in (ANNUAL, COMP, EXCLUDE) } def classify_account_type(code: str | None, overrides: dict[str, str]) -> str: """Classify an SF account-type code → ``annual`` / ``comp`` / ``exclude``.""" if not code: return EXCLUDE ov = overrides.get(code) if ov in (ANNUAL, COMP, EXCLUDE): return ov if code in _DEFAULT_ANNUAL_CODES: return ANNUAL if code in _DEFAULT_COMP_CODES: return COMP low = code.lower() # Sick/paternity hints win first so e.g. a "PaidSickLeave" never reads as # annual on the "annual" substring. if any(h in low for h in _EXCLUDE_HINTS): return EXCLUDE if any(h in low for h in _COMP_HINTS): return COMP if any(h in low for h in _ANNUAL_HINTS): return ANNUAL return EXCLUDE def _f(value: Decimal | float | None) -> float: return 0.0 if value is None else float(value) def _ratio(num: float, den: float) -> float | None: """Utilisation ratio, ``None`` when there is no quota to divide by.""" if den <= 0: return None return round(num / den, 4) # Below this fraction of the fiscal year elapsed, the pace index (utilisation # ÷ elapsed) is statistically meaningless — dividing by a near-zero elapsed # turns a tiny utilisation into an absurd multiple. Report ``None`` instead. _PACE_MIN_ELAPSED = 0.02 def _pace(util: float | None, elapsed: float) -> float | None: """Annualised pace index, or ``None`` when too early in the FY to mean anything (guards the FY-boundary divide-by-~0 artifact).""" if util is None or elapsed < _PACE_MIN_ELAPSED: return None return round(util / elapsed, 4) def _median(values: list[float]) -> float | None: if not values: return None s = sorted(values) n = len(s) mid = n // 2 if n % 2: return round(s[mid], 4) return round((s[mid - 1] + s[mid]) / 2, 4) def fiscal_year_elapsed_fraction( fiscal_year: int, start_month: int, today: date, ) -> float: """Day-based fraction of the fiscal year elapsed through ``today``. Clamped to ``(0, 1]``. Used only to annualise utilisation into the forward-looking pace index — never as a health threshold. """ fy_start = date(fiscal_year, start_month, 1) # Exclusive end = first day of the NEXT fiscal year, so ``total`` is the # full day-count of the FY (365/366) and matches the 1-indexed # ``elapsed`` below. The old ``date(fiscal_year, 12, 31)`` for a # January FY made ``total`` 364 (Jan-1→Dec-31 is 364 days) while # ``elapsed`` counts inclusively, so the fraction overshot by ~1 day. if start_month == 1: fy_end = date(fiscal_year + 1, 1, 1) else: fy_end = date(fiscal_year + 1, start_month, 1) total = (fy_end - fy_start).days if total <= 0: return 1.0 elapsed = (today - fy_start).days + 1 frac = elapsed / total if frac <= 0: return 0.0001 return min(1.0, round(frac, 6)) @dataclass class PersonLeaveMetric: """Per-person leave figures used to roll up every dimension.""" person_id: str display_name: str employee_no: str | None department: str | None org_path: str | None bu: str | None manager_person_id: str | None annual_entitled: float = 0.0 annual_used: float = 0.0 annual_balance: float = 0.0 forfeit_at_risk: float = 0.0 comp_balance: float = 0.0 @property def has_quota(self) -> bool: return self.annual_entitled > 0 @property def utilisation(self) -> float | None: return _ratio(self.annual_used, self.annual_entitled) @dataclass class GroupInsight: """One ranked row: a BU / department / lead's leave-health summary.""" dimension: str key: str label: str member_count: int covered_count: int annual_entitled_total: float annual_used_total: float annual_balance_total: float utilisation: float | None median_utilisation: float | None low_utiliser_count: int forfeit_at_risk_total: float comp_banked_total: float pace_index: float | None vs_company: float | None health: str # Lead-only: the lead's own person row (so the UI can show who owns it). lead_person_id: str | None = None @property def is_redacted(self) -> bool: return self.health in (HEALTH_INSUFFICIENT, HEALTH_UNCOVERED) def as_dict(self, *, redact: bool = True) -> dict: """Serialise. When ``redact`` (default) and the group is below the small-group privacy guard, every leave METRIC is nulled — only the label, counts and health band survive, so a 1-2 person "team" can never have its numbers (or an individual's leave) reconstructed from the ranking. The snapshot writer passes ``redact=False`` since that row is internal (the trend endpoint applies its own guard on read).""" base = { "dimension": self.dimension, "key": self.key, "label": self.label, "member_count": self.member_count, "covered_count": self.covered_count, "health": self.health, "lead_person_id": self.lead_person_id, } if redact and self.is_redacted: return { **base, "annual_entitled_total": None, "annual_used_total": None, "annual_balance_total": None, "utilisation": None, "median_utilisation": None, "low_utiliser_count": None, "low_utiliser_pct": None, "forfeit_at_risk_total": None, "comp_banked_total": None, "pace_index": None, "vs_company": None, } return { **base, "annual_entitled_total": round(self.annual_entitled_total, 2), "annual_used_total": round(self.annual_used_total, 2), "annual_balance_total": round(self.annual_balance_total, 2), "utilisation": self.utilisation, "median_utilisation": self.median_utilisation, "low_utiliser_count": self.low_utiliser_count, "low_utiliser_pct": _ratio( float(self.low_utiliser_count), float(self.covered_count), ), "forfeit_at_risk_total": round(self.forfeit_at_risk_total, 2), "comp_banked_total": round(self.comp_banked_total, 2), "pace_index": self.pace_index, "vs_company": self.vs_company, } @dataclass class InsightsContext: """Everything a single read needs, computed once.""" fiscal_year: int fiscal_year_start_month: int elapsed_fraction: float as_of: date thresholds: dict metrics: list[PersonLeaveMetric] company_utilisation: float | None company_entitled_total: float company_used_total: float company_pace_index: float | None # person_id → metric, for drill-downs by_person: dict[str, PersonLeaveMetric] = field(default_factory=dict) # child adjacency manager_person_id → [child person_id] (active only) children: dict[str, list[str]] = field(default_factory=dict) class LeaveInsightsService: """Compute leave-health roll-ups. Pure reads; no mutation.""" def __init__(self, db: AsyncSession): self.db = db async def _load_thresholds(self) -> dict: from app.integrations.service import get_platform_setting raw = await get_platform_setting( SETTING_THRESHOLDS, default=None, session=self.db, ) merged = dict(_DEFAULT_THRESHOLDS) if isinstance(raw, dict): for k, v in raw.items(): if k in merged and isinstance(v, (int, float)): merged[k] = v return merged async def _load_type_overrides(self) -> dict[str, str]: return await load_type_overrides(self.db) async def build_context( self, *, as_of: date | None = None, ) -> InsightsContext: """Load the active population + their leave accounts and compute per-person metrics, the company baseline, and the manager tree.""" now = datetime.now(UTC) # "Today" is anchored to the company operating timezone (Asia/Shanghai # HQ — same anchor the identity-sync 监控看板 and finance close use), # NOT UTC. The snapshot cron fires 07:10 Asia/Shanghai (= 23:10 UTC the # prior day), so a UTC-derived date would label the trend point — and # the live 临期未休 day-counting reference — one calendar day behind # the business day. today = as_of or now.astimezone(ZoneInfo("Asia/Shanghai")).date() start_month = await get_fiscal_year_start_month(self.db) thresholds = await self._load_thresholds() overrides = await self._load_type_overrides() # The reporting fiscal year is derived from the actual current annual # accounts (filled during the account loop below), NOT the wall clock: # at an FY boundary the SF sync that opens the new-year TimeAccount may # lag the calendar, so a clock-derived label would claim the new FY # while the only ``is_current`` data is still last year's — mislabelling # the numbers and (because elapsed≈0 on day 1) blowing up the pace # index. Counting the observed current-account fiscal years keeps the # label honest to the data. Wall clock is only the fallback. observed_fy: dict[int, int] = {} # 1. Active population + primary department/org_path in one pass. person_rows = ( await self.db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.employee_no, IdentityPerson.manager_person_id, IdentityOrgMembership.org_node_name, IdentityOrgMembership.org_path, ) .outerjoin( IdentityOrgMembership, (IdentityOrgMembership.person_id == IdentityPerson.id) & IdentityOrgMembership.is_primary.is_(True) & IdentityOrgMembership.deleted_at.is_(None), ) .where( IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), ) ) ).all() metrics: dict[str, PersonLeaveMetric] = {} children: dict[str, list[str]] = {} for pid, name, emp_no, mgr, dept, org_path in person_rows: pid_s = str(pid) metrics[pid_s] = PersonLeaveMetric( person_id=pid_s, display_name=name or emp_no or pid_s, employee_no=emp_no, department=dept, org_path=org_path, bu=bu_from_org_path(org_path, fallback_name=dept), manager_person_id=str(mgr) if mgr else None, ) if mgr: children.setdefault(str(mgr), []).append(pid_s) # 2. Their leave accounts → per-person aggregates. if metrics: account_rows = ( await self.db.execute( select(HrLeaveAccount).where( HrLeaveAccount.person_id.in_(list(metrics.keys())), HrLeaveAccount.deleted_at.is_(None), ) ) ).scalars().all() for a in account_rows: m = metrics.get(a.person_id) if m is None: continue kind = classify_account_type(a.account_type, overrides) if kind == ANNUAL: # Utilisation is a *this-fiscal-year* figure → only the # current annual bucket. A prior-year leftover account # must not dilute the rate. if a.is_current: m.annual_entitled += _f(a.entitled_days) m.annual_used += _f(a.used_days) m.annual_balance += _f(a.balance_days) # Same function the employee leave card calls, so the # board's 临期未休 total and the card's countdown can # never disagree about whether a bucket forfeits. if forfeits_at_period_end(a.account_type, overrides): m.forfeit_at_risk += _f(a.balance_days) if a.fiscal_year is not None: observed_fy[a.fiscal_year] = observed_fy.get(a.fiscal_year, 0) + 1 elif kind == COMP: # Comp/TOIL accounts are per-event with no "current" # window — sum every open (non-closed) bucket's balance. if not a.account_closed: m.comp_balance += _f(a.balance_days) metric_list = list(metrics.values()) # Derive the reporting fiscal year from the observed current annual # accounts (most common wins); fall back to the wall clock when there # are no current annual accounts at all (empty tenant / sync not run). if observed_fy: fiscal_year = max(observed_fy, key=lambda fy: (observed_fy[fy], fy)) else: fiscal_year = current_fiscal_year(start_month, now) elapsed = fiscal_year_elapsed_fraction(fiscal_year, start_month, today) # 3. Company baseline (covered population only). ent_total = sum(m.annual_entitled for m in metric_list) used_total = sum(m.annual_used for m in metric_list) company_util = _ratio(used_total, ent_total) company_pace = _pace(company_util, elapsed) return InsightsContext( fiscal_year=fiscal_year, fiscal_year_start_month=start_month, elapsed_fraction=elapsed, as_of=today, thresholds=thresholds, metrics=metric_list, company_utilisation=company_util, company_entitled_total=round(ent_total, 2), company_used_total=round(used_total, 2), company_pace_index=company_pace, by_person=metrics, children=children, ) # ── Aggregation ────────────────────────────────────────────────────── def _classify_health( self, util: float | None, covered: int, ctx: InsightsContext, ) -> str: min_size = int(ctx.thresholds["min_group_size"]) if covered == 0 or util is None: # Nobody in this group has a leave account — not a low-usage # problem, just out of the leave system's coverage (most non-CN # teams). Keep it distinct from the small-but-has-data case. return HEALTH_UNCOVERED if covered < min_size: return HEALTH_INSUFFICIENT company = ctx.company_utilisation if not company: # company has no quota at all → can't compare return HEALTH_INSUFFICIENT ratio = util / company if ratio >= ctx.thresholds["healthy_ratio"]: return HEALTH_HEALTHY if ratio >= ctx.thresholds["watch_ratio"]: return HEALTH_NORMAL return HEALTH_ATTENTION def _aggregate( self, dimension: str, key: str, label: str, members: list[PersonLeaveMetric], ctx: InsightsContext, *, lead_person_id: str | None = None, ) -> GroupInsight: covered = [m for m in members if m.has_quota] ent = sum(m.annual_entitled for m in covered) used = sum(m.annual_used for m in covered) bal = sum(m.annual_balance for m in covered) util = _ratio(used, ent) med = _median([m.utilisation for m in covered if m.utilisation is not None]) company = ctx.company_utilisation low_ratio = ctx.thresholds["low_utiliser_ratio"] low_threshold = (company or 0.0) * low_ratio low_count = sum( 1 for m in covered if m.utilisation is not None and m.utilisation < low_threshold ) pace = _pace(util, ctx.elapsed_fraction) vs_company = ( None if (util is None or not company) else round(util / company, 4) ) return GroupInsight( dimension=dimension, key=key, label=label, member_count=len(members), covered_count=len(covered), annual_entitled_total=ent, annual_used_total=used, annual_balance_total=bal, utilisation=util, median_utilisation=med, low_utiliser_count=low_count, forfeit_at_risk_total=sum(m.forfeit_at_risk for m in covered), comp_banked_total=sum(m.comp_balance for m in members), pace_index=pace, vs_company=vs_company, health=self._classify_health(util, len(covered), ctx), lead_person_id=lead_person_id, ) def _descendants(self, manager_id: str, ctx: InsightsContext) -> list[str]: """Full reporting subtree (BFS) of ``manager_id`` over the active population, excluding the manager themself. Cycle-safe.""" out: list[str] = [] seen: set[str] = {manager_id} queue = list(ctx.children.get(manager_id, [])) while queue: cur = queue.pop() if cur in seen: continue seen.add(cur) out.append(cur) queue.extend(ctx.children.get(cur, [])) return out @staticmethod def canon_group_key(value: str) -> str: """Canonical bucket key for a free-text BU / department name. ``org_node_name`` (and the snapshot table's ``group_key``) are ``utf8mb4_unicode_ci`` — case- and trailing-space-insensitive. So HRIS names that differ only in case or surrounding whitespace (``'PreSales'`` vs ``'Presales '``) are the SAME group to the DB but distinct Python dict keys. Bucketing by the raw string then emits two ranking rows that collapse onto one row of the snapshot table's case-insensitive unique index — a 1062 that aborts the whole daily cron (no snapshot for ANY group that day). Folding case + stripping whitespace here mirrors the collation, so one HRIS team is always one group end-to-end (ranking, drill-down, and trend).""" return value.strip().casefold() def _group_by_string( self, dimension: str, attr: str, ctx: InsightsContext, ) -> list[GroupInsight]: """Roll up by a free-text attribute (``bu`` / ``department``), folding case/whitespace variants of the same name into one group (see :meth:`canon_group_key`). Both the group ``key`` and its label use the most common original spelling — so a single-spelling org reads exactly as before, while variant spellings collapse into one group (one row of the snapshot table's case-insensitive unique index) instead of two that 1062-clash. ``members_of`` canonicalises both sides, so every spelling variant still resolves to this group on drill-down / trend.""" buckets: dict[str, list[PersonLeaveMetric]] = {} spellings: dict[str, Counter] = {} for m in ctx.metrics: raw = getattr(m, attr) if not raw: continue canon = self.canon_group_key(raw) buckets.setdefault(canon, []).append(m) spellings.setdefault(canon, Counter())[raw.strip()] += 1 out: list[GroupInsight] = [] for canon, members in buckets.items(): # Deterministic representative: most common spelling, ties broken # alphabetically so the same key is chosen run-to-run. label = min( spellings[canon].items(), key=lambda kv: (-kv[1], kv[0]), )[0] out.append(self._aggregate(dimension, label, label, members, ctx)) return out def group_by_bu(self, ctx: InsightsContext) -> list[GroupInsight]: return self._group_by_string(DIMENSION_BU, "bu", ctx) def group_by_department(self, ctx: InsightsContext) -> list[GroupInsight]: return self._group_by_string(DIMENSION_DEPARTMENT, "department", ctx) def group_by_lead(self, ctx: InsightsContext) -> list[GroupInsight]: out: list[GroupInsight] = [] for lead_id, direct in ctx.children.items(): lead = ctx.by_person.get(lead_id) if lead is None: # manager not in active population — skip continue sub_ids = self._descendants(lead_id, ctx) members = [ctx.by_person[i] for i in sub_ids if i in ctx.by_person] if not members: continue label = lead.display_name out.append( self._aggregate( DIMENSION_LEAD, lead_id, label, members, ctx, lead_person_id=lead_id, ) ) return out def group(self, dimension: str, ctx: InsightsContext) -> list[GroupInsight]: if dimension == DIMENSION_BU: return self.group_by_bu(ctx) if dimension == DIMENSION_DEPARTMENT: return self.group_by_department(ctx) if dimension == DIMENSION_LEAD: return self.group_by_lead(ctx) raise ValueError(f"unknown dimension: {dimension}") def members_of( self, dimension: str, key: str, ctx: InsightsContext, ) -> list[PersonLeaveMetric]: """The per-person rows behind one group, for the drill-down drawer. ``key`` is the canonical bucket key (see :meth:`canon_group_key`); canonicalise both sides so a member is matched to its group regardless of the spelling of its raw ``bu`` / ``department`` (and regardless of which spelling the caller echoes back).""" if dimension == DIMENSION_BU: ck = self.canon_group_key(key) return [m for m in ctx.metrics if m.bu and self.canon_group_key(m.bu) == ck] if dimension == DIMENSION_DEPARTMENT: ck = self.canon_group_key(key) return [ m for m in ctx.metrics if m.department and self.canon_group_key(m.department) == ck ] if dimension == DIMENSION_LEAD: sub_ids = self._descendants(key, ctx) return [ctx.by_person[i] for i in sub_ids if i in ctx.by_person] raise ValueError(f"unknown dimension: {dimension}") def below_privacy_guard( self, members: list[PersonLeaveMetric], ctx: InsightsContext, ) -> bool: """True when a group has fewer COVERED members than the small-group guard — the same test ``_classify_health`` uses to mark a group ``insufficient_data``. The drill-down (members) and trend endpoints call this so the per-person privacy protection the ranking enforces cannot be bypassed by drilling straight into a tiny group.""" covered = sum(1 for m in members if m.has_quota) return covered < int(ctx.thresholds["min_group_size"])