"""机器账号的策略标记 —— 建议 + 批量,永不自动落库。 ## 为什么需要它,以及为什么它只能是「策略」 台账里有 17 条成员资格属于 8 个机器账号(``sre-bot`` / ``ti-chi-bot`` / ``ti-srebot`` / ``tidb-ci-bot`` / ``tidbcloud-bot`` / ``pingcap-zendesk-bot`` / ``TestInfra-Bot`` / ``ti-community-*``)。它们在 「认不出是谁」里占着位置,而它们**永远也认不出来** —— 那不是一个待办, 那是一个分类。 第一反应是按 GitHub 的 ``type`` 字段判定,权威又省事。**实测不行**:上面 七个 handle 逐个查 ``GET /users/{login}``,**全部报 ``type=User``**。 GitHub 的 ``Bot`` 类型只给 GitHub App 身份(``dependabot[bot]`` 那种), 而这些都是人建的普通账号拿来当机器人用 —— 那是这家公司的惯例,不是 GitHub 的概念。 所以判定只能来自命名约定和资料名,而两者都是启发式。**因此这里只产出 建议和批量入口,写入永远经过人确认**:把一个真名叫 ``robot-liu`` 的员工 标成机器账号,后果是他离职时不被回收 —— 和挂错人一样,是一次静默的 权限泄漏。 ## 策略存在 platform_settings,不写死在代码里 新的 bot 命名惯例会出现(``*-srebot`` 就是后来才有的),而那不该等一次 发布。默认值在迁移里播种,``github_admin`` 可以改。 """ from __future__ import annotations import fnmatch import logging from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from . import constants as const from .models import GithubOrg, GithubOrgMembership logger = logging.getLogger(__name__) SETTING_KEY = "github_org.bot_handle_patterns" #: 迁移播种的同一份默认值。代码里也留一份,是为了让「设置行被删掉」 #: 退化成「用默认策略」而不是「策略为空、一个都不匹配」。 DEFAULT_PATTERNS: tuple[str, ...] = ( "*-bot", "bot-*", "*-bot-*", "*bot", "*robot*", "*-ci-bot", "*-srebot", ) #: 资料名里出现这些词,也算一条**较弱**的证据(``sre-bot`` 的 name 是 #: 「SRE bot」,``ti-chi-bot`` 的是「Ti Chi Robot」)。它单独不足以建议, #: 只在 handle 也命中时用来提高说服力 —— 界面上显示的是理由。 _NAME_HINTS = ("bot", "robot", "automation", "ci ") async def get_patterns(db: AsyncSession) -> list[str]: """当前的 bot 命名策略。 ⚠️ **从 586 起,这里不再读自己那个设置项,而是从标签规则里派生** —— 取所有 ``match=handle_glob`` 且 ``label=account:bot`` 的规则。 为什么改:586 引入了通用的标签规则表,而 bot 命名约定本来就是它的一个 特例。两个存储并存的必然结果是它们会不一致 —— 有人在标签规则里加了 ``*-runner``,而 bot 卡片还按旧的设置项算,于是同一个问题两个界面给 两个答案,然后谁都不敢信。所以合成一个来源,bot 这一层只是一次筛选。 旧设置项 ``github_org.bot_handle_patterns`` 留在库里不删(它是 585 的 播种数据,删了 downgrade 就不闭合),但**不再被读取**。 """ from . import labels as _labels # noqa: PLC0415 循环依赖:labels 用 svc rules = await _labels.get_rules(db) pats = [ str(r["value"]).strip() for r in rules if r.get("match") == _labels.MATCH_HANDLE_GLOB and r.get("label") == "account:bot" and str(r.get("value") or "").strip() ] if pats: return pats logger.warning( "github_org: no account:bot handle_glob rules configured — " "falling back to the built-in default patterns", ) return list(DEFAULT_PATTERNS) async def set_patterns( db: AsyncSession, patterns: list[str], *, actor_person_id: str | None, ) -> list[str]: """改策略。空列表被拒绝 —— 见 :func:`get_patterns` 关于「空策略等于 静默失效」。""" cleaned = [str(p).strip() for p in patterns if str(p).strip()] if not cleaned: raise ValueError("bot handle patterns must not be empty") # 写回**标签规则**这一个来源:替换掉其中 account:bot 的 handle_glob # 那几条,其余规则(employer / community 之类)原样保留。 from . import labels as _labels # noqa: PLC0415 # ``account:bot`` 是这条代码路径**硬依赖**的那一个标签 # (``ACCOUNT_LABEL_TO_KIND`` 就靠它派生 external_kind)。 # 迁移 586 会播种它,但这里仍然做一次幂等 ensure —— 否则调用方明明 # 只提交了「几个命名模式」,却会收到一条「规则指向了一个不存在的标签」 # 的报错,而那个标签他从没提过。**报错必须指向调用方真正做错的事**, # 而这件事调用方没做错。 if await _labels.get_label(db, "account:bot") is None: await _labels.upsert_label( db, key="account:bot", category="account", display_zh="机器账号", display_en="Machine account", description=( "不是人,是自动化用的账号。要问的是「这个自动化还在用吗、" "token 谁管」。" ), actor_person_id=actor_person_id, ) existing = await _labels.get_rules(db) kept = [ r for r in existing if not ( r.get("match") == _labels.MATCH_HANDLE_GLOB and r.get("label") == "account:bot" ) ] merged = kept + [ {"match": _labels.MATCH_HANDLE_GLOB, "value": p, "label": "account:bot"} for p in cleaned ] await _labels.set_rules(db, merged) return cleaned def match_reason( login: str, patterns: list[str], profile: dict[str, Any] | None, ) -> str | None: """这个 handle 为什么像机器账号。返回一句中文理由或 ``None``。 界面上显示的是理由而不是「匹配了」—— 人要凭理由判断,而不是凭一个 布尔值。 """ low = (login or "").lower() if not low: return None hit = next((p for p in patterns if fnmatch.fnmatch(low, p.lower())), None) if hit is None: return None reason = f"用户名符合机器账号命名约定({hit})" name = str((profile or {}).get("name") or "").lower() if name and any(h in name for h in _NAME_HINTS): reason += f",且资料名是「{(profile or {}).get('name')}」" acct = str((profile or {}).get("account_type") or "") if acct.lower() == "bot": # 这条在本租户里从没命中过(七个 bot 全报 User),但 GitHub App # 身份确实会是 Bot —— 留着,并且让它压过命名约定。 reason = "GitHub 账号类型就是 Bot(GitHub App 身份)" return reason async def preview( db: AsyncSession, *, patterns: list[str] | None = None, ) -> dict[str, Any]: """按策略扫一遍「还没认出是谁」的成员资格,列出像机器账号的那些。 **只读**。返回按 handle 聚合的候选,每条带理由和它涉及的组织 —— 一个 bot 通常在多个组织里(``ti-chi-bot`` 在 5 个),一次确认应该 覆盖它的所有行。 """ pats = patterns if patterns is not None else await get_patterns(db) rows = list( ( await db.execute( select(GithubOrgMembership, GithubOrg) .join(GithubOrg, GithubOrg.id == GithubOrgMembership.org_id) .where( GithubOrgMembership.person_id.is_(None), GithubOrgMembership.external_confirmed_at.is_(None), GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES), GithubOrg.deleted_at.is_(None), ) .order_by(GithubOrgMembership.github_login) ) ).all() ) by_handle: dict[str, dict[str, Any]] = {} for m, org in rows: prof = m.github_profile if isinstance(m.github_profile, dict) else None reason = match_reason(m.github_login, pats, prof) if reason is None: continue e = by_handle.setdefault( m.github_login_lower, { "handle": m.github_login, "reason": reason, "membership_ids": [], "org_slugs": [], "profile_name": (prof or {}).get("name"), }, ) e["membership_ids"].append(m.id) e["org_slugs"].append(org.slug) items = sorted(by_handle.values(), key=lambda x: x["handle"].lower()) for it in items: it["org_slugs"] = sorted(set(it["org_slugs"])) return { "patterns": pats, "items": items, "handles": len(items), "memberships": sum(len(i["membership_ids"]) for i in items), }