79 lines
4.5 KiB
Python
79 lines
4.5 KiB
Python
#!/usr/bin/python3
|
|
"""load2.py MINUTES — the proof (fixes 2026-10-21 Part 4): the SAME uploads as the freeze, repeated in cycles for MINUTES,
|
|
each cycle FRESH work so every app does its full processing again: Paperless emptied (documents + trash) then the 53
|
|
PDFs; a NEW Immich user per cycle (its photos are new to it, so thumbnails + the analysis run again); a NEW Nextcloud
|
|
folder. Three streams at once per cycle, each sequential. Env on stdin: IRISZ, ODON, COOKIE (the setup-gate cookie is
|
|
not needed after the first admin exists). Never prints a secret."""
|
|
import glob, os, subprocess, sys, threading, time, datetime as dt
|
|
import requests, urllib3
|
|
urllib3.disable_warnings()
|
|
IP = "192.168.0.139"; D = "tester4.test"
|
|
env = dict(l.split("=", 1) for l in sys.stdin.read().splitlines() if "=" in l)
|
|
end = time.time() + 60 * int(sys.argv[1])
|
|
lk = threading.Lock()
|
|
def log(*a):
|
|
with lk:
|
|
print(time.strftime("%H:%M:%SZ", time.gmtime()), *a, flush=True)
|
|
def s(host):
|
|
x = requests.Session(); x.verify = False; x.headers["Host"] = f"{host}.{D}"; return x
|
|
def u(p): return f"https://{IP}{p}"
|
|
def paperless(c):
|
|
x = s("paperless"); x.auth = ("irisz", env["IRISZ"])
|
|
ids = []
|
|
r = x.get(u("/api/documents/?page_size=500&fields=id"), timeout=120)
|
|
if r.status_code == 200:
|
|
ids = [d["id"] for d in r.json()["results"]]
|
|
if ids:
|
|
r = x.post(u("/api/documents/bulk_edit/"), json={"documents": ids, "method": "delete"}, timeout=300)
|
|
log(f"c{c} paperless delete {len(ids)}", r.status_code)
|
|
r = x.post(u("/api/trash/"), json={"action": "empty"}, timeout=300)
|
|
log(f"c{c} paperless empty trash", r.status_code)
|
|
ok = bad = 0
|
|
for f in sorted(glob.glob("/root/data/paperless/*")):
|
|
try:
|
|
r = x.post(u("/api/documents/post_document/"), files={"document": open(f, "rb")}, timeout=300)
|
|
ok += r.status_code == 200; bad += r.status_code != 200
|
|
except Exception as e:
|
|
bad += 1; log(f"c{c} paperless", os.path.basename(f), "ERR", str(e)[:100])
|
|
log(f"c{c} paperless DONE ok {ok} bad {bad}")
|
|
def immich(c):
|
|
a = s("photos")
|
|
r = a.post(u("/api/auth/login"), json={"email": "tester4+odon@felhom.eu", "password": env["ODON"]}, timeout=60)
|
|
a.headers["Authorization"] = "Bearer " + r.json()["accessToken"]
|
|
em = f"tester4+proof{c}@felhom.eu"
|
|
r = a.post(u("/api/admin/users"), json={"email": em, "password": env["ODON"], "name": f"proof{c}"}, timeout=60)
|
|
log(f"c{c} immich new user", r.status_code)
|
|
x = s("photos")
|
|
r = x.post(u("/api/auth/login"), json={"email": em, "password": env["ODON"]}, timeout=60)
|
|
if r.status_code not in (200, 201):
|
|
log(f"c{c} immich login", r.status_code); return
|
|
x.headers["Authorization"] = "Bearer " + r.json()["accessToken"]; ok = bad = 0
|
|
for f in sorted(glob.glob("/root/data/immich/**/*", recursive=True)):
|
|
if not os.path.isfile(f):
|
|
continue
|
|
ts = dt.datetime.fromtimestamp(os.stat(f).st_mtime, dt.timezone.utc).isoformat()
|
|
try:
|
|
r = x.post(u("/api/assets"), files={"assetData": open(f, "rb")}, data={"deviceAssetId": f, "deviceId": f"proof{c}", "fileCreatedAt": ts, "fileModifiedAt": ts}, timeout=600)
|
|
ok += r.status_code in (200, 201); bad += r.status_code not in (200, 201)
|
|
except Exception as e:
|
|
bad += 1; log(f"c{c} immich", os.path.basename(f), "ERR", str(e)[:100])
|
|
log(f"c{c} immich DONE ok {ok} bad {bad}")
|
|
def nextcloud(c):
|
|
obs = subprocess.run(["rclone", "obscure", "-"], input=env["ODON"], capture_output=True, text=True).stdout.strip()
|
|
e = dict(os.environ, RCLONE_CONFIG_NC_TYPE="webdav", RCLONE_CONFIG_NC_URL=f"https://cloud.{D}/remote.php/dav/files/odon",
|
|
RCLONE_CONFIG_NC_VENDOR="nextcloud", RCLONE_CONFIG_NC_USER="odon", RCLONE_CONFIG_NC_PASS=obs)
|
|
p = subprocess.run(["rclone", "copy", "--no-check-certificate", "--transfers", "4", "--max-size", "1G", "--stats", "0", "--stats-log-level", "NOTICE",
|
|
"/root/data/nextcloud/Csalad", f"nc:/Csalad_proof{c}"], env=e, capture_output=True, text=True)
|
|
errs = sum(1 for l in (p.stderr or "").splitlines() if "ERROR" in l)
|
|
log(f"c{c} nextcloud DONE rc {p.returncode} error-lines {errs}")
|
|
def loop(fn):
|
|
c = 0
|
|
while time.time() < end:
|
|
c += 1
|
|
fn(c)
|
|
log(fn.__name__, "cycles", c)
|
|
log("PROOF START", sys.argv[1], "min — each stream repeats on its own until the end")
|
|
ts = [threading.Thread(target=loop, args=(f,)) for f in (paperless, immich, nextcloud)]
|
|
[t.start() for t in ts]; [t.join() for t in ts]
|
|
log("PROOF END")
|