diff --git a/pyproject.toml b/pyproject.toml index 01fd5ae..0b1344a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -110,4 +110,4 @@ dev-dependencies = [ "griffe-typingdoc>=0.2", # YORE: EOL 3.10: Remove line. "tomli>=2.0; python_version < '3.11'", -] \ No newline at end of file +] diff --git a/src/insiders/__init__.py b/src/insiders/__init__.py index 07df4cd..3c05de9 100644 --- a/src/insiders/__init__.py +++ b/src/insiders/__init__.py @@ -6,3 +6,8 @@ from __future__ import annotations __all__: list[str] = [] + +# TODO: Reorganize code by platform (GitHub, Polar, etc). +# Then create an aggregation module (for example, augmenting GitHub issues with Polar data). +# Then create higher modules for each concept (sponsors, backlog, etc). +# Hook everything up in the CLI. diff --git a/src/insiders/backlog.py b/src/insiders/backlog.py new file mode 100644 index 0000000..6a633f8 --- /dev/null +++ b/src/insiders/backlog.py @@ -0,0 +1,174 @@ +"""Backlog management.""" + +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Callable + +import httpx + +if TYPE_CHECKING: + from insiders.sponsors import Sponsor + + +@dataclass +class Issue: + """An issue.""" + + repository: str + number: int + title: str + created: str + author: str + upvotes: int + pledged: int + + +IssueDict = dict[tuple[str, str], Issue] + + +def get_github_issues() -> IssueDict: + """Get issues from GitHub.""" + issues = {} + items = json.loads( + subprocess.getoutput( # noqa: S605 + "gh search issues " # noqa: S607 + "user:pawamoy org:mkdocstrings " + "sort:created state:open " + "--json repository,number,title,url,author,createdAt " + "--limit 1000", + ), + ) + for item in items: + iid = (item["repository"]["nameWithOwner"], item["number"]) + issues[iid] = Issue( + repository=item["repository"]["nameWithOwner"], + number=item["number"], + title=item["title"], + created=item["createdAt"], + author=item["author"]["login"], + upvotes=0, + pledged=0, + ) + return issues + + +def get_polar_issues(token: str, github_issues: IssueDict | None = None) -> IssueDict: + """Get issues from Polar.""" + issues = github_issues if github_issues is not None else {} + with httpx.Client() as client: + page = 1 + while True: + response = client.get( + "https://api.polar.sh/v1/issues/", + params={ + "external_organization_name": ["pawamoy", "mkdocstrings"], + # "is_badged": True, + "sorting": "-created_at", + "limit": 100, + "page": page, + }, + headers={ + "Accept": "application/json", + "Authorization": f"Bearer {token}", # Scope: issues:read, user:read. + }, + ) + data = response.json() + if not data["items"]: + break + page += 1 + for item in data["items"]: + repository_name = f'{item["repository"]["organization"]["name"]}/{item["repository"]["name"]}' + iid = (repository_name, item["number"]) + if iid in issues: # GitHub issues are the source of truth. + issues[iid].upvotes = item["reactions"]["plus_one"] + issues[iid].pledged = int(item["funding"]["pledges_sum"]["amount"] / 100) + return issues + + +class _Sort: + def __init__(self, *funcs: Callable[[Issue], Any]): + self.funcs = list(funcs) + + def add(self, func: Callable[[Issue], Any]) -> None: + self.funcs.append(func) + + def __call__(self, issue: Issue) -> tuple: + return tuple(func(issue) for func in self.funcs) + + +def get_backlog( + sponsors: list[Sponsor] | None = None, + min_tiers: int | None = None, + min_pledge: int | None = None, + polar_token: str | None = None, +) -> list[Issue]: + """Get the backlog of issues.""" + _sort = _Sort() + # TODO: Use max amount between user amount and their orgs' amounts. + # Example: if user is a member of org1 and org2, and user amount is 10, org1 amount is 20, and org2 amount is 30, + # then the user amount should be 30. + sponsors_dict = {sponsor.account.name: sponsor for sponsor in (sponsors or ())} + if sponsors is not None and min_tiers is not None: + _sort.add(lambda issue: sp.amount if (sp := sponsors_dict.get(issue.author)) and sp.amount >= min_tiers else 0) + elif sponsors is not None: + _sort.add(lambda issue: sp.amount if (sp := sponsors_dict.get(issue.author)) else 0) + if min_pledge is not None: + _sort.add(lambda issue: issue.pledged if issue.pledged >= min_pledge else 0) + else: + _sort.add(lambda issue: issue.pledged) + _sort.add(lambda issue: issue.upvotes) + _sort.add(lambda issue: issue.created) + + issues = get_github_issues() + if polar_token: + issues = get_polar_issues(polar_token, issues) + return sorted(issues.values(), key=_sort, reverse=True) + + +def print_backlog(backlog: list[Issue], *, pledges: bool = True, rich: bool = True) -> None: + """Print the backlog.""" + if rich: + from rich.console import Console + from rich.table import Table + + table = Table(title="Backlog") + table.add_column("Issue", style="underline", no_wrap=True) + table.add_column("Author", no_wrap=True) + if pledges: + table.add_column("Pledged", justify="right", no_wrap=True) + table.add_column("Upvotes", justify="right", no_wrap=True) + table.add_column("Title") + + if pledges: + for issue in backlog: + iid = f"{issue.repository}#{issue.number}" + table.add_row( + f"[link=https://github.com/{issue.repository}/issues/{issue.number}]{iid}[/link]", + f"[link=https://github.com/{issue.author}]{issue.author}[/link]", + f"💲{issue.pledged}", + f"👍{issue.upvotes}", + issue.title, + ) + else: + for issue in backlog: + iid = f"{issue.repository}#{issue.number}" + table.add_row( + f"[link=https://github.com/{issue.repository}/issues/{issue.number}]{iid}[/link]", + f"[link=https://github.com/{issue.author}]{issue.author}[/link]", + f"👍{issue.upvotes}", + issue.title, + ) + + console = Console() + console.print(table) + + else: + for issue in backlog: + iid = f"{issue.repository}#{issue.number}" + pledged = f"💲{issue.pledged} " if pledges else "" + upvotes = f"👍{issue.upvotes}" + pledged_upvotes = f"{pledged}{upvotes}" + print(f"{iid:44} {pledged_upvotes:12} {issue.author:26} {issue.title}") # noqa: T201 diff --git a/src/insiders/cli.py b/src/insiders/cli.py index 5ee357d..05dba71 100644 --- a/src/insiders/cli.py +++ b/src/insiders/cli.py @@ -253,6 +253,56 @@ class CommandPyPI(HelpOption): subcommand: An[cappa.Subcommands[CommandPyPIRegister], Doc("The selected subcommand.")] +@cappa.command( + name="sync", + help="Synchronize members of a team with current sponsors.", + description=cleandoc( + """ + Fetch current sponsors from GitHub, + then grant or revoke access to a GitHub team + for eligible sponsors. + """, + ), +) +@dataclass(kw_only=True) +class CommandSync(HelpOption): + """Command to sync team memberships with current sponsors.""" + + github_sponsored_account: An[ + str, + cappa.Arg(short=False, long=True), + Doc("""The sponsored account on GitHub Sponsors."""), + ] + polar_sponsored_account: An[ + str, + cappa.Arg(short=False, long=True), + Doc("""The sponsored account on Polar."""), + ] + min_amount: An[ + int, + cappa.Arg(short=False, long=True), + Doc("""Minimum amount to be considered an Insider."""), + ] + github_team: An[ + str, + cappa.Arg(short=False, long=True), + Doc("""The GitHub team to sync."""), + ] + github_privileged_users: An[ + list[str], + cappa.Arg(short=False, long=True), + Doc("""Users that should always be in the team."""), + ] + github_org_users: An[ + dict[str, list[str]], + cappa.Arg(short=False, long=True), + Doc("""A mapping of users belonging to sponsoring organizations."""), + ] + + def __call__(self) -> int: # noqa: D102 + raise NotImplementedError("Not implemented yet.") + + @cappa.command( name=NAME, help="Manage your Insiders projects.", diff --git a/src/insiders/sponsors.py b/src/insiders/sponsors.py new file mode 100644 index 0000000..84b3af4 --- /dev/null +++ b/src/insiders/sponsors.py @@ -0,0 +1,148 @@ +"""Sponsors management.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Literal + +import httpx + +GRAPHQL_SPONSORS = """ +query { + viewer { + sponsorshipsAsMaintainer( + first: 100, + after: %s + includePrivate: true, + orderBy: { + field: CREATED_AT, + direction: DESC + } + ) { + pageInfo { + hasNextPage + endCursor + } + nodes { + createdAt, + isOneTimePayment, + privacyLevel, + sponsorEntity { + ...on Actor { + __typename, + login, + avatarUrl, + url + } + }, + tier { + monthlyPriceInDollars + } + } + } + } +} +""" + +SupportedPlatform = Literal["github", "polar", "kofi", "patreon", "liberapay"] + + +@dataclass +class Account: + """A sponsor account.""" + + name: str + image: str + url: str + org: bool + platform: SupportedPlatform + + def as_dict(self) -> dict: + """Return account as a dictionary.""" + return { + "name": self.name, + "image": self.image, + "url": self.url, + "org": self.org, + "platform": self.platform, + } + + +@dataclass +class Sponsor: + """A sponsor.""" + + account: Account + private: bool + created: datetime + amount: int + + +def get_github_sponsors(token: str) -> list[Sponsor]: + """Get GitHub sponsors.""" + sponsors = [] + with httpx.Client() as client: + cursor = "null" + while True: + # Get sponsors data + payload = {"query": GRAPHQL_SPONSORS % cursor} + response = client.post( + "https://api.github.com/graphql", + json=payload, + headers={"Authorization": f"Bearer {token}"}, # Scope: admin:org, read:user. + ) + response.raise_for_status() + + # Post-process sponsors data + data = response.json()["data"] + for item in data["viewer"]["sponsorshipsAsMaintainer"]["nodes"]: + if item["isOneTimePayment"]: + continue + + # Determine account + account = Account( + name=item["sponsorEntity"]["login"], + image=item["sponsorEntity"]["avatarUrl"], + url=item["sponsorEntity"]["url"], + org=item["sponsorEntity"]["__typename"].lower() == "organization", + platform="github", + ) + + # Add sponsor + sponsors.append( + Sponsor( + account=account, + private=item["privacyLevel"].lower() == "private", + created=datetime.strptime(item["createdAt"], "%Y-%m-%dT%H:%M:%SZ"), # noqa: DTZ007 + amount=item["tier"]["monthlyPriceInDollars"], + ), + ) + + # Check for next page + if data["viewer"]["sponsorshipsAsMaintainer"]["pageInfo"]["hasNextPage"]: + cursor = f'"{data["viewer"]["sponsorshipsAsMaintainer"]["pageInfo"]["endCursor"]}"' + else: + break + + return sponsors + + +def get_polar_sponsors() -> list[Sponsor]: + """Get Polar sponsors.""" + raise NotImplementedError("Polar support is not implemented yet") + + +def get_kofi_sponsors() -> list[Sponsor]: + """Get Ko-fi sponsors.""" + raise NotImplementedError("Ko-fi support is not implemented yet") + + +def get_patreon_sponsors() -> list[Sponsor]: + """Get Patreon sponsors.""" + raise NotImplementedError("Patreon support is not implemented yet") + + +def get_liberapay_sponsors() -> list[Sponsor]: + """Get Liberapay sponsors.""" + raise NotImplementedError("Liberapay support is not implemented yet") diff --git a/src/insiders/sync.py b/src/insiders/sync.py new file mode 100644 index 0000000..6e94e27 --- /dev/null +++ b/src/insiders/sync.py @@ -0,0 +1,154 @@ +"""Sync GitHub sponsors with GitHub teams.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import httpx +from loguru import logger + +from insiders.sponsors import Sponsor, get_github_sponsors + +# TODO: Pass token through parameters instead of using global env var. +# permissions: admin:org and read:user +GITHUB_TOKEN = os.getenv("GITHUB_TOKEN") + + +# TODO: Maybe pass already instantiated client. +def get_github_team_members(org: str, team: str) -> set[str]: + """Get members of a GitHub team.""" + page = 1 + members = set() + while True: + response = httpx.get( + f"https://api.github.com/orgs/{org}/teams/{team}/members", + params={"per_page": 100, "page": page}, + headers={"Authorization": f"Bearer {GITHUB_TOKEN}"}, + ) + response.raise_for_status() + response_data = response.json() + members |= {member["login"] for member in response_data} + if len(response_data) < 100: # noqa: PLR2004 + break + page += 1 + return {user["login"] for user in response.json()} + + +# TODO: Maybe pass already instantiated client. +def get_github_team_invites(org: str, team: str) -> set[str]: + """Get pending invitations to a GitHub team.""" + response = httpx.get( + f"https://api.github.com/orgs/{org}/teams/{team}/invitations", + params={"per_page": 100}, + headers={"Authorization": f"Bearer {GITHUB_TOKEN}"}, + ) + response.raise_for_status() + return {user["login"] for user in response.json()} + + +# TODO: Maybe pass already instantiated client. +def grant_access_to_github_team(user: str, org: str, team: str) -> None: + """Grant access to a user to a GitHub team.""" + with httpx.Client() as client: + response = client.put( + f"https://api.github.com/orgs/{org}/teams/{team}/memberships/{user}", + headers={"Authorization": f"Bearer {GITHUB_TOKEN}"}, + ) + try: + response.raise_for_status() + except httpx.HTTPError as error: + logger.error(f"Couldn't add @{user} to {org}/{team} team: {error}") + if response.content: + response_body = response.json() + logger.error(f"{response_body['message']} See {response_body['documentation_url']}") + else: + logger.info(f"@{user} added to {org}/{team} team") + + +# TODO: Maybe pass already instantiated client. +def revoke_access_from_github_team(user: str, org: str, team: str) -> None: + """Revoke access from a user to a GitHub team.""" + with httpx.Client() as client: + response = client.delete( + f"https://api.github.com/orgs/{org}/teams/{team}/memberships/{user}", + headers={"Authorization": f"Bearer {GITHUB_TOKEN}"}, + ) + try: + response.raise_for_status() + except httpx.HTTPError as error: + logger.error(f"Couldn't remove @{user} from {org}/{team} team: {error}") + if response.content: + response_body = response.json() + logger.error(f"{response_body['message']} See {response_body['documentation_url']}") + else: + logger.info(f"@{user} removed from {org}/{team} team") + + +def sync_github_team( + team: str, + min_amount: int, + privileged_users: set[str], + org_users: dict[str, set[str]], + token: str, +) -> list[Sponsor]: + """Sync sponsors with members of a GitHub team.""" + sponsors = get_github_sponsors(token=token) + + eligible_orgs = { + sponsor.account.name for sponsor in sponsors if sponsor.account.org and sponsor.amount >= min_amount + } + eligible_users = { + sponsor.account.name for sponsor in sponsors if not sponsor.account.org and sponsor.amount >= min_amount + } + eligible_users |= privileged_users + for eligible_org in eligible_orgs: + eligible_users |= org_users.get(eligible_org, set()) + + # TODO: Fetch org users from GitHub directly: + # https://docs.github.com/en/rest/orgs/members?apiVersion=2022-11-28#list-organization-members. + # If the org sponsors for $10 dollars, do nothing. + # If the org sponsors for $50 dollars, and the org has more than 5 members, do nothing (can't decide). + # If the org sponsors for $100 dollars, and the org has 10 or less members, add them to the team. + + org, team = team.split("/", 1) + members = get_github_team_members(org, team) | get_github_team_invites(org, team) + # revoke accesses + for user in members: + if user not in eligible_users: + revoke_access_from_github_team(user, org, team) + # grant accesses + for user in eligible_users: + if user not in members: + grant_access_to_github_team(user, org, team) + + return sponsors + + +def update_numbers_file(sponsors: list[Sponsor], filepath: Path = Path("numbers.json")) -> None: + """Update the file storing sponsorship numbers.""" + with filepath.open("w") as f: + json.dump( + { + "total": sum(sponsor.amount for sponsor in sponsors), + "count": len(sponsors), + }, + f, + indent=2, + ) + + +def update_sponsors_file( + sponsors: list[Sponsor], + filepath: Path = Path("sponsors.json"), + *, + exclude_private: bool = True, +) -> None: + """Update the file storing sponsors info.""" + with filepath.open("w") as f: + json.dump( + [sponsor.account.as_dict() for sponsor in sponsors if not sponsor.private or not exclude_private], + f, + indent=2, + )