commit 8ce7aa5e3c20989f0b1bf433c1191e8aebba6592
Author: Jens Kristoffersson <jenkr@vgregion.se>
AuthorDate: Wed Sep 9 10:49:20 2026 +0200
Commit: Jens Kristoffersson <jenkr@vgregion.se>
CommitDate: Wed Sep 9 10:49:20 2026 +0200
Nya steg: ocr_read, prompt, lookup, merge_json, read_url
Fem nya nodtyper for verklig desktop-RPA (portning av Salma->ELVIS-flodet fran
AHK). Fas 1: motor + schema + tester.
- rpa/ocr.py (ny): OCR via winocr (Windows.Media.Ocr) + pymupdf for inskannade
PDF:er. pdf_to_text provar pypdf-textlager forst. image/window/region-kallor,
apply_pattern (namngiven grupp (?P<value>) eller hela traffen), value_near
(varde bredvid/under en etikett - portar ValueBelowLabel/ExtractLabelValue).
Saknat OCR-sprak -> OcrUnavailable med DISM-kommandot.
- rpa/prompt_dialog.py (ny): tkinter-dialog som egen process (samma monster som
desktop_capture). Lagen confirm / input / fields (visa+lat anvandaren ratta
variabler - t.ex. OCR-fel innan de skrivs in i ELVIS).
- rpa/flow_engine.py: _safe_export_path + _dict_lookup utbrutna. Nya _execute_step-
grenar. _NO_BROWSER_STEPS += ocr_read/prompt/lookup/merge_json (inte read_url).
Modul-docstring uppdaterad.
- rpa/schema.py: fem STEP_TYPES + validering (prompt mode=input krav pa var;
lookup krav pa from_var|file). Ny fal-konvention visible_when for villkorlig
faltvisning i frontend.
- requirements.txt: winocr + pymupdf (sys_platform == win32).
- Tester: test_ocr.py, test_new_steps.py (+ test_schema/test_skip_browser). 139 grona.
---
requirements.txt | 6 ++
rpa/flow_engine.py | 166 +++++++++++++++++++++++++++++++++++
rpa/ocr.py | 213 +++++++++++++++++++++++++++++++++++++++++++++
rpa/prompt_dialog.py | 124 ++++++++++++++++++++++++++
rpa/schema.py | 107 +++++++++++++++++++++++
tests/test_new_steps.py | 124 ++++++++++++++++++++++++++
tests/test_ocr.py | 78 +++++++++++++++++
tests/test_schema.py | 24 +++++
tests/test_skip_browser.py | 13 +++
9 files changed, 855 insertions(+)
diff --git a/requirements.txt b/requirements.txt
index 37011c6..0463a5a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -12,3 +12,9 @@ mcp[cli]>=2.1
pywinauto>=0.6.8; sys_platform == "win32"
pyautogui>=0.9.54; sys_platform == "win32"
opencv-python>=4.8; sys_platform == "win32"
+
+# OCR (ocr_read-steget) - bara Windows. winocr = Windows.Media.Ocr; kräver att
+# svenskt OCR-språk är installerat i Windows. pymupdf renderar inskannade
+# PDF-sidor till bilder. Saknas de ger ocr_read ett tydligt fel, ingen krasch.
+winocr>=0.0.15; sys_platform == "win32"
+pymupdf>=1.24; sys_platform == "win32"
diff --git a/rpa/flow_engine.py b/rpa/flow_engine.py
index 41f81e1..d116c3e 100644
--- a/rpa/flow_engine.py
+++ b/rpa/flow_engine.py
@@ -48,6 +48,19 @@ Nodtyper (fält per typ):
run_script {path, args?, timeout?, var?} -- kör ett separat Python-script,
-- loggar stdout/stderr, sparar ev.
-- stdout (JSON om möjligt) i var
+ ocr_read {source, file?/window?/x,y,w,h?/image?, lang?, var, pattern?, near?, near_mode?}
+ -- OCR (Windows) av en inskannad PDF /
+ -- fönster / skärmområde / bildfil -> var
+ -- (pattern = regex-utplock, near = värde
+ -- bredvid/under en etikett)
+ prompt {mode, message, var?, fields?, timeout?} -- interaktiv ruta (confirm/input/fields):
+ -- OK/Avbryt, fråga efter ett värde, eller
+ -- visa+låt användaren rätta variabler
+ lookup {from_var?/file?, key, var, ci?, on_missing?} -- slår upp key i en JSON-map (från en
+ -- inläst variabel eller en fil i exports/)
+ merge_json {file, key, value} -- data/exports/<file>[key] = value
+ -- (skapar filen; kom-ihåg-tabeller)
+ read_url {var} -- aktuell sidas URL -> var
screenshot {name?} -- sparar skärmdump i data/screenshots
list {value, var} -- CSS-selector, sparar antal+selector i var
(samt en textförhandsvisning av de första 3
@@ -161,12 +174,38 @@ _DESKTOP_STEPS = frozenset({
# separat (dess kropp = egna noder). "if" avgörs av sina villkor.
_NO_BROWSER_STEPS = frozenset({
"set_var", "increment", "clear_var", "save_var", "load_var", "run_script", "wait",
+ "ocr_read", "prompt", "lookup", "merge_json",
})
_BAD_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
+def _safe_export_path(filename: str) -> Path:
+ """En fil i data/exports/ – inga sökvägar tillåtna (delas av save_var/load_var/
+ merge_json/lookup)."""
+ if not filename:
+ raise FlowError("Filnamn saknas.")
+ if Path(filename).name != filename:
+ raise FlowError(f"Ogiltigt filnamn: {filename!r} (inga sökvägar tillåtna).")
+ return config.EXPORT_DIR / filename
+
+
+_MISSING = object()
+
+
+def _dict_lookup(table: dict, key: str, ci: bool):
+ """Slår upp `key` i `table` (ev. skiftlägesokänsligt). _MISSING om ingen rad."""
+ if key in table:
+ return table[key]
+ if ci:
+ kl = key.lower()
+ for k, v in table.items():
+ if isinstance(k, str) and k.lower() == kl:
+ return v
+ return _MISSING
+
+
def _safe_filename(name: str, fallback: str = "fil") -> str:
"""Rensar bort sökvägsdelar och otillåtna tecken ur ett filnamn."""
name = unicodedata.normalize("NFC", name or "")
@@ -685,6 +724,133 @@ def _execute_step(step, t, page, log, vars_, item_locator, download_dir, desktop
except json.JSONDecodeError:
vars_[var_name] = result.stdout.strip()
+ elif t == "ocr_read":
+ from . import ocr
+
+ var_name = step.get("var")
+ if not var_name:
+ raise FlowError("ocr_read kräver 'var'.")
+ source = step.get("source", "pdf")
+ lang = _interpolate(step.get("lang") or ocr.DEFAULT_LANG, vars_)
+ words: list = []
+ if source == "pdf":
+ fname = _interpolate(step.get("file", ""), vars_)
+ path = Path(fname)
+ if not path.is_absolute():
+ path = download_dir / fname
+ text = ocr.pdf_to_text(path, lang)
+ elif source == "window":
+ if desktop_holder["session"] is None:
+ desktop_holder["session"] = DesktopSession(log)
+ text, words = ocr.window_to_text(
+ desktop_holder["session"], _interpolate(step.get("window", ""), vars_) or None, lang
+ )
+ elif source == "region":
+ bbox = tuple(int(float(step.get(k) or 0)) for k in ("x", "y", "w", "h"))
+ text, words = ocr.region_to_text(bbox, lang)
+ elif source == "image":
+ text, words = ocr.image_file_to_text(_interpolate(step.get("image", ""), vars_), lang)
+ else:
+ raise FlowError(f"ocr_read: okänd source {source!r}.")
+
+ near = step.get("near")
+ pattern = step.get("pattern")
+ if near:
+ value = ocr.value_near(text, words, _interpolate(near, vars_), step.get("near_mode", "same_line"))
+ elif pattern:
+ value = ocr.apply_pattern(text, pattern)
+ else:
+ value = text
+ vars_[var_name] = value
+ log(f"OCR ({source}) -> '{var_name}': {str(value)[:80]!r}")
+
+ elif t == "prompt":
+ mode = step.get("mode", "confirm")
+ spec: dict = {"mode": mode, "message": _interpolate(step.get("message", ""), vars_)}
+ if mode == "input":
+ spec["default"] = _interpolate(step.get("default", ""), vars_)
+ elif mode == "fields":
+ names = [n.strip() for n in str(step.get("fields", "")).split(",") if n.strip()]
+ spec["fields"] = names
+ spec["values"] = {n: str(vars_.get(n, "")) for n in names}
+ timeout_s = float(step.get("timeout", 300))
+ log(f"Väntar på användaren ({mode}) …")
+ try:
+ proc = subprocess.run(
+ [sys.executable, "-m", "rpa.prompt_dialog"],
+ input=json.dumps(spec, ensure_ascii=False),
+ cwd=str(config.ROOT_DIR), capture_output=True, text=True, timeout=timeout_s,
+ )
+ except subprocess.TimeoutExpired as exc:
+ raise FlowError(f"prompt: ingen inmatning inom {timeout_s:.0f}s.") from exc
+ last = (proc.stdout or "").strip().splitlines()[-1:] or ["{}"]
+ try:
+ answer = json.loads(last[0])
+ except json.JSONDecodeError:
+ raise FlowError(f"prompt: oväntat svar från dialogen: {proc.stdout} {proc.stderr}") from None
+ if not answer.get("ok"):
+ raise FlowError("Avbrutet av användaren i prompt-steget.")
+ if mode == "input":
+ if not step.get("var"):
+ raise FlowError("prompt (mode=input) kräver 'var'.")
+ vars_[step["var"]] = answer.get("value", "")
+ elif mode == "fields":
+ for n, val in (answer.get("values") or {}).items():
+ vars_[n] = val
+ log(" användaren bekräftade")
+
+ elif t == "lookup":
+ var_name = step.get("var")
+ key = _interpolate(step.get("key", ""), vars_)
+ if not var_name or not key:
+ raise FlowError("lookup kräver 'key' och 'var'.")
+ from_var = step.get("from_var")
+ if from_var:
+ table = vars_.get(from_var)
+ src_desc = f"variabeln {from_var!r}"
+ else:
+ table = json.loads(_safe_export_path(_interpolate(step.get("file", ""), vars_)).read_text(encoding="utf-8"))
+ src_desc = "filen"
+ if not isinstance(table, dict):
+ raise FlowError(f"lookup: {src_desc} innehåller ingen JSON-map.")
+ ci = str(step.get("ci", "true")).lower() != "false"
+ found = _dict_lookup(table, key, ci)
+ if found is _MISSING:
+ on_missing = step.get("on_missing", "error")
+ if on_missing == "error":
+ raise FlowError(f"lookup: ingen rad för nyckeln {key!r}.")
+ vars_[var_name] = "" if on_missing == "empty" else on_missing
+ log(f"lookup: {key!r} saknas -> '{var_name}' = {vars_[var_name]!r}")
+ else:
+ vars_[var_name] = found
+ log(f"lookup: {key!r} -> '{var_name}'")
+
+ elif t == "merge_json":
+ path = _safe_export_path(_interpolate(step.get("file", ""), vars_))
+ key = _interpolate(step.get("key", ""), vars_)
+ if not key:
+ raise FlowError("merge_json kräver 'key'.")
+ value = _interpolate(step.get("value", ""), vars_)
+ data: dict = {}
+ if path.exists():
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except json.JSONDecodeError:
+ data = {}
+ if not isinstance(data, dict):
+ raise FlowError(f"merge_json: {path.name} innehåller ingen JSON-map.")
+ data[key] = value
+ config.EXPORT_DIR.mkdir(parents=True, exist_ok=True)
+ path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+ log(f"merge_json: {path.name}[{key!r}] = {value!r}")
+
+ elif t == "read_url":
+ var_name = step.get("var")
+ if not var_name:
+ raise FlowError("read_url kräver 'var'.")
+ vars_[var_name] = page.url
+ log(f"read_url -> '{var_name}': {page.url}")
+
elif t == "screenshot":
name = step.get("name") or "screenshot"
target = config.SCREENSHOT_DIR / f"{name}.png"
diff --git a/rpa/ocr.py b/rpa/ocr.py
new file mode 100644
index 0000000..9e61e9d
--- /dev/null
+++ b/rpa/ocr.py
@@ -0,0 +1,213 @@
+"""OCR för ocr_read-steget (rpa/flow_engine.py).
+
+Läser text ur en inskannad PDF, ett fönster, ett skärmområde eller en bildfil.
+Windows-only: winocr (Windows.Media.Ocr) + pymupdf för PDF-rendering. Importeras
+bakom try/except så flow_engine kan importeras överallt – anropas ocr_read utan
+paketen kastas OcrUnavailable i stället för ImportError.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+from . import config
+
+try: # Windows-only
+ import winocr
+except Exception: # noqa: BLE001
+ winocr = None
+
+try:
+ import pymupdf
+except Exception: # noqa: BLE001
+ pymupdf = None
+
+try:
+ from PIL import Image, ImageGrab
+except Exception: # noqa: BLE001
+ Image = None
+ ImageGrab = None
+
+DEFAULT_LANG = "sv-SE"
+
+# Om textlagret i en PDF ger minst så här många tecken hoppar vi OCR helt.
+_TEXT_LAYER_MIN = 40
+
+
+class OcrUnavailable(RuntimeError):
+ """winocr/pymupdf/PIL saknas eller OCR-språket är inte installerat i Windows."""
+
+
+class OcrError(RuntimeError):
+ """Filen/fönstret/området kunde inte OCR:as."""
+
+
+def _need_winocr() -> None:
+ if winocr is None:
+ raise OcrUnavailable(
+ "winocr kunde inte laddas – OCR fungerar bara på Windows. "
+ "Installera med: pip install -r requirements.txt"
+ )
+
+
+def _need_pil() -> None:
+ if Image is None:
+ raise OcrUnavailable("Pillow kunde inte laddas – pip install -r requirements.txt")
+
+
+# --------------------------------------------------------------------------
+# Låg nivå: en PIL-bild -> text + ordboxar
+# --------------------------------------------------------------------------
+
+def image_to_text(pil_image, lang: str = DEFAULT_LANG) -> tuple[str, list[dict]]:
+ """(text, words). words = [{text, x, y, w, h}] i bildpixlar."""
+ _need_winocr()
+ try:
+ result = winocr.recognize_pil_sync(pil_image, lang)
+ except AssertionError as exc:
+ # winocr assertar med exakt DISM-kommandot när språket saknas.
+ raise OcrUnavailable(
+ f"OCR-språket {lang!r} är inte installerat i Windows. Installera det "
+ "(admin-terminal):\n " + str(exc) + "\neller via Inställningar → Tid "
+ "och språk → Språk → lägg till språket → Språkalternativ → 'Optisk "
+ "teckenläsning'."
+ ) from exc
+ except Exception as exc: # noqa: BLE001
+ raise OcrError(f"OCR misslyckades: {exc}") from exc
+
+ lines = result.get("lines", []) or []
+ words: list[dict] = []
+ for line in lines:
+ for w in line.get("words", []) or []:
+ r = w.get("bounding_rect") or {}
+ words.append({
+ "text": w.get("text", ""),
+ "x": r.get("x", 0), "y": r.get("y", 0),
+ "w": r.get("width", 0), "h": r.get("height", 0),
+ })
+ # Bygg texten radvis (winocrs "text" slår ihop rader med mellanslag – sämre
+ # för radbaserad extraktion i value_near).
+ text = "\n".join(l.get("text", "") for l in lines) or (result.get("text") or "")
+ return text, words
+
+
+# --------------------------------------------------------------------------
+# Källor
+# --------------------------------------------------------------------------
+
+def pdf_to_text(path: str | Path, lang: str = DEFAULT_LANG) -> str:
+ """pypdf-textlager först; är det tomt renderas sidorna och OCR:as."""
+ p = Path(path)
+ if not p.exists():
+ raise OcrError(f"PDF:en finns inte: {p}")
+
+ layer = _pdf_text_layer(p)
+ if len(layer.strip()) >= _TEXT_LAYER_MIN:
+ return layer
+
+ if pymupdf is None:
+ raise OcrUnavailable(
+ "pymupdf kunde inte laddas – behövs för att OCR:a inskannade PDF:er. "
+ "pip install -r requirements.txt"
+ )
+ _need_pil()
+ parts: list[str] = []
+ doc = pymupdf.open(str(p))
+ try:
+ for page in doc:
+ pix = page.get_pixmap(dpi=300)
+ mode = "RGBA" if pix.alpha else "RGB"
+ img = Image.frombytes(mode, (pix.width, pix.height), pix.samples)
+ parts.append(image_to_text(img.convert("RGB"), lang)[0])
+ finally:
+ doc.close()
+ return "\n\n".join(parts).strip()
+
+
+def _pdf_text_layer(p: Path) -> str:
+ try:
+ from pypdf import PdfReader
+
+ reader = PdfReader(str(p))
+ return "\n".join((page.extract_text() or "") for page in reader.pages)
+ except Exception: # noqa: BLE001
+ return ""
+
+
+def window_to_text(session, title: str | None, lang: str = DEFAULT_LANG) -> tuple[str, list[dict]]:
+ """session = rpa.desktop.DesktopSession. Fångar fönstret och OCR:ar."""
+ _need_pil()
+ win = session._window(title or None) # noqa: SLF001 – avsiktlig återanvändning
+ img = win.capture_as_image()
+ return image_to_text(img, lang)
+
+
+def region_to_text(bbox: tuple[int, int, int, int], lang: str = DEFAULT_LANG) -> tuple[str, list[dict]]:
+ """bbox = (x, y, w, h) i skärmkoordinater."""
+ _need_pil()
+ if ImageGrab is None:
+ raise OcrUnavailable("PIL.ImageGrab saknas.")
+ x, y, w, h = bbox
+ img = ImageGrab.grab(bbox=(x, y, x + w, y + h), all_screens=True)
+ return image_to_text(img, lang)
+
+
+def image_file_to_text(path: str | Path, lang: str = DEFAULT_LANG) -> tuple[str, list[dict]]:
+ _need_pil()
+ p = Path(path)
+ if not p.is_absolute():
+ p = config.DESKTOP_IMAGE_DIR / p
+ if not p.exists():
+ raise OcrError(f"Bildfilen finns inte: {p}")
+ return image_to_text(Image.open(str(p)), lang)
+
+
+# --------------------------------------------------------------------------
+# Extraktion ur OCR-text
+# --------------------------------------------------------------------------
+
+def apply_pattern(text: str, pattern: str) -> str:
+ """re.search mot texten. Har mönstret en namngiven grupp `(?P<value>…)` returneras
+ den, annars hela träffen. Ingen träff -> ''."""
+ m = re.search(pattern, text, re.IGNORECASE | re.MULTILINE)
+ if not m:
+ return ""
+ if "value" in (m.groupdict() or {}):
+ return m.group("value") or ""
+ return m.group(0)
+
+
+def value_near(text: str, words: list[dict], label: str, mode: str = "same_line") -> str:
+ """Värdet bredvid/under en etikett.
+
+ same_line: resten av den rad som innehåller `label` (efter label + ev. ':').
+ below: närmaste rad UNDER etiketten, ungefär i samma kolumn (kräver ordboxar).
+ """
+ lines = [ln for ln in text.splitlines()]
+ if mode == "same_line":
+ for ln in lines:
+ if label.lower() in ln.lower():
+ idx = ln.lower().index(label.lower()) + len(label)
+ rest = ln[idx:].strip(" :\t-")
+ if rest:
+ return rest
+ return ""
+
+ # below: hitta etikettordens y/x, ta närmaste ordrad under
+ lbl_words = [w for w in words if label.lower() in w["text"].lower()]
+ if not lbl_words and " " in label:
+ first = label.split()[0].lower()
+ lbl_words = [w for w in words if w["text"].lower() == first]
+ if not lbl_words:
+ return ""
+ ly = min(w["y"] for w in lbl_words)
+ lx = min(w["x"] for w in lbl_words)
+ lh = max(w["h"] for w in lbl_words)
+ below = [w for w in words if w["y"] > ly + lh - 5 and abs(w["x"] - lx) < 300]
+ if not below:
+ return ""
+ row_y = min(w["y"] for w in below)
+ row = [w for w in below if abs(w["y"] - row_y) < 15]
+ row.sort(key=lambda w: w["x"])
+ return " ".join(w["text"] for w in row).strip()
diff --git a/rpa/prompt_dialog.py b/rpa/prompt_dialog.py
new file mode 100644
index 0000000..0cfacf4
--- /dev/null
+++ b/rpa/prompt_dialog.py
@@ -0,0 +1,124 @@
+"""Interaktiv dialog för prompt-steget (rpa/flow_engine.py).
+
+Körs som en egen process (tkinter vill ha huvudtråden – samma mönster som
+rpa/desktop_capture.py). Läser en spec som JSON på stdin, visar en ruta och
+skriver resultatet som JSON på stdout.
+
+Spec:
+ {"mode": "confirm" | "input" | "fields",
+ "message": "...", # färdig-interpolerad text
+ "fields": ["namn", ...], # mode=fields
+ "values": {"namn": "värde", ...}} # aktuella värden (mode=fields/input)
+
+Resultat:
+ confirm -> {"ok": true|false}
+ input -> {"ok": true, "value": "..."} | {"ok": false}
+ fields -> {"ok": true, "values": {...}} | {"ok": false}
+
+ python -m rpa.prompt_dialog (läser stdin – för manuellt test:
+ echo {"mode":"confirm","message":"Hej"} | python -m rpa.prompt_dialog)
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+
+
+def _run(spec: dict) -> dict:
+ try:
+ import tkinter as tk
+ from tkinter import ttk
+ except Exception as exc: # noqa: BLE001 # pragma: no cover
+ return {"ok": False, "error": f"tkinter saknas: {exc}"}
+
+ try: # DPI-medveten (som desktop_capture)
+ import ctypes
+
+ ctypes.windll.shcore.SetProcessDpiAwareness(2)
+ except Exception: # noqa: BLE001
+ pass
+
+ mode = spec.get("mode", "confirm")
+ message = str(spec.get("message", ""))
+ field_names = list(spec.get("fields", []) or [])
+ values = dict(spec.get("values", {}) or {})
+
+ state: dict = {"ok": False}
+
+ root = tk.Tk()
+ root.title("RPA – bekräfta")
+ root.attributes("-topmost", True)
+ root.resizable(False, False)
+ frm = ttk.Frame(root, padding=16)
+ frm.grid(sticky="nsew")
+
+ ttk.Label(frm, text=message, wraplength=460, justify="left").grid(
+ row=0, column=0, columnspan=2, sticky="w", pady=(0, 12)
+ )
+
+ entries: dict[str, tk.Entry] = {}
+ single: tk.Entry | None = None
+ row = 1
+
+ if mode == "input":
+ single = ttk.Entry(frm, width=48)
+ single.insert(0, str(spec.get("default", "")))
+ single.grid(row=row, column=0, columnspan=2, sticky="ew", pady=(0, 12))
+ single.focus_set()
+ row += 1
+ elif mode == "fields":
+ for name in field_names:
+ ttk.Label(frm, text=name).grid(row=row, column=0, sticky="w", padx=(0, 8), pady=2)
+ e = ttk.Entry(frm, width=42)
+ e.insert(0, str(values.get(name, "")))
+ e.grid(row=row, column=1, sticky="ew", pady=2)
+ entries[name] = e
+ row += 1
+ ttk.Label(frm, text="").grid(row=row, column=0, pady=6)
+ row += 1
+
+ def ok(_e=None):
+ state["ok"] = True
+ if mode == "input" and single is not None:
+ state["value"] = single.get()
+ if mode == "fields":
+ state["values"] = {n: e.get() for n, e in entries.items()}
+ root.destroy()
+
+ def cancel(_e=None):
+ state["ok"] = False
+ root.destroy()
+
+ btns = ttk.Frame(frm)
+ btns.grid(row=row, column=0, columnspan=2, sticky="e", pady=(8, 0))
+ ttk.Button(btns, text="Avbryt", command=cancel).grid(row=0, column=0, padx=(0, 8))
+ ok_btn = ttk.Button(btns, text="OK", command=ok)
+ ok_btn.grid(row=0, column=1)
+ if mode != "input":
+ ok_btn.focus_set()
+
+ frm.columnconfigure(1, weight=1)
+ root.bind("<Return>", ok)
+ root.bind("<Escape>", cancel)
+ root.protocol("WM_DELETE_WINDOW", cancel)
+ root.after(80, lambda: (root.lift(), root.focus_force()))
+ root.eval("tk::PlaceWindow . center")
+ root.mainloop()
+
+ return state
+
+
+def main() -> None:
+ raw = sys.stdin.read()
+ try:
+ spec = json.loads(raw) if raw.strip() else {"mode": "confirm", "message": raw}
+ except json.JSONDecodeError:
+ spec = {"mode": "confirm", "message": raw}
+ result = _run(spec)
+ print(json.dumps(result, ensure_ascii=False))
+ sys.exit(0 if result.get("ok") else 1)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/rpa/schema.py b/rpa/schema.py
index 833ce71..e296c8e 100644
--- a/rpa/schema.py
+++ b/rpa/schema.py
@@ -248,6 +248,16 @@ STEP_TYPES: list[dict] = [
"help": "Pausar tills ett element via text/CSS/roll finns.",
"selector": True, "fields": _sel(dict(_TIMEOUT_FIELD)),
},
+ {
+ "type": "read_url", "category": "navigation",
+ "label_sv": "Läs aktuell URL", "label_en": "Read current URL",
+ "help": "Sparar den aktuella sidans URL i en variabel (t.ex. för att plocka ut ett id ur adressen med lookup/set_var).",
+ "fields": [
+ {"key": "var", "kind": "text", "required": True,
+ "label_sv": "Spara som variabel", "label_en": "Save as variable"},
+ ],
+ "produces_vars": ["var"],
+ },
# ---- Interaktion ----
{
"type": "click", "category": "interaction",
@@ -435,6 +445,84 @@ STEP_TYPES: list[dict] = [
],
"produces_vars": ["var"],
},
+ {
+ "type": "ocr_read", "category": "data",
+ "label_sv": "OCR – läs text ur bild/PDF", "label_en": "OCR – read text from image/PDF",
+ "help": "Optisk teckenläsning (Windows). Källa: en inskannad PDF (data/downloads/), ett fönster, ett skärmområde eller en bildfil. Med 'pattern' plockas bara en regex-träff ut (namngiven grupp (?P<value>…) om den finns, annars hela träffen). Med 'near' returneras värdet bredvid/under en etikett.",
+ "fields": [
+ {"key": "source", "kind": "select", "options": ["pdf", "window", "region", "image"], "default": "pdf",
+ "label_sv": "Källa", "label_en": "Source"},
+ {"key": "file", "kind": "text", "interpolate": True, "visible_when": {"source": "pdf"},
+ "label_sv": "PDF-filnamn i data/downloads/ (stödjer {variabel})",
+ "label_en": "PDF filename in data/downloads/ (supports {variable})"},
+ {"key": "window", "kind": "text", "interpolate": True, "visible_when": {"source": "window"},
+ "label_sv": "Fönstertitel (regex, tomt = aktuellt)", "label_en": "Window title (regex, empty = current)"},
+ {"key": "x", "kind": "number", "default": 0, "visible_when": {"source": "region"},
+ "label_sv": "X", "label_en": "X"},
+ {"key": "y", "kind": "number", "default": 0, "visible_when": {"source": "region"},
+ "label_sv": "Y", "label_en": "Y"},
+ {"key": "w", "kind": "number", "default": 0, "visible_when": {"source": "region"},
+ "label_sv": "Bredd", "label_en": "Width"},
+ {"key": "h", "kind": "number", "default": 0, "visible_when": {"source": "region"},
+ "label_sv": "Höjd", "label_en": "Height"},
+ {"key": "image", "kind": "text", "interpolate": True, "visible_when": {"source": "image"},
+ "label_sv": "Bildfil (data/desktop_images/ eller absolut sökväg)",
+ "label_en": "Image file (data/desktop_images/ or absolute path)"},
+ {"key": "lang", "kind": "text", "default": "sv-SE",
+ "label_sv": "OCR-språk", "label_en": "OCR language"},
+ {"key": "var", "kind": "text", "required": True,
+ "label_sv": "Spara resultat i variabel", "label_en": "Save result to variable"},
+ {"key": "pattern", "kind": "text",
+ "label_sv": "Regex-utplock (valfritt)", "label_en": "Regex extraction (optional)"},
+ {"key": "near", "kind": "text", "interpolate": True,
+ "label_sv": "Etikett att läsa värdet bredvid (valfritt)",
+ "label_en": "Label to read the value next to (optional)"},
+ {"key": "near_mode", "kind": "select", "options": ["same_line", "below"], "default": "same_line",
+ "label_sv": "Var värdet står", "label_en": "Where the value sits"},
+ ],
+ "produces_vars": ["var"],
+ },
+ {
+ "type": "prompt", "category": "data",
+ "label_sv": "Fråga användaren (paus)", "label_en": "Ask the user (pause)",
+ "help": "Pausar körningen och visar en ruta på skärmen. confirm = OK/Avbryt (Avbryt stoppar flödet). input = ett textfält -> variabel. fields = visa de angivna variablerna redigerbara, så användaren kan rätta t.ex. OCR-fel innan de skrivs in.",
+ "fields": [
+ {"key": "mode", "kind": "select", "options": ["confirm", "input", "fields"], "default": "confirm",
+ "label_sv": "Läge", "label_en": "Mode"},
+ {"key": "message", "kind": "text", "interpolate": True, "required": True,
+ "label_sv": "Meddelande (stödjer {variabel})", "label_en": "Message (supports {variable})"},
+ {"key": "var", "kind": "text", "visible_when": {"mode": "input"},
+ "label_sv": "Spara svaret i variabel", "label_en": "Save answer to variable"},
+ {"key": "fields", "kind": "text", "visible_when": {"mode": "fields"},
+ "label_sv": "Variabler att visa (komma-separerade)", "label_en": "Variables to show (comma-separated)"},
+ {"key": "timeout", "kind": "number", "default": 300,
+ "label_sv": "Timeout (sekunder)", "label_en": "Timeout (seconds)"},
+ ],
+ "produces_vars": ["var"],
+ },
+ {
+ "type": "lookup", "category": "data",
+ "label_sv": "Slå upp i tabell", "label_en": "Look up in table",
+ "help": "Slår upp 'key' i en JSON-map. Kartan kommer antingen från en inläst variabel (from_var, via 'Läs in variabel') eller en JSON-fil i data/exports/. Resultatet (kan vara ett objekt – då fungerar {var.fält} i efterföljande steg) sparas i 'var'.",
+ "fields": [
+ {"key": "from_var", "kind": "text",
+ "label_sv": "Karta från variabel", "label_en": "Map from variable"},
+ {"key": "file", "kind": "text", "interpolate": True,
+ "label_sv": "…eller JSON-fil i data/exports/", "label_en": "…or JSON file in data/exports/"},
+ {"key": "key", "kind": "text", "interpolate": True, "required": True,
+ "label_sv": "Nyckel (stödjer {variabel}, ex. {grupp}|{specialitet})",
+ "label_en": "Key (supports {variable}, e.g. {grupp}|{specialitet})"},
+ {"key": "var", "kind": "text", "required": True,
+ "label_sv": "Spara resultat i variabel", "label_en": "Save result to variable"},
+ {"key": "ci", "kind": "bool", "default": True,
+ "label_sv": "Skiftlägesokänslig matchning", "label_en": "Case-insensitive match"},
+ {"key": "on_missing", "kind": "text", "default": "error",
+ "label_sv": "Om nyckeln saknas: 'error', 'empty' eller ett defaultvärde",
+ "label_en": "If key missing: 'error', 'empty' or a default value"},
+ ],
+ "consumes_vars": ["from_var"],
+ "produces_vars": ["var"],
+ },
# ---- Kontrollflöde ----
{
"type": "if", "category": "control",
@@ -506,6 +594,19 @@ STEP_TYPES: list[dict] = [
),
"produces_vars": ["var"],
},
+ {
+ "type": "merge_json", "category": "files",
+ "label_sv": "Skriv nyckel till JSON-fil", "label_en": "Write key to JSON file",
+ "help": "data/exports/<file>[key] = value (skapar filen om den saknas). För kom-ihåg-tabeller: fråga användaren om ett värde med prompt och spara det här, så finns det nästa körning (läs tillbaka med 'Läs in variabel' + lookup).",
+ "fields": [
+ {"key": "file", "kind": "text", "interpolate": True, "required": True,
+ "label_sv": "JSON-filnamn i data/exports/", "label_en": "JSON filename in data/exports/"},
+ {"key": "key", "kind": "text", "interpolate": True, "required": True,
+ "label_sv": "Nyckel (stödjer {variabel})", "label_en": "Key (supports {variable})"},
+ {"key": "value", "kind": "text", "interpolate": True,
+ "label_sv": "Värde (stödjer {variabel})", "label_en": "Value (supports {variable})"},
+ ],
+ },
# ---- Desktop (Windows) ----
{
"type": "desktop_launch", "category": "desktop",
@@ -870,6 +971,12 @@ def _validate_node_fields(node: dict, label: str, errors: list[str], strict: boo
for ei, m in enumerate(entries):
if not isinstance(m, dict) or any(_is_blank(m.get(k)) for k in req):
errors.append(f"{label}.{key}[{ei}]: kräver {', '.join(req)}.")
+ elif t == "prompt":
+ if strict and node.get("mode") == "input" and _is_blank(node.get("var")):
+ errors.append(f"{label}.var: obligatoriskt för prompt med mode='input'.")
+ elif t == "lookup":
+ if strict and _is_blank(node.get("from_var")) and _is_blank(node.get("file")):
+ errors.append(f"{label}: lookup kräver 'from_var' eller 'file'.")
def _validate_condition(cond: Any, path: str, errors: list[str]) -> None:
diff --git a/tests/test_new_steps.py b/tests/test_new_steps.py
new file mode 100644
index 0000000..fe1d666
--- /dev/null
+++ b/tests/test_new_steps.py
@@ -0,0 +1,124 @@
+"""lookup / merge_json / read_url / prompt – nya steg i rpa/flow_engine.py."""
+
+import json
+
+import pytest
+
+from rpa import flow_engine
+from rpa.flow_engine import FlowError, _execute_step
+
+
+def run_step(step, vars_, *, page=None):
+ _execute_step(step, step["type"], page, lambda _m: None, vars_, None, None, {"session": None})
+ return vars_
+
+
+# --- lookup -------------------------------------------------------------
+
+TABLE = {"MEDS Gastro|IBD": {"team": "MAGMOT", "orsak": "IBD"}, "X|Y": "platt"}
+
+
+def test_lookup_from_var_ci_and_dict_result():
+ v = run_step(
+ {"type": "lookup", "from_var": "cfg", "key": "{g}|{s}", "var": "row", "ci": True},
+ {"cfg": TABLE, "g": "meds gastro", "s": "ibd"},
+ )
+ assert v["row"] == {"team": "MAGMOT", "orsak": "IBD"}
+ # dict-resultat -> {row.field} funkar i efterföljande steg
+ run_step({"type": "set_var", "var": "t", "text": "{row.team}"}, v)
+ assert v["t"] == "MAGMOT"
+
+
+def test_lookup_case_sensitive_miss():
+ with pytest.raises(FlowError, match="ingen rad"):
+ run_step({"type": "lookup", "from_var": "cfg", "key": "x|y", "var": "r", "ci": "false"}, {"cfg": TABLE})
+
+
+def test_lookup_on_missing_empty_and_default():
+ v = run_step({"type": "lookup", "from_var": "cfg", "key": "nope", "var": "r", "on_missing": "empty"}, {"cfg": TABLE})
+ assert v["r"] == ""
+ v = run_step({"type": "lookup", "from_var": "cfg", "key": "nope", "var": "r", "on_missing": "MED"}, {"cfg": TABLE})
+ assert v["r"] == "MED"
+
+
+def test_lookup_from_file(tmp_path, monkeypatch):
+ monkeypatch.setattr(flow_engine.config, "EXPORT_DIR", tmp_path)
+ (tmp_path / "m.json").write_text(json.dumps(TABLE), encoding="utf-8")
+ v = run_step({"type": "lookup", "file": "m.json", "key": "X|Y", "var": "r"}, {})
+ assert v["r"] == "platt"
+
+
+def test_lookup_requires_key_and_var():
+ with pytest.raises(FlowError):
+ run_step({"type": "lookup", "from_var": "cfg", "var": "r"}, {"cfg": TABLE})
+
+
+# --- merge_json --------------------------------------------------------
+
+def test_merge_json_creates_and_updates(tmp_path, monkeypatch):
+ monkeypatch.setattr(flow_engine.config, "EXPORT_DIR", tmp_path)
+ run_step({"type": "merge_json", "file": "reg.json", "key": "A", "value": "1"}, {})
+ run_step({"type": "merge_json", "file": "reg.json", "key": "{k}", "value": "{val}"}, {"k": "B", "val": "2"})
+ data = json.loads((tmp_path / "reg.json").read_text(encoding="utf-8"))
+ assert data == {"A": "1", "B": "2"}
+
+
+def test_merge_json_rejects_path(monkeypatch, tmp_path):
+ monkeypatch.setattr(flow_engine.config, "EXPORT_DIR", tmp_path)
+ with pytest.raises(FlowError, match="Ogiltigt filnamn"):
+ run_step({"type": "merge_json", "file": "../evil.json", "key": "a", "value": "b"}, {})
+
+
+# --- read_url --------------------------------------------------------
+
+def test_read_url():
+ page = type("P", (), {"url": "https://salma.vgregion.se/incident/12345"})()
+ v = run_step({"type": "read_url", "var": "u"}, {}, page=page)
+ assert v["u"].endswith("/12345")
+
+
+# --- prompt (subprocess mockad) --------------------------------------
+
+def _fake_proc(stdout):
+ return type("R", (), {"stdout": stdout, "stderr": "", "returncode": 0})()
+
+
+def test_prompt_confirm_ok(monkeypatch):
+ monkeypatch.setattr(flow_engine.subprocess, "run", lambda *a, **k: _fake_proc('{"ok": true}'))
+ run_step({"type": "prompt", "mode": "confirm", "message": "Fortsätt?"}, {}) # ingen exception
+
+
+def test_prompt_confirm_cancel_aborts(monkeypatch):
+ monkeypatch.setattr(flow_engine.subprocess, "run", lambda *a, **k: _fake_proc('{"ok": false}'))
+ with pytest.raises(FlowError, match="Avbrutet av användaren"):
+ run_step({"type": "prompt", "mode": "confirm", "message": "x"}, {})
+
+
+def test_prompt_input_sets_var(monkeypatch):
+ monkeypatch.setattr(flow_engine.subprocess, "run", lambda *a, **k: _fake_proc('{"ok": true, "value": "99999-999"}'))
+ v = run_step({"type": "prompt", "mode": "input", "message": "Nummer?", "var": "nr"}, {})
+ assert v["nr"] == "99999-999"
+
+
+def test_prompt_fields_writes_back_edited_values(monkeypatch):
+ seen = {}
+
+ def fake_run(*a, **k):
+ seen["input"] = json.loads(k["input"])
+ return _fake_proc('{"ok": true, "values": {"pnr": "19850101-1234", "datum": "20260115"}}')
+
+ monkeypatch.setattr(flow_engine.subprocess, "run", fake_run)
+ v = run_step(
+ {"type": "prompt", "mode": "fields", "message": "Kontrollera", "fields": "pnr, datum"},
+ {"pnr": "1985010I-1234", "datum": "20260115"},
+ )
+ assert seen["input"]["fields"] == ["pnr", "datum"]
+ assert v["pnr"] == "19850101-1234" # användaren rättade OCR-felet
+
+
+def test_prompt_interpolates_message(monkeypatch):
+ seen = {}
+ monkeypatch.setattr(flow_engine.subprocess, "run",
+ lambda *a, **k: (seen.update(json.loads(k["input"])), _fake_proc('{"ok": true}'))[1])
+ run_step({"type": "prompt", "mode": "confirm", "message": "Spara {namn}?"}, {"namn": "Anna"})
+ assert seen["message"] == "Spara Anna?"
diff --git a/tests/test_ocr.py b/tests/test_ocr.py
new file mode 100644
index 0000000..bf01464
--- /dev/null
+++ b/tests/test_ocr.py
@@ -0,0 +1,78 @@
+"""rpa/ocr.py – textlager-PDF, mönster-utplock, värde-nära-etikett."""
+
+import pytest
+
+from rpa import ocr
+
+
+def test_apply_pattern_whole_match_by_default():
+ txt = "Rad 1\nPersonnummer 19850101-1234 skräp\nRad 3"
+ assert ocr.apply_pattern(txt, r"(19|20)?\d{6}[-+]?\d{4}") == "19850101-1234"
+ assert ocr.apply_pattern(txt, r"\d{4}-\d{2}-\d{2}") == ""
+
+
+def test_apply_pattern_named_group_wins():
+ txt = "Läkare: Anna Andersson (leg)"
+ assert ocr.apply_pattern(txt, r"Läkare:\s*(?P<value>[^()]+)").strip() == "Anna Andersson"
+
+
+def test_apply_pattern_no_match_empty():
+ assert ocr.apply_pattern("inget här", r"\d{10}") == ""
+
+
+def test_value_near_same_line():
+ txt = "Avsändare: Vårdcentralen Nordstan\nStatus Öppet"
+ assert ocr.value_near(txt, [], "Avsändare", "same_line") == "Vårdcentralen Nordstan"
+ assert ocr.value_near(txt, [], "Status", "same_line") == "Öppet"
+ assert ocr.value_near(txt, [], "Saknas", "same_line") == ""
+
+
+def test_value_near_below_uses_word_boxes():
+ words = [
+ {"text": "Personnummer", "x": 10, "y": 10, "w": 90, "h": 12},
+ {"text": "19850101-1234", "x": 12, "y": 40, "w": 100, "h": 12},
+ {"text": "Namn", "x": 300, "y": 10, "w": 40, "h": 12},
+ ]
+ text = "Personnummer Namn\n19850101-1234"
+ assert ocr.value_near(text, words, "Personnummer", "below") == "19850101-1234"
+
+
+def test_pdf_text_layer_path(tmp_path):
+ """En PDF med textlager ska läsas utan OCR (winocr rörs aldrig)."""
+ pymupdf = pytest.importorskip("pymupdf")
+ p = tmp_path / "remiss.pdf"
+ doc = pymupdf.open()
+ page = doc.new_page()
+ page.insert_text((72, 72), "Personnummer: 19850101-1234\nRemissdatum: 2026-01-15", fontsize=12)
+ doc.save(str(p))
+ doc.close()
+
+ text = ocr.pdf_to_text(p, "sv-SE")
+ assert "19850101-1234" in text
+ assert "2026-01-15" in text
+
+
+def test_image_to_text_parses_winocr_shape(monkeypatch):
+ fake = {
+ "text": "Personnummer: 19850101-1234",
+ "lines": [{
+ "text": "Personnummer: 19850101-1234",
+ "words": [
+ {"text": "Personnummer:", "bounding_rect": {"x": 1, "y": 2, "width": 3, "height": 4}},
+ {"text": "19850101-1234", "bounding_rect": {"x": 5, "y": 2, "width": 6, "height": 4}},
+ ],
+ }],
+ }
+ monkeypatch.setattr(ocr, "winocr", type("M", (), {"recognize_pil_sync": staticmethod(lambda *a: fake)}))
+ text, words = ocr.image_to_text(object(), "sv-SE")
+ assert text == "Personnummer: 19850101-1234"
+ assert words[1]["text"] == "19850101-1234" and words[1]["x"] == 5
+
+
+def test_image_to_text_missing_language_is_ocr_unavailable(monkeypatch):
+ def boom(*_a):
+ raise AssertionError('Add-WindowsCapability -Online -Name "Language.OCR~~~sv-SE~0.0.1.0"')
+
+ monkeypatch.setattr(ocr, "winocr", type("M", (), {"recognize_pil_sync": staticmethod(boom)}))
+ with pytest.raises(ocr.OcrUnavailable, match="Add-WindowsCapability"):
+ ocr.image_to_text(object(), "sv-SE")
diff --git a/tests/test_schema.py b/tests/test_schema.py
index 32a3ba5..ff50f00 100644
--- a/tests/test_schema.py
+++ b/tests/test_schema.py
@@ -1,5 +1,7 @@
from pathlib import Path
+import pytest
+
from rpa import migrate, schema
ROOT = Path(__file__).resolve().parent.parent
@@ -112,6 +114,28 @@ def test_call_flow_bad_io_entry_rejected():
assert any(".out[0]" in e for e in schema.validate_graph(flow, strict=False))
+@pytest.mark.parametrize("name", ["ocr_read", "prompt", "lookup", "merge_json", "read_url"])
+def test_new_steps_registered_with_default_handles(name):
+ assert name in schema.STEP_TYPE_NAMES
+ assert schema.node_handles(name)["outputs"] == ["out"]
+
+
+def test_prompt_input_needs_var_strict_only():
+ node = {"id": "n1", "type": "prompt", "mode": "input", "message": "?"}
+ assert any(".var" in e for e in schema.validate_graph({"nodes": [node], "edges": []}, strict=True))
+ assert schema.validate_graph({"nodes": [node], "edges": []}, strict=False) == []
+
+
+def test_lookup_needs_a_source_strict_only():
+ node = {"id": "n1", "type": "lookup", "key": "k", "var": "v"}
+ assert any("from_var" in e for e in schema.validate_graph({"nodes": [node], "edges": []}, strict=True))
+
+
+def test_ocr_read_source_enum_checked():
+ node = {"id": "n1", "type": "ocr_read", "var": "v", "source": "telepathy"}
+ assert any(".source" in e for e in schema.validate_graph({"nodes": [node], "edges": []}, strict=False))
+
+
def test_subflow_graph_is_validated():
flow = {
"nodes": [], "edges": [],
diff --git a/tests/test_skip_browser.py b/tests/test_skip_browser.py
index 7f6c3f9..7f242fb 100644
--- a/tests/test_skip_browser.py
+++ b/tests/test_skip_browser.py
@@ -45,6 +45,19 @@ def test_if_with_element_condition_needs_browser():
])) is True
+def test_ocr_prompt_lookup_dont_need_browser():
+ assert graph_engine.graph_needs_browser(nodes([
+ {"type": "ocr_read", "source": "pdf", "file": "r.pdf", "var": "t"},
+ {"type": "prompt", "mode": "confirm", "message": "?"},
+ {"type": "lookup", "from_var": "cfg", "key": "k", "var": "v"},
+ {"type": "merge_json", "file": "x.json", "key": "k", "value": "v"},
+ ])) is False
+
+
+def test_read_url_needs_browser():
+ assert graph_engine.graph_needs_browser(nodes([{"type": "read_url", "var": "u"}])) is True
+
+
def test_call_flow_needs_browser_depends_on_target():
resolve_pure = lambda tgt: ([{"id": "s", "type": "set_var", "var": "y", "text": "1"}], [], "s", [])
resolve_web = lambda tgt: ([{"id": "s", "type": "goto", "url": "https://x"}], [], "s", [])