Create the unnotifier

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"),
+    ]
+)

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()