94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple file watcher using stat() instead of watchdog (no external deps).
|
|
|
|
This lightweight watcher can be used when you don't want the `watchdog`
|
|
dependency. Configure the directories to watch using the `WATCH_PATHS`
|
|
environment variable (comma-separated). Configure which container to restart
|
|
using `TARGET_CONTAINER` (defaults to `ur_gia`).
|
|
|
|
Example:
|
|
|
|
WATCH_PATHS=/code/core TARGET_CONTAINER=ur_gia
|
|
|
|
Touching a file under the watched path will trigger a restart of the target
|
|
container; e.g. `touch /code/core/__restart__` will cause the watcher to act.
|
|
"""
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
|
|
def get_mtime(path):
|
|
"""Recursively get the most recent mtime in a directory tree."""
|
|
max_mtime = 0
|
|
for root, dirs, files in os.walk(path):
|
|
# Skip pycache and hidden dirs
|
|
dirs[:] = [d for d in dirs if not d.startswith(".") and d != "__pycache__"]
|
|
for file in files:
|
|
if file.endswith((".pyc", ".pyo")):
|
|
continue
|
|
try:
|
|
mtime = os.path.getmtime(os.path.join(root, file))
|
|
max_mtime = max(max_mtime, mtime)
|
|
except OSError:
|
|
pass
|
|
return max_mtime
|
|
|
|
|
|
def restart_ur():
|
|
"""Restart target container (defaults to `ur_gia`)."""
|
|
target = os.environ.get("TARGET_CONTAINER", "ur_gia")
|
|
print(f'[{time.strftime("%H:%M:%S")}] Restarting {target}...', flush=True)
|
|
cmd = f"podman restart {target} 2>/dev/null || docker restart {target} 2>/dev/null"
|
|
result = subprocess.run(cmd, shell=True, capture_output=True)
|
|
if result.returncode == 0:
|
|
print(f'[{time.strftime("%H:%M:%S")}] {target} restarted', flush=True)
|
|
else:
|
|
print(f'[{time.strftime("%H:%M:%S")}] restart failed', flush=True)
|
|
|
|
|
|
def main():
|
|
# In the container the repository is mounted at /code
|
|
# Allow overriding watched paths via environment variable `WATCH_PATHS`.
|
|
# Default is `/code/core,/code/app`.
|
|
paths_env = os.environ.get("WATCH_PATHS", "/code/core,/code/app")
|
|
paths = [p.strip() for p in paths_env.split(",") if p.strip()]
|
|
last_mtimes = {}
|
|
|
|
for path in paths:
|
|
if os.path.exists(path):
|
|
print(f"Watching: {path}", flush=True)
|
|
last_mtimes[path] = get_mtime(path)
|
|
else:
|
|
print(f"Not found: {path}", flush=True)
|
|
|
|
print(f'[{time.strftime("%H:%M:%S")}] Watcher started', flush=True)
|
|
restart_debounce = 0
|
|
|
|
try:
|
|
while True:
|
|
time.sleep(2)
|
|
restart_debounce -= 2
|
|
|
|
for path in paths:
|
|
if not os.path.exists(path):
|
|
continue
|
|
current_mtime = get_mtime(path)
|
|
if current_mtime > last_mtimes.get(path, 0):
|
|
print(
|
|
f'[{time.strftime("%H:%M:%S")}] Changes in {path}', flush=True
|
|
)
|
|
last_mtimes[path] = current_mtime
|
|
if restart_debounce <= 0:
|
|
restart_ur()
|
|
restart_debounce = 5 # Don't restart more than every 5s
|
|
except KeyboardInterrupt:
|
|
print("Watcher stopped", flush=True)
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|