98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Submit a copy job to rclone's RC API and watch it live.
|
|
This is the telemetry prototype for the monitoring UI: every number printed
|
|
here comes from an HTTP endpoint the future FastAPI backend can poll.
|
|
|
|
Usage:
|
|
python3 watch_batch.py <srcFs> <dstFs> [bwlimit]
|
|
python3 watch_batch.py /sources/batch_test synodrive:upload/BatchTest 3M
|
|
|
|
Endpoints used:
|
|
POST /core/bwlimit throttle (demo only, so transfers are watchable)
|
|
POST /sync/copy submit job (_async=true -> jobid)
|
|
POST /job/status is the job finished?
|
|
POST /core/stats LIVE: totals, speed, ETA, in-flight files, queue
|
|
POST /core/transferred HISTORY: per-file started_at/completed_at/error
|
|
"""
|
|
import base64, json, re, sys, time, urllib.request
|
|
from datetime import datetime
|
|
|
|
RC = "http://localhost:5572"
|
|
AUTH = "Basic " + base64.b64encode(b"admin:rc-admin-123").decode()
|
|
|
|
|
|
def rc(path, body=None):
|
|
req = urllib.request.Request(
|
|
RC + path,
|
|
data=json.dumps(body or {}).encode(),
|
|
headers={"Content-Type": "application/json", "Authorization": AUTH},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=15) as r:
|
|
return json.loads(r.read())
|
|
|
|
|
|
def parse_ts(s):
|
|
# rclone timestamps may carry nanoseconds; trim to microseconds
|
|
s = re.sub(r"(\.\d{6})\d+", r"\1", s)
|
|
return datetime.fromisoformat(s)
|
|
|
|
|
|
def fmt_mb(b):
|
|
return f"{b / 1e6:6.1f}MB"
|
|
|
|
|
|
def main():
|
|
src, dst = sys.argv[1], sys.argv[2]
|
|
if len(sys.argv) > 3:
|
|
rc("/core/bwlimit", {"rate": sys.argv[3]})
|
|
print(f"bwlimit set to {sys.argv[3]}/s (demo throttle)")
|
|
|
|
job = rc("/sync/copy", {"srcFs": src, "dstFs": dst, "_async": True})
|
|
jid = job["jobid"]
|
|
group = f"job/{jid}"
|
|
print(f"submitted jobid={jid} {src} -> {dst}\n")
|
|
|
|
while True:
|
|
s = rc("/core/stats", {"group": group})
|
|
done = s.get("transfers", 0)
|
|
total = s.get("totalTransfers", 0)
|
|
moving = s.get("transferring") or []
|
|
queued = max(total - done - len(moving), 0)
|
|
eta = s.get("eta")
|
|
print(
|
|
f"scheduled:{total:3} done:{done:3} moving:{len(moving)} "
|
|
f"queued:{queued:3} {fmt_mb(s.get('bytes', 0))}/{fmt_mb(s.get('totalBytes', 0))} "
|
|
f"{s.get('speed', 0) / 1e6:5.2f}MB/s ETA:{'-' if eta is None else str(int(eta)) + 's'}"
|
|
)
|
|
for t in moving:
|
|
print(
|
|
f" ↑ {t.get('name', '?'):32} {t.get('percentage', 0):3}% "
|
|
f"{t.get('speed', 0) / 1e6:5.2f}MB/s eta {t.get('eta', '-')}s"
|
|
)
|
|
if rc("/job/status", {"jobid": jid}).get("finished"):
|
|
break
|
|
time.sleep(2)
|
|
|
|
# ---- per-file history: this is what goes into MongoDB later ----
|
|
hist = rc("/core/transferred", {"group": group}).get("transferred", [])
|
|
files = [h for h in hist if not h.get("checked")] # actual transfers, not skip-checks
|
|
print(f"\n=== per-file report ({len(files)} transferred) ===")
|
|
print(f"{'file':34} {'size':>8} {'seconds':>8} {'MB/s':>6} status")
|
|
for h in sorted(files, key=lambda h: h.get("started_at", "")):
|
|
try:
|
|
secs = (parse_ts(h["completed_at"]) - parse_ts(h["started_at"])).total_seconds()
|
|
except Exception:
|
|
secs = 0.0
|
|
size = h.get("size", 0)
|
|
rate = size / 1e6 / secs if secs > 0 else 0
|
|
status = "ERROR: " + h["error"] if h.get("error") else "ok"
|
|
print(f"{h.get('name', '?'):34} {fmt_mb(size)} {secs:8.2f} {rate:6.2f} {status}")
|
|
|
|
rc("/core/bwlimit", {"rate": "off"})
|
|
print("\nbwlimit removed; done.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|