513 lines
17 KiB
Python
513 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""Orca/Anycubic-Nachbearbeitung: G-code nach Grotti laden und Senden-Seite öffnen.
|
||
|
||
Wenn SLIC3R_PP_OUTPUT_NAME unter dem Slice-Ordner liegt (auch Unterordner),
|
||
wird die Datei dorthin geschrieben und der Watch neu gelesen — nicht als Upload.
|
||
Sonst: Upload über die API.
|
||
|
||
Config: data/orca-senden.json mit url und token (grt_…).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import mimetypes
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import uuid
|
||
import webbrowser
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.request import Request, urlopen
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
CONFIG_PATH = REPO_ROOT / "data" / "orca-senden.json"
|
||
DEBUG_LOG = REPO_ROOT / "data" / "orca-senden-last.log"
|
||
BROWSER_LOCK = REPO_ROOT / "data" / "orca-senden-browser.lock"
|
||
FOCUS_PATH = REPO_ROOT / "data" / "orca-senden-focus.json"
|
||
APP_PROFILE = REPO_ROOT / "data" / "grotti-app-profile"
|
||
ALLOWED_SUFFIXES = (".gcode", ".gcode.3mf", ".3mf", ".gcode.pp", ".pp")
|
||
BROWSER_DEBOUNCE_S = 25.0
|
||
# Anycubic: MMDD-HHMM-Name.gcode — Zeitstempel raus, sonst jedes Export ein neues Fenster
|
||
_TIME_PREFIX_RE = re.compile(r"^\d{4}-\d{4}-")
|
||
_CHROMIUM_CANDIDATES = (
|
||
Path(r"C:\Program Files\Google\Chrome\Application\chrome.exe"),
|
||
Path(r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe"),
|
||
Path(os.path.expandvars(r"%LOCALAPPDATA%\Microsoft\Edge\Application\msedge.exe")),
|
||
Path(os.path.expandvars(r"%LOCALAPPDATA%\Google\Chrome\Application\chrome.exe")),
|
||
)
|
||
|
||
|
||
def show_error(message: str) -> None:
|
||
text = str(message or "Unbekannter Fehler.")
|
||
if sys.platform == "win32":
|
||
try:
|
||
import ctypes
|
||
|
||
ctypes.windll.user32.MessageBoxW(0, text, "Grotti · Orca senden", 0x10)
|
||
return
|
||
except Exception:
|
||
pass
|
||
print(text, file=sys.stderr)
|
||
|
||
|
||
def write_debug(argv: list[str], extra: str = "") -> None:
|
||
try:
|
||
DEBUG_LOG.parent.mkdir(parents=True, exist_ok=True)
|
||
lines = [
|
||
"argv:",
|
||
*[repr(item) for item in argv],
|
||
"",
|
||
"env:",
|
||
"SLIC3R_PP_HOST=%s" % (os.environ.get("SLIC3R_PP_HOST") or ""),
|
||
"SLIC3R_PP_OUTPUT_NAME=%s" % (os.environ.get("SLIC3R_PP_OUTPUT_NAME") or ""),
|
||
"",
|
||
extra,
|
||
]
|
||
DEBUG_LOG.write_text("\n".join(lines), encoding="utf-8")
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def load_config(path: Path = CONFIG_PATH) -> dict[str, str]:
|
||
if not path.is_file():
|
||
raise FileNotFoundError(
|
||
"Config fehlt: %s\nLege url und token (Einstellungen → API-Token) an." % path
|
||
)
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError) as exc:
|
||
raise ValueError("Config ungültig: %s" % exc) from exc
|
||
if not isinstance(data, dict):
|
||
raise ValueError("Config muss ein JSON-Objekt sein.")
|
||
url = str(data.get("url") or "").strip().rstrip("/")
|
||
token = str(data.get("token") or "").strip()
|
||
if not url:
|
||
raise ValueError("Config: url fehlt (z. B. http://127.0.0.1:8770).")
|
||
if not token:
|
||
raise ValueError("Config: token fehlt (grt_… aus Einstellungen).")
|
||
browser = str(data.get("browser") or "app").strip().lower() or "app"
|
||
if browser not in ("app", "tab"):
|
||
browser = "app"
|
||
return {"url": url, "token": token, "browser": browser}
|
||
|
||
|
||
def is_gcode_path(path: Path) -> bool:
|
||
name = path.name.lower()
|
||
return any(name.endswith(suffix) for suffix in ALLOWED_SUFFIXES)
|
||
|
||
|
||
def output_target() -> Path | None:
|
||
env_name = (os.environ.get("SLIC3R_PP_OUTPUT_NAME") or "").strip().strip('"')
|
||
if not env_name:
|
||
return None
|
||
return Path(env_name)
|
||
|
||
|
||
def upload_name_for(path: Path) -> str:
|
||
"""Schöner Dateiname: SLIC3R_PP_OUTPUT_NAME, sonst Pfad ohne .pp."""
|
||
target = output_target()
|
||
if target is not None and target.name:
|
||
base = target.name
|
||
if not is_gcode_path(Path(base)) and path.name.lower().endswith(".pp"):
|
||
if not base.lower().endswith((".gcode", ".3mf")):
|
||
base = base + ".gcode"
|
||
return base
|
||
name = path.name
|
||
if name.lower().endswith(".pp"):
|
||
name = name[:-3]
|
||
if not name.lower().endswith((".gcode", ".3mf")):
|
||
name = name + ".gcode"
|
||
return name
|
||
|
||
|
||
def find_gcode_arg(argv: list[str]) -> Path:
|
||
"""Pfad aus argv oder SLIC3R_PP_OUTPUT_NAME; Temp-Dateien enden oft auf .pp."""
|
||
candidates: list[Path] = []
|
||
for raw in argv[1:]:
|
||
text = (raw or "").strip().strip('"')
|
||
if not text or text.startswith("-"):
|
||
continue
|
||
candidates.append(Path(text))
|
||
for candidate in reversed(candidates):
|
||
if not is_gcode_path(candidate):
|
||
continue
|
||
if candidate.is_file():
|
||
return candidate.resolve()
|
||
raise FileNotFoundError("Datei fehlt: %s" % candidate)
|
||
for candidate in reversed(candidates):
|
||
if candidate.is_file():
|
||
return candidate.resolve()
|
||
env_path = output_target()
|
||
if env_path is not None and env_path.is_file():
|
||
return env_path.resolve()
|
||
detail = "argv=%s" % (argv[1:] or ["(leer)"])
|
||
raise ValueError(
|
||
"Kein G-code-Pfad in den Argumenten.\n"
|
||
"G-code exportieren (nicht nur slicen).\n"
|
||
"%s\nDetails: data/orca-senden-last.log" % detail
|
||
)
|
||
|
||
|
||
def send_page_url(base_url: str, file_id: int) -> str:
|
||
return "%s/dateien?file=%s" % (base_url.rstrip("/"), int(file_id))
|
||
|
||
|
||
def app_page_url(base_url: str) -> str:
|
||
"""Stabile App-URL — Datei kommt über orca-fokus, kein neues Fenster pro Slice."""
|
||
return "%s/dateien?orca=1" % base_url.rstrip("/")
|
||
|
||
|
||
def find_chromium() -> Path | None:
|
||
for path in _CHROMIUM_CANDIDATES:
|
||
try:
|
||
if path.is_file():
|
||
return path
|
||
except OSError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def app_already_running() -> bool:
|
||
"""True wenn Chrome/Edge schon mit grotti-app-profile läuft (Profil-Sperre)."""
|
||
needle = "grotti-app-profile"
|
||
if sys.platform == "win32":
|
||
try:
|
||
creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||
out = subprocess.check_output(
|
||
[
|
||
"powershell",
|
||
"-NoProfile",
|
||
"-Command",
|
||
(
|
||
"Get-CimInstance Win32_Process | "
|
||
"Where-Object { "
|
||
"($_.Name -match '^(chrome|msedge)\\.exe$') -and "
|
||
"($_.CommandLine -like '*%s*') "
|
||
"} | Select-Object -First 1 -ExpandProperty ProcessId"
|
||
)
|
||
% needle,
|
||
],
|
||
timeout=8,
|
||
stderr=subprocess.DEVNULL,
|
||
creationflags=creationflags,
|
||
)
|
||
return bool((out or b"").strip())
|
||
except (OSError, subprocess.SubprocessError):
|
||
pass
|
||
for name in ("SingletonLock", "SingletonSocket", "lockfile"):
|
||
path = APP_PROFILE / name
|
||
try:
|
||
if path.exists() or path.is_symlink():
|
||
return True
|
||
except OSError:
|
||
continue
|
||
return False
|
||
|
||
|
||
def open_as_app(url: str) -> bool:
|
||
"""Chrome/Edge --app=… → eigenes Fenster auf dem Desktop, kein Browser-Tab."""
|
||
binary = find_chromium()
|
||
if binary is None:
|
||
return False
|
||
try:
|
||
APP_PROFILE.mkdir(parents=True, exist_ok=True)
|
||
subprocess.Popen(
|
||
[
|
||
str(binary),
|
||
"--app=%s" % url,
|
||
"--user-data-dir=%s" % str(APP_PROFILE.resolve()),
|
||
"--no-first-run",
|
||
"--no-default-browser-check",
|
||
],
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
close_fds=True,
|
||
)
|
||
return True
|
||
except OSError as exc:
|
||
write_debug(sys.argv, "app_open_fail=%s" % exc)
|
||
return False
|
||
|
||
|
||
def _api_json(
|
||
base_url: str,
|
||
token: str,
|
||
method: str,
|
||
path: str,
|
||
*,
|
||
data: bytes | None = None,
|
||
headers: dict[str, str] | None = None,
|
||
timeout: float = 120.0,
|
||
) -> Any:
|
||
hdrs = {
|
||
"Authorization": "Bearer %s" % token,
|
||
"Accept": "application/json",
|
||
}
|
||
if headers:
|
||
hdrs.update(headers)
|
||
request = Request(
|
||
"%s/api/v1%s" % (base_url.rstrip("/"), path),
|
||
data=data,
|
||
method=method,
|
||
headers=hdrs,
|
||
)
|
||
try:
|
||
with urlopen(request, timeout=timeout) as response:
|
||
raw = response.read()
|
||
if not raw:
|
||
return None
|
||
return json.loads(raw.decode("utf-8"))
|
||
except HTTPError as exc:
|
||
detail = ""
|
||
try:
|
||
detail = json.loads(exc.read().decode("utf-8")).get("error") or ""
|
||
except Exception:
|
||
detail = str(exc.reason or exc)
|
||
raise RuntimeError("API %s fehlgeschlagen (%s): %s" % (path, exc.code, detail or exc)) from exc
|
||
except URLError as exc:
|
||
raise RuntimeError(
|
||
"Grotti nicht erreichbar (%s). Läuft der Server?" % (exc.reason or exc)
|
||
) from exc
|
||
|
||
|
||
def fetch_watch_dir(base_url: str, token: str) -> Path | None:
|
||
payload = _api_json(base_url, token, "GET", "/einstellungen")
|
||
if not isinstance(payload, dict):
|
||
return None
|
||
text = str(payload.get("watch_dir") or "").strip()
|
||
if not text:
|
||
return None
|
||
path = Path(text)
|
||
return path if path.is_dir() else None
|
||
|
||
|
||
def under_watch(watch_dir: Path, target: Path) -> tuple[str, str] | None:
|
||
"""(folder_relativ, dateiname) wenn target unter watch_dir liegt."""
|
||
try:
|
||
rel = target.resolve().relative_to(watch_dir.resolve())
|
||
except ValueError:
|
||
return None
|
||
parts = rel.parts
|
||
if not parts:
|
||
return None
|
||
name = parts[-1]
|
||
folder = "/".join(parts[:-1]) if len(parts) > 1 else ""
|
||
return folder, name
|
||
|
||
|
||
def place_in_watch(src: Path, dest: Path) -> Path:
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copyfile(str(src), str(dest))
|
||
return dest.resolve()
|
||
|
||
|
||
def rescan(base_url: str, token: str) -> None:
|
||
_api_json(base_url, token, "POST", "/dateien/neu-lesen", data=b"{}", headers={"Content-Type": "application/json"})
|
||
|
||
|
||
def find_library_file(base_url: str, token: str, *, name: str, folder: str) -> dict[str, Any] | None:
|
||
rows = _api_json(base_url, token, "GET", "/dateien")
|
||
if not isinstance(rows, list):
|
||
return None
|
||
folder_key = folder or ""
|
||
matches = []
|
||
for row in rows:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
if (row.get("name") or "") != name:
|
||
continue
|
||
row_folder = row.get("folder") or ""
|
||
if row.get("source") == "watch":
|
||
if row_folder == folder_key or (not folder_key and row_folder in ("", None)):
|
||
matches.append(row)
|
||
elif not folder_key and row_folder in ("", "Upload"):
|
||
matches.append(row)
|
||
if not matches:
|
||
# Name reicht, neueste ID
|
||
matches = [row for row in rows if isinstance(row, dict) and (row.get("name") or "") == name]
|
||
if not matches:
|
||
return None
|
||
return max(matches, key=lambda row: int(row.get("id") or 0))
|
||
|
||
|
||
def _multipart_body(filename: str, content: bytes, field: str = "file") -> tuple[bytes, str]:
|
||
boundary = "----GrottiOrca%s" % uuid.uuid4().hex
|
||
guessed = mimetypes.guess_type(filename)[0] or "application/octet-stream"
|
||
head = (
|
||
"--%s\r\n"
|
||
'Content-Disposition: form-data; name="%s"; filename="%s"\r\n'
|
||
"Content-Type: %s\r\n"
|
||
"\r\n"
|
||
) % (boundary, field, filename.replace('"', ""), guessed)
|
||
tail = "\r\n--%s--\r\n" % boundary
|
||
body = head.encode("utf-8") + content + tail.encode("ascii")
|
||
return body, "multipart/form-data; boundary=%s" % boundary
|
||
|
||
|
||
def upload_file(
|
||
base_url: str,
|
||
token: str,
|
||
path: Path,
|
||
filename: str | None = None,
|
||
timeout: float = 120.0,
|
||
) -> dict[str, Any]:
|
||
content = path.read_bytes()
|
||
name = filename or upload_name_for(path)
|
||
body, content_type = _multipart_body(name, content)
|
||
payload = _api_json(
|
||
base_url,
|
||
token,
|
||
"POST",
|
||
"/dateien",
|
||
data=body,
|
||
headers={"Content-Type": content_type},
|
||
timeout=timeout,
|
||
)
|
||
if not isinstance(payload, dict) or "id" not in payload:
|
||
raise RuntimeError("Unerwartete Antwort vom Upload.")
|
||
return payload
|
||
|
||
|
||
def browser_key(name: str) -> str:
|
||
"""Gleicher Slice trotz Zeitstempel / -2 / Leerzeichen → ein Browserfenster."""
|
||
stem = Path(name or "").stem.lower()
|
||
stem = _TIME_PREFIX_RE.sub("", stem)
|
||
stem = re.sub(r"-\d+$", "", stem)
|
||
return re.sub(r"[^a-z0-9]+", "", stem)
|
||
|
||
|
||
def write_focus(file_id: int, url: str, key: str = "") -> None:
|
||
"""Offene Dateien-Seite kann auf die neueste Orca-Datei umschalten."""
|
||
try:
|
||
FOCUS_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
FOCUS_PATH.write_text(
|
||
json.dumps(
|
||
{
|
||
"file_id": int(file_id),
|
||
"url": url,
|
||
"key": key,
|
||
"ts": time.time(),
|
||
},
|
||
ensure_ascii=False,
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
except (OSError, TypeError, ValueError):
|
||
pass
|
||
|
||
|
||
def open_browser(
|
||
url: str,
|
||
key: str = "",
|
||
file_id: int | None = None,
|
||
mode: str = "app",
|
||
) -> None:
|
||
"""Anycubic/Orca starten das Skript oft 2× — nur ein Fenster pro Modell.
|
||
|
||
mode=app (Standard): Chrome/Edge als App-Fenster auf dem Desktop.
|
||
mode=tab: klassischer Browser-Tab.
|
||
"""
|
||
token = (key or url).strip() or url
|
||
now = time.time()
|
||
if file_id is not None:
|
||
write_focus(int(file_id), url, key=token)
|
||
try:
|
||
BROWSER_LOCK.parent.mkdir(parents=True, exist_ok=True)
|
||
if BROWSER_LOCK.is_file():
|
||
prev = json.loads(BROWSER_LOCK.read_text(encoding="utf-8"))
|
||
if (
|
||
isinstance(prev, dict)
|
||
and prev.get("key") == token
|
||
and now - float(prev.get("ts") or 0) < BROWSER_DEBOUNCE_S
|
||
):
|
||
write_debug(
|
||
sys.argv,
|
||
"browser_skip key=%s age=%.1fs focus=%s"
|
||
% (token, now - float(prev.get("ts") or 0), file_id),
|
||
)
|
||
return
|
||
BROWSER_LOCK.write_text(
|
||
json.dumps({"key": token, "ts": now, "url": url, "mode": mode}, ensure_ascii=False),
|
||
encoding="utf-8",
|
||
)
|
||
except (OSError, TypeError, ValueError, json.JSONDecodeError):
|
||
pass
|
||
if (mode or "app").strip().lower() != "tab":
|
||
# Zweites Chrome mit gleichem Profil = leeres Profil / erneutes Login.
|
||
# Offenes Fenster pollt orca-fokus und wechselt die Datei selbst.
|
||
if app_already_running():
|
||
write_debug(sys.argv, "browser_reuse focus=%s" % file_id)
|
||
return
|
||
if open_as_app(url):
|
||
write_debug(sys.argv, "browser_app url=%s" % url)
|
||
return
|
||
webbrowser.open(url)
|
||
|
||
|
||
def deliver(base_url: str, token: str, src: Path) -> int:
|
||
"""Datei in Slice-Unterordner legen oder hochladen; gibt file_id zurück."""
|
||
watch = fetch_watch_dir(base_url, token)
|
||
target = output_target()
|
||
if watch is not None and target is not None:
|
||
placed = under_watch(watch, target)
|
||
if placed is not None:
|
||
folder, name = placed
|
||
dest = place_in_watch(src, target if target.suffix else target.with_suffix(".gcode"))
|
||
# Name aus Zielpfad, falls ohne Endung
|
||
if dest.name:
|
||
name = dest.name
|
||
rescan(base_url, token)
|
||
found = find_library_file(base_url, token, name=name, folder=folder)
|
||
if not found:
|
||
raise RuntimeError(
|
||
"Datei liegt unter %s, aber Grotti kennt sie noch nicht. "
|
||
"Unter Dateien „Ordner neu lesen“." % dest
|
||
)
|
||
return int(found["id"])
|
||
saved = upload_file(base_url, token, src, filename=upload_name_for(src))
|
||
return int(saved["id"])
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
argv = list(sys.argv if argv is None else argv)
|
||
write_debug(argv)
|
||
try:
|
||
config = load_config()
|
||
path = find_gcode_arg(argv)
|
||
write_debug(
|
||
argv,
|
||
"chosen=%s\nupload_as=%s\noutput=%s"
|
||
% (path, upload_name_for(path), output_target() or ""),
|
||
)
|
||
file_id = deliver(config["url"], config["token"], path)
|
||
name = upload_name_for(path)
|
||
write_debug(argv, "file_id=%s\nupload_as=%s" % (file_id, name))
|
||
mode = str(config.get("browser") or "app").strip().lower()
|
||
# App: stabile URL; Tab: Deep-Link mit file=
|
||
open_url = (
|
||
send_page_url(config["url"], file_id)
|
||
if mode == "tab"
|
||
else app_page_url(config["url"])
|
||
)
|
||
open_browser(
|
||
open_url,
|
||
key=browser_key(name),
|
||
file_id=file_id,
|
||
mode=mode,
|
||
)
|
||
except Exception as exc:
|
||
write_debug(argv, "error=%s" % exc)
|
||
show_error(str(exc))
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|