feat: Implement inotify watch with debounced Immich library scan

Co-authored-by: aider (gemini/gemini-2.5-pro) <aider@aider.chat>
This commit is contained in:
2025-11-25 14:53:34 -07:00
parent 1c463c5113
commit 8c71e3cdfc

48
main.py
View File

@@ -5,13 +5,59 @@ logging.basicConfig(
level=logging.DEBUG if DEBUG else logging.INFO) level=logging.DEBUG if DEBUG else logging.INFO)
import secrets import secrets
import threading
import requests
from inotify_simple import INotify, flags
WATCH_PATHS = [ WATCH_PATHS = [
'/tmp/watch-test', '/tmp/watch-test',
] ]
SCAN_DELAY_SECONDS = 5
timer = None
def trigger_scan():
"""Trigger a library scan in Immich."""
logging.info("Triggering Immich library scan.")
url = f"https://photos.dns.t0.vc/api/libraries/{secrets.LIBRARY_UUID}/scan"
headers = {
'Accept': 'application/json',
'x-api-key': secrets.IMMICH_API_KEY,
}
try:
response = requests.post(url, headers=headers)
response.raise_for_status()
logging.info(f"Scan triggered successfully: {response.status_code}")
except requests.exceptions.RequestException as e:
logging.error(f"Failed to trigger scan: {e}")
def main(): def main():
pass global timer
watch_flags = flags.MODIFY | flags.CREATE | flags.DELETE | flags.MOVED_FROM | flags.MOVED_TO
with INotify() as inotify:
for path in WATCH_PATHS:
logging.info(f"Watching {path}")
inotify.add_watch(path, watch_flags)
try:
while True:
events = inotify.read(timeout=1000)
if events:
for event in events:
logging.debug(f"Event: {event!r}")
if timer:
timer.cancel()
timer = threading.Timer(SCAN_DELAY_SECONDS, trigger_scan)
timer.start()
logging.info(f"Modification detected. Triggering scan in {SCAN_DELAY_SECONDS} seconds.")
except KeyboardInterrupt:
logging.info("Shutting down...")
finally:
if timer:
timer.cancel()
if __name__ == '__main__': if __name__ == '__main__':