"""Who a campaign goes to. **The filter engine is not new.** ``app.core.person_filters`` already runs the platform's person-population semantics — departments with subtree expansion, countries, legal entities, employment statuses, job titles, email suffixes, an explicit ``person_ids`` allow-list, and/or mode, plus the unconditional HR test-record (P99XX) exclusion. Identity Sync (push) and the Directory / SCIM consumers (pull) both evaluate through it. Reusing it means an awareness audience and a sync scope cannot disagree about what "研发部门,含子部门" means. Two things this module adds on top: 1. **The candidate loader.** Identity Sync's loader lives inside another domain (and domain modules do not import each other), but more to the point it answers a *different* question: it carries sync-specific exclusions — purged rows, no-push flags, non-corporate-email junk — that exist because that pool feeds vendor directories. The pool here is "people who should read a security notice", which is active employees with a work email. Same evaluator, different candidate set, deliberately. 2. **``exclude_person_ids``.** The shared ``FilterConfig`` has no exclusion dimension, and it is simultaneously the pull-surface *authorisation boundary* — widening that engine for a content feature is the wrong trade. Exclusion is applied here, after ``matches()``, where it can only ever narrow. **Resolution happens once, at send time, and is snapshotted.** People change departments; an audience re-derived on every read would silently rewrite who a campaign was sent to. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.locale import resolve_lang from app.core.person_filters import ( CandidatePerson, FilterConfig, FilterError, expand_department_subtree, matches, ) from app.domains.security_awareness.catalog import MAX_AUDIENCE from app.identity.person_models import ( IdentityOrgMembership, IdentityOrgNode, IdentityPerson, ) logger = logging.getLogger(__name__) #: Statuses a campaign may reach. ``left`` is absent and stays absent: #: pushing a security comic to somebody who has departed is noise at #: best, and the notification outbox has its own leaver gate that would #: mark the row dead anyway. ``leaver_pending`` IS included — a person #: working out their notice period is still an employee, and the last #: weeks are exactly when data-handling reminders matter. REACHABLE_STATUSES: tuple[str, ...] = ("active", "on_leave", "leaver_pending") @dataclass(slots=True) class AudienceMember: """One resolved recipient, with the snapshot the target row stores.""" person: IdentityPerson lang: str snapshot: dict[str, Any] @dataclass(slots=True) class AudiencePreview: """What the admin sees before pressing send.""" total: int sample: list[dict[str, str | None]] by_department: list[dict[str, Any]] by_country: list[dict[str, Any]] #: Non-fatal notes: dimensions that matched nobody, people dropped #: for want of a work email, and so on. An audience of zero has a #: reason, and the admin is entitled to it before they schedule. notes: list[str] def parse_audience(raw: dict[str, Any] | None) -> tuple[FilterConfig, set[str]]: """Split a stored ``audience_config`` into the shared filter plus this domain's exclusion overlay. Raises :class:`FilterError` on a malformed config — the same exception type the shared engine raises, so callers have one thing to catch. """ payload = dict(raw or {}) raw_excludes = payload.pop("exclude_person_ids", []) or [] if not isinstance(raw_excludes, list): raise FilterError("exclude_person_ids must be a list") excludes: set[str] = set() for item in raw_excludes: try: excludes.add(str(uuid.UUID(str(item)))) except (ValueError, AttributeError) as exc: raise FilterError( f"exclude_person_ids entry is not a uuid: {item!r}" ) from exc return FilterConfig.from_dict(payload), excludes class AudienceResolver: """Loads the candidate pool once, then answers for many campaigns. Built per dispatch tick: the org tree and the person pool are the expensive parts and they do not change between two campaigns firing in the same minute. """ def __init__(self, db: AsyncSession): self.db = db self._candidates: list[CandidatePerson] | None = None self._people: dict[uuid.UUID, IdentityPerson] = {} self._membership: dict[uuid.UUID, IdentityOrgMembership] = {} self._children_index: dict[str, list[str]] | None = None self._skipped_no_email = 0 async def _load(self) -> None: if self._candidates is not None: return people = (await self.db.execute( select(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status.in_(REACHABLE_STATUSES), ) )).scalars().all() self._people = {p.id: p for p in people} memberships = (await self.db.execute( select(IdentityOrgMembership).where( IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), IdentityOrgMembership.person_id.in_(self._people.keys()), ) )).scalars().all() if self._people else [] self._membership = {m.person_id: m for m in memberships} candidates: list[CandidatePerson] = [] skipped = 0 for p in people: if not p.work_email: # No mailbox, no Feishu identity resolved by email, no # way to reach them. Counted and reported rather than # silently dropped — "the audience is 12 smaller than I # expected" needs an answer. skipped += 1 continue m = self._membership.get(p.id) candidates.append(CandidatePerson( person_id=p.id, work_email=p.work_email, employment_status=p.employment_status, country=p.country, job_title=p.job_title, employee_no=p.employee_no, company_code=p.company_code, employee_type_code=p.employee_type_code, hire_date=p.hire_date, hris_person_id=p.hris_person_id, org_path=m.org_path if m else None, org_node_source_id=m.org_node_source_id if m else None, )) self._candidates = candidates self._skipped_no_email = skipped nodes = (await self.db.execute( select(IdentityOrgNode).where(IdentityOrgNode.deleted_at.is_(None)) )).scalars().all() index: dict[str, list[str]] = {} for n in nodes: if n.parent_org_node_source_id: index.setdefault(n.parent_org_node_source_id, []).append( n.org_node_source_id, ) self._children_index = index async def resolve( self, audience_config: dict[str, Any] | None, ) -> list[AudienceMember]: """Every person this audience selects, in a stable order. Ordered by work email so two resolutions of the same audience produce the same list — which is what makes a dispatch reproducible and a preview honest about what will be sent. """ await self._load() cfg, excludes = parse_audience(audience_config) dept_ids = expand_department_subtree( cfg.departments, children_index=self._children_index or {}, include_subtree=cfg.include_subtree, ) out: list[AudienceMember] = [] for cand in self._candidates or []: if str(cand.person_id) in excludes: continue if not matches(cand, cfg, department_source_ids=dept_ids): continue person = self._people.get(cand.person_id) if person is None: # pragma: no cover - pool is built from it continue membership = self._membership.get(cand.person_id) out.append(AudienceMember( person=person, lang=resolve_lang(person), snapshot={ "display_name": person.display_name, "work_email": person.work_email, "department": membership.org_path if membership else None, "country": person.country, "employee_no": person.employee_no, }, )) out.sort(key=lambda m: (m.person.work_email or "").lower()) return out async def preview( self, audience_config: dict[str, Any] | None, *, sample_size: int = 12, ) -> AudiencePreview: """Dry-run the audience for the admin console. Returns counts, a small named sample, and per-department / per-country breakdowns — because "1 人" and "487 人" are both plausible outcomes of a mis-built filter and the number alone does not tell an admin which one they built. """ members = await self.resolve(audience_config) notes: list[str] = [] if self._skipped_no_email: notes.append(f"no_work_email:{self._skipped_no_email}") if not members: notes.append("empty_audience") if len(members) > MAX_AUDIENCE: notes.append(f"over_cap:{len(members)}") depts: dict[str, int] = {} countries: dict[str, int] = {} for m in members: dept = m.snapshot.get("department") or "—" depts[dept] = depts.get(dept, 0) + 1 country = m.snapshot.get("country") or "—" countries[country] = countries.get(country, 0) + 1 return AudiencePreview( total=len(members), sample=[ { # Name and email stay SEPARATE fields end to end # (#3417) — nothing on this path joins them. "person_id": str(m.person.id), "name": m.snapshot.get("display_name"), "email": m.snapshot.get("work_email"), "department": m.snapshot.get("department"), "lang": m.lang, } for m in members[:sample_size] ], by_department=[ {"key": k, "count": v} for k, v in sorted(depts.items(), key=lambda kv: (-kv[1], kv[0])) ], by_country=[ {"key": k, "count": v} for k, v in sorted( countries.items(), key=lambda kv: (-kv[1], kv[0]), ) ], notes=notes, )