This commit is contained in:
Reid 'arrdem' McKenzie 2023-07-27 17:23:39 -06:00
parent ad9816f55d
commit 5895e7f404
2 changed files with 62 additions and 7 deletions

View file

@ -0,0 +1,26 @@
# Github Unnotifier
So I work in a monorepo, to which I have elevated access.
This means I get a LOT of work associated notifications on Github because approval requests land on my team's desk.
However, since we are a team, many of these notificaitons don't require that I action them.
I would like to not be notified at all, but Github's notification filtering features are non-existent so I'm doing this the hard way.
This script at present does the simple thing - it looks through notifications for activity on resolved PRs and marks such activity as read automatically.
Unfortunately, that's the best you can really do.
## Usage
You're gonna need a `config.toml` like
``` toml
[gh-unnotifier]
api_key = "your secret value"
```
And then you can `bazel run //projects/gh-unnotifier -- maintain --config $(realpath config.toml)`.
By default, `maintain` will go over your notifications once a minute and mark anything as read which:
1. Relates to a closed (merged or abandoned) PR
2. Does not have an outstanding review request against EITHER the user OR one of the user's teams (someone else already reviewed it)
It would be nice if there was a way to quickly ascertain what review permissions the configured user has and whether requested reviews to relevant groups have already been provided, but so far that's a pipe dream.

View file

@ -49,6 +49,22 @@ class Client(object):
resp.raise_for_status() resp.raise_for_status()
return resp.json() return resp.json()
def get_pr_reviewers(self, pr):
url = pr["url"] + "/requested_reviewers"
resp = self._session.get(url, headers=self.HEADERS)
resp.raise_for_status()
return resp.json()
def get_user(self):
resp = self._session.get(f"{self.BASE}/user")
resp.raise_for_status()
return resp.json()
def get_user_teams(self):
resp = self._session.get(f"{self.BASE}/user/teams")
resp.raise_for_status()
return resp.json()
@click.group() @click.group()
def cli(): def cli():
@ -83,13 +99,22 @@ def parse_seconds(text: str) -> timedelta:
@cli.command() @cli.command()
@click.option("--config", "config_path", type=Path, default=lambda: Path(os.getenv("BUILD_WORKSPACE_DIRECTORY", "")) / "projects/gh-unnotifier/config.toml") @click.option("--config", "config_path", type=Path, default=lambda: Path(os.getenv("BUILD_WORKSPACE_DIRECTORY", "")) / "projects/gh-unnotifier/config.toml")
@click.option("--schedule", "schedule", default="15 seconds", type=parse_seconds) @click.option("--schedule", "schedule", default="1 minute", type=parse_seconds)
def maintain(config_path: Path, schedule: timedelta): def maintain(config_path: Path, schedule: timedelta):
with open(config_path, "rb") as fp: with open(config_path, "rb") as fp:
config = tomllib.load(fp) config = tomllib.load(fp)
client = Client(config["gh-unnotifier"]["api_key"]) client = Client(config["gh-unnotifier"]["api_key"])
org_shitlist = config["gh-unnotifier"].get("org_shitlist", [])
user = client.get_user()
user_teams = {it["slug"] for it in client.get_user_teams()}
mark = None mark = None
def _resolve(notif):
client.read(notif)
client.unsubscribe(notif)
click.echo(f"Resolved {notif['id']}")
while True: while True:
try: try:
prev = mark prev = mark
@ -107,15 +132,19 @@ def maintain(config_path: Path, schedule: timedelta):
pr = None pr = None
if "/pulls/" in subject["url"]: if "/pulls/" in subject["url"]:
pr = client.get_pr(url=subject["url"]) pr = client.get_pr(url=subject["url"])
if pr["state"] == "closed":
client.read(notif)
client.unsubscribe(notif)
click.echo(f"Resolved {notif['id']}") if pr["state"] == "closed":
_resolve(notif)
continue continue
click.echo("Napping...") reviewers = client.get_pr_reviewers(pr)
sleep((tick - datetime.now(timezone.utc)).total_seconds()) if any(org in subject["url"] for org in org_shitlist) and not any(it["login"] == user["login"] for it in reviewers.get("users", [])) and not any(it["slug"] in user_teams for it in reviewers.get("teams", [])):
_resolve(notif)
continue
duration = (tick - datetime.now(timezone.utc)).total_seconds()
if duration > 0:
sleep(duration)
except KeyboardInterrupt: except KeyboardInterrupt:
break break