contrib/forge_teams.py

#!/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():
            unmanaged = " (unmanaged)" if name in unmanaged_teams else ""
            print(f"{name}{unmanaged}")
            for m in team.members:
                print(f"  {m['forgeid']}")
    return 0 if not error_count else 1


if __name__ == "__main__":
    sys.exit(main())