Syncthing Monitoring via API

syncthing_sync.py

@@ -0,0 +1,69 @@
+import os
+import xml.etree.ElementTree as ET
+import requests
+
+# Global variables for API access - These variables do not change and are set once when the app starts.
+API_URL = os.getenv('SYNCTHING_API_URL', 'http://127.0.0.1:8384')
+API_KEY = None
+
+def get_api_key(config_file):
+    """ Retrieve the API key from the local Syncthing configuration XML file. """
+    try:
+        tree = ET.parse(config_file)
+        root = tree.getroot()
+        return root.find('.//apikey').text
+    except FileNotFoundError:
+        raise Exception("Configuration file not found.")
+    except AttributeError:
+        raise Exception("API key not found in the configuration file.")
+   
+
+def syncthing_request(url):
+    """ Generic function for Syncthing API GET requests. """
+    headers = {'X-Api-Key': API_KEY}
+    response = requests.get(url, headers=headers)
+    response.raise_for_status()
+    return response.json()
+   
+
+def check_folder_idle(folder_id):
+    """ Check if the specified folder is idle. """
+    status_url = f"{API_URL}/rest/db/status?folder={folder_id}"
+    status = syncthing_request(status_url)
+    return status['state'] == 'idle'
+    
+def long_poll_events(last_event_id, folder_id):
+    """ Long poll the events endpoint filtering by the last known event ID and specific events. """
+    events_url = f"{API_URL}/rest/events?since={last_event_id}&events=StateChanged&timeout=60"
+    events = syncthing_request(events_url)
+    return [event for event in events if event['data']['folder'] == folder_id]
+    
+def main():
+    global API_KEY
+    folder_id = os.getenv('SYNCTHING_FOLDER_ID')
+    config_file = os.getenv('SYNCTHING_CONFIG_FILE')
+
+    if not folder_id:
+        raise ValueError("Folder ID must be provided as an environment variable.")
+
+    API_KEY = get_api_key(config_file)
+
+    # Start by getting the latest event ID
+    current_events = syncthing_request(f"{API_URL}/rest/events?limit=1")
+    last_event_id = current_events[0]['id'] if current_events else 0
+
+    # Check initial state
+    is_idle = check_folder_idle(folder_id)
+
+    while not is_idle:
+        events = long_poll_events(last_event_id, folder_id)
+        for event in events:
+            last_event_id = event['id']
+            is_idle = event['data']['to'] == 'idle'
+            if is_idle:
+                print("Folder transitioned to idle.")
+                break
+    
+
+if __name__ == '__main__':
+    main()