62 lines
No EOL
2.1 KiB
Python
62 lines
No EOL
2.1 KiB
Python
# Kemoverse - a gacha-style bot for the Fediverse.
|
|
# Copyright © 2025 Waifu
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see https://www.gnu.org/licenses/.
|
|
|
|
import subprocess
|
|
import os
|
|
from watchdog.observers import Observer
|
|
from watchdog.events import FileSystemEventHandler
|
|
import time
|
|
|
|
class ReloadHandler(FileSystemEventHandler):
|
|
def __init__(self, script_path):
|
|
self.script_path = script_path
|
|
self.process = None
|
|
self.last_modified = 0
|
|
self.debounce_interval = 1 # Wait 1 second to avoid double triggers
|
|
self.start_process()
|
|
|
|
def start_process(self):
|
|
if self.process:
|
|
self.process.terminate()
|
|
print(f"🔁 Starting {self.script_path}...")
|
|
self.process = subprocess.Popen(["python3", self.script_path])
|
|
|
|
def on_modified(self, event):
|
|
if event.src_path.endswith(".py"):
|
|
current_time = time.time()
|
|
if current_time - self.last_modified > self.debounce_interval:
|
|
print(f"Detected change in {event.src_path}")
|
|
self.start_process()
|
|
self.last_modified = current_time
|
|
|
|
if __name__ == "__main__":
|
|
script_path = "bot/bot_app.py"
|
|
if not os.path.exists(script_path):
|
|
print(f"Error: {script_path} does not exist")
|
|
exit(1)
|
|
|
|
event_handler = ReloadHandler(script_path)
|
|
observer = Observer()
|
|
observer.schedule(event_handler, path=".", recursive=True)
|
|
observer.start()
|
|
|
|
try:
|
|
while True:
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
print("Stopping...")
|
|
observer.stop()
|
|
observer.join() |