From 64c46222008c35816e555af62709624f02ca5513 Mon Sep 17 00:00:00 2001 From: Reid 'arrdem' McKenzie Date: Thu, 27 Jul 2023 16:34:26 -0600 Subject: [PATCH] Create the unnotifier --- projects/gh-unnotifier/BUILD | 9 ++ .../src/python/ghunnotif/__main__.py | 125 ++++++++++++++++++ 2 files changed, 134 insertions(+) create mode 100644 projects/gh-unnotifier/BUILD create mode 100644 projects/gh-unnotifier/src/python/ghunnotif/__main__.py diff --git a/projects/gh-unnotifier/BUILD b/projects/gh-unnotifier/BUILD new file mode 100644 index 0000000..a820dab --- /dev/null +++ b/projects/gh-unnotifier/BUILD @@ -0,0 +1,9 @@ +py_project( + name = "gh-unnotifier", + main = "src/python/ghunnotif/__main__.py", + main_deps = [ + py_requirement("click"), + py_requirement("requests"), + py_requirement("pytimeparse"), + ] +) diff --git a/projects/gh-unnotifier/src/python/ghunnotif/__main__.py b/projects/gh-unnotifier/src/python/ghunnotif/__main__.py new file mode 100644 index 0000000..d4a7e44 --- /dev/null +++ b/projects/gh-unnotifier/src/python/ghunnotif/__main__.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 + +import os +from pathlib import Path +import tomllib +from typing import Optional +from datetime import datetime, timedelta, timezone +from time import sleep + +import click +import requests +import pytimeparse + +class Client(object): + BASE = "https://api.github.com" + HEADERS = {"Accept": "application/vnd.github+json"} + + def __init__(self, token): + self._token = token + self._session = requests.Session() + self._session.headers["Authorization"] = f"Bearer {token}" + + def get_notifications(self, page=1): + resp = self._session.get(f"{self.BASE}/notifications", headers=self.HEADERS, params={"page": page, "all": "false"}) + resp.raise_for_status() + return resp.json() + + def get_all_notifications(self): + page = 1 + while True: + results = self.get_notifications(page=page) + if not results: + return + yield from results + page += 1 + + def read(self, notif): + return self._session.patch(notif["url"]).raise_for_status() + + def unsubscribe(self, notif): + return self._session.delete(notif["subscription_url"]).raise_for_status() + + def get_pr(self, + url: Optional[str] = None, + repo: Optional[str] = None, + id: Optional[int] = None): + url = url or f"{self.BASE}/{repo}/pulls/{id}" + resp = self._session.get(url, headers=self.HEADERS) + resp.raise_for_status() + return resp.json() + + +@click.group() +def cli(): + pass + + +@cli.command() +@click.option("--config", "config_path", type=Path, default=lambda: Path(os.getenv("BUILD_WORKSPACE_DIRECTORY", "")) / "projects/gh-unnotifier/config.toml") +def oneshot(config_path: Path): + with open(config_path, "rb") as fp: + config = tomllib.load(fp) + + client = Client(config["gh-unnotifier"]["api_key"]) + for notif in client.get_all_notifications(): + subject = notif["subject"] + pr = None + if "/pulls/" in subject["url"]: + pr = client.get_pr(url=subject["url"]) + if pr["state"] == "closed": + client.read(notif) + client.unsubscribe(notif) + + print("Resolved", notif["id"]) + continue + + print(notif["id"], notif["subscription_url"], notif["reason"], notif["subject"], pr) + + +def parse_seconds(text: str) -> timedelta: + return timedelta(seconds=pytimeparse.parse(text)) + + +@cli.command() +@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) +def maintain(config_path: Path, schedule: timedelta): + with open(config_path, "rb") as fp: + config = tomllib.load(fp) + + client = Client(config["gh-unnotifier"]["api_key"]) + mark = None + while True: + try: + prev = mark + mark = datetime.now(timezone.utc) + tick = mark + schedule + assert tick - schedule == mark + for notif in client.get_all_notifications(): + subject = notif["subject"] + + # Don't waste time on notifications which haven't changed since the last scrub + updated_at = datetime.fromisoformat(notif["updated_at"]) + if prev and updated_at < prev: + continue + + pr = None + if "/pulls/" in 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']}") + continue + + click.echo("Napping...") + sleep((tick - datetime.now(timezone.utc)).total_seconds()) + + except KeyboardInterrupt: + break + + +if __name__ == "__main__": + cli()