nix/modules/software/janitor/janitor-backend.nix

{
  den.aspects.janitor-backend = {
    nixos = { pkgs, lib, config, ... }:
    let
      cfg = config.services.janitor;

      janitorScript = pkgs.writeText "janitor.py" ''
        import os
        import sys
        import json
        import shutil
        import time
        import logging
        from pathlib import Path

        logging.basicConfig(
            level=logging.INFO,
            format="[janitor] %(message)s",
            stream=sys.stdout,
        )
        log = logging.getLogger("janitor")

        CONFIG_PATH = os.environ.get("JANITOR_CONFIG")


        def load_config():
            if not CONFIG_PATH or not os.path.exists(CONFIG_PATH):
                raise FileNotFoundError(f"JANITOR_CONFIG not set or missing: {CONFIG_PATH}")
            with open(CONFIG_PATH) as f:
                return json.load(f)


        def get_destination(extension, rules):
            ext = extension.lower().lstrip(".")
            for folder, extensions in rules.items():
                if ext in extensions:
                    return folder
            return None


        def resolve_dest(dest_key, watch_dir):
            if os.path.isabs(dest_key):
                return Path(dest_key)
            return watch_dir.parent / dest_key


        def unique_path(target):
            if not target.exists():
                return target
            stem, suffix = target.stem, target.suffix
            counter = 1
            while True:
                candidate = target.parent / f"{stem}_{counter}{suffix}"
                if not candidate.exists():
                    return candidate
                counter += 1


        def main():
            try:
                config = load_config()
            except Exception as e:
                log.error("Failed to load config: %s", e)
                sys.exit(1)

            grace_period = config.get("grace_period", 60)
            rules = config.get("rules", {})
            now = time.time()

            for watch_dir_str in config.get("watched_dirs", []):
                watch_dir = Path(os.path.expanduser(watch_dir_str))

                if not watch_dir.exists():
                    log.warning("Watched dir does not exist: %s", watch_dir)
                    continue

                for item in watch_dir.iterdir():
                    if not item.is_file() or item.name.startswith("."):
                        continue

                    age = now - item.stat().st_mtime
                    if age < grace_period:
                        continue

                    dest_key = get_destination(item.suffix, rules)
                    if dest_key is None:
                        continue

                    target_dir = resolve_dest(dest_key, watch_dir)
                    target_dir.mkdir(parents=True, exist_ok=True)

                    target = unique_path(target_dir / item.name)

                    try:
                        shutil.move(str(item), str(target))
                        log.info("Moved %s -> %s", item.name, target)
                    except Exception as e:
                        log.error("Failed to move %s: %s", item.name, e)


        if __name__ == "__main__":
            main()
      '';

      janitorConfig = pkgs.writeText "janitor_config.json" (builtins.toJSON {
        grace_period = cfg.gracePeriod;
        watched_dirs = cfg.watchedDirs;
        rules = cfg.rules;
      });

    in {
      options.services.janitor = {
        enable = lib.mkEnableOption "file sorting janitor";

        interval = lib.mkOption {
          type = lib.types.str;
          default = "5min";
          description = "How often to run the janitor (systemd time span, e.g. \"5min\", \"1h\").";
        };

        gracePeriod = lib.mkOption {
          type = lib.types.int;
          default = 60;
          description = "Seconds a file must remain unmodified before it is eligible to be moved.";
        };

        watchedDirs = lib.mkOption {
          type = lib.types.listOf lib.types.str;
          default = [ "~/Downloads" ];
          description = "Directories to scan and sort. Supports ~ expansion.";
        };

        rules = lib.mkOption {
          type = lib.types.attrsOf (lib.types.listOf lib.types.str);
          default = { };
          description = ''
            Mapping of destination folder path to a list of file extensions (without leading dot).
            Destinations are relative to the parent of the watched directory, or absolute if they
            start with /.

            Example:
              rules = {
                "Pictures/Downloads" = [ "jpg" "png" "gif" "webp" ];
                "Videos/Downloads"   = [ "mp4" "mkv" "webm" ];
                "/mnt/archive"       = [ "zip" "tar" ];
              };
          '';
          example = lib.literalExpression ''
            {
              "Pictures/Downloads" = [ "jpg" "jpeg" "png" "gif" "webp" "avif" ];
              "Videos/Downloads"   = [ "mp4" "mkv" "mov" "webm" "avi" ];
              "Music/Downloads"    = [ "mp3" "flac" "wav" "ogg" "opus" "m4a" ];
              "Documents/Downloads" = [ "pdf" "doc" "docx" "odt" "txt" "md" "epub" ];
            }
          '';
        };

        user = lib.mkOption {
          type = lib.types.str;
          default = "bug";
          description = "User account the janitor service runs as.";
        };
      };

      config = lib.mkIf cfg.enable {
        systemd.services.janitor = {
          description = "File sorting janitor";

          environment.JANITOR_CONFIG = "${janitorConfig}";

          serviceConfig = {
            Type = "oneshot";
            User = cfg.user;
            ExecStart = "${pkgs.python3}/bin/python3 ${janitorScript}";
          };
        };

        systemd.timers.janitor = {
          wantedBy = [ "timers.target" ];

          timerConfig = {
            OnBootSec = "10m";
            OnUnitActiveSec = cfg.interval;
          };
        };
      };
    };
  };
}