Spaces:
Running
Running
File size: 3,830 Bytes
b08d8a8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
#!/usr/bin/env python3
"""
File Watcher
Monitors sheets_output/ directory and triggers CSV-to-JSON conversion on changes
"""
import time
import json
import subprocess
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# Paths
BASE_DIR = Path(__file__).parent.parent
CONFIG_FILE = BASE_DIR / "config.json"
CONVERTER_SCRIPT = BASE_DIR / "scripts" / "csv_to_json.py"
class CSVChangeHandler(FileSystemEventHandler):
"""Handles CSV file changes"""
def __init__(self, debounce_seconds=2):
self.debounce_seconds = debounce_seconds
self.last_modified = {}
def on_modified(self, event):
"""Called when a file is modified"""
if event.is_directory:
return
# Only process CSV files
if not event.src_path.endswith('.csv'):
return
# Debounce: ignore rapid consecutive changes
now = time.time()
if event.src_path in self.last_modified:
if now - self.last_modified[event.src_path] < self.debounce_seconds:
return
self.last_modified[event.src_path] = now
print(f"\n📝 Detected change: {Path(event.src_path).name}")
self.trigger_conversion()
def on_created(self, event):
"""Called when a file is created"""
if not event.is_directory and event.src_path.endswith('.csv'):
print(f"\n📄 New file: {Path(event.src_path).name}")
self.trigger_conversion()
def trigger_conversion(self):
"""Run the CSV-to-JSON converter"""
print("🔄 Running CSV-to-JSON conversion...")
try:
result = subprocess.run(
['python3', str(CONVERTER_SCRIPT)],
capture_output=True,
text=True,
cwd=str(BASE_DIR)
)
if result.returncode == 0:
print(result.stdout)
else:
print(f"❌ Conversion failed:")
print(result.stderr)
except Exception as e:
print(f"❌ Error running converter: {e}")
def load_config():
"""Load configuration"""
if not CONFIG_FILE.exists():
return {
'file_watcher': {
'enabled': True,
'watch_directory': 'sheets_output',
'debounce_seconds': 2
}
}
with open(CONFIG_FILE, 'r') as f:
return json.load(f)
def main():
"""Start file watcher"""
config = load_config()
watcher_config = config.get('file_watcher', {})
if not watcher_config.get('enabled', True):
print("⚠️ File watcher is disabled in config.json")
return
watch_dir = BASE_DIR / watcher_config.get('watch_directory', 'sheets_output')
debounce = watcher_config.get('debounce_seconds', 2)
if not watch_dir.exists():
print(f"❌ Watch directory not found: {watch_dir}")
print(f"Creating directory: {watch_dir}")
watch_dir.mkdir(parents=True, exist_ok=True)
print("👀 File Watcher Started")
print(f"📂 Watching: {watch_dir}")
print(f"⏱️ Debounce: {debounce} seconds")
print("\nWaiting for CSV file changes... (Press Ctrl+C to stop)\n")
event_handler = CSVChangeHandler(debounce_seconds=debounce)
observer = Observer()
observer.schedule(event_handler, str(watch_dir), recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("\n\n👋 Stopping file watcher...")
observer.stop()
observer.join()
print("✅ File watcher stopped")
if __name__ == "__main__":
main()
|