nojhan/watchyap
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat: allow multiple targets

Browse source
This commit is contained in:
Johann Dreo 2025-08-10 14:17:44 +02:00
commit b5ca18992d
2 changed files with 115 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

51
README.md
View file

@ -1,7 +1,7 @@
FlickSave -- automatic snapshot each time you hit save FlickSave -- automatic snapshot each time you hit save
====================================================== ======================================================
FlickSave automagically perform an action each time you touch a watched file. FlickSave automagically perform an action each time you touch watched files.
For instance, FlickSave can backup a timestamped snapshot of a file that you edit with any program, without getting in your way. For instance, FlickSave can backup a timestamped snapshot of a file that you edit with any program, without getting in your way.
@ -15,29 +15,30 @@ You can even do an automatic Git commit the same way.
## Usage ## Usage
You should start FlickSave with the file you want to save as an argument You should start FlickSave with the files you want to save as argument(s)
and let it run during the time you want to let it perform your action. and let it run during the time you want to perform your action(s).
Once started, it will watch for modifications on the given file Once started, it will watch for modifications on the given file(s)
and, for instance, create the snapshots by himself each time something happens. and, for instance, create the snapshots by himself each time something happens.
Currently, FlickSave can perform the following actions: Currently, FlickSave can perform the following actions:
- `--save`: save a timestamped snapshot of the watched file, - `--save`: save a timestamped snapshot of the watched file(s),
- `--inkscape`: export a timestamped PNG of a watched SVG, - `--inkscape`: export a timestamped PNG(s) of a watched SVG(s),
- `--git`: commit the modifications of the watched file, - `--git`: commit the modifications of the watched file(s),
- `--log`: print a message stating that the watched file was touched. - `--log`: print a message stating that the watched file(s) were touched.
- `--dbus`: send a notification to the system's D-Bus. - `--dbus`: send a notification to the system's D-Bus.
You can pass multiple actions. You can pass multiple actions.
For instance, you can both export a PNG and do a Git commit. For instance, you can both export a PNG and do a Git commit.
By default, timestamped snapshots have the same name than the targeted file, with the addition of the timestamp. By default, timestamped snapshots have the same name than the targeted files,
with the addition of the timestamp.
For instance, with a target file named `test.svg`, a snapshot will look like `test_2017-11-28T21:02:59.svg`. For instance, with a target file named `test.svg`, a snapshot will look like `test_2017-11-28T21:02:59.svg`.
### Synopsis ### Synopsis
`flicksave.py [-h] [--save] [--inkscape] [--git] [--log] [--dbus] [-d DIRECTORY] [-y DELAY] [-s SEPARATOR] [-t TEMPLATE] [-w] [-v {DEBUG,INFO,WARNING,ERROR}] [-e {opened,moved,deleted,created,modified,closed}...] target` `flicksave.py [-h] [--save] [--inkscape] [--git] [--log] [--dbus] [-d DIRECTORY] [-y DELAY] [-s SEPARATOR] [-t TEMPLATE] [-w] [-v {DEBUG,INFO,WARNING,ERROR}] [-e {opened,moved,deleted,created,modified,closed}...] target [target ...]`
### Required positional argument ### Required positional argument
@ -73,23 +74,33 @@ As usual, hit `Ctrl-C` (or close the terminal) to stop FlickSave.
If you want to specify a directory in which to put your snapshots, with no more than one separate file every minute: If you want to specify a directory in which to put your snapshots, with no more than one separate file every minute:
$ flicksave --inkscape -d flicksave -y 60 my_file.svg $ flicksave --inkscape --directory flicksave --delay 60 my_file.svg
You can call sevral actions, for instance export a PNG and commit the source SVG:
$ flicksave --inkscape --git my_file.svg
You can use your shell to create the file list:
$ flicksave --git *.py
And even handle files in subdirectories:
$ flicksave --log */*.log
You may want to save both the file before and after it was modified by any You may want to save both the file before and after it was modified by any
program: program:
$ flicksave --save --events opened closed --no-overwrite my_file $ flicksave --save --events opened closed --no-overwrite my_file
If you want to see what's going on and when the action(s) are called, You can export PNGs and commit their source across sub-directories,
ask it to be more verbose: and be notified when it's done:
$ touch test.txt $ flicksave --inkscape --git --dbus --events closed */*.svg
$ flicksave --log -v INFO test.txt &
[1] 4303 You can also prepare the list of files to watch by using a subcommand:
echo "." >> test.txt
2017-11-29T21:15:51 -- ./test.txt -> ./test_2017-11-29T21:15:51.txt $ flicksave --log $(find . -type f -name *.png | grep -v test)
$ kill 4303
[1]+ Complété ./flicksave.py -v INFO test.txt
## Authors ## Authors

142
flicksave.py
View file

@ -2,6 +2,7 @@
#encoding: utf-8 #encoding: utf-8
import os import os
import sys
import time import time
import glob import glob
import shutil import shutil
@ -12,8 +13,8 @@ import subprocess
try: try:
from sdbus_block.notifications import FreedesktopNotifications from sdbus_block.notifications import FreedesktopNotifications
except Exception as e: except Exception as e:
logging.error(e) print("WARNING:", e, file=sys.stderr)
logging.error("Suitable `sdbus` module cannot be loaded, the --dbus action is disabled.") print("WARNING: Suitable `sdbus` module cannot be loaded, the --dbus action is disabled.", file=sys.stderr)
HAS_DBUS = False HAS_DBUS = False
else: else:
HAS_DBUS = True HAS_DBUS = True
@ -39,27 +40,66 @@ class Flick:
class Flicker: class Flicker:
"""Build a new timestamped file name.""" """Build a new timestamped file name."""
def __init__(self, target, delay=10): def __init__(self, targets, delay=10):
self.target = target self.targets = targets
self.delay = delay self.delay = delay
self.last_date = None self.last_date = {}
def __iter__(self): def __call__(self):
return self for target in self.targets:
name,ext = os.path.splitext(target)
date_now = datetime.datetime.now()
def next(self): if target in self.last_date:
name,ext = os.path.splitext(self.target) logging.debug(f"Target already actionned at {self.last_date[target]}")
date_now = datetime.datetime.now() logging.debug("Current date: %s", date_now.isoformat())
assert(self.last_date[target] <= date_now)
if self.last_date: if date_now - self.last_date[target] < datetime.timedelta(seconds=self.delay):
logging.debug("Current date: %s", date_now.isoformat()) logging.debug("Delay passed, current delta: %s < %s", date_now - self.last_date[target],datetime.timedelta(seconds=self.delay))
assert(self.last_date <= date_now) yield Flick(target, self.last_date[target], ext)
self.last_date[target] = date_now
else:
logging.debug("Delay not passed")
else:
self.last_date[target] = date_now
yield Flick(target, date_now, ext)
if date_now - self.last_date < datetime.timedelta(seconds=self.delay):
logging.debug("Current delta: %s < %s", date_now - self.last_date,datetime.timedelta(seconds=self.delay))
return Flick(self.target, self.last_date, ext)
return Flick(self.target, date_now, ext) class Handler(FileSystemEventHandler):
"""Event handler, will call a sequence of operators at each event matching the target."""
def __init__(self, operators, flicker, watched_events = ["modified"] ):
self.flicker = flicker
self.ops = operators
self.watched_events = watched_events
def on_any_event(self, event):
logging.debug(f"##############################\n\t\t\tRECEIVED EVENT:")
logging.debug(f"{event}")
super(Handler,self).on_any_event(event)
for flick in self.flicker():
logging.debug(f"Created flick: {flick}")
# Watchdog cannot watch single files (FIXME bugreport?),
# so we filter it in this event handler.
if (not event.is_directory) and os.path.abspath(event.src_path) == os.path.abspath(flick.target) and event.event_type in self.watched_events:
logging.debug("Handle event")
logging.debug("New flick for %s: %s", event.src_path, flick)
for op in self.ops:
logging.debug("Calling %s", op)
op(os.path.abspath(event.src_path), flick, event)
else:
is_dir = ""
if event.is_directory:
is_dir = " is a directory"
is_not_target = ""
if os.path.abspath(event.src_path) != os.path.abspath(flick.target):
is_not_target = " is not a target"
is_not_watched = ""
if event.event_type not in self.watched_events:
is_not_watched = " is not a watched event"
logging.debug("Not handling event:" + ", ".join([i for i in [is_dir, is_not_target, is_not_watched] if i != ""]))
class Operator: class Operator:
@ -81,6 +121,9 @@ class Save(Operator):
self.date_sep = date_sep self.date_sep = date_sep
self.no_overwrite = no_overwrite self.no_overwrite = no_overwrite
# Alternate last_date, xtracted from the file's timestamp.
self.last_date = None
# Make a glob search expression with the date template. # Make a glob search expression with the date template.
self.fields = {'Y':4,'m':2,'d':2,'H':2,'M':2,'S':2} self.fields = {'Y':4,'m':2,'d':2,'H':2,'M':2,'S':2}
self.glob_template = self.date_template self.glob_template = self.date_template
@ -211,7 +254,7 @@ if HAS_DBUS:
return "Log()" return "Log()"
def __call__(self, target, flick, event, alt_ext = None): def __call__(self, target, flick, event, alt_ext = None):
logging.info("Event(s) seen for {}: {}".format(target,flick)) logging.info(f"File {target} was {event.event_type}")
notif = FreedesktopNotifications() notif = FreedesktopNotifications()
hints = notif.create_hint( hints = notif.create_hint(
urgency = 0, urgency = 0,
@ -226,45 +269,25 @@ if HAS_DBUS:
) )
class Handler(FileSystemEventHandler): def flicksave(globbed_targets, operators=None, delay=10, watched=["modified"]):
"""Event handler, will call a sequence of operators at each event matching the target."""
def __init__(self, operators, flicker, watched_types = ["modified"] ):
self.flicker = flicker
self.ops = operators
self.watched_types = watched_types
def on_any_event(self, event):
logging.debug("Received event %s",event)
super(Handler,self).on_any_event(event)
# Watchdog cannot watch single files (FIXME bugreport?),
# so we filter it in this event handler.
if (not event.is_directory) and os.path.abspath(event.src_path) == os.path.abspath(self.flicker.target) and event.event_type in self.watched_types:
logging.debug("Handle event")
flick = self.flicker.next()
logging.debug("New flicker for %s: %s", event.src_path, flick)
for op in self.ops:
logging.debug("Calling %s", op)
op(os.path.abspath(event.src_path), flick, event)
else:
logging.debug("Not handling event: file={}, is_directory={}, watched_types={}".format(os.path.abspath(event.src_path),event.is_directory, self.watched_types))
def flicksave(target, operators=None, delay=10, watched=["modified"]):
"""Start the watch thread.""" """Start the watch thread."""
# Handle files specified without a directory.
root = os.path.dirname(target)
if not root:
root = '.'
target = os.path.join(root,target)
flicker = Flicker(target, delay) targets = [os.path.abspath(p) for p in globbed_targets]
logging.debug(f"Watching {len(targets)} files.")
root = os.path.commonpath(targets)
if os.path.isfile(root):
root = os.path.dirname(root)
logging.debug(f"Watching at root directory: `{root}`.")
flicker = Flicker(targets, delay)
handler = Handler(operators, flicker, watched) handler = Handler(operators, flicker, watched)
# Start the watch thread. # Start the watch thread.
observer = Observer() observer = Observer()
observer.schedule(handler, root) observer.schedule(handler, root)
logging.debug("Start watching...")
observer.start() observer.start()
try: try:
while True: while True:
@ -303,8 +326,8 @@ if __name__=="__main__":
""") """)
# Required argument. # Required argument.
parser.add_argument("target", parser.add_argument("targets", nargs="+",
help="The file to save each time it's modified.") help="The files to save each time it's modified.")
# Optional arguments. # Optional arguments.
parser.add_argument("-d","--directory", default='.', parser.add_argument("-d","--directory", default='.',
@ -336,16 +359,16 @@ if __name__=="__main__":
existing = { existing = {
"save": "save":
["Save a snapshot of the target file.", ["Save a snapshot of the target files.",
None], None],
"inkscape": "inkscape":
["Save a PNG snapshot of the file, using inkscape.", ["Save a PNG snapshot of the files, using inkscape.",
None], None],
"git": "git":
["Commit the target file if it has been modified [the repository should be set-up].", ["Commit the target files if it has been modified [the repository should be set-up].",
None], None],
"log": "log":
["Log when the target file is modified.", ["Log when the target files are touched.",
None], None],
} }
if HAS_DBUS: if HAS_DBUS:
@ -355,8 +378,6 @@ if __name__=="__main__":
def help(name): def help(name):
return existing[name][0] return existing[name][0]
def instance(name):
return existing[name][1]
for name in existing: for name in existing:
parser.add_argument("--"+name, help=help(name), action="store_true") parser.add_argument("--"+name, help=help(name), action="store_true")
@ -365,6 +386,8 @@ if __name__=="__main__":
logging.basicConfig(level=log_as[asked.verbose], format='%(asctime)s -- %(message)s', datefmt=asked.template) logging.basicConfig(level=log_as[asked.verbose], format='%(asctime)s -- %(message)s', datefmt=asked.template)
logging.debug("Load actions...")
# Add instances, now that we have all parameters. # Add instances, now that we have all parameters.
available = existing; available = existing;
available["save"][1] = Save(asked.directory, asked.separator, asked.template, asked.no_overwrite) available["save"][1] = Save(asked.directory, asked.separator, asked.template, asked.no_overwrite)
@ -387,6 +410,8 @@ if __name__=="__main__":
operators = [] operators = []
requested = vars(asked) requested = vars(asked)
def instance(name):
return existing[name][1]
for it in [iz for iz in requested if iz in available and requested[iz]==True]: for it in [iz for iz in requested if iz in available and requested[iz]==True]:
operators.append(instance(it)) operators.append(instance(it))
@ -404,5 +429,6 @@ if __name__=="__main__":
logging.debug("\t%s", op) logging.debug("\t%s", op)
# Start it. # Start it.
flicksave(asked.target, operators, asked.delay, asked.events) logging.debug(asked.targets)
flicksave(asked.targets, operators, asked.delay, asked.events)