notifier: add initial code

nixpkgs_notifier.py

@@ -0,0 +1,234 @@
+import sys
+import io
+import pycurl
+from bs4 import BeautifulSoup
+from pathlib import Path
+import os
+import notify2
+from time import sleep
+import subprocess
+import json
+
+state_dir = Path(
+    os.environ.get("XDG_STATE_HOME", Path.home() / ".local" / "state")
+) / "nixpkgs-notify"
+state_dir.mkdir(parents=True, exist_ok=True)
+
+trackedPRPath = state_dir / "tracked.txt"
+trackedPRFile = Path(trackedPRPath)
+
+config_dir = Path(
+    os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")
+) / "nixpkgs-notify"
+
+configPath = config_dir / "config.json"
+configFile = Path(configPath)
+
+# parse config file
+DEFAULT_CONFIG = {
+    "configTime": 3600,
+    "configFetchTime": 1,
+    "localNotify": True,
+    "matrix": {
+        "enable": False,
+        "room": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+        "ping": False,
+        "userPing": "@someone",
+        "userPingServer": "matrix.org"
+    }
+}
+
+def load_config():
+    config_dir.mkdir(parents=True, exist_ok=True)
+
+    if not configFile.exists():
+        with configFile.open("w") as f:
+            json.dump(DEFAULT_CONFIG, f, indent=2)
+        return DEFAULT_CONFIG
+
+    try:
+        with configFile.open("r") as f:
+            return json.load(f)
+    except json.JSONDecodeError:
+        print("Config file is broken. Using default config.")
+        return DEFAULT_CONFIG
+
+def addTracker(PRnumber):
+    result = fetchStatus(PRnumber)
+    if result == "accepted":
+        print("PR " + str(PRnumber) + " was already merged into nixos-unstable.")
+    elif result == "not_found":
+        print("PR " + str(PRnumber) + " not found.")
+    elif result == "unknown":
+        print("PR " + str(PRnumber) + " returned unknown status.")
+    elif result == "pending":
+        with trackedPRFile.open("a") as f:
+            f.write(f"{PRnumber}\n")
+
+def removeTracker(PRnumber):
+    if not trackedPRFile.exists():
+        return
+
+    with trackedPRFile.open("r") as f:
+        lines = f.readlines()
+
+    with trackedPRFile.open("w") as f:
+        for line in lines:
+            if line.strip() != str(PRnumber):
+                f.write(line)
+
+def localNotify(PRnumber):
+    notify2.init("nixpkgs-tracker-notify")
+
+    notification = notify2.Notification(
+        "nixpkgs PR merged",
+        f"PR #{PRnumber} is now available in nixos-unstable.\n"
+        f"https://nixpk.gs/pr-tracker.html?pr={PRnumber}",
+    )
+
+    notification.show()
+
+def matrix_check_whoami():
+    try:
+        result = subprocess.run(
+            ["matrix-commander-rs", "--whoami"],
+            capture_output=True,
+            text=True,
+            check=True
+        )
+
+        output = result.stdout.strip()
+
+        if output.startswith("@"):
+            return output  # OK, proceed
+        else:
+            print(f"Error: invalid identity returned: {output}")
+            return None
+
+    except FileNotFoundError:
+        print("Error: matrix-commander-rs not found in PATH.")
+        return None
+
+    except subprocess.CalledProcessError as e:
+        print(f"Error: command failed with exit code {e.returncode}")
+        print(e.stderr.strip())
+        return None
+
+def matrixNotify(PRnumber):
+    if matrix_check_whoami():
+        command = "matrix-commander-rs -m \"PR " + str(PRnumber) + " is now available in nixos-unstable.\nhttps://nixpk.gs/pr-tracker.html?pr=" + str(PRnumber)
+        if matrixPing:
+            command += '\n<a href=\\"https://matrix.to/#/' + matrixUserToPing + ':' + matrixUserServer + '\\">' + matrixUserToPing + '</a>\" --html'
+        else:
+            command += "\""
+        command += " -r \"" + matrixRoom + "\""
+        subprocess.run(
+            command,
+            shell=True,
+            capture_output=True,
+            text=True
+        )
+def fetchStatus(PRnumber):
+    buf = io.BytesIO()
+
+    c = pycurl.Curl()
+    c.setopt(c.URL, f"https://nixpk.gs/pr-tracker.html?pr={PRnumber}")
+    c.setopt(c.WRITEDATA, buf)
+    c.perform()
+    c.close()
+
+    soup = BeautifulSoup(buf.getvalue(), "html.parser")
+
+    # Check if the PR does not exist
+    no_pr_msg = soup.find(
+        string=lambda s: s and f"No such nixpkgs PR #{PRnumber}" in s
+    )
+    if no_pr_msg:
+        return "not_found"
+
+    # Check nixos-unstable status
+    for a in soup.find_all("a"):
+        if a.get_text(strip=True) == "nixos-unstable":
+            span = a.find_parent("li").find("span")
+            classes = span.get("class", [])
+
+            if "state-accepted" in classes:
+                return "accepted"
+            elif "state-pending" in classes:
+                return "pending"
+            elif "state-rejected" in classes:
+                return "rejected"
+            else:
+                return "unknown"
+
+    return "unknown"
+
+def main():
+    # fetch config
+    cfg = load_config()
+    configTime = cfg["configTime"]
+    configFetchTime = cfg["configFetchTime"]
+    shouldLocalNotify = cfg["localNotify"]
+    shouldMatrixNotify = cfg["matrix"]["enable"]
+    matrixRoom = cfg["matrix"]["room"]
+    matrixPing = cfg["matrix"]["ping"]
+    matrixUserToPing = cfg["matrix"]["userPing"]
+    matrixUserServer = cfg["matrix"]["userPingServer"]
+    
+    
+    if len(sys.argv) < 2:
+        print("Missing command")
+        sys.exit(1)
+    if sys.argv[1] == "add":
+        for PR in sys.argv:
+            if PR != sys.argv[0] and PR != sys.argv[1]:
+                try:
+                    PR = int(PR)
+                    addTracker(PR)
+                except ValueError:
+                    print(f"{PR} is not an int")
+                    continue
+            sleep(configFetchTime) # avoid spamming requests
+    
+    
+    elif sys.argv[1] == "remove" or sys.argv[1] == "rm":
+        for PR in sys.argv:
+            if PR != sys.argv[0] and PR != sys.argv[1]:
+                try:
+                    PR = int(PR)
+                    removeTracker(PR)
+                except ValueError:
+                    print(f"{PR} is not an int")
+                    continue
+            sleep(configFetchTime) # avoid spamming requests
+    
+    elif sys.argv[1] == "list" or sys.argv[1] == "ls":
+        if not trackedPRFile.exists():
+            print("No tracked PRs.")
+        else:
+            print("Tracked PRs:")
+            with trackedPRFile.open("r") as f:
+                for line in f:
+                    pr = line.strip()
+                    if pr:
+                        print(f"  {pr}")
+    
+    elif sys.argv[1] == "listen":
+        while True:
+            if trackedPRFile.exists():
+                with trackedPRFile.open("r") as f:
+                    for line in f:
+                        PR = int(line.strip())
+                        if fetchStatus(PR) == "accepted":
+                            if shouldLocalNotify:
+                                localNotify(PR)
+                            if shouldMatrixNotify:
+                                matrixNotify(PR)
+                            removeTracker(PR)
+                        sleep(configFetchTime) # avoids to send request to quickly to the website
+            sleep(configTime)
+    
+    else:
+        print("Unknown argument \"" + sys.argv[1] + "\"\nAccepted arguments are:\n    add\n    remove or rm\n    list or ls\n    listen")
+    
+