79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple file watcher using stat() instead of watchdog (no external deps).
|
|
Watches core/ and app/ for changes and restarts ur_gia.
|
|
"""
|
|
import os
|
|
import sys
|
|
import time
|
|
import subprocess
|
|
|
|
|
|
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 ur_gia container."""
|
|
print(f'[{time.strftime("%H:%M:%S")}] Restarting ur_gia...', flush=True)
|
|
result = subprocess.run(
|
|
'podman restart ur_gia 2>/dev/null || docker restart ur_gia 2>/dev/null',
|
|
shell=True,
|
|
capture_output=True,
|
|
)
|
|
if result.returncode == 0:
|
|
print(f'[{time.strftime("%H:%M:%S")}] ur_gia restarted', flush=True)
|
|
else:
|
|
print(f'[{time.strftime("%H:%M:%S")}] restart failed', flush=True)
|
|
|
|
|
|
def main():
|
|
paths = ['/code/GIA/core', '/code/GIA/app']
|
|
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()
|