"""Atlassian production-site activity directory. Single source of truth for "when did each person last actually use a given Atlassian product on a *paid production* site?". Used by the ``software_seat.idle_reclaim`` job and reusable by any license-utilisation report. Why this lives in its own module (not inside the reclaim handler) ───────────────────────────────────────────────────────────────── The previous idle-seat attempt computed "90 days unused" wrong because it trusted whatever ``last_active`` came back without thinking about *which site* it came from. Atlassian's Org Admin API (``GET /admin/v1/orgs/{orgId}/users``) returns one ``product_access`` entry per (product × site). PingCAP's org contains, besides the paid production sites: * ``tidb-sandbox.atlassian.net`` — a sandbox whose ``last_active`` is frozen years in the past; treating it as a real signal made daily users (hu.guo, charles.m) look like they hadn't logged in for two years. * external personal sites (``yanyanqing``, ``ecosystem``, ``fluxai``) that have nothing to do with the company's licensed seats. So the only honest signal is: **for each (person, product), take the MAX ``last_active`` across the PRODUCTION sites that grant that product, ignoring sandbox / external sites entirely.** That computation is encapsulated here so the reclaim job and any future report share one correct implementation. Production-site policy (overridable via the ``software_seat.idle_reclaim.production_sites`` platform setting): * Jira Software / Jira Service Management → ``tidb.atlassian.net`` * Confluence → ``tidb.atlassian.net`` + ``pingcap-us.atlassian.net`` ``last_active`` parsing: the field is nanosecond-precision UTC (``2026-05-26T14:24:25.994292581Z``) which ``datetime.fromisoformat`` chokes on. We only ever need day granularity for an idle-days count, so we take the first 10 characters (the ``YYYY-MM-DD`` date) and parse that. A blank / missing ``last_active`` resolves to ``None`` — and a ``None`` means "cannot determine", never "idle" (admins, API-only users, mobile-only users, and telemetry gaps all legitimately produce blanks). """ from __future__ import annotations import logging from collections.abc import Iterable from dataclasses import dataclass, field from datetime import UTC, date, datetime from typing import TYPE_CHECKING, Any from sqlalchemy.ext.asyncio import AsyncSession if TYPE_CHECKING: from app.integrations.drivers.jira_software import JiraSoftwareDriver logger = logging.getLogger(__name__) # ── Product families we track (the Jira three-piece set) ────────── PRODUCT_JIRA_SOFTWARE = "jira-software" PRODUCT_JIRA_SERVICEDESK = "jira-servicedesk" PRODUCT_CONFLUENCE = "confluence" ALL_TRACKED_PRODUCTS = ( PRODUCT_JIRA_SOFTWARE, PRODUCT_JIRA_SERVICEDESK, PRODUCT_CONFLUENCE, ) # Production-site allowlist per product family. A ``product_access`` # entry only counts toward the idle computation when its site host is # in the set for that product. Everything not listed (tidb-sandbox, # yanyanqing, ecosystem, fluxai, …) is ignored — that is the whole # point of this module. DEFAULT_PRODUCTION_SITES: dict[str, frozenset[str]] = { PRODUCT_JIRA_SOFTWARE: frozenset({"tidb.atlassian.net"}), PRODUCT_JIRA_SERVICEDESK: frozenset({"tidb.atlassian.net"}), PRODUCT_CONFLUENCE: frozenset( {"tidb.atlassian.net", "pingcap-us.atlassian.net"}, ), } _PRODUCTION_SITES_SETTING_KEY = "software_seat.idle_reclaim.production_sites" # Atlassian org/site-admin role tokens (best-effort; the reclaim job's # primary admin guard is admin-group membership, this is a bonus net). _ADMIN_ROLE_TOKENS = frozenset( {"org-admin", "organization-admin", "site-admin", "site-administrator"}, ) # ── Six-bucket classification (since 2026-05-29) ─────────────────── # # Every Atlassian account holding a license-tracked product family # (returned by ``AtlassianActivityDirectory.all_license_holders``) is # classified into EXACTLY ONE of these buckets so the displayed identity # ``vendor_license_holders == sum(buckets)`` always holds. The buckets # are ordered by precedence — the first matching rule wins so the # partition is exclusive. # # Priority order: # 1. ``HOLDER_NO_EMAIL`` — account has no email (can't be matched or # drilled into; count-only). # 2. ``HOLDER_VENDOR_INACTIVE`` — vendor account_status != active. # Atlassian still bills these as occupying a seat until offboarding # releases them. Takes precedence over Helix-status checks because # a deactivated vendor account isn't a useful seat regardless of # who it once belonged to. # 3. ``HOLDER_OFF_PAID_SITE`` — vendor active, but the account only # holds the product on NON-paid sites (sandbox / personal). Mostly # relevant for marketplace addons (Tempo, ScriptRunner) that bill # org-wide. Takes precedence over Helix-status because a sandbox # seat isn't a "real" employee seat regardless of the person's # employment status. # 4. ``HOLDER_LEAVER_RESIDUAL`` — vendor active on paid site, email # matches a Helix person whose ``employment_status`` is ``left`` # / ``leaver_pending``. Offboarding leak — surfaced loudly. # 5. ``HOLDER_EXTERNAL`` — vendor active on paid site, no Helix # person matches the email (partners, bots, contractors). # 6. ``HOLDER_EMPLOYEE_ACTIVE`` — vendor active on paid site, email # matches an active Helix person. The healthy baseline; the only # bucket that results in an active ``software_seat_allocations`` # row. HOLDER_NO_EMAIL = "no_email" HOLDER_VENDOR_INACTIVE = "vendor_inactive" HOLDER_OFF_PAID_SITE = "off_paid_site" HOLDER_LEAVER_RESIDUAL = "leaver_residual" HOLDER_EXTERNAL = "external" HOLDER_EMPLOYEE_ACTIVE = "employees_active" ALL_HOLDER_BUCKETS = ( HOLDER_EMPLOYEE_ACTIVE, HOLDER_LEAVER_RESIDUAL, HOLDER_EXTERNAL, HOLDER_VENDOR_INACTIVE, HOLDER_OFF_PAID_SITE, HOLDER_NO_EMAIL, ) # Helix employment statuses that mean "this person has left the company" # for the purpose of the leaver-residual bucket. ``leaver_pending`` is # included so an in-flight offboarding's residual seats are flagged as # soon as the offboarding-blocked guard fires (not only after termination). _LEAVER_EMPLOYMENT_STATUSES = frozenset({"left", "leaver_pending"}) def classify_holder( acct: AtlassianAccountActivity, product_family: str, paid_sites: frozenset[str], *, helix_employment_status: str | None, ) -> str: """Bucket one license holder into one of :data:`ALL_HOLDER_BUCKETS`. ``paid_sites`` is the family's paid-site allowlist (the plan's ``site_urls`` after normalisation — usually ``{tidb.atlassian.net}`` for jira-software/JSM and ``{tidb.atlassian.net, pingcap-us.atlassian.net}`` for confluence). ``helix_employment_status`` is the Helix person's ``employment_status`` if the holder's email matched a person, ``None`` if no person matched. Precedence order is documented at the bucket constants above — the first matching rule wins, so the partition stays exclusive even when a single holder could "fit" multiple buckets (e.g. a deactivated leaver's account is ``HOLDER_VENDOR_INACTIVE``, NOT ``HOLDER_LEAVER_RESIDUAL``, because we want to highlight the actually-active leaks). """ if not acct.email: return HOLDER_NO_EMAIL if not acct.is_active: return HOLDER_VENDOR_INACTIVE family_sites = acct.access_sites_by_product.get( product_family, frozenset(), ) if not (family_sites & paid_sites): return HOLDER_OFF_PAID_SITE if helix_employment_status in _LEAVER_EMPLOYMENT_STATUSES: return HOLDER_LEAVER_RESIDUAL if helix_employment_status is None: return HOLDER_EXTERNAL return HOLDER_EMPLOYEE_ACTIVE @dataclass(slots=True) class AtlassianAccountActivity: """One managed Atlassian account's production-site activity.""" account_id: str email: str | None account_status: str | None roles: tuple[str, ...] = () # product family → latest production-site ``last_active`` date, or # ``None`` when no production entry carried a parseable date. A # family absent from this dict means the account has no production # access to that product at all. prod_last_active_by_product: dict[str, date] = field(default_factory=dict) # product family → the set of site hosts on which this account holds # the product, regardless of ``last_active`` (presence, not recency). # This is the seat-COUNT signal: a license holder counts even when # ``last_active`` is blank (admins / API-only / mobile / telemetry # gaps), so it must NOT be derived from ``prod_last_active_by_product`` # (which drops null-date entries). Hosts are stored unfiltered (incl. # sandbox / external); the count caller intersects with the *plan's* # paid-production ``site_urls`` so sandbox is excluded at query time. access_sites_by_product: dict[str, frozenset[str]] = field(default_factory=dict) raw: dict[str, Any] = field(default_factory=dict) @property def is_active(self) -> bool: return (self.account_status or "").lower() == "active" @property def is_org_or_site_admin(self) -> bool: return any(r in _ADMIN_ROLE_TOKENS for r in self.roles) def last_active_for(self, product_family: str) -> date | None: return self.prod_last_active_by_product.get(product_family) def has_product_on_sites( self, product_family: str, sites: frozenset[str], ) -> bool: """True when this account holds ``product_family`` on any host in ``sites``. ``sites`` must already be normalized lowercase hosts.""" return bool(self.access_sites_by_product.get(product_family, frozenset()) & sites) @dataclass(slots=True) class AtlassianActivityDirectory: """Indexed snapshot of the whole org's production-site activity.""" by_email: dict[str, AtlassianAccountActivity] by_account_id: dict[str, AtlassianAccountActivity] fetched_at: datetime total_accounts: int def lookup_by_emails( self, *emails: str | None, ) -> AtlassianAccountActivity | None: """Return the first account matching any of ``emails`` (case- insensitive). Caller passes work_email + aliases.""" for e in emails: if not e: continue hit = self.by_email.get(e.strip().lower()) if hit is not None: return hit return None def license_holders( self, product_family: str, sites: Iterable[str], ) -> list[AtlassianAccountActivity]: """Every **active** account that holds ``product_family`` on any of the (paid-production) ``sites``. Used by the **import / drift-revoke** side of the seat-sync (we only want to mirror seats for live employees on the paid site; sandbox holders and deactivated accounts should NOT result in active Helix allocations). For the **reconciliation count** — "how many seats does Atlassian's billing UI show?" — use :meth:`all_license_holders` instead, which mirrors the vendor's billing semantics (no status / no site filter). Crucially counts holders even when ``last_active`` is blank: a license is a license regardless of recency. ``sites`` is the plan's paid-production host list — passing ``["tidb.atlassian.net"]`` for Jira Software, or ``["pingcap-us.atlassian.net"]`` for the US Confluence subscription, yields exactly that subscription's import / drift-revoke set. """ norm = {normalize_site_host(s) for s in sites} norm.discard("") if not norm: return [] frozen = frozenset(norm) return [ a for a in self.by_account_id.values() if a.is_active and a.has_product_on_sites(product_family, frozen) ] def all_license_holders( self, product_family: str, ) -> list[AtlassianAccountActivity]: """Every account holding ``product_family`` ANYWHERE in the org, regardless of ``account_status`` or site. This is the count that matches Atlassian's Admin UI / billing page (which charges for every distinct user holding the product on the org's billed subscriptions — including ``inactive`` accounts whose offboarding hasn't fully released the seat yet, and accounts on sandbox / personal sites that marketplace addons like Tempo bill org-wide). Caller classifies each into one of six buckets via :meth:`classify_holder` — see that method's docstring for the bucket definitions and the identity rule. Use :meth:`license_holders` for the import / drift-revoke side (where we DO want the active-on-paid-site filter); use this method for the count side. """ return [ a for a in self.by_account_id.values() if a.access_sites_by_product.get(product_family) ] # ── Product / site classification ───────────────────────────────── def product_family_for_access_key(key: str | None) -> str | None: """Map an Atlassian ``product_access[].key`` to a tracked family. Atlassian product keys are stable strings (``jira-software``, ``jira-servicedesk``, ``confluence``, ``jira-core``, ``jira-product-discovery``, …). We match by substring rather than exact equality so a future key rename (``jira-servicedesk`` → ``jira-service-management``) keeps classifying correctly. Order matters: check service-desk and confluence before the broad ``software`` test so ``jira-servicedesk`` is not swallowed by a naive ``jira`` match. ``jira-core`` / discovery / unknown keys return ``None`` (out of scope for the three-piece set). """ k = (key or "").strip().lower() if not k: return None if ( "servicedesk" in k or "service-desk" in k or "service-management" in k or "servicemanagement" in k ): return PRODUCT_JIRA_SERVICEDESK if "confluence" in k: return PRODUCT_CONFLUENCE if "jira-software" in k or k == "software": return PRODUCT_JIRA_SOFTWARE return None def product_family_for_group(group_name: str | None) -> str | None: """Map a binding's ``group_name`` to a tracked product family. ``jira-software-users`` → jira-software, ``jira-servicedesk-users`` → jira-servicedesk, ``confluence-users`` → confluence. Same substring discipline as :func:`product_family_for_access_key`. Returns ``None`` for groups outside the three-piece set so the reclaim job skips plans it can't classify (rather than guessing). """ g = (group_name or "").strip().lower() if not g: return None if ( "servicedesk" in g or "service-desk" in g or "service-management" in g or "servicemanagement" in g ): return PRODUCT_JIRA_SERVICEDESK if "confluence" in g: return PRODUCT_CONFLUENCE if "software" in g: return PRODUCT_JIRA_SOFTWARE return None def normalize_site_host(url: str | None) -> str: """Reduce a ``product_access[].url`` to a bare lowercase host. Atlassian sometimes returns ``tidb.atlassian.net`` and sometimes ``https://tidb.atlassian.net/`` depending on API version — normalise both to ``tidb.atlassian.net`` so the production-site set membership test is robust. """ v = (url or "").strip().lower() if not v: return "" v = v.removeprefix("https://").removeprefix("http://") v = v.split("/", 1)[0].split("?", 1)[0].strip() return v.rstrip(".") def parse_last_active_date(value: Any) -> date | None: """Parse Atlassian's nanosecond UTC ``last_active`` to a ``date``. Takes ONLY the first 10 chars (``YYYY-MM-DD``) — feeding the full nanosecond string to ``datetime.fromisoformat`` raises. Blank / non-string / unparseable → ``None`` (treated as "undetermined" by callers, never as "idle"). """ if not value or not isinstance(value, str): return None head = value.strip()[:10] try: return date.fromisoformat(head) except ValueError: return None def _get_field(u: dict[str, Any], *keys: str) -> Any: """Read a field from an Atlassian user record, tolerating both the flat shape and the ``{"attributes": {...}}`` nested shape (the Org Admin API has shipped both over its lifetime).""" raw_attrs = u.get("attributes") attrs: dict[str, Any] = raw_attrs if isinstance(raw_attrs, dict) else {} for key in keys: if key in u and u[key] is not None: return u[key] if key in attrs and attrs[key] is not None: return attrs[key] return None def _compute_prod_last_active( product_access: Any, production_sites: dict[str, frozenset[str]], ) -> dict[str, date]: """For each tracked product family, the MAX ``last_active`` across that family's production sites. Sandbox / external sites are skipped because their host is not in the family's allowlist. """ result: dict[str, date] = {} if not isinstance(product_access, list): return result for entry in product_access: if not isinstance(entry, dict): continue family = product_family_for_access_key(entry.get("key")) if family is None: continue allowed = production_sites.get(family) or frozenset() host = normalize_site_host(entry.get("url")) if host not in allowed: continue d = parse_last_active_date(entry.get("last_active")) if d is None: continue cur = result.get(family) if cur is None or d > cur: result[family] = d return result def _compute_access_sites(product_access: Any) -> dict[str, frozenset[str]]: """For each tracked product family, the set of site hosts on which the account holds the product — regardless of ``last_active``. Unlike :func:`_compute_prod_last_active` this does NOT drop entries with a blank ``last_active`` (a holder is a holder) and does NOT filter to a production-site allowlist (the seat-count caller intersects with the plan's own ``site_urls``). Sandbox / external hosts are therefore *present* here but harmless — they never appear in a plan's paid- production ``site_urls`` so they're excluded at count time. """ out: dict[str, set[str]] = {} if not isinstance(product_access, list): return {} for entry in product_access: if not isinstance(entry, dict): continue family = product_family_for_access_key(entry.get("key")) if family is None: continue host = normalize_site_host(entry.get("url")) if not host: continue out.setdefault(family, set()).add(host) return {fam: frozenset(hosts) for fam, hosts in out.items()} def build_account_activity( user_record: dict[str, Any], production_sites: dict[str, frozenset[str]], ) -> AtlassianAccountActivity: """Parse one ``/admin/v1/orgs/{orgId}/users`` item.""" account_id = str( _get_field(user_record, "account_id", "id") or "", ).strip() raw_email = _get_field(user_record, "email", "emailAddress") email = str(raw_email).strip().lower() if raw_email else None account_status = _get_field(user_record, "account_status", "status") account_status = ( str(account_status).strip().lower() if account_status else None ) roles_raw = _get_field(user_record, "roles") or [] roles = tuple( str(r).strip().lower() for r in roles_raw if r ) if isinstance(roles_raw, list) else () product_access = _get_field(user_record, "product_access", "productAccess") prod_la = _compute_prod_last_active(product_access, production_sites) access_sites = _compute_access_sites(product_access) return AtlassianAccountActivity( account_id=account_id, email=email, account_status=account_status, roles=roles, prod_last_active_by_product=prod_la, access_sites_by_product=access_sites, raw=user_record, ) async def _load_production_sites( db: AsyncSession, ) -> dict[str, frozenset[str]]: """Production-site allowlist, defaults merged with any platform setting override. The override value is ``{"jira-software": ["host", …], "confluence": [...], …}``; missing families fall through to the hard defaults so a partial override can't accidentally widen an unspecified product to "all sites". """ from app.integrations.models import PlatformSetting sites: dict[str, frozenset[str]] = dict(DEFAULT_PRODUCTION_SITES) row = await db.get(PlatformSetting, _PRODUCTION_SITES_SETTING_KEY) if row is None or not isinstance(row.value, dict): return sites for family, hosts in row.value.items(): fam = str(family).strip().lower() if fam not in ALL_TRACKED_PRODUCTS: continue if not isinstance(hosts, list): continue normalized = { normalize_site_host(h) for h in hosts if isinstance(h, str) } normalized.discard("") if normalized: sites[fam] = frozenset(normalized) return sites async def fetch_atlassian_activity( db: AsyncSession, *, driver: JiraSoftwareDriver | None = None, ) -> AtlassianActivityDirectory: """Pull the whole org's directory and index it by email + account id. ``driver`` may be injected (tests / reuse); otherwise the primary Jira driver is resolved via the coordinator. Raises ``RuntimeError`` if no Jira integration is configured or the Admin API is unreachable — callers (the reclaim job) must NOT treat that as "everyone idle". """ if driver is None: from app.core.coordinators.jira_seat_coordinator import ( load_jira_driver, ) driver = await load_jira_driver(db) if driver is None: raise RuntimeError( "Jira integration not configured or not enabled — cannot " "fetch Atlassian activity directory.", ) production_sites = await _load_production_sites(db) users = await driver.list_org_users_with_activity() by_email: dict[str, AtlassianAccountActivity] = {} by_account_id: dict[str, AtlassianAccountActivity] = {} for u in users: acct = build_account_activity(u, production_sites) if acct.account_id: by_account_id[acct.account_id] = acct if acct.email: # Primary wins on the (rare) duplicate-email collision — # last write wins, which for managed accounts is stable. by_email[acct.email] = acct return AtlassianActivityDirectory( by_email=by_email, by_account_id=by_account_id, fetched_at=datetime.now(UTC), total_accounts=len(users), ) async def fetch_admin_account_ids( driver: JiraSoftwareDriver, group_names: list[str], ) -> set[str]: """Union of Atlassian accountIds in any of the named admin groups. The idle-reclaim job's hard guardrail: never auto-revoke a seat held by an Atlassian administrator. A group that doesn't exist on the site (404) is skipped silently — an org may not have created every conventional admin group. Any other failure (auth, 5xx) raises so the caller can refuse the whole reclaim tick rather than proceed with an incomplete admin set. """ admin_ids: set[str] = set() for group_name in group_names: name = (group_name or "").strip() if not name: continue try: members = await driver.list_group_members(name) except RuntimeError as e: if "not found" in str(e).lower() or "404" in str(e): logger.info( "admin group %r not found on site — skipping in " "idle-reclaim admin guard", name, ) continue raise for m in members: if not isinstance(m, dict): continue acct = str(m.get("accountId") or "").strip() if acct: admin_ids.add(acct) return admin_ids __all__ = [ "ALL_HOLDER_BUCKETS", "ALL_TRACKED_PRODUCTS", "DEFAULT_PRODUCTION_SITES", "HOLDER_EMPLOYEE_ACTIVE", "HOLDER_EXTERNAL", "HOLDER_LEAVER_RESIDUAL", "HOLDER_NO_EMAIL", "HOLDER_OFF_PAID_SITE", "HOLDER_VENDOR_INACTIVE", "PRODUCT_CONFLUENCE", "PRODUCT_JIRA_SERVICEDESK", "PRODUCT_JIRA_SOFTWARE", "AtlassianAccountActivity", "AtlassianActivityDirectory", "build_account_activity", "classify_holder", "fetch_admin_account_ids", "fetch_atlassian_activity", "normalize_site_host", "parse_last_active_date", "product_family_for_access_key", "product_family_for_group", ]