diff --git a/nixpkgs_notifier.py b/nixpkgs_notifier.py new file mode 100644 index 0000000..0257f85 --- /dev/null +++ b/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' + matrixUserToPing + '\" --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") + +