hjp
/
kitsune
1
0
Fork 0
Code Issues 1 Pull Requests Releases Wiki Activity

Watch multiple files or directories instead of a single directory

This commit is contained in:
Peter J. Holzer 2021-12-18 11:25:56 +01:00 committed by Peter J. Holzer
parent 17c6e87a89
commit 8954588ab4
1 changed files with 32 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
kitsune
View File

@ -3,44 +3,57 @@
import datetime import datetime
import os import os
import sys import sys
import stat
import time import time
class WatchedFile: class WatchedFile:
def __init__(self, name): def __init__(self, path):
self.name = name self.path = path
self.reopen() self.reopen()
def reopen(self): def reopen(self):
self.fd = open(self.name, errors='replace') self.fd = open(self.path, errors='replace')
self.fileno = self.fd.fileno() self.fileno = self.fd.fileno()
self.last_ts = 0 self.last_ts = 0
def format_ts(ts_ns): def format_ts(ts_ns):
return datetime.datetime.fromtimestamp(ts_ns / 1E9).strftime("%H:%M:%S.%f") return datetime.datetime.fromtimestamp(ts_ns / 1E9).strftime("%H:%M:%S.%f")
def watch(dir): def watch(args):
os.chdir(dir) watched_dirs = []
last_ts = 0
watched_files = {} watched_files = {}
filename_length = 0 filename_length = 0
for a in args:
st = os.stat(a)
if stat.S_ISDIR(st.st_mode):
watched_dirs.append(a)
elif stat.S_ISREG(st.st_mode):
watched_files[a] = WatchedFile(a)
if len(a) > filename_length:
filename_length = len(a)
else:
print(a, "is not a file or directory - skipping", file=sys.stderr)
last_ts = 0
while True: while True:
# are there new files? # are there new files?
st = os.stat(".") for d in watched_dirs:
if st.st_mtime_ns > last_ts: st = os.stat(d)
last_ts = st.st_mtime_ns if st.st_mtime_ns > last_ts:
for de in os.scandir("."): last_ts = st.st_mtime_ns
if de.is_file(): for de in os.scandir(d):
if de.name not in watched_files: if de.is_file():
watched_files[de.name] = WatchedFile(de.name) if de.path not in watched_files:
if len(de.name) > filename_length: watched_files[de.path] = WatchedFile(de.path)
filename_length = len(de.name) if len(de.path) > filename_length:
filename_length = len(de.path)
# has any of the files changed # has any of the files changed
for f in watched_files.values(): for f in watched_files.values():
# XXX - we should also detect replaced files. # XXX - we should also detect replaced files.
st = os.stat(f.fileno) st = os.stat(f.fileno)
try: try:
stf = os.stat(f.name) stf = os.stat(f.path)
if stf.st_ino != st.st_ino: if stf.st_ino != st.st_ino:
f.reopen() f.reopen()
except FileNotFoundError: except FileNotFoundError:
@ -55,10 +68,10 @@ def watch(dir):
if lines[-1] == "": if lines[-1] == "":
lines.pop() lines.pop()
for ln in lines: for ln in lines:
print(f"{f.name:{filename_length}}", format_ts(f.last_ts), ln) print(f"{f.path:{filename_length}}", format_ts(f.last_ts), ln)
time.sleep(0.1) time.sleep(0.1)
if __name__ == "__main__": if __name__ == "__main__":
dir = sys.argv[1] if len(sys.argv) > 1 else "." args = sys.argv[1:] if len(sys.argv) > 1 else ["."]
watch(dir) watch(args)