"""自动加入规则:哪些人天然就该在哪个 GitHub 组织里。 规则的 CRUD、命中预览,以及入职时的执行。匹配语义在 ``app/core/github_org_match``(部门按**子树**、按路径段比较),放在 core 是因为入职协调器和这个域都要用同一套判断,而域模块之间不许互相 import。 ## 三道闸,顺序不能换 一条规则命中一个人,并**不等于**现在就该把他加进去: 1. **组织级总闸** ``github_orgs.auto_join_enabled``。规则表可能配了但 组织没开 —— 这让「先把规则配好、确认命中范围对了再开闸」成为可能, 而不是配完的一瞬间就开始给人发邀请。 2. **人员资格** ``is_blocked_from_new_grant``:待离职/已离职/软删/HRIS 墓碑一律不给。入职钩子理论上碰不到这些人,但规则也会被「重跑对账」 调用,那条路上会。 3. **handle 必须有**。这是最常见的拦截点:生产上在职员工只有 63% 登记了 GitHub handle。**没有 handle 不是跳过,是一件待办** —— 静默跳过意味着 这个人永远不会被自动加入,而且没有任何地方说过这件事。 第 3 点的处理是这个模块最重要的设计:缺 handle 时**发一条通知给本人** (去补登记),并把这次「本来该加入但没能加入」记进审计。补上之后,下一次 入职钩子或对账会自然把他加进去 —— 不需要任何人记得回来处理。 """ from __future__ import annotations import logging import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import ConflictError, NotFoundError, ValidationError from app.core.github_org_match import rule_matches_person, split_department_path from app.identity.person_models import IdentityPerson from . import constants as const from . import execution from . import service as svc from .models import GithubOrg, GithubOrgAutoJoinRule from .validators import bounded_text logger = logging.getLogger(__name__) def _now() -> datetime: return datetime.now(UTC) # ════════════════════════════════════════════════════════════════════ # 规则 CRUD # ════════════════════════════════════════════════════════════════════ async def list_rules( db: AsyncSession, *, org_id: str | None = None, include_inactive: bool = True, ) -> list[GithubOrgAutoJoinRule]: q = select(GithubOrgAutoJoinRule).where( GithubOrgAutoJoinRule.deleted_at.is_(None), ) if org_id: q = q.where(GithubOrgAutoJoinRule.org_id == str(org_id)) if not include_inactive: q = q.where(GithubOrgAutoJoinRule.is_active.is_(True)) return list( (await db.execute(q.order_by(GithubOrgAutoJoinRule.created_at))) .scalars().all() ) async def get_rule(db: AsyncSession, rule_id: str) -> GithubOrgAutoJoinRule: row = await db.get(GithubOrgAutoJoinRule, str(rule_id)) if row is None or row.deleted_at is not None: raise NotFoundError("GithubOrgAutoJoinRule", str(rule_id)) return row def _normalize_department_path(raw: str | None) -> str | None: """把部门路径归一成「剥根 + 单空格斜杠」的存储形态。 管理员从组织树上复制过来的路径可能带租户根、可能空格不一致。不归一的 后果不是报错,而是一条**永远匹配不上**的规则 —— 那种沉默的失败比一次 校验错误糟糕得多。 """ segments = split_department_path(raw) return " / ".join(segments) if segments else None async def create_rule( db: AsyncSession, *, org_id: str, match_department_path: str | None = None, match_job_title: str | None = None, match_location: str | None = None, notes: str | None = None, is_active: bool = True, actor_person_id: str | None = None, ) -> GithubOrgAutoJoinRule: org = await svc.get_org(db, org_id) dept = _normalize_department_path(match_department_path) title = bounded_text(match_job_title, 128, "match_job_title") location = bounded_text(match_location, 64, "match_location") if not (dept or title or location): # 三个维度全空 = 命中所有人。这几乎一定是误配,而后果是把全公司 # 加进一个 GitHub 组织。要真想这么做,得显式地写一条通配部门规则 # (例如整个租户根下的第一级),而不是留三个空。 raise ValidationError( "an auto-join rule must constrain at least one of " "department / job title / location — a rule with none would " "match every employee", ) existing = await list_rules(db, org_id=org.id) for row in existing: if ( row.match_department_path == dept and row.match_job_title == title and row.match_location == location ): raise ConflictError( "an identical auto-join rule already exists for this " "organization", ) rule = GithubOrgAutoJoinRule( id=str(uuid.uuid4()), org_id=org.id, match_department_path=dept, match_job_title=title, match_location=location, is_active=bool(is_active), notes=bounded_text(notes, 2000, "notes"), created_by=str(actor_person_id) if actor_person_id else None, ) db.add(rule) await db.flush() return rule async def update_rule( db: AsyncSession, rule: GithubOrgAutoJoinRule, *, changes: dict[str, Any], ) -> dict[str, dict[str, Any]]: """PATCH 语义,返回按值的差异(审计要按值记,不是按字段名)。""" diff: dict[str, dict[str, Any]] = {} def _apply(field: str, new: Any) -> None: old = getattr(rule, field) if old != new: diff[field] = {"from": old, "to": new} setattr(rule, field, new) if "match_department_path" in changes: _apply( "match_department_path", _normalize_department_path(changes["match_department_path"]), ) if "match_job_title" in changes: _apply( "match_job_title", bounded_text(changes["match_job_title"], 128, "match_job_title"), ) if "match_location" in changes: _apply( "match_location", bounded_text(changes["match_location"], 64, "match_location"), ) if "is_active" in changes: _apply("is_active", bool(changes["is_active"])) if "notes" in changes: _apply("notes", bounded_text(changes["notes"], 2000, "notes")) if not ( rule.match_department_path or rule.match_job_title or rule.match_location ): raise ValidationError( "an auto-join rule must constrain at least one dimension", ) return diff async def delete_rule(db: AsyncSession, rule: GithubOrgAutoJoinRule) -> None: """软删。台账上引用它的成员记录不动 —— 规则消失不该把已经发生的事实 一起带走(FK 是 ``SET NULL``,来源仍记在 ``source`` 上)。""" rule.deleted_at = _now() rule.is_active = False # ════════════════════════════════════════════════════════════════════ # 命中 # ════════════════════════════════════════════════════════════════════ def _person_segments(person: IdentityPerson) -> list[str]: return list(person.department_chain or []) async def rules_matching_person( db: AsyncSession, person: IdentityPerson, ) -> list[GithubOrgAutoJoinRule]: """命中这个人的所有**启用**规则(不判组织总闸,那是执行时的事)。""" segments = _person_segments(person) return [ rule for rule in await list_rules(db, include_inactive=False) if rule_matches_person( rule, department_segments_of_person=segments, job_title=person.job_title, location=person.work_location_code, ) ] async def preview_rule( db: AsyncSession, rule: GithubOrgAutoJoinRule, *, limit: int = 200, ) -> dict[str, Any]: """这条规则现在覆盖哪些在职员工。 在 Python 里遍历而不是拼 SQL:部门子树按**路径段**比较,而 ``department_chain`` 是一个 Python 属性。为它写第二套 SQL 前缀匹配, 等于让预览和真正的执行用两套语义 —— 而它们迟早会漂移(裸前缀匹配会让 ``Engineering`` 吃掉 ``Engineering Support``,那正是对外 API 修过一次 的坑)。574 个人在内存里过一遍是毫秒级的。 """ from sqlalchemy.orm import selectinload people = list( ( await db.execute( select(IdentityPerson) .where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ) .options(selectinload(IdentityPerson.org_memberships)) ) ).scalars().all() ) matched = [ p for p in people if rule_matches_person( rule, department_segments_of_person=_person_segments(p), job_title=p.job_title, location=p.work_location_code, ) ] without_handle = [p for p in matched if not (p.github_id or "").strip()] return { "match_count": len(matched), # 「命中了但没登记 handle」是这条规则真正的落地率缺口 —— 生产上 # 在职员工只有 63% 有 handle,所以它必须和命中数一起显示,否则 # 管理员会以为配好规则就等于这些人都会进组。 "without_handle_count": len(without_handle), "sample": [ { "person_id": str(p.id), "display_name": p.display_name, "work_email": p.work_email, "department": p.primary_department, "github_id": p.github_id or None, } for p in matched[:limit] ], "truncated": len(matched) > limit, } # ════════════════════════════════════════════════════════════════════ # 执行 # ════════════════════════════════════════════════════════════════════ #: 缺 handle 的处置结果,与「加成功了」「跳过了」分开计数。 APPLY_JOINED = "joined" APPLY_ALREADY = "already" APPLY_NO_HANDLE = "no_handle" APPLY_ORG_DISABLED = "org_disabled" APPLY_INELIGIBLE = "ineligible" APPLY_FAILED = "failed" async def apply_rules_for_person( db: AsyncSession, person: IdentityPerson, *, trace_id: str, notify_missing_handle: bool = True, ) -> dict[str, Any]: """把所有命中的规则作用到一个人身上。 幂等:已经在组织里的人不会被重复邀请(执行层按 ``(org, handle)`` 收敛)。可以安全地重复调用 —— 入职钩子、hire-day cron、以及管理员 手工重跑都会调它。 """ from app.core.grant_eligibility import ( # noqa: PLC0415 is_blocked_from_new_grant, ) summary: dict[str, Any] = { "person": person.work_email, APPLY_JOINED: 0, APPLY_ALREADY: 0, APPLY_NO_HANDLE: 0, APPLY_ORG_DISABLED: 0, APPLY_INELIGIBLE: 0, APPLY_FAILED: 0, "orgs": [], } if is_blocked_from_new_grant(person): summary[APPLY_INELIGIBLE] = 1 return summary rules = await rules_matching_person(db, person) if not rules: return summary handle = (person.github_id or "").strip() audit = AuditService(db) orgs_seen: set[str] = set() for rule in rules: if rule.org_id in orgs_seen: # 多条规则命中同一个组织是正常的(部门规则 + 职位规则)—— # 取并集,不重复执行。 continue orgs_seen.add(rule.org_id) org = await db.get(GithubOrg, rule.org_id) if org is None or org.deleted_at is not None or not org.is_active: continue if not org.auto_join_enabled: # 组织级总闸没开。这让「先配规则、确认命中范围、再开闸」成为 # 可能,而不是配完的一瞬间就开始发邀请。 summary[APPLY_ORG_DISABLED] += 1 continue if not handle: summary[APPLY_NO_HANDLE] += 1 summary["orgs"].append( {"org": org.slug, "result": APPLY_NO_HANDLE}, ) # 🔴 **不静默跳过。** 没有 handle 意味着这个人永远不会被自动 # 加入,而且没有任何地方说过这件事。记审计 + 通知本人去补登记; # 补上之后下一次钩子或对账会自然把他加进去。 await audit.log( trace_id=trace_id, actor_type="system", action="github_org.auto_join.blocked_no_handle", status="failure", resource_type="identity_person", resource_id=str(person.id), detail={ "org": org.slug, "rule_id": rule.id, "reason": "person has no github_id on file", }, ) continue # 按**人**查,不按 handle 查。一个换过 GitHub 账号的人,按新 # handle 查不到他用旧 handle 建立的那条成员资格,于是会被再邀请 # 一次 —— 同一个组织里留下这个人的两条记录。「他已经有访问权了 # 吗」这个问题的答案永远按人算。 existing = await svc.find_person_membership_in_org( db, org_id=org.id, person_id=str(person.id), ) if existing is not None: summary[APPLY_ALREADY] += 1 entry: dict[str, Any] = { "org": org.slug, "result": APPLY_ALREADY, } if existing.github_login_lower != handle.lower(): # 他在里面用的是另一个 handle。不自动去撤旧的 —— 那是一个 # 需要人判断的动作(旧账号可能还在用)。但必须说出来, # 否则台账上会有一条谁也不会去看的陈旧记录。 entry["handle_mismatch"] = { "in_org_as": existing.github_login, "now_registered_as": handle, } await audit.log( trace_id=trace_id, actor_type="system", action="github_org.auto_join.handle_mismatch", status="success", resource_type="github_org_membership", resource_id=existing.id, detail={ "org": org.slug, "person": person.work_email, "in_org_as": existing.github_login, "now_registered_as": handle, }, ) summary["orgs"].append(entry) continue outcome = await execution.execute_approved( db, org=org, github_login=handle, person=person, source=const.SOURCE_AUTO_JOIN, trace_id=trace_id, auto_join_rule_id=rule.id, ) if outcome.outcome == execution.OUTCOME_FAILED: summary[APPLY_FAILED] += 1 summary["orgs"].append( { "org": org.slug, "result": APPLY_FAILED, "error": outcome.error, }, ) else: summary[APPLY_JOINED] += 1 summary["orgs"].append( {"org": org.slug, "result": outcome.outcome}, ) if summary[APPLY_NO_HANDLE] and notify_missing_handle: await _notify_missing_handle(db, person=person, summary=summary) return summary async def _notify_missing_handle( db: AsyncSession, *, person: IdentityPerson, summary: dict[str, Any], ) -> None: """告诉本人:有组织本该自动把你加进去,但我们没有你的 GitHub 用户名。 尽力而为 —— 通知失败不该让整次入职钩子失败。 """ orgs = [ o["org"] for o in summary.get("orgs", []) if o.get("result") == APPLY_NO_HANDLE ] if not orgs or person.employment_status != "active": return try: from app.notifications.service import ( # noqa: PLC0415 NotificationEnvelope, NotificationService, ) _org_lines = "\n".join(f" · {o}" for o in orgs) await NotificationService(db).dispatch( NotificationEnvelope( recipient=person, title="补一下 GitHub 用户名,就能自动加入这些组织", body_text=( "按你所在部门的规则,你本该被自动加入以下 GitHub 组织:\n" + _org_lines + "\n\n但我们这边没有你的 GitHub 用户名,所以这一步没能" "完成。到 Helix 的「个人设置 → 基本信息」登记一下 " "GitHub ID(提交后由你的 HR Ops 确认),之后同类规则" "会自动生效。" ), template_key="github_org_missing_handle", # 迁移 579 之前这里指的是 ``/me/github``,而文案说「申请一次 # 就顺便把用户名登记上了」—— 那句话当时并不成立:加入申请单 # 刻意不回写 ``identity_persons.github_id``(会被下一次 HRIS # 同步盖掉)。现在有了真正的登记入口,指过去。 url="/me/settings?tab=profile", # PRE-FORMATTED, like ``{software_list}`` on the hire-day # template. The registered template renders ``{orgs}`` # through ``format_map``; handing it the raw list would # print a Python repr — ``['tidb', 'pingcap']`` — into an # employee-facing notification. context={"orgs": _org_lines}, card_subtitle="差一个 GitHub 用户名", related_resource_type="github_org", ) ) except Exception as e: # noqa: BLE001 logger.warning( "github_org: missing-handle notification failed for %s: %s", person.work_email, e, )