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

Seek to end unless --start given

This commit is contained in:
Peter J. Holzer 2022-03-18 23:52:42 +01:00 committed by Peter J. Holzer
parent d17173aaff
commit 9ab32e5b6b
1 changed files with 14 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
kitsune
View File

@ -9,19 +9,24 @@ import stat
import time
class WatchedFile:
def __init__(self, path):
def __init__(self, path, seek_to_end):
self.path = path
self.seek_to_end = seek_to_end
self.reopen()
def reopen(self):
self.fd = open(self.path, errors='replace')
self.fileno = self.fd.fileno()
self.last_ts = 0
if self.seek_to_end:
self.fd.seek(0, 2)
self.seek_to_end = False
def format_ts(ts_ns):
return datetime.datetime.fromtimestamp(ts_ns / 1E9).strftime("%H:%M:%S.%f")
def watch(args):
seek_to_end = not args.start
watched_dirs = []
watched_files = {}
filename_length = 0
@ -30,7 +35,7 @@ def watch(args):
if stat.S_ISDIR(st.st_mode):
watched_dirs.append(a)
elif stat.S_ISREG(st.st_mode):
watched_files[a] = WatchedFile(a)
watched_files[a] = WatchedFile(a, seek_to_end)
if len(a) > filename_length:
filename_length = len(a)
else:
@ -47,7 +52,7 @@ def watch(args):
if de.is_file():
if args.match_filename is None or fnmatch.fnmatch(de.name, args.match_filename):
if de.path not in watched_files:
watched_files[de.path] = WatchedFile(de.path)
watched_files[de.path] = WatchedFile(de.path, seek_to_end)
if len(de.path) > filename_length:
filename_length = len(de.path)
@ -65,6 +70,10 @@ def watch(args):
# or just mark as deleted?
pass
if st.st_mtime_ns > f.last_ts:
if st.st_size < f.fd.tell():
# We are beyond the end of the file, so it has probably
# been truncated and rewritten - read from beginning
f.fd.seek(0, 0)
f.last_ts = st.st_mtime_ns
new_content = f.fd.read()
lines = new_content.split("\n")
@ -73,9 +82,11 @@ def watch(args):
for ln in lines:
print(f"{f.path:{filename_length}}", format_ts(f.last_ts), ln)
time.sleep(0.1)
seek_to_end = False
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--start", action='store_true')
ap.add_argument("--match-filename")
ap.add_argument("files", nargs="*", default=["."])
args = ap.parse_args()