More tuning of the unnotifier

projects/gh-unnotifier/src/python/ghunnotif/__main__.py

@@ -6,7 +6,7 @@ import tomllib
 from typing import Optional
 from datetime import datetime, timedelta, timezone
 from time import sleep
-from pprint import pformat
+from pprint import pprint
 
 import click
 import requests
@@ -149,8 +149,6 @@ def parse_seconds(text: str) -> timedelta:
     "--config",
     "config_path",
     type=Path,
-    default=lambda: Path(os.getenv("BUILD_WORKSPACE_DIRECTORY", ""))
-    / "projects/gh-unnotifier/config.toml",
 )
 @click.option("--schedule", "schedule", default="30 seconds", type=parse_seconds)
 def maintain(config_path: Path, schedule: timedelta):
@@ -162,70 +160,95 @@ def maintain(config_path: Path, schedule: timedelta):
     team_shitlist = config["gh-unnotifier"].get("team_shitlist", [])
     user = client.get_user()
     user_teams = {it["slug"] for it in client.get_user_teams()}
-    mark = None
+    mark = savepoint = prev = None
 
     def _resolve(notif, reason):
         client.read(notif)
         client.unsubscribe(notif)
         click.echo(f"Resolved {notif['id']} {reason} ({notif['subject']})")
 
+    def _pr(subject, notif):
+        pr = client.get_pr(url=subject["url"])
+
+        if pr.get("merged", False):
+            _resolve(notif, "Merged")
+            return
+
+        if pr.get("state") == "closed":
+            _resolve(notif, "Closed")
+            return
+
+        if notif["reason"] == "review_requested":
+            reviewers = client.get_pr_reviewers(pr)
+            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
+                    and it["slug"] not in team_shitlist
+                    for it in reviewers.get("teams", [])
+                )
+            ):
+                _resolve(notif, "Reviewed")
+                return
+
+        print("Oustanding PR notification")
+        pprint({"subject": subject, "notif": notif, "pr": pr})
+
+    def _mention(subject, notif):
+        pr = client.get_pr(url=subject["url"])
+
+        reviewers = client.get_pr_reviewers(pr)
+        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
+                and it["slug"] not in team_shitlist
+                for it in reviewers.get("teams", [])
+            )
+        ):
+            _resolve(notif, "Ignoring team mention")
+            return
+
+    def _issue(subject, notif):
+        issue = client.get_issue(url=subject["url"])
+        if issue["state"] == "closed":
+            comments = client.get_comments(url=issue["comments_url"])
+            if not any(
+                it["user"]["login"] == user["login"] for it in comments
+            ):
+                _resolve(notif, "Unengaged issue closed")
+
     while True:
         try:
+            savepoint = prev
             prev = mark
             mark = datetime.now(timezone.utc)
             tick = mark + schedule
             assert tick - schedule == mark
             for notif in client.get_all_notifications(since=prev):
+                notif_timestamp = datetime.fromisoformat(notif["updated_at"])
+                if (mark - notif_timestamp).days > 3:
+                    _resolve(notif, "More than 3 days old")
+
                 subject = notif["subject"]
 
-                pr = None
+                match subject["type"]:
-                if subject["type"] == "PullRequest":
+                    case "PullRequest":
-                    if notif["reason"] == "review_requested":
+                        _pr(subject, notif)
-                        pr = client.get_pr(url=subject["url"])
 
-                        reviewers = client.get_pr_reviewers(pr)
+                    case "mention":
-                        if (
+                        _mention(subject, notif)
-                            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
-                                and it["slug"] not in team_shitlist
-                                for it in reviewers.get("teams", [])
-                            )
-                        ):
-                            _resolve(notif, "Reviewed")
-                            continue
 
-                    elif notif["reason"] == "team_mention":
+                    case "Issue":
-                        pr = client.get_pr(url=subject["url"])
+                        _issue(subject, notif)
-
-                        reviewers = client.get_pr_reviewers(pr)
-                        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
-                                and it["slug"] not in team_shitlist
-                                for it in reviewers.get("teams", [])
-                            )
-                        ):
-                            _resolve(notif, "Ignoring team mention")
-                            continue
-
-                elif subject["type"] == "Issue":
-                    issue = client.get_issue(url=subject["url"])
-                    if issue["state"] == "closed":
-                        comments = client.get_comments(url=issue["comments_url"])
-                        if not any(
-                            it["user"]["login"] == user["login"] for it in comments
-                        ):
-                            _resolve(notif, "Unengaged issue closed")
 
             duration = (tick - datetime.now(timezone.utc)).total_seconds()
             if duration > 0:
@@ -234,6 +257,11 @@ def maintain(config_path: Path, schedule: timedelta):
         except KeyboardInterrupt:
             break
 
+        except requests.exceptions.HTTPError as e:
+            print("Encoutered exception", e, "backing off")
+            prev = savepoint
+            sleep(schedule.total_seconds())
+
 
 if __name__ == "__main__":
     cli()