[PATCH v1 2/2] forge: add a script to manage teams on the forge

From: Richard Earnshaw <[email protected]>

This is a forge management script that is intended to be used
to manage most teams on the forge by using the forge API
to synchronize team membership with the information in
MAINTAINERS.yml.

At present the script will not create teams that do not
exist, this has to be done manually, but once that has been
done it can be used to synchronize the list of members.

The forge limits the ability to make changes to members of
the 'gcc' orgainzation Owners team, so don't expect to be
able to run this if you are not a member of that team.

contrib/ChangeLog:

	* forge_teams.py: New file.
---
 contrib/forge_teams.py | 417 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 417 insertions(+)
 create mode 100755 contrib/forge_teams.py

diff --git a/contrib/forge_teams.py b/contrib/forge_teams.py
new file mode 100755
index 0000000000000..b7c17779fff1a
--- /dev/null
+++ b/contrib/forge_teams.py
@@ -0,0 +1,417 @@
+#!/usr/bin/env python3
+
+# Manage maintainer teams on the forge
+
+# Copyright (C) 2026 Free Software Foundation, Inc.
+#
+# This file is part of GCC.
+#
+# GCC is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3, or (at your option)
+# any later version.
+#
+# GCC is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with GCC; see the file COPYING.  If not, write to
+# the Free Software Foundation, 51 Franklin Street, Fifth Floor,
+# Boston, MA 02110-1301, USA.
+
+# This script is used to syncronize the data in the MAINTAINERS.yml
+# file with the teams used by the forge for notifications.
+
+# The script is set up with all defaults for the sourceware forge
+# except for an API token.  The token's minimum permissions are
+#   write:organization
+# but only owners of the 'gcc' organization will be able to generate
+# useable write tokens.
+# You can enter the API token either via the environment variable
+# APIKEY, or when requested.
+
+
+import getpass
+import json
+import math
+import os
+import pprint
+import sys
+
+import urllib.parse
+import urllib.request
+import http.client
+
+from optparse import OptionParser
+
+import maintainer_utils as maintutils
+
+defaults = {
+    "PROJECT": "gcc",
+    "REPO": "gcc",
+    "FORGE": "https://forge.sourceware.org/api/v1",
+}
+
+# These are teams in the forge that this script will not attempt to manage
+# because the relevant data is not part of MAINTAINERS.yml
+unmanaged_teams = [
+    'Automation',
+    'Collaborators',
+    'Maintainers.Global',   # Remove once fully implemented
+    'Owners',
+    'Release.Managers',
+    'Reviewers',            # Remove once fully implemented
+    'Write.After.Approval', # Remove once fully implemented
+]
+
+opts = None
+error_count = 0
+
+
+def verbose(msg) -> None:
+    if opts.verbose:
+        print(msg)
+    return
+
+
+class Forge:
+    def __init__(self):
+        self._setup()
+
+    def _setup(self) -> None:
+        self.project = os.getenv("PROJECT", defaults['PROJECT'])
+        self.repo = os.getenv("REPO", defaults['REPO'])
+        self.forge = os.getenv("FORGE", defaults['FORGE'])
+        self.apikey = os.getenv("APIKEY") or getpass.getpass(
+            prompt="API key: "
+        )
+        verbose(f"Accessing {self.forge}/ for org {self.project}.")
+        return
+
+    def _send_request(
+        self,
+        url,
+        method='GET',
+    ) -> http.client.HTTPResponse:
+        # For safety while developing.
+        if method == 'GET' or not opts.dry_run:
+            headers = {
+                'Authorization': f"token {self.apikey}",
+                'accept': 'application/json',
+                'Content-Type': 'application/json',
+            }
+            request = urllib.request.Request(
+                url,
+                headers=headers,
+                method=method,
+            )
+            return urllib.request.urlopen(request)
+        return None
+
+    # The rest API does not guarantee to return all the data with a
+    # single request.  We support fetching by pages of up to limit.
+    # Note that the server may return less than this if increased too
+    # far. 50 seems a reasonable size; we don't expect there to be
+    # more than a couple of pages at that size.
+    def _get_teams_page(self, page=1, limit=50) -> tuple[
+        list,
+        int,
+    ]:
+        fetch = urllib.parse.urlencode(
+            {
+                "page": page,
+                "limit": limit,
+            }
+        )
+        url = f"{self.forge}/orgs/{self.project}/teams?{fetch}"
+        verbose(f"Requesting {url}")
+        reply = self._send_request(url)
+        item_count = int(reply.headers.get("X-Total-Count", -1))
+        pages = math.ceil(item_count / limit) if item_count > 0 else 1
+        return json.load(reply), pages
+
+    def _get_team_members_page(self, id, page=1, limit=50) -> tuple[
+        list,
+        int,
+    ]:
+        fetch = urllib.parse.urlencode(
+            {
+                "page": page,
+                "limit": limit,
+            }
+        )
+        url = f"{self.forge}/teams/{id}/members?{fetch}"
+        verbose(f"GET {url}")
+        reply = self._send_request(url)
+        item_count = int(reply.headers.get("X-Total-Count", -1))
+        pages = math.ceil(item_count / limit) if item_count > 0 else 1
+        return json.load(reply), pages
+
+    def fetch_teams(self) -> list:
+        teams, pages = self._get_teams_page()
+        if pages == 1:
+            return teams
+        for page in range(2, pages + 1):
+            teams_page, _ = self._get_teams_page(page=page)
+            teams.extend(teams_page)
+        return teams
+
+    def fetch_team_members(self, id) -> list:
+        members, pages = self._get_team_members_page(id)
+        if pages == 1:
+            return members
+        for page in range(2, pages + 1):
+            members_page, _ = self._get_team_members_page(id, page=page)
+            members.extend(members_page)
+        return members
+
+    def manage_team_member(self, id, login, remove=False) -> None:
+        url = f"{self.forge}/teams/{id}/members/{login}"
+        action = 'DELETE' if remove else 'PUT'
+        verbose(f"{action} {url}")
+        reply = self._send_request(url, method=action)
+
+
+class DesiredTeam:
+    def __init__(
+        self,
+        teamname,
+        role_filter,
+        user_data,
+        subsystem=None,
+    ) -> None:
+        self.name = teamname
+        self.members = []
+        self.accounts = set()
+        self.role_filter = role_filter
+        self.add_users(subsystem, user_data)
+
+    @staticmethod
+    def subsystem_filter(subsystem_name, roles) -> bool:
+        return any(
+            isinstance(role, dict)
+            and (
+                role.get('Maintainer') == subsystem_name
+                or role.get('Reviewer') == subsystem_name
+            )
+            for role in roles
+        )
+
+    @staticmethod
+    def maintainers_filter(subsystem_name, roles) -> bool:
+        return any(
+            role == 'Global'
+            or (isinstance(role, dict) and 'Maintainer' in role)
+            for role in roles
+        )
+
+    @staticmethod
+    def global_reviewers_filter(subsystem_name, roles) -> bool:
+        return any(
+            role == 'Global'
+            for role in roles
+        )
+
+    @staticmethod
+    def reviewers_filter(subsystem_name, roles) -> bool:
+        return any(
+            isinstance(role, dict) and 'Reviewer' in role
+            for role in roles
+        )
+
+    @staticmethod
+    def write_after_filter(subsystem_name, roles) -> bool:
+        return any(
+            role == "WriteAfter"
+            for role in roles
+        )
+
+    def add_users(self, subsystem_name, user_data) -> None:
+        new_members = [
+            u
+            for u in user_data
+            if (forgeid := u.get("forgeid")) is not None
+            and forgeid not in self.accounts
+            and self.role_filter(subsystem_name, u['roles'])
+        ]
+        self.members.extend(new_members)
+        self.accounts.update(u["forgeid"] for u in new_members)
+
+
+class DesiredTeamList:
+    def __init__(self, data) -> None:
+        # Start with the static teams that are not based on specific
+        # subsytems.
+        self.teams = {
+            'Maintainers.Global': DesiredTeam(
+                "Maintainers.Global",
+                DesiredTeam.global_reviewers_filter,
+                data['users'],
+            ),
+            'Maintainers': DesiredTeam(
+                "Maintainers", DesiredTeam.maintainers_filter, data['users']
+            ),
+            'Reviewers': DesiredTeam(
+                "Reviewers", DesiredTeam.reviewers_filter, data['users']
+            ),
+            'Write.After.Approval': DesiredTeam(
+                "Write.After.Approval",
+                DesiredTeam.write_after_filter,
+                data['users'],
+            ),
+        }
+        # Now add teams for subsystems where a forge team has been
+        # set up.  This is a many<->many mapping: some subsystems
+        # will impact multiple teams and some teams my cover multiple
+        # subsystems and there may be overlaps in team membership.
+        for subsystem in filter(
+                lambda s: 'teams' in s,
+                data['subsystems']
+        ):
+            for t in subsystem['teams']:
+                if t in self.teams:
+                    self.teams[t].add_users(
+                        subsystem['name'],
+                        data['users'],
+                    )
+                else:
+                    self.teams[t] = DesiredTeam(
+                        t,
+                        DesiredTeam.subsystem_filter,
+                        data['users'],
+                        subsystem=subsystem['name'],
+                    )
+
+    def get_by_name(self, name) -> DesiredTeam:
+        return self.teams.get(name)
+
+
+class ExistingTeam:
+    def __init__(self, forge, team) -> None:
+        self._forge = forge
+        self._team = team
+        self._members = {
+            m['login']: m
+            for m in forge.fetch_team_members(team['id'])
+        }
+
+    def get_member_names(self) -> list:
+        return self._members.keys()
+
+    def has_member(self, name) -> bool:
+        return self._members.get(name) != None
+
+    def add_member(self, name) -> None:
+        self._forge.manage_team_member(self._team['id'], name)
+
+    def remove_member(self, name) -> None:
+        self._forge.manage_team_member(self._team['id'], name, remove=True)
+
+
+class ExistingMaintainers:
+    def __init__(self, forge):
+        self._forge = forge
+        self._teams = {
+            t['name']: ExistingTeam(forge, t)
+            for t in forge.fetch_teams()
+            if not t['name'] in unmanaged_teams
+        }
+
+    def get_by_name(self, name) -> ExistingTeam:
+        return self._teams.get(name)
+
+    def get_team_names(self) -> list:
+        return self._teams.keys()
+
+
+def update_teams(forge, needed_teams, existing_teams) -> None:
+    global error_count
+    for name, team in needed_teams.teams.items():
+        # Skip any teams we don't manage
+        if name in unmanaged_teams:
+            continue
+        existing = existing_teams.get_by_name(name)
+        if existing:
+            verbose(f"Checking membership of {name}")
+            # First add any missing members; while doing this, build up
+            # a set of forgeids in the team so that we can quickly search
+            # the team list for pass 2
+            member_names = set()
+            for member in team.members:
+                member_names.add(member['forgeid'])
+                if not existing.has_member(member['forgeid']):
+                    # Add new member
+                    verbose(
+                        f"  missing: {member['cn']} ({member['forgeid']})"
+                    )
+                    existing.add_member(member['forgeid'])
+                else:
+                    verbose(f"  OK: {member['cn']} ({member['forgeid']})")
+            # Secondly, remove any excess members
+            for member in existing.get_member_names():
+                if not member in member_names:
+                    verbose(f"  remove: {member}")
+                    existing.remove_member(member)
+        else:
+            error_count += 1
+            print(f"Team '{name}' does not exist on the forge")
+    # Scan the existing teams and report any that do not exist in
+    # either MAINTAINERS.yml or in the ignore list.
+    for existing in existing_teams.get_team_names():
+        if (
+            not needed_teams.get_by_name(existing)
+            and existing not in unmanaged_teams
+        ):
+            error_count += 1
+            if len(existing_teams.get_by_name(existing).get_member_names()):
+                print(f"Unexpected team '{existing}' found on forge")
+            else:
+                print(
+                    f"Forge team '{existing}' has no members and can be removed"
+                )
+
+
+def main() -> int:
+    global opts
+    optp = OptionParser("Usage: %prog [<options>] <maintainers.yml>")
+    optp.add_option(
+        "-v", "--verbose",
+        action="store_true",
+        dest="verbose",
+        default=False,
+    )
+    optp.add_option(
+        "-n", "--dry-run",
+        action="store_true",
+        dest="dry_run",
+        default=False,
+    )
+    opts, args = optp.parse_args()
+    if len(args) != 1:
+        optp.print_help()
+        return 1
+
+    maint_data = maintutils.load(args[0])
+    maintutils.validate(maint_data)
+    needed_teams = DesiredTeamList(maint_data)
+    forge = Forge()
+    existing_teams = ExistingMaintainers(forge)
+    update_teams(forge, needed_teams, existing_teams)
+    if opts.verbose:
+        print("Existing managed teams and members:")
+        for team in existing_teams.get_team_names():
+            print(team)
+            for m in existing_teams.get_by_name(team).get_member_names():
+                print(f"  {m}")
+        print("\nDesired teams and members:")
+        for name, team in needed_teams.teams.items():
+            print(name)
+            for m in team.members:
+                print(f"  {m['forgeid']}")
+    return 0 if not error_count else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
-- 
2.54.0