Syncthing status

syncstat

#!/usr/bin/env python3

"""Syncthing status."""

# https://nelsonslog.wordpress.com/2021/10/01/syncthing-status-from-the-command-line/
# https://svn.blender.org/svnroot/bf-blender/trunk/blender/build_files/scons/tools/bcolors.py
# https://docs.syncthing.net/dev/rest.html

import json
import re
import sys
import urllib.request
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta, timezone
from pathlib import Path

# Configure the URL endpoint and API key.
settings = Path.home() / ".config/syncthing/config.xml"
with open(settings, "r", encoding="utf8") as xmlfile:
    tree = ET.parse(xmlfile)
root = tree.getroot()
host = "http://" + root.find("gui/address").text
api_key = root.find("gui/apikey").text

def st_api(command):
    """Call the SyncThing REST API and return as JSON."""
    req = urllib.request.Request(f"{host}/{command}")
    req.add_header("X-API-Key", api_key)
    with urllib.request.urlopen(req, timeout=2) as response:
        return json.loads(response.read())

class bcolors:
    """Color definitions."""
    OKGREEN = "\033[92m"
    WARNING = "\033[93m"
    FAIL = "\033[91m"
    ENDC = "\033[0m"

def main():
    """Execute."""
    now = datetime.now(timezone.utc)
    device_report = []
    try:
        # Explicit error handling for first REST connection.
        my_id = st_api("rest/system/status")["myID"]
    except Exception:
        print(f"Syncthing: Error connecting to {host}.")
        sys.exit(1)

    # Check completion for all remote devices
    devices = st_api("rest/config/devices")
    stats = st_api("rest/stats/device")
    for device in devices:
        device_id = device["deviceID"]
        if device_id == my_id:
            continue
        completion = st_api(f"rest/db/completion?device={device_id}")
        last_seen = datetime.fromisoformat(stats[device_id]["lastSeen"])
        delta = abs(now - last_seen).total_seconds()
        connected = bool(delta < 120)
        device_report.append((device["name"],
                              completion["completion"],
                              connected))

    # Print devices and completion percentage
    for name, completion, connected in device_report:
        if completion == 100 and connected:
            color = bcolors.OKGREEN
            status = "Up to Date"
        elif connected:
            color = bcolors.WARNING
            status = f"Syncing ({completion:.0f} %)"
        else:
            color = bcolors.FAIL
            status = "Disconnected"
        print(f"{color}{name[:20]:<20} {status}{bcolors.ENDC}")

    print()
    try:
        event = st_api("rest/events/disk?since=0&limit=1")[0]
    except TimeoutError:
        print("No events.")
    else:
        event_type = {"RemoteChangeDetected": "↓",
                      "LocalChangeDetected": "↑"}[event["type"]]
        delta = abs(now - datetime.fromisoformat(
            "".join(re.findall(r"(.*T[0-9:]*\.?[0-9]{,3})[0-9]*(.*)",
                               event["time"])[0])))
        delta = timedelta(seconds=round(delta.total_seconds()))
        path = event["data"]["path"].split("/")[-1]
        max_length = 20
        if len(path) > max_length:
            path = path[:max_length - 1] + "…"
        else:
            path = path[:max_length]
        if not any(completion == 100 and connected
                   for _, completion, connected in device_report):
            print(f"{bcolors.FAIL}Not synchronized to any device."
                  f"{bcolors.ENDC}")
        else:
            print(f"{event_type} {path} ({delta} ago)")

if __name__ == '__main__':
    main()