9e3bd8a2d0
Although we had only 190 developers, the script was accusing a total of 365 developers (with 188 inactive). There were two problems here: * The gitea API was only using the `limit` paramater when an explicit page was passed. * There was a bug with the get_next_page code, that would skip the last page. I now made the page always an explicit argument passed to fetch_single. With this fix we now get: 190 developers with 98 inactive.
254 lines
6.2 KiB
Python
Executable File
254 lines
6.2 KiB
Python
Executable File
# SPDX-FileCopyrightText: 2023 Blender Foundation
|
|
#
|
|
# SPDX-License-Identifier: GPL-2.0-or-later
|
|
|
|
# Script to get all the inactive gitea developers
|
|
# Usage: GITEA_API_TOKEN=<yourtoken> python3 gitea_inactive_developers.py
|
|
#
|
|
# The API Token have the "read:org" or "admin:org" scope.
|
|
#
|
|
# Potential errors:
|
|
# * 403 Client Error: That means the token doesn't have the right scope.
|
|
# * 500 Server Error: The token is invalid.
|
|
|
|
import csv
|
|
import logging
|
|
import os
|
|
import requests
|
|
import sys
|
|
import yarl
|
|
from retry import retry as retry_decorator
|
|
import datetime
|
|
import dataclasses as dc
|
|
import iso8601
|
|
|
|
from typing import (
|
|
cast,
|
|
Callable,
|
|
Dict,
|
|
Iterable,
|
|
List,
|
|
NewType,
|
|
Optional,
|
|
Tuple,
|
|
Type,
|
|
TypeVar,
|
|
Union,
|
|
)
|
|
from requests.structures import CaseInsensitiveDict
|
|
|
|
logger = logging.getLogger(__file__)
|
|
|
|
|
|
@dc.dataclass
|
|
class TeamMember():
|
|
id: int
|
|
login: str
|
|
full_name: str
|
|
last_login: datetime.datetime
|
|
|
|
def __str__(self) -> str:
|
|
return "{id};{login};{full_name};{last_login};{url}\n".format(
|
|
id=self.id,
|
|
login=self.login,
|
|
full_name=self.full_name,
|
|
last_login=self.last_login,
|
|
url=gitea_domain + self.login,
|
|
)
|
|
|
|
|
|
Page = NewType('Page', int)
|
|
|
|
F = TypeVar('F', bound=Callable[..., object])
|
|
T = TypeVar('T', bound=object)
|
|
|
|
retry: Callable[[F], F] = retry_decorator(
|
|
tries=10, delay=1, backoff=2, logger=logger)
|
|
|
|
|
|
def assert_cast(typ: Type[T], obj: object) -> T:
|
|
assert isinstance(obj, typ), f'object is not of type {typ}: {obj}'
|
|
# NOTE: MYPY warns the cast is redundant, we might consider removing it.
|
|
return cast(T, obj) # type: ignore
|
|
|
|
|
|
def get_date_object(date_string: str) -> datetime.datetime:
|
|
result = iso8601.parse_date(date_string)
|
|
assert isinstance(result, datetime.datetime)
|
|
return result
|
|
|
|
|
|
results_per_page = 25
|
|
|
|
|
|
def get_next_page(headers: CaseInsensitiveDict[str], page: Page) -> Optional[Page]:
|
|
"""
|
|
Parse the header looking for reference to next.
|
|
"""
|
|
total_count = int(assert_cast(str, headers.get('X-Total-Count')))
|
|
|
|
# If current page already accounts to all the results we need
|
|
# there is no need for extra pages.
|
|
if page * results_per_page >= total_count:
|
|
return None
|
|
|
|
next_page = Page(page + 1)
|
|
return next_page
|
|
|
|
|
|
@retry
|
|
def fetch_single(
|
|
api_url: yarl.URL,
|
|
api_token: str,
|
|
method: str,
|
|
data: Dict[str, str],
|
|
page: Page,
|
|
) -> Tuple[List[object], Optional[Page]]:
|
|
"""Generic function to query a single item from the API.
|
|
|
|
Returns:
|
|
A dictionary containing the item data.
|
|
"""
|
|
headers = {
|
|
'accept': 'application/json',
|
|
'Authorization': 'token ' + api_token,
|
|
}
|
|
|
|
params: Dict[str, Union[str, int]] = {
|
|
'limit': results_per_page,
|
|
'page': page,
|
|
**data,
|
|
}
|
|
|
|
logger.info(f"Calling {method} ({params=}).")
|
|
response = requests.get(str(api_url / method), params=params, headers=headers)
|
|
response.raise_for_status()
|
|
response_json = response.json()
|
|
|
|
next_page = get_next_page(response.headers, page)
|
|
return response_json, None if next_page is None else next_page
|
|
|
|
|
|
def fetch_all(
|
|
api_url: yarl.URL,
|
|
api_token: str,
|
|
method: str,
|
|
data: Dict[str, str],
|
|
) -> Iterable[object]:
|
|
"""Generic function to query lists from API.
|
|
|
|
Yields:
|
|
response_data - the result of fetch_single()
|
|
"""
|
|
page = Page(1)
|
|
while page is not None:
|
|
response_data, page = fetch_single(api_url, api_token, method, data, page)
|
|
yield from response_data if response_data is not None else ()
|
|
|
|
|
|
def fetch_team_members(
|
|
api_url: yarl.URL,
|
|
api_token: str,
|
|
organization_name: str,
|
|
team_name: str,
|
|
) -> List[TeamMember]:
|
|
"""Query API for all the members of a team.
|
|
|
|
Yields:
|
|
TeamMember objects.
|
|
"""
|
|
|
|
method = "orgs/{org}/teams".format(org=organization_name)
|
|
team_id = None
|
|
|
|
for team in cast(
|
|
Iterable[Dict[object, object]],
|
|
fetch_all(
|
|
api_url,
|
|
api_token,
|
|
method,
|
|
data={},
|
|
),
|
|
):
|
|
if team.get('name') != team_name:
|
|
continue
|
|
|
|
team_id = team.get('id')
|
|
break
|
|
|
|
if team_id is None:
|
|
logger.error('No team found with name: ' + team_name)
|
|
sys.exit(2)
|
|
|
|
method = "teams/{id}/members".format(id=team_id)
|
|
users = list()
|
|
|
|
for member in cast(
|
|
Iterable[Dict[object, object]],
|
|
fetch_all(
|
|
api_url,
|
|
api_token,
|
|
method,
|
|
data={},
|
|
),
|
|
):
|
|
users.append(
|
|
TeamMember(
|
|
id=assert_cast(int, member.get('id')),
|
|
login=assert_cast(str, member.get('login')),
|
|
full_name=assert_cast(str, member.get('full_name')),
|
|
last_login=get_date_object(
|
|
assert_cast(str, member.get('last_login'))),
|
|
))
|
|
|
|
return users
|
|
|
|
|
|
def is_inactive(member: TeamMember) -> bool:
|
|
"""
|
|
Returns whether the member is no longer active.
|
|
|
|
Users are active when they logged in the past 2 years.
|
|
"""
|
|
tzinfo = member.last_login.tzinfo
|
|
two_years_ago = datetime.datetime.now(tzinfo) - datetime.timedelta(days=2 * 365)
|
|
return member.last_login < two_years_ago
|
|
|
|
|
|
teams = (
|
|
"Developers",
|
|
"Add-ons",
|
|
"Translation",
|
|
"Documentation",
|
|
"Technical-Artists",
|
|
"Contributors",
|
|
)
|
|
|
|
|
|
api_token = os.environ['GITEA_API_TOKEN']
|
|
gitea_domain = "https://projects.blender.org/"
|
|
api_url = yarl.URL(gitea_domain + 'api/v1/')
|
|
organization_name = "blender"
|
|
|
|
|
|
def main() -> None:
|
|
for team_name in teams:
|
|
logger.warning(team_name)
|
|
members = fetch_team_members(
|
|
api_url, api_token, organization_name, team_name)
|
|
|
|
inactive_members = [str(m) for m in members if is_inactive(m)]
|
|
|
|
logger.warning(" Total members:" + str(len(members)))
|
|
logger.warning(" Inactive members: " + str(len(inactive_members)))
|
|
|
|
file_name = team_name.lower() + ".csv"
|
|
with open(file_name, 'w', newline='') as csv_file:
|
|
csv_file.write("id;login;full_name;last_login;url\n")
|
|
csv_file.writelines(inactive_members)
|
|
logger.warning(" Output file: " + file_name)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|