97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Kopiert ein beliebiges .rsc auf crs326-core und importiert es via SSH.
|
|
|
|
Usage: python3 import-core-rsc.py <pfad/zur/datei.rsc> [--host 10.40.41.211]
|
|
|
|
Creds kommen aus ~/.claude.json (mcpServers.mikromcp.env) envPrefix CRS326_CORE_.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
CLAUDe_JSON = Path.home() / ".claude.json"
|
|
RSC_PATH = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else None
|
|
HOST = sys.argv[2] if len(sys.argv) > 2 and sys.argv[1] == "--host" else None
|
|
|
|
|
|
def usage():
|
|
print(__doc__, file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
|
|
if RSC_PATH is None:
|
|
usage()
|
|
|
|
if not RSC_PATH.exists():
|
|
sys.exit(f"Datei nicht gefunden: {RSC_PATH}")
|
|
|
|
if HOST is None:
|
|
HOST = "10.40.41.211"
|
|
|
|
# Creds aus Claude-Settings laden
|
|
env = json.loads(CLAUDe_JSON.read_text())["mcpServers"]["mikromcp"]["env"]
|
|
get = lambda key, default=None: env.get(key, default)
|
|
|
|
user = get("CRS326_CORE_USER", "admin")
|
|
pw = get("CRS326_CORE_PASS")
|
|
if not pw:
|
|
sys.exit("CRS326_CORE_PASS fehlt in ~/.claude.json")
|
|
|
|
# Remote-Dateiname: alphanumerisch + Unterstrich/Punkt/Minus
|
|
remote_name = re.sub(r"[^A-Za-z0-9_.-]", "_", RSC_PATH.name)
|
|
|
|
|
|
def ssh_cmd(remote_command: str) -> list:
|
|
return [
|
|
"sshpass", "-p", pw,
|
|
"ssh",
|
|
"-o", "StrictHostKeyChecking=no",
|
|
"-o", "UserKnownHostsFile=/dev/null",
|
|
"-o", "PubkeyAuthentication=no",
|
|
"-o", "LogLevel=ERROR",
|
|
f"{user}@{HOST}",
|
|
remote_command,
|
|
]
|
|
|
|
|
|
def run_silent(cmd: list):
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
out = (r.stdout + r.stderr).replace(pw, "***")
|
|
raise SystemExit(f"SSH-Befehl fehlgeschlagen (rc={r.returncode}):\n{out}")
|
|
return r
|
|
|
|
|
|
# 1) Eventuell vorhandene alte Remote-Datei entfernen
|
|
run_silent(ssh_cmd(f"/file/remove [find name=\"{remote_name}\"]; :delay 500ms"))
|
|
|
|
# 2) SCP
|
|
scp = [
|
|
"sshpass", "-p", pw,
|
|
"scp",
|
|
"-o", "StrictHostKeyChecking=no",
|
|
"-o", "UserKnownHostsFile=/dev/null",
|
|
"-o", "PubkeyAuthentication=no",
|
|
"-o", "LogLevel=ERROR",
|
|
str(RSC_PATH),
|
|
f"{user}@{HOST}:/{remote_name}",
|
|
]
|
|
r = subprocess.run(scp, capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
out = (r.stdout + r.stderr).replace(pw, "***")
|
|
raise SystemExit(f"SCP fehlgeschlagen (rc={r.returncode}):\n{out}")
|
|
|
|
# 3) Import — mit verbose, damit wir die Zeile sehen, wenn ein Fehler auftritt
|
|
import_cmd = f"/import file-name={remote_name} verbose=yes"
|
|
res = subprocess.run(ssh_cmd(import_cmd), capture_output=True, text=True)
|
|
out = (res.stdout + res.stderr).replace(pw, "***")
|
|
if res.returncode != 0:
|
|
raise SystemExit(f"Import fehlgeschlagen (rc={res.returncode}):\n{out}")
|
|
|
|
print(out)
|
|
print(f"OK: {RSC_PATH.name} importiert auf {HOST} als /{remote_name}")
|