add ebuild for version 0.0.2

.gitignore

@@ -115,5 +115,8 @@ dmypy.json
 # Pyre type checker
 .pyre/
 
+# Temporary Vim files
+.*.swp
+
 # config file
 /config.py

MANIFEST.in

@@ -1,3 +1,2 @@
 include LICENSE README.md
-recursive-include pyinotifyd/docs *
 recursive-include pyinotifyd/misc *

README.md

@@ -1,25 +1,26 @@
 # pyinotifyd
-A daemon to monitore filesystems events with inotify on Linux and execute tasks (Python methods or Shell commands) with an optional delay. It is also possible to cancel delayed tasks.
+A daemon to monitor filesystems events with inotify on Linux and run tasks like filesystem operations (copy, move or delete), a shell commands or custom async python methods.  
+It is possible to schedule tasks with a delay, delayed tasks can be cancelled again in case a certain event occurs. A useful example would be to run tasks only if a file has not changed within a certain amount of time.
 
-## Requirements
+# Requirements
 * [pyinotify](https://github.com/seb-m/pyinotify)
 
-## Installation
+# Installation
 ```sh
 # install pyinotifyd with pip
 pip install pyinotifyd
 
-# install systemd service and create config directory
+# install service files and config
 pyinotifyd --install
 
-# uninstall systemd service
+# uninstall service files and unmodified config
 pyinotifyd --uninstall
 ```
 
-### Autostart
+## Autostart
 The following init systems are supported.
 
-#### systemd
+### systemd
 ```sh
 # start the daemon at boot time
 systemctl enable pyinotifyd.service
@@ -28,7 +29,7 @@ systemctl enable pyinotifyd.service
 systemctl start pyinotifyd.service
 ```
 
-#### OpenRC (Gentoo)
+### OpenRC (Gentoo)
 ```sh
 # start the daemon at boot time
 rc-update add pyinotifyd default
@@ -37,29 +38,53 @@ rc-update add pyinotifyd default
 rc-service pyinotifyd start 
 ```
 
-## Configuration
-The config file **/etc/pyinotifyd/config.py** is written in Python syntax. pyinotifyd reads and executes its content, that means you can add your custom Python code to the config file.
+# Configuration
+The config file **/etc/pyinotifyd/config.py** is written in python syntax. pyinotifyd reads and executes its content, that means you can write your custom async python methods directly into the config file.  
+The basic idea is to instantiate one or multiple schedulers and map specific inotify events to schedulers with the help of event maps. Then, watch the given paths for events and run tasks as defined in the event maps.
 
-### Tasks
-Tasks are Python methods that are called in case an event occurs. They can be bound directly to an event type in an event map. Although this is the easiest and quickest way, it is usually better to add a task to a scheduler and bind the scheduler to event types.
+## Schedulers
+pyinotifyd has different schedulers to schedule tasks with an optional delay. The advantages of using a scheduler are consistent logging and the possibility to cancel delayed tasks. Furthermore, schedulers have the ability to differentiate between files and directories.
 
-#### Simple
-This is a very basic example task that just logs each event and task_id:
+### TaskScheduler
+Schedule a custom python method *job* with an optional *delay* in seconds. Skip scheduling of tasks for files and/or directories according to *files* and *dirs* arguments. If there already is a scheduled task, re-schedule it with *delay*. Use *logname* in log messages.  
+All arguments except for *job* are optional.
 ```python
-async def task(event, task_id):
+# Please note that pyinotifyd uses pythons asyncio for asynchronous task execution.
+# Do not run anything inside the custom python method that blocks the daemon.
+#
+# Bad:  time.sleep(10)
+# Good: await asyncio.sleep(10)
+
+async def custom_job(event, task_id):
+    await asyncio.sleep(10)
     logging.info(f"{task_id}: execute example task: {event}")
+
+task_sched = TaskScheduler(
+    job=custom_job,
+    files=True,
+    dirs=False,
+    delay=0,
+    logname="sched")
 ```
 
-#### FileManager
-FileManager moves, copy or deletes files and/or directories following a list of *rules*. 
-
-A rule holds an *action* (move, copy or delete) and a regular expression *src_re*. The FileManager task will be executed if *src_re* matches the path of an event. 
-If the action is copy or move, the destination path *dst_re* is mandatory and if *action* is delete and *rec* is set to True, non-empty directories will be deleted recursively.  
-With *auto_create* set to True, possibly missing subdirectories in *dst_re* are created automatically. Regex subgroups or named-subgroups may be used in *src_re* and *dst_re*.  
-Set the mode of moved/copied files/directories with *filemode* and *dirmode*. Ownership of moved/copied files/directories is set with *user* and *group*. Mode and ownership is also set to automatically created subdirectories.  
-Log messages with *logname*.  
+### ShellScheduler
+Schedule a shell command *cmd*. Replace  **{maskname}**, **{pathname}** and **{src_pathname}** in *cmd* with the actual values of occuring events. This scheduler is based on TaskScheduler and has the same optional arguments.
 ```python
-rule = Rule(
+# Please note that **{src_pathname}** is only present for IN_MOVED_TO events and only in the case where the IN_MOVED_FROM events are watched too.
+# If it is not present, the command line argument will be an empty string.
+shell_sched = ShellScheduler(
+    cmd="/usr/local/bin/task.sh {maskname} {pathname} {src_pathname}")
+```
+
+### FileManagerScheduler
+Move, copy or delete files and/or directories following the list of *rules*, the first matching rule is executed.  
+This scheduler is based on TaskScheduler and has the same optional arguments.  
+
+A rule holds an *action* (move, copy or delete) and a regular expression *src_re*. The *action* will be executed if *src_re* matches the path of an event. In case where *action* is copy or move, use *dst_re* as destination path. Subgroups and/or named-subgroups may be used in *src_re* and *dst_re*.  
+Automatically create possibly missing sub-directories if *auto_create* is set to True. Set the mode and ownership of moved or copied files/directories and newly created sub-directories to *filemode* and *dirmode*. Override destination files if *override* is set to True.  
+If *action* is delete, delete non-empty directories if *rec* is set to True.  
+```python
+move_rule = FileManagerRule(
     action="move",
     src_re="^/src_path/(?P<path>.*).to_move$",
     dst_re="/dst_path/\g<path>",
@@ -68,49 +93,43 @@ rule = Rule(
     filemode=None,
     dirmode=None,
     user=None,
-    group=None)
+    group=None,
+    override=False)
 
-fm = FileManager(
-    rules=[rule],
-    logname="filemgr")
+delete_rule = FileManagerRule(
+    action="delete",
+    src_re="^/src_path/(?P<path>.*).to_delete$",
+    rec=False)
+
+file_sched = FileManagerScheduler(
+    rules=[move_rule, delete_rule])
 ```
-FileManager provides a task **fm.task**.
 
-### Schedulers
-pyinotifyd has different schedulers to schedule tasks with an optional delay. The advantages of using a scheduler are consistent logging and the possibility to cancel delayed tasks. Furthermore, schedulers have the ability to differentiate between files and directories.
-
-#### TaskScheduler
-TaskScheduler schedules *task* with an optional *delay* in seconds. Use the *files* and *dirs* arguments to schedule tasks only for files and/or directories. 
-Log messages with *logname*. All arguments except for *task* are optional.
+## Event maps
+Map specific events to one or multiple schedulers. Ignore the event if the scheduler is set to None. Use **Cancel** to cancel a scheduled task within a scheduler.  
+This is an example which schedules tasks for newly created files if they are not modified, moved or deleted within the delay time of the scheduler.
 ```python
-s = TaskScheduler(
-    task=task,
-    files=True,
-    dirs=False,
-    delay=0,
-    logname="sched")
-```
-TaskScheduler provides two tasks which can be bound to an event in an event map.
-* **s.schedule** 
-  Schedule a task. If there is already a scheduled task, it will be canceled first.
-* **s.cancel** 
-  Cancel a scheduled task.
+event_map = {
+    "IN_ACCESS": None,
+    "IN_ATTRIB": None,
+    "IN_CLOSE_NOWRITE": None,
+    "IN_CLOSE_WRITE": task_sched,
+    "IN_CREATE": task_sched,
+    "IN_DELETE": Cancel(task_sched),
+    "IN_DELETE_SELF": Cancel(task_sched),
+    "IN_IGNORED": None,
+    "IN_MODIFY": Cancel(task_sched),
+    "IN_MOVE_SELF": None,
+    "IN_MOVED_FROM": Cancel(task_sched),
+    "IN_MOVED_TO": task_sched,
+    "IN_OPEN": None,
+    "IN_Q_OVERFLOW": None,
+    "IN_UNMOUNT": Cancel(task_sched)}
 
-#### ShellScheduler
-ShellScheduler schedules Shell command *cmd*. The placeholders **{maskname}**, **{pathname}** and **{src_pathname}** are replaced with the actual values of the event. ShellScheduler has the same optional arguments as TaskScheduler and provides the same tasks.
-```python
-s1 = ShellScheduler(
-    cmd="/usr/local/bin/task.sh {maskname} {pathname} {src_pathname}")
+# It is possible to instantiate an event map with a default scheduler set for every event,
+event_map = EventMap(default_sched=task_sched)
 ```
-### Event maps
-EventMap maps event types to tasks. It is possible to set a list of tasks to run multiple tasks on a single event. If the task of an event type is set to None, it is ignored. 
-This is an example:
-```python
-event_map = EventMap({
-        "IN_CLOSE_NOWRITE": [s.schedule, s1.schedule],
-        "IN_CLOSE_WRITE": s.schedule})
-```
-The following event types are available:
+The following events are available:
 * **IN_ACCESS**: a file was accessed
 * **IN_ATTRIB**: a metadata changed
 * **IN_CLOSE_NOWRITE**: an unwritable file was closed
@@ -127,38 +146,49 @@ The following event types are available:
 * **IN_Q_OVERFLOW**: the event queue overflown. This event is not associated with any watch descriptor
 * **IN_UNMOUNT**: when backing filesystem was unmounted. Notified to each watch of this filesystem
 
-### Watches
-Watch watches *path* for event types in *event_map* and execute the corresponding task(s). If *rec* is True, a watch will be added on each subdirectory in *path*. If *auto_add* is True, a watch will be added automatically on newly created subdirectories in *path*.
-```python
-watch = Watch(
-    path="/tmp",
-    event_map=event_map,
-    rec=False,
-    auto_add=False)
-```
-
-### Pyinotifyd
-pyinotifyd expects an instance of Pyinotifyd named **pyinotifyd** defined in the config file. The options are a list of *watches* and the *shutdown_timeout*. pyinotifyd will wait *shutdown_timeout* seconds for pending tasks to complete during shutdown. Log messages with *logname*.
+## Pyinotifyd
+pyinotifyd requires you to define a variable called **pyinotifyd** within the config file, which contains an instance of the Pyinotifyd class. Set the optional list of *watches* and the *shutdown_timeout*. Pyinotifyd will wait *shutdown_timeout* seconds for pending tasks to complete before shutdown. Use *logname* in log messages.  
 ```python
 pyinotifyd = Pyinotifyd(
-    watches=[watch],
+    watches=[],
     shutdown_timeout=30,
     logname="daemon")
 ```
 
-### Logging
+### Watches
+A watch connects the *path* to an *event_map*. Automatically add a watch on each sub-directories in *path* if *rec* is set to True. If *auto_add* is True, a watch will be added automatically on newly created sub-directories in *path*.
+```python
+# Add a watch directly to Pyinotifyd.
+pyinotifyd.add_watch(
+    path="/src_path",
+    event_map=event_map,
+    rec=False,
+    auto_add=False)
+
+# Or instantiate and add it
+w = Watch(
+    path="/src_path",
+    event_map=event_map,
+    rec=False,
+    auto_add=False)
+
+pyinotifyd.add_watch(
+    watch=w)
+```
+
+## Logging
 Pythons [logging](https://docs.python.org/3/howto/logging.html) framework is used to log messages (see https://docs.python.org/3/howto/logging.html).  
 
 Configure the global loglevel. This is the default:
 ```python
 logging.getLogger().setLevel(logging.WARNING)
 ```
-It is possible to configure the loglevel per log name. This is an example for logname **TaskScheduler**:
+It is possible to configure the loglevel per *logname*. This is an example for logname **sched**:
 ```python
-logging.getLogger("TaskScheduler").setLevel(logging.INFO)
+logging.getLogger("sched").setLevel(logging.INFO)
 ```
 
-#### Syslog
+### Syslog
 Add this to your config file to send log messages to a local syslog server.
 ```python
 # send log messages to the Unix socket of the syslog server.
@@ -173,63 +203,57 @@ syslog.setFormatter(
 # set the log level for syslog messages
 syslog.setLevel(logging.INFO)
 
-# enable syslog
+# enable syslog for pyinotifyd
 logging.getLogger().addHandler(syslog)
 
-# or enable syslog just for TaskScheduler
-logging.getLogger("TaskManager").addHandler(syslog)
+# or enable syslog just for the daemon
+logging.getLogger("daemon").addHandler(syslog)
 ```
 
-## Examples
+# Examples
 
-### Schedule Python task for all events
+## Schedule python method for all events on files and directories
 ```python
-async def task(event, task_id):
+async def custom_job(event, task_id):
     logging.info(f"{task_id}: execute example task: {event}")
 
-s = TaskScheduler(
-    task=task,
+task_sched = TaskScheduler(
+    job=custom_job,
     files=True,
     dirs=True)
 
 event_map = EventMap(
-    default_task=s.schedule)
+    default_sched=task_sched)
 
-watch = Watch(
-    path="/tmp",
+pyinotifyd = Pyinotifyd()
+pyinotifyd.add_watch(
+	path="/src_path",
 	event_map=event_map,
 	rec=True,
 	auto_add=True)
-
-pyinotifyd_config = PyinotifydConfig(
-    watches=[watch],
-    shutdown_timeout=5)
 ```
 
-### Schedule Shell commands for specific events on files
+## Schedule shell commands for specific events on files
 ```python
-s = ShellScheduler(
+shell_sched = ShellScheduler(
     cmd="/usr/local/sbin/task.sh {pathname}",
     files=True,
     dirs=False)
 
-event_map = EventMap(
-    {"IN_WRITE_CLOSE": s.schedule})
+event_map = {
+    "IN_WRITE_CLOSE": shell_sched}
 
-watch = Watch(
-    path="/tmp",
+pyinotifyd = Pyinotifyd()
+pyinotifyd.add_watch(
+	path="/src_path",
 	event_map=event_map,
 	rec=True,
 	auto_add=True)
-
-pyinotifyd_config = PyinotifydConfig(
-    watches=[watch],
-    shutdown_timeout=5)
 ```
 
-### Move, copy or delete newly created files after a delay
+## Move, copy or delete newly created files after a delay
 ```python
-move_rule = Rule(
+move_rule = FileManagerRule(
     action="move",
     src_re="^/src_path/(?P<path>.*)\.to_move$",
     dst_re="/dst_path/\g<path>",
@@ -237,7 +261,7 @@ move_rule = Rule(
     filemode=0o644,
     dirmode=0o755)
 
-copy_rule = Rule(
+copy_rule = FileManagerRule(
     action="copy",
     src_re="^/src_path/(?P<path>.*)\.to_copy$",
     dst_re="/dst_path/\g<path>",
@@ -245,37 +269,33 @@ copy_rule = Rule(
     filemode=0o644,
     dirmode=0o755)
 
-delete_rule = Rule(
+delete_rule = FileManagerRule(
     action="delete",
     src_re="^/src_path/(?P<path>.*)\.to_delete$",
     rec=False)
 
-fm = FileManager(
-    rules=[move_rule, copy_rule, delete_rule])
-
-s = TaskScheduler(
-    task=fm.task,
-    delay=30,
+file_sched = FileManagerScheduler(
+    rules=[move_rule, copy_rule, delete_rule],
+    delay=60,
     files=True,
     dirs=False)
 
-event_map = EventMap({
-        "IN_CLOSE_WRITE": s.schedule,
-        "IN_DELETE": s.cancel,
-        "IN_DELETE_SELF": s.cancel,
-        "IN_MODIFY": s.cancel,
-        "IN_MOVED_TO": s.schedule,
-        "IN_UNMOUNT": s.cancel})
+event_map = {
+    "IN_CLOSE_WRITE": file_sched,
+    "IN_CREATE": file_sched,
+    "IN_DELETE": Cancel(file_sched),
+    "IN_DELETE_SELF": Cancel(file_sched),
+    "IN_MODIFY": Cancel(file_sched),
+    "IN_MOVED_FROM": Cancel(file_sched),
+    "IN_MOVED_TO": file_sched,
+    "IN_UNMOUNT": Cancel(file_sched)}
 
-watch = Watch(
+# Please note that the shutdown timeout should be greater than the greatest scheduler delay,
+# otherwise pending tasks may get cancelled during shutdown.
+pyinotifyd = Pyinotifyd(shutdown_timeout=35)
+pyinotifyd.add_watch(
 	path="/src_path",
 	event_map=event_map,
 	rec=True,
 	auto_add=True)
-
-# note that shutdown_timeout should be greater than the greatest scheduler delay,
-# otherwise pending tasks may get cancelled during shutdown.
-pyinotifyd_config = PyinotifydConfig(
-    watches=[watch],
-    shutdown_timeout=35)
 ```

distribution/gentoo/dev-python/pyinotifyd/pyinotifyd-0.0.2.ebuild

@@ -0,0 +1,47 @@
+# Copyright 2020 Gentoo Authors
+# Distributed under the terms of the GNU General Public License v2
+
+EAPI=7
+PYTHON_COMPAT=( python3_{7,8,9} )
+DISTUTILS_USE_SETUPTOOLS=rdepend
+
+SCM=""
+if [ "${PV#9999}" != "${PV}" ] ; then
+	SCM="git-r3"
+	EGIT_REPO_URI="https://github.com/spacefreak86/${PN}"
+fi
+
+inherit ${SCM} distutils-r1 systemd
+
+DESCRIPTION="Monitore filesystems events and execute Python methods or Shell commands."
+HOMEPAGE="https://github.com/spacefreak86/pymodmilter"
+if [ "${PV#9999}" != "${PV}" ] ; then
+	SRC_URI=""
+	KEYWORDS=""
+	# Needed for tests
+	S="${WORKDIR}/${PN}"
+	EGIT_CHECKOUT_DIR="${S}"
+else
+	SRC_URI="https://github.com/spacefreak86/${PN}/archive/${PV}.tar.gz -> ${P}.tar.gz"
+	KEYWORDS="amd64 x86"
+fi
+
+LICENSE="GPL-3"
+SLOT="0"
+
+IUSE="systemd"
+
+RDEPEND="dev-python/pyinotify[${PYTHON_USEDEP}]"
+
+python_install_all() {
+	distutils-r1_python_install_all
+
+	dodir /etc/${PN}
+	insinto /etc/${PN}
+	newins ${PN}/misc/config.py.example config.py
+
+	use systemd && systemd_dounit ${PN}/misc/${PN}.service
+
+	newinitd ${PN}/misc/openrc/${PN}.initd ${PN}
+	newconfd ${PN}/misc/openrc/${PN}.confd ${PN}
+}

distribution/gentoo/dev-python/pyinotifyd/pyinotifyd-9999.ebuild

@@ -0,0 +1,47 @@
+# Copyright 2020 Gentoo Authors
+# Distributed under the terms of the GNU General Public License v2
+
+EAPI=7
+PYTHON_COMPAT=( python3_{7,8,9} )
+DISTUTILS_USE_SETUPTOOLS=rdepend
+
+SCM=""
+if [ "${PV#9999}" != "${PV}" ] ; then
+	SCM="git-r3"
+	EGIT_REPO_URI="https://github.com/spacefreak86/${PN}"
+fi
+
+inherit ${SCM} distutils-r1 systemd
+
+DESCRIPTION="Monitore filesystems events and execute Python methods or Shell commands."
+HOMEPAGE="https://github.com/spacefreak86/pymodmilter"
+if [ "${PV#9999}" != "${PV}" ] ; then
+	SRC_URI=""
+	KEYWORDS=""
+	# Needed for tests
+	S="${WORKDIR}/${PN}"
+	EGIT_CHECKOUT_DIR="${S}"
+else
+	SRC_URI="https://github.com/spacefreak86/${PN}/archive/${PV}.tar.gz -> ${P}.tar.gz"
+	KEYWORDS="amd64 x86"
+fi
+
+LICENSE="GPL-3"
+SLOT="0"
+
+IUSE="systemd"
+
+RDEPEND="dev-python/pyinotify[${PYTHON_USEDEP}]"
+
+python_install_all() {
+	distutils-r1_python_install_all
+
+	dodir /etc/${PN}
+	insinto /etc/${PN}
+	newins ${PN}/misc/config.py.example config.py
+
+	use systemd && systemd_dounit ${PN}/misc/${PN}.service
+
+	newinitd ${PN}/misc/openrc/${PN}.initd ${PN}
+	newconfd ${PN}/misc/openrc/${PN}.confd ${PN}
+}

distribution/pypi/build.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+set -e
+
+PYTHON=$(which python)
+
+script_dir=$(dirname "$(readlink -f -- "$BASH_SOURCE")")
+pkg_dir=$(realpath "${script_dir}"/../..)
+
+cd "${pkg_dir}"
+${PYTHON} setup.py clean
+${PYTHON} setup.py sdist bdist_wheel

distribution/pypi/distribute.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+set -e
+
+TWINE=$(which twine)
+
+script_dir=$(dirname "$(readlink -f -- "$BASH_SOURCE")")
+pkg_dir=$(realpath "${script_dir}/../..")
+
+cd "${pkg_dir}/dist"
+msg="Select version to distribute (cancel with CTRL+C):"
+echo "${msg}"
+select version in $(find . -maxdepth 1 -type f -name "pyinotifyd-*.*.*.tar.gz" -printf "%f\n" | sed "s#\.tar\.gz##g"); do
+  [ -n "${version}" ] && break
+  echo -e "\ninvalid choice\n\n${msg}"
+done
+${TWINE} upload "${version}"{.tar.gz,-*.whl}

pyinotifyd/__init__.py

@@ -14,6 +14,13 @@
 # along with pyinotifyd.  If not, see <http://www.gnu.org/licenses/>.
 #
 
+__all__ = [
+    "EventMap"
+    "Watch",
+    "Pyinotifyd",
+    "DaemonInstance",
+    "scheduler"]
+
 import argparse
 import asyncio
 import logging
@@ -22,21 +29,171 @@ import pyinotify
 import signal
 import sys
 
-from pyinotifyd.watch import Watch, EventMap
-from pyinotifyd._install import install, uninstall
+from pyinotify import ProcessEvent
 
-__version__ = "0.0.1"
+from pyinotifyd._install import install, uninstall
+from pyinotifyd.scheduler import TaskScheduler, Cancel
+
+__version__ = "0.0.2"
+
+
+class _SchedulerList:
+    def __init__(self, schedulers=[], loop=None):
+        if not isinstance(schedulers, list):
+            schedulers = [schedulers]
+
+        self._schedulers = schedulers
+        self._loop = (loop or asyncio.get_event_loop())
+
+    def process_event(self, event):
+        for scheduler in self._schedulers:
+            self._loop.create_task(scheduler.process_event(event))
+
+    def schedulers(self):
+        return self._schedulers
+
+
+class EventMap(ProcessEvent):
+    flags = {
+        **pyinotify.EventsCodes.OP_FLAGS,
+        **pyinotify.EventsCodes.EVENT_FLAGS}
+
+    def my_init(self, event_map=None, default_sched=None, loop=None,
+                logname="eventmap"):
+        self._map = {}
+        self._loop = (loop or asyncio.get_event_loop())
+
+        if default_sched is not None:
+            for flag in EventMap.flags:
+                self.set(flag, default_sched)
+
+        if event_map is not None:
+            assert isinstance(event_map, dict), \
+                f"event_map: expected {type(dict)}, got {type(event_map)}"
+            for flag, schedulers in event_map.items():
+                self.set_scheduler(flag, schedulers)
+
+        self._log = logging.getLogger((logname or __name__))
+
+    def set_scheduler(self, flag, schedulers):
+        assert flag in EventMap.flags, \
+            f"event_map: invalid flag: {flag}"
+        if schedulers is not None:
+            if not isinstance(schedulers, list):
+                schedulers = [schedulers]
+
+            instances = []
+            for scheduler in schedulers:
+                if issubclass(type(scheduler), TaskScheduler) or \
+                        isinstance(scheduler, Cancel):
+                    instances.append(scheduler)
+                else:
+                    instances.append(
+                        TaskScheduler(scheduler, loop=self._loop))
+
+            self._map[flag] = _SchedulerList(instances, loop=self._loop)
+
+        elif flag in self._map:
+            del self._map[flag]
+
+    def process_default(self, event):
+        msg = "received event"
+        for attr in [
+                "dir", "mask", "maskname", "pathname", "src_pathname", "wd"]:
+            value = getattr(event, attr, None)
+            if attr == "mask":
+                value = hex(value)
+            if value:
+                msg += f", {attr}={value}"
+
+        self._log.debug(msg)
+        maskname = event.maskname.split("|")[0]
+        if maskname in self._map:
+            self._map[maskname].process_event(event)
+
+    def schedulers(self):
+        schedulers = []
+        for scheduler_list in self._map.values():
+            schedulers.extend(
+                scheduler_list.schedulers())
+
+        return list(set(schedulers))
+
+
+class Watch:
+    def __init__(self, path, event_map=None, default_sched=None, rec=False,
+                 auto_add=False, logname="watch", loop=None):
+        assert isinstance(path, str), \
+            f"path: expected {type('')}, got {type(path)}"
+
+        if isinstance(event_map, EventMap):
+            self._event_map = event_map
+        else:
+            self._event_map = EventMap(
+                event_map=event_map, default_sched=default_sched)
+
+        assert isinstance(rec, bool), \
+            f"rec: expected {type(bool)}, got {type(rec)}"
+        assert isinstance(auto_add, bool), \
+            f"auto_add: expected {type(bool)}, got {type(auto_add)}"
+        logname = (logname or __name__)
+        self._loop = loop
+
+        self._path = path
+        self._rec = rec
+        self._auto_add = auto_add
+
+        self._watch_manager = pyinotify.WatchManager()
+        self._notifier = None
+        self._log = logging.getLogger(logname)
+
+    def path(self):
+        return self._path
+
+    def event_map(self):
+        return self._event_map
+
+    def start(self, loop=None):
+        loop = (loop or self._loop)
+        self._watch_manager.add_watch(self._path, pyinotify.ALL_EVENTS,
+                                      rec=self._rec, auto_add=self._auto_add,
+                                      do_glob=True)
+
+        self._notifier = pyinotify.AsyncioNotifier(
+            self._watch_manager, loop, default_proc_fun=self._event_map)
+
+    def stop(self):
+        self._notifier.stop()
+
+        self._notifier = None
 
 
 class Pyinotifyd:
-    def __init__(self, watches=[], shutdown_timeout=30, logname="daemon"):
+    name = "pyinotifyd"
+
+    def __init__(self, watches=[], shutdown_timeout=30, logname="daemon",
+                 loop=None):
         self.set_watches(watches)
         self.set_shutdown_timeout(shutdown_timeout)
         logname = (logname or __name__)
+        self._loop = (loop or asyncio.get_event_loop())
+
         self._log = logging.getLogger(logname)
-        self._loop = asyncio.get_event_loop()
-        self._notifiers = []
-        self._wm = pyinotify.WatchManager()
+
+    @staticmethod
+    def from_cfg_file(config_file):
+        config = {}
+        name = Pyinotifyd.name
+        exec("import logging", {}, config)
+        exec(f"from {name} import Pyinotifyd, Watch", {}, config)
+        exec(f"from {name}.scheduler import *", {}, config)
+        with open(config_file, "r") as fh:
+            exec(fh.read(), {}, config)
+        instance = config[f"{name}"]
+        assert isinstance(instance, Pyinotifyd), \
+            f"{name}: expected {type(Pyinotifyd)}, " \
+            f"got {type(instance)}"
+        return instance
 
     def set_watches(self, watches):
         if not isinstance(watches, list):
@@ -46,9 +203,15 @@ class Pyinotifyd:
             assert isinstance(watch, Watch), \
                 f"watches: expected {type(Watch)}, got {type(watch)}"
 
-        self._watches = watches
+        self._watches = []
+        self._watches.extend(watches)
 
-    def add_watch(self, *args, **kwargs):
+    def add_watch(self, *args, watch=None, **kwargs):
+        if watch:
+            assert isinstance(watch, Watch), \
+                f"watch: expected {type(Watch)}, got {type(watch)}"
+            self._watches.append(watch)
+        else:
             self._watches.append(Watch(*args, **kwargs))
 
     def set_shutdown_timeout(self, timeout):
@@ -57,58 +220,116 @@ class Pyinotifyd:
             f"got {type(timeout)}"
         self._shutdown_timeout = timeout
 
+    def schedulers(self):
+        schedulers = []
+        for w in self._watches:
+            schedulers.extend(w.event_map().schedulers())
+        return list(set(schedulers))
+
     def start(self, loop=None):
-        if not loop:
-            loop = self._loop
+        loop = (loop or self._loop)
 
-        self._log.info("starting")
         if len(self._watches) == 0:
-            self._log.warning("no watches configured, the daemon will not do anything")
+            self._log.warning(
+                "no watches configured, the daemon will not do anything")
+
         for watch in self._watches:
-            self._log.info(f"start watching '{watch.path}' for inotify events")
-            self._notifiers.append(watch.event_notifier(self._wm, loop))
+            self._log.info(
+                f"start listening for inotify events on '{watch.path()}'")
+            watch.start(loop)
 
-    def stop(self):
-        self._log.info("stop watching for inotify events")
-        for notifier in self._notifiers:
-            notifier.stop()
+    def pause(self):
+        for scheduler in self.schedulers():
+            scheduler.pause()
 
-        self._notifiers = []
-        return self._shutdown_timeout
+    async def shutdown(self):
+        schedulers = self.schedulers()
+
+        tasks = [s.shutdown(self._shutdown_timeout) for s in set(schedulers)]
+        if tasks:
+            await asyncio.gather(*tasks)
+
+        for watch in self._watches:
+            self._log.debug(
+                f"stop listening for inotify events on '{watch.path()}'")
+            watch.stop()
 
 
-async def _shutdown(signame, daemon, log):
-    log.info(f"got signal {signame}, graceful shutdown")
-    timeout = daemon.stop()
+class DaemonInstance:
+    def __init__(self, instance, logname="daemon"):
+        self._instance = instance
+        self._shutdown = False
+        self._log = logging.getLogger(logname)
+
+    def start(self):
+        self._instance.start()
+
+    async def shutdown(self, signame):
+        if self._shutdown:
+            self._log.warning(
+                f"got signal {signame}, but shutdown already in progress")
+            return
+
+        self._log.info(f"got signal {signame}, shutdown")
+        self._shutdown = True
+
+        try:
+            await self._instance.shutdown()
+
             pending = [t for t in asyncio.all_tasks()
                        if t is not asyncio.current_task()]
-    if len(pending) > 0:
-        log.info(
-            f"waiting {timeout}s for remaining tasks to complete")
-        try:
-            future = asyncio.gather(*pending)
-            await asyncio.wait_for(future, timeout)
-        except asyncio.TimeoutError:
-            log.warning("forcefully terminate remaining tasks")
-            future.cancel()
-            future.exception()
 
-    log.info("shutdown complete")
+            for task in pending:
+                task.cancel()
+
+            try:
+                await asyncio.gather(*pending)
+            except asyncio.CancelledError:
+                pass
+        except Exception as e:
+            self._log.exception(f"error during shutdown: {e}")
+
         asyncio.get_event_loop().stop()
+        self._shutdown = False
+        self._log.info("shutdown complete")
+
+    async def reload(self, signame, config_file, debug=False):
+        if self._shutdown:
+            self._log.info(
+                f"got signal {signame}, but shutdown already in progress")
+            return
+
+        self._log.info(f"got signal {signame}, reload config file")
+        try:
+            instance = Pyinotifyd.from_cfg_file(config_file)
+        except Exception as e:
+            logging.exception(
+                f"unable to reload config file '{config_file}': {e}")
+        else:
+            if debug:
+                logging.getLogger().setLevel(logging.DEBUG)
+
+            old_instance = self._instance
+
+            old_instance.pause()
+            instance.start()
+            asyncio.create_task(old_instance.shutdown())
+
+            self._instance = instance
 
 
 def main():
-    myname = "pyinotifyd"
+    name = Pyinotifyd.name
 
     parser = argparse.ArgumentParser(
-        description=myname,
+        description=name,
         formatter_class=lambda prog: argparse.HelpFormatter(
             prog, max_help_position=45, width=140))
     parser.add_argument(
         "-c",
         "--config",
-        help=f"path to config file (default: /etc/{myname}/config.py)",
-        default=f"/etc/{myname}/config.py")
+        help=f"path to config file (default: /etc/{name}/config.py)",
+        default=f"/etc/{name}/config.py")
     parser.add_argument(
         "-d",
         "--debug",
@@ -129,12 +350,12 @@ def main():
     exclusive.add_argument(
         "-i",
         "--install",
-        help="install systemd service file",
+        help="install service files and config",
         action="store_true")
     exclusive.add_argument(
         "-u",
         "--uninstall",
-        help="uninstall systemd service file",
+        help="uninstall service files and unmodified config",
         action="store_true")
     exclusive.add_argument(
         "-t",
@@ -145,7 +366,7 @@ def main():
     args = parser.parse_args()
 
     if args.version:
-        print(f"{myname} ({version})")
+        print(f"{name} ({__version__})")
         sys.exit(0)
 
     if args.list:
@@ -167,23 +388,20 @@ def main():
     root_logger.addHandler(ch)
 
     if args.install:
-        sys.exit(install(myname))
+        sys.exit(install(name))
 
     if args.uninstall:
-        sys.exit(uninstall(myname))
+        sys.exit(uninstall(name))
 
     try:
-        config = {}
-        exec(f"from {myname}.scheduler import *", config)
-        exec(f"from {myname}.filemanager import *", config)
-        with open(args.config, "r") as c:
-            exec(c.read(), globals(), config)
-        daemon = config[f"{myname}"]
-        assert isinstance(daemon, Pyinotifyd), \
-            f"{myname}: expected {type(Pyinotifyd)}, " \
-            f"got {type(daemon)}"
+        pyinotifyd = Pyinotifyd.from_cfg_file(args.config)
+        daemon = DaemonInstance(pyinotifyd)
     except Exception as e:
-        logging.exception(f"config file '{args.config}': {e}")
+        if args.debug:
+            logging.exception(f"config file: {e}")
+        else:
+            logging.error(f"config file: {e}")
+
         sys.exit(1)
 
     if args.configtest:
@@ -192,17 +410,22 @@ def main():
 
     if args.debug:
         root_logger.setLevel(loglevel)
+
     formatter = logging.Formatter(
-        f"%(asctime)s - {myname}/%(name)s - %(levelname)s - %(message)s")
+        f"%(asctime)s - {name}/%(name)s - %(levelname)s - %(message)s")
     ch.setFormatter(formatter)
 
-    log = logging.getLogger(myname)
     loop = asyncio.get_event_loop()
     for signame in ["SIGINT", "SIGTERM"]:
         loop.add_signal_handler(
             getattr(signal, signame),
-            lambda: asyncio.ensure_future(
-                _shutdown(signame, daemon, log)))
+            lambda: loop.create_task(
+                daemon.shutdown(signame)))
+
+    loop.add_signal_handler(
+        getattr(signal, "SIGHUP"),
+        lambda: loop.create_task(
+            daemon.reload("SIGHUP", args.config, args.debug)))
 
     daemon.start()
     loop.run_forever()

pyinotifyd/_install.py

@@ -99,7 +99,7 @@ def install(name):
     logging.info("install configuration file")
     config_dir = f"/etc/{name}"
     if os.path.isdir(config_dir):
-        logging.info(f" => directory {config_dir} exists already")
+        logging.info(f" => directory {config_dir} already exists")
     else:
         try:
             logging.info(f" => create directory {config_dir}")
@@ -109,7 +109,7 @@ def install(name):
             sys.exit(3)
 
     files = [
-        (f"{pkg_dir}/docs/config.py.example", f"{config_dir}/config.py")]
+        (f"{pkg_dir}/misc/config.py.default", f"{config_dir}/config.py")]
     for src, dst in files:
         _copy_missing_file(src, dst)
 

pyinotifyd/docs/config.py.example

@@ -1,83 +0,0 @@
-#!/usr/bin/env python3
-
-####################################
-#  Example usage of TaskScheduler  #
-####################################
-
-#async def custom_task(event, task_id):
-#    logging.info(f"{task_id}: execute example task: {event}")
-#
-#s = TaskScheduler(
-#    task=custom_task,
-#    files=True,
-#    dirs=False)
-
-
-#####################################################
-#  Example usage of TaskScheduler with FileManager  #
-#####################################################
-
-#rules=[{
-#    "action": "move",
-#    "src_re": r"^(?P<path>.*)",
-#    "dst_re": r"\g<path>.processed",
-#    "auto_create": True,
-#    "filemode": 0o755,
-#    "dirmode": 0o644,
-#    "user": "root",
-#    "goup": "root"}]
-#
-#fm = FileManager(
-#    rules=rules)
-#
-#s = TaskScheduler(
-#    task=fm.task,
-#    delay=10,
-#    files=True,
-#    dirs=False)
-
-
-#####################################
-#  Example usage of ShellScheduler  #
-#####################################
-
-#cmd = "/usr/local/bin/task.sh {maskname} {pathname} {src_pathname}"
-#s = ShellScheduler(
-#    cmd=cmd)
-
-
-###################
-#  Example watch  #
-###################
-
-#event_map = EventMap({
-#    "IN_ACCESS": None,
-#    "IN_ATTRIB": None,
-#    "IN_CLOSE_NOWRITE": None,
-#    "IN_CLOSE_WRITE": s.schedule,
-#    "IN_CREATE": None,
-#    "IN_DELETE": s.cancel,
-#    "IN_DELETE_SELF": s.cancel,
-#    "IN_IGNORED": None,
-#    "IN_MODIFY": s.cancel,
-#    "IN_MOVE_SELF": None,
-#    "IN_MOVED_FROM": s.cancel,
-#    "IN_MOVED_TO": s.schedule,
-#    "IN_OPEN": None,
-#    "IN_Q_OVERFLOW": None,
-#    "IN_UNMOUNT": s.cancel})
-#
-#watch = Watch(
-#    path="/tmp",
-#    event_map=event_map,
-#    rec=True,
-#    auto_add=True)
-
-
-########################
-#  Example pyinotifyd  #
-########################
-
-pyinotifyd = Pyinotifyd(
-    watches=[],
-    shutdown_timeout=30)

pyinotifyd/filemanager.py

@@ -1,193 +0,0 @@
-#!/usr/bin/env python3
-
-# pyinotifyd is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# pyinotifyd is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with pyinotifyd.  If not, see <http://www.gnu.org/licenses/>.
-#
-
-import logging
-import os
-import re
-import shutil
-
-
-class Rule:
-    valid_actions = ["copy", "move", "delete"]
-
-    def __init__(self, action, src_re, dst_re="", auto_create=False,
-                 dirmode=None, filemode=None, user=None, group=None,
-                 rec=False):
-        assert action in self.valid_actions, \
-            f"action: expected [{Rule.valid_actions.join(', ')}], got{action}"
-        self.action = action
-
-        self.src_re = re.compile(src_re)
-
-        assert isinstance(dst_re, str), \
-            f"dst_re: expected {type('')}, got {type(dst_re)}"
-        self.dst_re = dst_re
-
-        assert isinstance(auto_create, bool), \
-            f"auto_create: expected {type(bool)}, got {type(auto_create)}"
-        self.auto_create = auto_create
-
-        if dirmode is not None:
-            assert isinstance(dirmode, int), \
-                f"dirmode: expected {type(int)}, got {type(dirmode)}"
-        self.dirmode = dirmode
-
-        if filemode is not None:
-            assert isinstance(filemode, int), \
-                f"filemode: expected {type(int)}, got {type(filemode)}"
-        self.filemode = filemode
-
-        if user is not None:
-            assert isinstance(user, str), \
-                f"user: expected {type('')}, got {type(user)}"
-        self.user = user
-
-        if group is not None:
-            assert isinstance(group, str), \
-                f"group: expected {type('')}, got {type(group)}"
-        self.group = group
-
-        assert isinstance(rec, bool), \
-            f"rec: expected {type(bool)}, got {type(rec)}"
-        self.rec = rec
-
-
-class FileManager:
-    def __init__(self, rules, logname="filemgr"):
-        if not isinstance(rules, list):
-            rules = [rules]
-
-        for rule in rules:
-            assert isinstance(rule, Rule), \
-                f"rules: expected {type(Rule)}, got {type(rule)}"
-
-        self._rules = rules
-        self._log = logging.getLogger((logname or __name__))
-
-    def add_rule(self, *args, **kwargs):
-        self._rules.append(Rule(*args, **kwargs))
-
-    async def _chmod_and_chown(self, path, mode, chown, task_id):
-        if mode is not None:
-            self._log.debug(f"{task_id}: chmod {oct(mode)} '{path}'")
-            os.chmod(path, mode)
-
-        if chown is not None:
-            changes = ""
-            if chown[0] is not None:
-                changes = chown[0]
-
-            if chown[1] is not None:
-                changes = f"{changes}:{chown[1]}"
-
-            self._log.debug(f"{task_id}: chown {changes} '{path}'")
-            shutil.chown(path, *chown)
-
-    async def _set_mode_and_owner(self, path, rule, task_id):
-        if (rule.user is rule.group is None):
-            chown = None
-        else:
-            chown = (rule.user, rule.group)
-
-        work_on_dirs = not (rule.dirmode is chown is None)
-        work_on_files = not (rule.filemode is chown is None)
-
-        if os.path.isdir(path):
-            await self._chmod_and_chown(path, rule.dirmode, chown, task_id)
-            if work_on_dirs or work_on_files:
-                for root, dirs, files in os.walk(path):
-                    if work_on_dirs:
-                        for p in [os.path.join(root, d) for d in dirs]:
-                            await self._chmod_and_chown(
-                                p, rule.dirmode, chown, task_id)
-
-                    if work_on_files:
-                        for p in [os.path.join(root, f) for f in files]:
-                            await self._chmod_and_chown(
-                                p, rule.filemode, chown, task_id)
-        else:
-            await self._chmod_and_chown(path, rule.filemode, chown, task_id)
-
-    async def task(self, event, task_id):
-        path = event.pathname
-        match = None
-        for rule in self._rules:
-            match = rule.src_re.match(path)
-            if match:
-                break
-
-        if not match:
-            self._log.debug(
-                f"{task_id}: path '{path}' matches no rule in ruleset")
-            return
-
-        try:
-            if rule.action in ["copy", "move"]:
-                dst = rule.src_re.sub(rule.dst_re, path)
-                if not dst:
-                    raise RuntimeError(
-                        f"{task_id}: unable to {rule.action} '{path}', "
-                        f"resulting destination path is empty")
-
-                if os.path.exists(dst):
-                    raise RuntimeError(
-                        f"{task_id}: unable to move file from '{path} "
-                        f"to '{dst}', dstination path exists already")
-
-                dst_dir = os.path.dirname(dst)
-                if not os.path.isdir(dst_dir) and rule.auto_create:
-                    self._log.info(
-                        f"{task_id}: create directory '{dst_dir}'")
-                    first_subdir = dst_dir
-                    while not os.path.isdir(first_subdir):
-                        parent = os.path.dirname(first_subdir)
-                        if not os.path.isdir(parent):
-                            first_subdir = parent
-                        else:
-                            break
-                    os.makedirs(dst_dir)
-                    await self._set_mode_and_owner(first_subdir, rule, task_id)
-
-                self._log.info(
-                    f"{task_id}: {rule.action} '{path}' to '{dst}'")
-                if rule.action == "copy":
-                    if os.path.isdir(path):
-                        shutil.copytree(path, dst)
-                    else:
-                        shutil.copy2(path, dst)
-
-                else:
-                    os.rename(path, dst)
-
-                await self._set_mode_and_owner(dst, rule, task_id)
-
-            elif rule.action == "delete":
-                self._log.info(
-                    f"{task_id}: {rule.action} '{path}'")
-                if os.path.isdir(path):
-                    if rule.rec:
-                        shutil.rmtree(path)
-                    else:
-                        shutil.rmdir(path)
-
-                else:
-                    os.remove(path)
-
-        except RuntimeError as e:
-            self._log.error(f"{task_id}: {e}")
-
-        except Exception as e:
-            self._log.exception(f"{task_id}: {e}")

pyinotifyd/misc/config.py.default

@@ -0,0 +1,84 @@
+##########################
+#  TaskScheduler config  #
+##########################
+
+#async def custom_job(event, task_id):
+#    logging.info(f"{task_id}: execute example task: {event}")
+#
+#task_sched = TaskScheduler(
+#    job=custom_job,
+#    files=True,
+#    dirs=False,
+#    delay=10)
+
+
+###########################
+#  ShellScheduler config  #
+###########################
+
+#shell_sched = ShellScheduler(
+#    cmd="/usr/local/bin/task.sh {maskname} {pathname} {src_pathname}",
+#    files=True,
+#    dirs=False,
+#    delay=10)
+
+
+#################################
+#  FileManagerScheduler config  #
+#################################
+
+#move_rule = Rule(
+#    action="move",
+#    src_re="^/src_path/(?P<path>.*).to_move",
+#    dst_re="/dst_path/\g<path>.moved",
+#    auto_create=True,
+#    filemode=0o755,
+#    dirmode=0o644,
+#    user="root",
+#    goup="root",
+#    overwrite=False)
+
+#delete_rule = Rule(
+#    action="delete",
+#    src_re="^/src_path/(?P<path>.*).to_delete",
+#    rec=False)
+
+#file_sched = FileManagerScheduler(
+#    rules=[move_rule, delete_rule],
+#    files=True,
+#    dirs=False,
+#    delay=10)
+
+
+#####################
+#  EventMap config  #
+#####################
+
+#event_map = {
+#    "IN_ACCESS": None,
+#    "IN_ATTRIB": None,
+#    "IN_CLOSE_NOWRITE": None,
+#    "IN_CLOSE_WRITE": task_sched,
+#    "IN_CREATE": task_sched,
+#    "IN_DELETE": Cancel(task_sched),
+#    "IN_DELETE_SELF": Cancel(task_sched),
+#    "IN_IGNORED": None,
+#    "IN_MODIFY": Cancel(task_sched),
+#    "IN_MOVE_SELF": None,
+#    "IN_MOVED_FROM": Cancel(task_sched),
+#    "IN_MOVED_TO": task_sched,
+#    "IN_OPEN": None,
+#    "IN_Q_OVERFLOW": None,
+#    "IN_UNMOUNT": Cancel(task_sched)}
+
+
+#######################
+#  pyinotifyd config  #
+#######################
+
+#pyinotifyd = Pyinotifyd(shutdown_timeout=15)
+#pyinotifyd.add_watch(
+#    path="/watched/directory",
+#    event_map = event_map,
+#    rec=True,
+#    auto_add=True)

pyinotifyd/misc/openrc/pyinotifyd.confd

@@ -4,5 +4,8 @@
 # USER="daemon"
 # USER="daemon:nobody"
 
-# Optional parameters for pymodmilter
+# Set the shutdown timeout
+# SHUTDOWN_TIMEOUT=300
+
+# Optional command line options
 # PYINOTIFYD_OPTS=""

pyinotifyd/misc/openrc/pyinotifyd.initd

@@ -2,14 +2,16 @@
 
 user=${USER:-root}
 pyinotifyd_opts="${PYINOTIFYD_OPTS:-}"
+shutdown_timeout="${SHUTDOWN_TIMEOUT:-300}"
 
 pidfile="/run/${RC_SVCNAME}.pid"
 command="/usr/bin/pyinotifyd"
 command_args="${pyinotifyd_opts}"
 command_background=true
 start_stop_daemon_args="--user ${user}"
+retry="SIGTERM/${shutdown_timeout}"
 
-extra_commands="configtest"
+extra_commands="configtest reload"
 
 depend() {
 	need net
@@ -44,3 +46,9 @@ stop_pre() {
 		checkconfig || return $?
 	fi
 }
+
+reload() {
+	ebegin "Reloading ${SVCNAME}"
+	start-stop-daemon --signal HUP --pidfile "${pidfile}"
+	eend $?
+}

pyinotifyd/misc/systemd/pyinotifyd.service

@@ -5,6 +5,7 @@ After=fs.target
 [Service]
 Type=simple
 ExecStart=/usr/bin/pyinotifyd
+ExecReload=/bin/kill -HUP $MAINPID
 TimeoutStopSec=300
 
 [Install]

pyinotifyd/scheduler.py

@@ -12,131 +12,208 @@
 # along with pyinotifyd.  If not, see <http://www.gnu.org/licenses/>.
 #
 
+__all__ = [
+    "TaskScheduler",
+    "Cancel",
+    "ShellScheduler",
+    "FileManagerRule",
+    "FileManagerScheduler"]
+
 import asyncio
 import logging
+import os
+import re
+import shutil
+
+from dataclasses import dataclass
+from inspect import iscoroutinefunction
 from shlex import quote as shell_quote
 from uuid import uuid4
 
 
-class _Task:
-    def __init__(self, event, delay, task_id, task, callback=None,
-                 logname="task"):
-        self._event = event
-        self._path = event.pathname
-        self._delay = delay
-        self._task_id = task_id
-        self._job = task
-        self._callback = callback
+class SchedulerLogger(logging.LoggerAdapter):
+    def process(self, msg, kwargs):
+        if "event" in self.extra:
+            event = self.extra["event"]
+            msg = f"{msg}, mask={event.maskname}, path={event.pathname}"
 
-        self._task = None
-        self._log = logging.getLogger((logname or __name__))
+        if "id" in self.extra:
+            task_id = self.extra["id"]
+            msg = f"{msg}, task_id={task_id}"
 
-    async def _start(self):
-        if self._delay > 0:
-            await asyncio.sleep(self._delay)
-
-        if self._callback is not None:
-            self._callback(self._event)
-
-        self._task = None
-
-        self._log.info(f"execute task {self._task_id}")
-        await asyncio.shield(self._job(self._event, self._task_id))
-        self._log.info(f"task {self._task_id} finished")
-
-    def start(self):
-        if self._task is None:
-            self._task = asyncio.create_task(self._start())
-
-    def cancel(self):
-        if self._task is not None:
-            self._task.cancel()
-            self._task = None
-
-    def restart(self):
-        self.cancel()
-        self.start()
+        return msg, kwargs
 
 
 class TaskScheduler:
-    def __init__(self, task, files, dirs, delay=0, logname="sched"):
-        assert callable(task), \
-            f"task: expected callable, got {type(task)}"
-        self._task = task
 
-        assert isinstance(delay, int), \
-            f"delay: expected {type(int)}, got {type(delay)}"
-        self._delay = delay
+    @dataclass
+    class TaskState:
+        id: str = str(uuid4())
+        task: asyncio.Task = None
+        cancelable: bool = True
 
+    def __init__(self, job, files=True, dirs=False, delay=0, logname="sched",
+                 loop=None):
+        assert iscoroutinefunction(job), \
+            f"job: expected coroutine, got {type(job)}"
         assert isinstance(files, bool), \
             f"files: expected {type(bool)}, got {type(files)}"
-        self._files = files
-
         assert isinstance(dirs, bool), \
             f"dirs: expected {type(bool)}, got {type(dirs)}"
+        assert isinstance(delay, int), \
+            f"delay: expected {type(int)}, got {type(delay)}"
+
+        self._job = job
+        self._files = files
         self._dirs = dirs
+        self._delay = delay
+        self._log = logging.getLogger((logname or __name__))
+        self._loop = (loop or asyncio.get_event_loop())
 
         self._tasks = {}
-        self._logname = (logname or __name__)
-        self._log = logging.getLogger(self._logname)
+        self._pause = False
 
-    def _task_started(self, event):
-        path = event.pathname
-        if path in self._tasks:
-            del self._tasks[path]
+    def pause(self):
+        self._log.info("pause scheduler")
+        self._pause = True
 
-    def schedule(self, event):
-        self._log.debug(f"received {event}")
+    async def shutdown(self, timeout=None):
+        self._pause = True
+        pending = [t.task for t in self._tasks.values()]
+        if pending:
+            if timeout is None:
+                self._log.info(
+                    f"wait for {len(pending)} "
+                    f"remaining task(s) to complete")
+            else:
+                self._log.info(
+                    f"wait {timeout} seconds for {len(pending)} "
+                    f"remaining task(s) to complete")
+            done, pending = await asyncio.wait([*pending], timeout=timeout,
+                                               loop=self._loop)
+            if pending:
+                self._log.warning(
+                    f"shutdown timeout exceeded, "
+                    f"cancel {len(pending)} remaining task(s)")
+                for task in pending:
+                    task.cancel()
+                try:
+                    await asyncio.gather(*pending, loop=self._loop)
+                except asyncio.CancelledError:
+                    pass
+            else:
+                self._log.info("all remainig tasks completed")
 
-        if (not event.dir and not self._files) or \
-                (event.dir and not self._dirs):
+    async def _run_job(self, event, task_state, restart=False):
+        logger = SchedulerLogger(self._log, {
+            "event": event,
+            "id": task_state.id})
+
+        if self._delay > 0:
+            task_state.task = self._loop.create_task(
+                asyncio.sleep(self._delay, loop=self._loop))
+            try:
+                if restart:
+                    prefix = "re-"
+                else:
+                    prefix = ""
+
+                logger.info(f"{prefix}schedule task, delay={self._delay}")
+
+                await task_state.task
+            except asyncio.CancelledError:
                 return
 
-        path = event.pathname
-        maskname = event.maskname.split("|", 1)[0]
+        logger.info("start task")
 
-        if path in self._tasks:
-            task = self._tasks[path]
-            self._log.info(
-                f"received event {maskname} on '{path}', "
-                f"re-schedule task {task.task_id} (delay={self._delay}s)")
-            task.restart()
+        task_state.task = self._loop.create_task(
+            self._job(event, task_state.id))
+
+        try:
+            task_state.cancelable = False
+            await task_state.task
+        except asyncio.CancelledError:
+            logger.warning("ongoing task cancelled")
         else:
-            task_id = str(uuid4())
-            self._log.info(
-                f"received event {maskname} on '{path}', "
-                f"schedule task {task_id} (delay={self._delay}s)")
-            task = _Task(
-                event, self._delay, task_id, self._task,
-                callback=self._task_started, logname=self._logname)
-            self._tasks[path] = task
-            task.start()
+            self._log.info("task finished")
+        finally:
+            del self._tasks[event.pathname]
 
-    def cancel(self, event):
-        self._log.debug(f"received {event}")
+    async def process_event(self, event):
+        if not ((not event.dir and self._files) or
+                (event.dir and self._dirs)):
+            return
 
-        path = event.pathname
-        maskname = event.maskname.split("|", 1)[0]
-        if path in self._tasks:
-            task = self._tasks[path]
-            self._log.info(
-                f"received event {maskname} on '{path}', "
-                f"cancel scheduled task {task.task_id}")
-            task.cancel()
-            del self._tasks[path]
+        restart = False
+        try:
+            task_state = self._tasks[event.pathname]
+
+            logger = SchedulerLogger(self._log, {
+                "event": event,
+                "id": task_state.id})
+
+            if task_state.cancelable:
+                task_state.task.cancel()
+                if not self._pause:
+                    restart = True
+                else:
+                    logger.info("scheduled task cancelled")
+
+            else:
+                logger.warning("skip event due to ongoing task")
+                return
+
+        except KeyError:
+            task_state = TaskScheduler.TaskState()
+            self._tasks[event.pathname] = task_state
+
+        if not self._pause:
+            await self._run_job(event, task_state, restart)
+
+    async def process_cancel_event(self, event):
+        try:
+            task_state = self._tasks[event.pathname]
+        except KeyError:
+            return
+
+        logger = SchedulerLogger(self._log, {
+            "event": event,
+            "id": task_state.id})
+
+        if task_state.cancelable:
+            task_state.task.cancel()
+            logger.info("scheduled task cancelled")
+            task_state.task = None
+            del self._tasks[event.pathname]
+        else:
+            logger.warning("skip event due to ongoing task")
+
+
+class Cancel:
+    def __init__(self, task, *args, **kwargs):
+        assert issubclass(type(task), TaskScheduler), \
+            f"task: expected {type(TaskScheduler)}, got {type(task)}"
+
+        setattr(self, "process_event", task.process_cancel_event)
+
+    def pause(self):
+        pass
+
+    async def shutdown(self, timeout=None):
+        pass
 
 
 class ShellScheduler(TaskScheduler):
-    def __init__(self, cmd, task=None, *args, **kwargs):
+    def __init__(self, cmd, job=None, *args, **kwargs):
+        super().__init__(*args, **kwargs, job=self._shell_job)
+
         assert isinstance(cmd, str), \
             f"cmd: expected {type('')}, got {type(cmd)}"
+
         self._cmd = cmd
 
-        super().__init__(*args, task=self.task, **kwargs)
-
-    async def task(self, event, task_id):
+    async def _shell_job(self, event, task_id):
         maskname = event.maskname.split("|", 1)[0]
-
         if hasattr(event, "src_pathname"):
             src_pathname = event.src_pathname
         else:
@@ -146,6 +223,213 @@ class ShellScheduler(TaskScheduler):
             "{pathname}", shell_quote(event.pathname)).replace(
                 "{src_pathname}", shell_quote(src_pathname))
 
-        self._log.info(f"{task_id}: execute shell command: {cmd}")
-        proc = await asyncio.create_subprocess_shell(cmd)
+        logger = SchedulerLogger(self._log, {
+            "event": event,
+            "id": task_id})
+
+        logger.info(f"execute shell command, cmd={cmd}")
+        try:
+            proc = await asyncio.create_subprocess_shell(cmd, loop=self._loop)
             await proc.communicate()
+        except Exception as e:
+            logger.error(e)
+
+
+class FileManagerRule:
+    valid_actions = ["copy", "move", "delete"]
+
+    def __init__(self, action, src_re, dst_re="", auto_create=False,
+                 overwrite=False, dirmode=None, filemode=None, user=None,
+                 group=None, rec=False):
+        valid = f"{', '.join(FileManagerRule.valid_actions)}"
+        assert action in self.valid_actions, \
+            f"action: expected [{valid}], got{action}"
+        assert isinstance(src_re, str), \
+            f"src_re: expected {type('')}, got {type(src_re)}"
+        assert isinstance(dst_re, str), \
+            f"dst_re: expected {type('')}, got {type(dst_re)}"
+        assert isinstance(auto_create, bool), \
+            f"auto_create: expected {type(bool)}, got {type(auto_create)}"
+        assert isinstance(overwrite, bool), \
+            f"auto_create: expected {type(bool)}, got {type(auto_create)}"
+        assert dirmode is None or isinstance(dirmode, int), \
+            f"dirmode: expected {type(int)}, got {type(dirmode)}"
+        assert filemode is None or isinstance(filemode, int), \
+            f"filemode: expected {type(int)}, got {type(filemode)}"
+        assert user is None or isinstance(user, str), \
+            f"user: expected {type('')}, got {type(user)}"
+        assert group is None or isinstance(group, str), \
+            f"group: expected {type('')}, got {type(group)}"
+        assert isinstance(rec, bool), \
+            f"rec: expected {type(bool)}, got {type(rec)}"
+
+        self.action = action
+        self.src_re = re.compile(src_re)
+        self.dst_re = dst_re
+        self.auto_create = auto_create
+        self.overwrite = overwrite
+        self.dirmode = dirmode
+        self.filemode = filemode
+        self.user = user
+        self.group = group
+        self.rec = rec
+
+
+class FileManagerScheduler(TaskScheduler):
+    def __init__(self, rules, job=None, *args, **kwargs):
+        super().__init__(*args, **kwargs, job=self._manager_job)
+
+        if not isinstance(rules, list):
+            rules = [rules]
+
+        for rule in rules:
+            assert isinstance(rule, FileManagerRule), \
+                f"rules: expected {type(FileManagerRule)}, got {type(rule)}"
+
+        self._rules = rules
+
+    def _get_rule_by_event(self, event):
+        rule = None
+        for r in self._rules:
+            if r.src_re.match(event.pathname):
+                rule = r
+                break
+
+        return rule
+
+    async def process_event(self, event):
+        if not ((not event.dir and self._files) or
+                (event.dir and self._dirs)):
+            return
+
+        if self._get_rule_by_event(event):
+            await super().process_event(event)
+        else:
+            logger = SchedulerLogger(self._log, {"event": event})
+            logger.debug("no rule in ruleset matches")
+
+    async def _chmod_and_chown(self, path, mode, chown, logger=None):
+        logger = (logger or self._log)
+
+        if mode is not None:
+            logger.debug(f"chmod {oct(mode)}")
+            os.chmod(path, mode)
+
+        if chown is not None:
+            changes = ""
+            if chown[0] is not None:
+                changes = chown[0]
+
+            if chown[1] is not None:
+                changes = f"{changes}:{chown[1]}"
+
+            logger.debug(f"chown {changes}")
+            shutil.chown(path, *chown)
+
+    async def _set_mode_and_owner(self, path, rule, logger=None):
+        logger = (logger or self._log)
+
+        if (rule.user is rule.group is None):
+            chown = None
+        else:
+            chown = (rule.user, rule.group)
+
+        if os.path.isdir(path):
+            mode = rule.dirmode
+        else:
+            mode = rule.filemode
+
+        await self._chmod_and_chown(path, mode, chown, logger)
+
+        if not os.path.isdir(path):
+            return
+
+        work_on_dirs = not (rule.dirmode is chown is None)
+        work_on_files = not (rule.filemode is chown is None)
+
+        if work_on_dirs or work_on_files:
+            for root, dirs, files in os.walk(path):
+                if work_on_dirs:
+                    for p in [os.path.join(root, d) for d in dirs]:
+                        await self._chmod_and_chown(
+                            p, rule.dirmode, chown, logger)
+
+                if work_on_files:
+                    for p in [os.path.join(root, f) for f in files]:
+                        await self._chmod_and_chown(
+                            p, rule.filemode, chown, logger)
+
+    async def _manager_job(self, event, task_id):
+        rule = self._get_rule_by_event(event)
+        if not rule:
+            return
+
+        logger = SchedulerLogger(self._log, {"id": task_id})
+
+        try:
+            path = event.pathname
+            if rule.action in ["copy", "move"]:
+                dst = rule.src_re.sub(rule.dst_re, path)
+                if not dst:
+                    raise RuntimeError(
+                        f"unable to {rule.action} '{path}', "
+                        f"resulting destination path is empty")
+
+                if os.path.exists(dst) and not rule.overwrite:
+                    raise RuntimeError(
+                        f"unable to {rule.action} file from '{path} "
+                        f"to '{dst}', path already exists")
+
+                dst_dir = os.path.dirname(dst)
+                if not os.path.isdir(dst_dir) and rule.auto_create:
+                    logger.info(f"create directory '{dst_dir}'")
+                    first_subdir = dst_dir
+                    while not os.path.isdir(first_subdir):
+                        parent = os.path.dirname(first_subdir)
+                        if not os.path.isdir(parent):
+                            first_subdir = parent
+                        else:
+                            break
+
+                    try:
+                        os.makedirs(dst_dir)
+                        await self._set_mode_and_owner(
+                            first_subdir, rule, logger)
+                    except Exception as e:
+                        raise RuntimeError(e)
+
+                logger.info(f"{rule.action} '{path}' to '{dst}'")
+
+                try:
+                    if rule.action == "copy":
+                        if os.path.isdir(path):
+                            shutil.copytree(path, dst)
+                        else:
+                            shutil.copy2(path, dst)
+
+                    else:
+                        os.rename(path, dst)
+
+                    await self._set_mode_and_owner(dst, rule, logger)
+                except Exception as e:
+                    raise RuntimeError(e)
+
+            elif rule.action == "delete":
+                logger.info(f"{rule.action} '{path}'")
+                try:
+                    if os.path.isdir(path):
+                        if rule.rec:
+                            shutil.rmtree(path)
+                        else:
+                            shutil.rmdir(path)
+
+                    else:
+                        os.remove(path)
+                except Exception as e:
+                    raise RuntimeError(e)
+
+        except RuntimeError as e:
+            logger.error(e)
+
+        except Exception as e:
+            logger.exception(e)

pyinotifyd/watch.py

@@ -1,115 +0,0 @@
-#!/usr/bin/env python3
-
-# pyinotifyd is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# pyinotifyd is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with pyinotifyd.  If not, see <http://www.gnu.org/licenses/>.
-#
-
-import asyncio
-import pyinotify
-
-
-class EventMap:
-    flags = {
-        **pyinotify.EventsCodes.OP_FLAGS,
-        **pyinotify.EventsCodes.EVENT_FLAGS}
-
-    def __init__(self, event_map=None, default_task=None):
-        self._map = {}
-
-        if default_task is not None:
-            assert callable(default_task), \
-                f"default_task: expected callable, got {type(default_task)}"
-            for flag in EventMap.flags:
-                self.set(flag, default_task)
-
-        if event_map is not None:
-            assert isinstance(event_map, dict), \
-                f"event_map: expected {type(dict)}, got {type(event_map)}"
-            for flag, task in event_map.items():
-                self.set(flag, task)
-
-    def items(self):
-        return self._map.items()
-
-    def set(self, flag, values):
-        assert flag in EventMap.flags, \
-            f"event_map: invalid flag: {flag}"
-        if values is not None:
-            if not isinstance(values, list):
-                values = [values]
-
-            for value in values:
-                assert callable(value), \
-                    f"event_map: {flag}: expected callable, got {type(value)}"
-
-            self._map[flag] = values
-        elif flag in self._map:
-            del self._map[flag]
-
-
-class _TaskList:
-    def __init__(self, tasks=[]):
-        if not isinstance(tasks, list):
-            tasks = [tasks]
-
-        self._tasks = tasks
-
-    def add(self, task):
-        self._tasks.append(task)
-
-    def remove(self, task):
-        self._tasks.remove(task)
-
-    def execute(self, event):
-        for task in self._tasks:
-            task(event)
-
-
-class Watch:
-    def __init__(self, path, event_map, rec=False, auto_add=False):
-        assert isinstance(path, str), \
-            f"path: expected {type('')}, got {type(path)}"
-        self.path = path
-
-        if isinstance(event_map, EventMap):
-            self.event_map = event_map
-        elif isinstance(event_map, dict):
-            self.event_map = EventMap(event_map)
-        else:
-            raise AssertionError(
-                f"event_map: expected {type(EventMap)} or {type(dict)}, "
-                f"got {type(event_map)}")
-
-        assert isinstance(rec, bool), \
-            f"rec: expected {type(bool)}, got {type(rec)}"
-        self.rec = rec
-
-        assert isinstance(auto_add, bool), \
-            f"auto_add: expected {type(bool)}, got {type(auto_add)}"
-        self.auto_add = auto_add
-
-    def event_notifier(self, wm, loop=asyncio.get_event_loop()):
-        handler = pyinotify.ProcessEvent()
-        mask = False
-        for flag, values in self.event_map.items():
-            setattr(handler, f"process_{flag}", _TaskList(values).execute)
-            if mask:
-                mask = mask | EventMap.flags[flag]
-            else:
-                mask = EventMap.flags[flag]
-
-        wm.add_watch(
-            self.path, mask, rec=self.rec, auto_add=self.auto_add,
-            do_glob=True)
-
-        return pyinotify.AsyncioNotifier(wm, loop, default_proc_fun=handler)