"""Per-vendor collectors for :mod:`app.integrations.usage_signals`. One function per provider that CAN answer "when did this person last use it?". Providers that cannot are declared in ``NO_USAGE_SIGNAL`` with a reason instead of getting a collector that fakes one. Each collector is responsible for exactly two things: pulling the raw roster, and turning each vendor's own last-login field into a ``(state, timestamp)`` pair via :func:`~app.integrations.usage_signals.parse_vendor_timestamp`. Matching to a Helix person, persistence, and the freshness gate are the caller's job — so a new vendor is one function, not a new pipeline. """ from __future__ import annotations import logging from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.integrations.usage_signals import ( SIGNAL_UNKNOWN, CollectorResult, UsageObservation, parse_vendor_timestamp, ) logger = logging.getLogger(__name__) #: Google and Zoom are single-product tenants from the seat model's point #: of view — one account, one last-login. Atlassian is not (a person can #: be a daily Confluence user and never open Jira), so it keys by product #: family. PRODUCT_GOOGLE_WORKSPACE = "workspace" PRODUCT_ZOOM = "zoom" async def collect_google_workspace( db: AsyncSession, driver: Any, ) -> CollectorResult: """Google Directory ``lastLoginTime``. Suspended accounts are included deliberately: a suspended account still holds its license until somebody removes it, so it is exactly the kind of seat this signal exists to surface. Beware the sentinel — Google returns ``1970-01-01T00:00:00.000Z`` for an account that has never signed in. ``parse_vendor_timestamp`` maps that to ``never`` rather than "56 years idle"; without that, every brand-new hire would be the most idle person in the tenant. """ out = CollectorResult() try: users = await driver.list_all_users(include_suspended=True) except Exception as exc: # noqa: BLE001 — reported, not swallowed out.error = f"google_workspace roster read failed: {exc}" return out for u in users or []: email = str(u.get("primaryEmail") or "").strip().lower() if not email: continue state, ts = parse_vendor_timestamp(u.get("lastLoginTime")) out.observations.append(UsageObservation( person_email=email, product_key=PRODUCT_GOOGLE_WORKSPACE, state=state, last_active_at=ts, evidence={ "raw_last_login": u.get("lastLoginTime"), "suspended": bool(u.get("suspended")), "org_unit": u.get("orgUnitPath"), # Admin guardrail input. Directory always returns both # flags, so ``False`` here is a real "not an admin", not # a missing field — which is what lets the reclaim job # require a POSITIVE determination before acting. "vendor_admin": bool(u.get("isAdmin")) or bool( u.get("isDelegatedAdmin"), ), "vendor_account_active": not bool(u.get("suspended")), }, )) return out async def collect_zoom(db: AsyncSession, driver: Any) -> CollectorResult: """Zoom ``last_login_time``. Both ``active`` and ``inactive`` statuses are pulled: an inactive Zoom user can still occupy a licensed seat, and the seat is the thing being reclaimed. Zoom truncates some fields in bulk listings depending on account plan (the eviction flow already compensates with a per-user ``get_user``). A truncated field arrives here as blank, which ``parse_vendor_timestamp`` maps to ``unknown`` — never to "idle" — so a plan-dependent truncation can only ever cost us a reclaim, not cause a wrong one. """ out = CollectorResult() users: list[dict[str, Any]] = [] for status in ("active", "inactive"): try: batch = await driver.list_users(status=status) except Exception as exc: # noqa: BLE001 out.error = f"zoom roster read failed ({status}): {exc}" return out users.extend(batch or []) for u in users: email = str(u.get("email") or "").strip().lower() if not email: continue state, ts = parse_vendor_timestamp(u.get("last_login_time")) out.observations.append(UsageObservation( person_email=email, product_key=PRODUCT_ZOOM, state=state, last_active_at=ts, evidence={ "raw_last_login": u.get("last_login_time"), "zoom_status": u.get("status"), "zoom_type": u.get("type"), "zoom_role_id": u.get("role_id"), "zoom_role_name": u.get("role_name"), # Fails CLOSED on purpose: only Zoom's explicit "Member" # role (role_id 2) counts as not-an-admin. An absent # role field, or any encoding this code does not # recognise, is treated as admin and therefore never # auto-reclaimed. Being wrong in that direction costs a # missed reclaim; being wrong the other way strips an # account owner's license. "vendor_admin": _zoom_is_admin(u), "vendor_account_active": ( str(u.get("status") or "").lower() == "active" ), }, )) return out def _zoom_is_admin(user: dict[str, Any]) -> bool: """True unless Zoom explicitly says this account is a plain Member. Zoom encodes the account role as ``role_id`` (``"2"`` = Member) and sometimes ``role_name``. Anything else — Owner, Admin, a custom role, or the field being absent entirely — is treated as an admin so the reclaim job leaves it alone. """ role_name = str(user.get("role_name") or "").strip().lower() if role_name: return role_name not in ("member", "members") role_id = str(user.get("role_id") or "").strip() if role_id: return role_id != "2" return True async def collect_atlassian(db: AsyncSession, driver: Any) -> CollectorResult: """Atlassian production-site activity, per product family. Delegates to :mod:`app.integrations.atlassian_activity` rather than re-deriving anything: that module already encodes the expensive lesson (take the MAX ``last_active`` across PAID PRODUCTION sites only — a sandbox's frozen timestamp made daily users look two-years-idle). Duplicating the computation here would be a second place for that bug to come back. A product family absent from ``prod_last_active_by_product`` means the account has no production access to it, so no row is emitted — absence of access is not absence of use. """ from app.integrations.atlassian_activity import ( ALL_TRACKED_PRODUCTS, fetch_atlassian_activity, ) out = CollectorResult() try: directory = await fetch_atlassian_activity(db, driver=driver) except Exception as exc: # noqa: BLE001 out.error = f"atlassian activity read failed: {exc}" return out for email, acct in directory.by_email.items(): clean = (email or "").strip().lower() if not clean: continue for product in ALL_TRACKED_PRODUCTS: if product not in acct.access_sites_by_product: continue day = acct.last_active_for(product) if day is None: # Holds the product but carries no parseable production # date — admins, API-only and mobile-only accounts land # here. Unknown, never idle. state, ts = SIGNAL_UNKNOWN, None else: state, ts = parse_vendor_timestamp(day.isoformat()) out.observations.append(UsageObservation( person_email=clean, product_key=product, state=state, last_active_at=ts, evidence={ "account_id": acct.account_id, "account_status": acct.account_status, "sites": sorted(acct.access_sites_by_product.get(product, ())), # Org/site admin only. The reclaim job ALSO unions # this with the configured admin-group membership it # fetches live, so this key is the floor, not the # whole guardrail. "vendor_admin": acct.is_org_or_site_admin, "vendor_account_active": acct.is_active, }, )) return out __all__ = [ "PRODUCT_GOOGLE_WORKSPACE", "PRODUCT_ZOOM", "collect_atlassian", "collect_google_workspace", "collect_zoom", ]