104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
import sys
|
|
import time
|
|
import threading
|
|
import os
|
|
import subprocess
|
|
from watchdog.observers import Observer
|
|
from watchdog.events import FileSystemEventHandler
|
|
|
|
DEBOUNCE_DELAY = 2 # seconds to wait after last modification
|
|
DEBUG = os.environ.get("DEBUG", "false").lower() == "true"
|
|
|
|
class WatcherHandler(FileSystemEventHandler):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.debounce_timers = {} # path -> Timer
|
|
self.processed_folders = set() # already processed folders
|
|
|
|
def on_any_event(self, event):
|
|
# Print all events
|
|
if event.event_type == "moved":
|
|
print(f"{'DIR' if event.is_directory else 'FILE'} MOVED: {event.src_path} -> {event.dest_path}")
|
|
else:
|
|
print(f"{'DIR' if event.is_directory else 'FILE'} {event.event_type.upper()}: {event.src_path}")
|
|
|
|
# Only schedule processing for directories
|
|
if event.is_directory and event.event_type in ("created", "moved", "modified"):
|
|
path_to_process = event.dest_path if event.event_type == "moved" else event.src_path
|
|
self.schedule_process(path_to_process)
|
|
|
|
def schedule_process(self, path):
|
|
"""Debounce processing of a folder."""
|
|
if path in self.debounce_timers:
|
|
self.debounce_timers[path].cancel()
|
|
|
|
timer = threading.Timer(DEBOUNCE_DELAY, self.process_new_dir, args=[path])
|
|
self.debounce_timers[path] = timer
|
|
timer.start()
|
|
|
|
def process_new_dir(self, path):
|
|
"""Called after folder is stable for DEBOUNCE_DELAY seconds."""
|
|
if path in self.debounce_timers:
|
|
del self.debounce_timers[path]
|
|
|
|
if not os.path.exists(path):
|
|
if DEBUG:
|
|
print(f"⚠️ Skipping {path}, folder no longer exists")
|
|
return
|
|
|
|
# Check if folder has .m4b files
|
|
m4b_files = [
|
|
f for f in os.listdir(path)
|
|
if os.path.isfile(os.path.join(path, f)) and f.lower().endswith(".m4b") and not f.startswith("._")
|
|
]
|
|
|
|
if not m4b_files:
|
|
if DEBUG:
|
|
print(f"❌ No .m4b files in {path}, skipping")
|
|
return
|
|
|
|
# Avoid re-processing the same folder
|
|
if path in self.processed_folders:
|
|
if DEBUG:
|
|
print(f"⚠️ Folder already processed: {path}")
|
|
return
|
|
self.processed_folders.add(path)
|
|
|
|
print(f"📂 Folder ready: {path} contains {len(m4b_files)} .m4b file(s)")
|
|
|
|
# Run main.py -da
|
|
try:
|
|
subprocess.run(["python3", "main.py", "-da", path], check=True)
|
|
print(f"✅ Processed folder: {path}")
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"❌ Error processing folder {path}: {e}")
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print(f"Usage: {sys.argv[0]} /path/to/watch")
|
|
sys.exit(1)
|
|
|
|
watch_path = sys.argv[1]
|
|
|
|
if not os.path.exists(watch_path):
|
|
print(f"❌ Path does not exist: {watch_path}")
|
|
sys.exit(1)
|
|
|
|
event_handler = WatcherHandler()
|
|
observer = Observer()
|
|
observer.schedule(event_handler, watch_path, recursive=True)
|
|
observer.start()
|
|
|
|
print(f"👀 Watching: {watch_path}")
|
|
try:
|
|
while True:
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
observer.stop()
|
|
observer.join()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|