"""Kör de enskilda stegen i ett flöde. Själva grafvandringen (kanter, if/loop/
call_flow) ligger i rpa/graph_engine.py; run_flow() här är dispatchern som
normaliserar indata (gammal {steps}-lista migreras) och sätter upp webbläsare/
trace/close-mode. `_execute_step` kör en enskild nod.
Nodtyper (fält per typ):
goto {url}
wait {seconds}
wait_for_load {}
go_back {} -- webbläsarens "bakåt" (historik), t.ex.
efter att ha klickat in på en post från
en lista, för att fortsätta en loop
wait_for_login {timeout}
wait_for_selector {by, value, role?, scope?, nth?, timeout}
click {by, value, role?, scope?, nth?}
type {by, value, role?, scope?, nth?, text} -- skriver text i fält (fill)
press_key {key, by?, value?, role?, scope?, nth?} -- utan by/value: tangent mot sidan
select_option {by, value, role?, scope?, nth?, option} -- väljer alternativ i <select>
check {by, value, role?, scope?, nth?, checked} -- kryssar i/ur checkbox/radio
extract_text {by, value, role?, scope?, nth?, var} -- läser textinnehåll till variabel
extract_row {var, scope?, fields: [{name, by, value, role?}, ...]}
-- läser FLERA namngivna fält (celler) från
-- aktuell post till EN rad-post (dict) och
-- lägger den i listvariabeln var - ger
-- strukturerade kolumner (som i Power
-- Automates "Extract data from table")
-- istället för en textklump. save_var
-- (CSV) skriver då riktiga kolumnrubriker
-- från fältnamnen; load_var läser tillbaka
-- samma struktur.
set_var {var, text} -- sätter variabel till ett literalt värde
clear_var {var} -- tar bort en variabel (nästa list_append/
-- save_attachment/extract_row skapar den på
-- nytt som tom lista). Lägg först i en loop
-- för att nollställa en samlar-lista per varv.
increment {var, by?} -- ökar en heltalsvariabel med "by" (default 1,
-- negativt värde minskar); saknad/icke-numerisk
-- variabel behandlas som 0. Praktiskt i en loop
-- för att särskilja filer: {var} blir 1, 2, 3 ...
list_append {var, by?, value?, role?, scope?, nth?, text?}
-- lägger till i en listvariabel (skapas
-- vid behov); via selector (som extract_text)
-- eller literalt 'text' om ingen selector satt
save_var {var, filename, format?, mode?} -- sparar variabel till data/exports/
-- (format: "json"|"csv", gissas från filnamn;
-- mode: "overwrite" (default)|"append")
load_var {var, filename} -- läser in variabel från data/exports/
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, pattern?} -- aktuell sidas URL -> var (pattern =
-- regex-utplock, t.ex. ett ärende-id)
stop {message} -- avbryter körningen med ett felmeddelande
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
träffarna, till hjälp vid felsökning i UI:ts
variabelpanel - loop-steget bryr sig bara om
selector/count)
loop {over, item_var?, index_var?} (utgångar each/done - se längre ner)
-- itererar över "list"-resultat (DOM-element,
scope=item fungerar som vanligt) ELLER över en
vanlig listvariabel (t.ex. från "Lägg till i
lista"/"Läs in variabel") - då finns varje varvs
värde i variabeln item_var (default "item"), som
kan användas som {item} i by=text/role-fält
(se interpolering nedan). Varvnumret (1, 2, 3 ...)
läggs i index_var (default "index") oavsett
looptyp - {index} kan användas i t.ex. filnamn.
download {by, value, role?, scope?, nth?}
save_attachment {by, value, role?, scope?, nth?, filename?, var?, timeout?}
-- laddar ner en bilagelänk och sparar filen i
-- data/downloads/. Hittar en <a> (elementet självt
-- eller en inuti det) och: har den en riktig href
-- hämtas filen direkt via webbläsarens session
-- (behåller inloggning); är det en javascript:-länk
-- klickas den och en nedladdning fångas - eller,
-- om en popup öppnas (t.ex. bildbilagor), hämtas
-- filen ur popupen (dess URL, en fil-/nedladdnings-
-- länk i den, eller ett klick som startar nedladdn.).
-- filename (stödjer {variabel}) annars filnamnet
-- från URL:en; krockar döps om till "namn (2).ext".
-- var: lägger den sparade sökvägen i en LISTA
-- (bra att loopa bilagor och sen skicka listan till
-- ett merge/skriv ut-script).
capture_pdf {by?, value?, role?, scope?, nth?, filename?, var?, timeout?}
-- för "Skriv ut"-knappar som öppnar webbläsarens
-- skrivardialog i stället för att ladda ner en fil.
-- Neutraliserar window.print, klickar ev. triggern
-- (by/value...) och fångar PDF:en på tre sätt i tur
-- och ordning: (1) ett PDF-nätverkssvar, (2) en
-- inbäddad PDF i sidan (embed/iframe/object/a med
-- .pdf), (3) Chrome "printToPDF" av sidan som den
-- ser ut. Sparar i data/downloads/ (filename,
-- default "utskrift.pdf", stödjer {variabel}) och
-- lägger ev. sökvägen i var.
if {branches: [{condition}, ...]} -- utgångar case0..caseN (första sanna) + else;
-- grenarnas noder kopplas med kanter (graph_engine)
loop {over, item_var?, index_var?} -- utgångar each (per varv) / done; kroppen kopplar
-- tillbaka till loopens in
call_flow {target, in?: [{name, value}], out?: [{from, to}]}
-- kör ett annat flöde ("target" = flödesnamn) eller
-- en lokal subgraf ("#<subflow-id>") som ett steg;
-- 'in' mappar mot subflödets inputs ({variabel} ok),
-- 'out' kopierar subflödets slutvariabler tillbaka
condition (för "if"-grenar): {kind, ...}
kind: "element_exists" | "element_not_exists" -- {by, value, role?, scope?, nth?}
| "var_equals" | "var_not_equals" | "var_contains" -- {var, text}
by: "text" | "css" | "role"
scope: "page" (default) | "item" -- "item" begränsar sökningen till aktuell loop-post
interpolering: text-fält (by=text/role-värden, "type"/"set_var"/list_append-litteralens
"text") kan innehålla {variabelnamn} eller {variabelnamn.fält} (för att läsa ett
namngivet fält ur en dict-variabel, t.ex. en rad från "Extrahera rad"), vilket
ersätts med textvärdet vid körning. Fungerar INTE i CSS-selectors (by=css),
URL:er eller filnamn - utom capture_pdf/save_attachment-stegens "filename".
label: valfri nyckel på vilken nod som helst - namnger noden (visas i körloggen)
enabled: valfri nyckel på vilken nod som helst - satt till false hoppar
körningen över noden och följer "out".
on_error: valfri nyckel på vilken nod som helst - "stop" (default) avbryter
körningen om noden kastar fel/timeout; "continue" loggar felet och
följer "out" ändå (gör noden "hit or miss").
"""
import base64
import csv
import json
import re
import shlex
import subprocess
import sys
import tempfile
import time
import unicodedata
from pathlib import Path
from typing import Callable
from urllib.parse import unquote, urljoin, urlsplit
from playwright.sync_api import Page, sync_playwright
from . import config
from .auth import wait_for_login
from .browser import get_page, launch_context
from .desktop import DesktopSession
LogFn = Callable[[str], None]
class FlowError(Exception):
pass
# Desktop-stegtyper (Windows). Hanteras samlat av _execute_desktop_step som
# skapar en DesktopSession lat vid första desktop-steget. Måste hållas i synk
# med rpa/schema.py (tests/test_schema.py:test_engine_and_schema_step_types_match).
_DESKTOP_STEPS = frozenset({
"desktop_launch", "desktop_connect", "desktop_wait_for_window", "desktop_focus_window",
"desktop_click", "desktop_type", "desktop_press_key", "desktop_read_text",
"desktop_screenshot", "desktop_click_image", "desktop_wait_for_image", "desktop_click_xy",
})
# Nodtyper som inte rör webbläsaren alls (utöver desktop_*). "loop" hanteras
# 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", "stop",
})
_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 "")
name = _BAD_FILENAME_CHARS.sub("_", name.strip()).strip(". ")
return name or fallback
def _unique_path(directory: Path, filename: str) -> Path:
"""Ger en sökväg som inte redan finns: "namn.ext", "namn (2).ext", ..."""
target = directory / filename
if not target.exists():
return target
stem, suffix = target.stem, target.suffix
n = 2
while (directory / f"{stem} ({n}){suffix}").exists():
n += 1
return directory / f"{stem} ({n}){suffix}"
_FILE_LINK_SELECTOR = (
'a[download], a[href*="/fileservice/"], a[href*="/file/"], '
'a[href$=".pdf"], a[href$=".jpg"], a[href$=".jpeg"], a[href$=".png"], '
'a[href$=".gif"], a[href$=".tif"], a[href$=".tiff"], a[href$=".doc"], '
'a[href$=".docx"], img[src]:not([src^="data:"])'
)
def _grab_file_from_popup(popup, context, timeout_ms: float, log: LogFn):
"""Hämtar filen ur en popup som öppnats av en bilagelänk.
Popupen kan vara (a) själva filen, (b) en sida med en nedladdningslänk,
(c) en sida som startar nedladdningen när man klickar. Returnerar
(bytes|None, filnamn|None).
"""
try:
popup.wait_for_load_state("domcontentloaded")
except Exception: # noqa: BLE001
pass
def _fetch(url: str):
u = urljoin(popup.url, url)
if u.startswith("blob:"):
b64 = popup.evaluate(
"async (u) => { const r = await fetch(u); const buf = await (await r.blob())"
".arrayBuffer(); let s=''; const a=new Uint8Array(buf);"
" for (let i=0;i<a.length;i++) s+=String.fromCharCode(a[i]); return btoa(s); }",
u,
)
return base64.b64decode(b64), None
r = context.request.get(u, timeout=timeout_ms)
if not r.ok:
log(f" Popup-URL svarade {r.status}: {u}")
return None, None
return r.body(), unquote(urlsplit(u).path.rsplit("/", 1)[-1]) or None
# (a) popupens egen URL är filen
popup_path = urlsplit(popup.url).path.lower()
if "/fileservice/" in popup_path or popup_path.rsplit(".", 1)[-1] in (
"pdf", "jpg", "jpeg", "png", "gif", "tif", "tiff", "doc", "docx"
):
data, name = _fetch(popup.url)
if data is not None:
return data, name
# (b) en fil-/nedladdningslänk i popupen
try:
info = popup.evaluate(
"(sel) => { const el = document.querySelector(sel);"
" if (!el) return null;"
" return {url: el.href || el.src || null,"
" name: (el.getAttribute && el.getAttribute('download')) || (el.textContent||'').trim() || null}; }",
_FILE_LINK_SELECTOR,
)
except Exception: # noqa: BLE001
info = None
if info and info.get("url"):
data, name = _fetch(info["url"])
if data is not None:
return data, (info.get("name") or name)
# (c) klicka första länken/knappen och fånga en nedladdning i popupen
try:
with popup.expect_download(timeout=timeout_ms) as dl_info:
popup.locator("a, button").first.click()
d = dl_info.value
tmp = Path(tempfile.mkdtemp()) / (d.suggested_filename or "bilaga")
d.save_as(str(tmp))
return tmp.read_bytes(), (d.suggested_filename or None)
except Exception: # noqa: BLE001
return None, None
def _interpolate(text: str, vars_: dict) -> str:
"""Ersätter {variabelnamn} eller {variabelnamn.fält} med textvärdet.
Punkt-notation läser ett namngivet fält ur en dict-variabel (t.ex. en rad
från "Extrahera rad" eller en inläst CSV-rad). Okända namn/fält lämnas
orörda i texten.
"""
def repl(m: re.Match) -> str:
parts = m.group(1).split(".")
val = vars_
for part in parts:
if isinstance(val, dict) and part in val:
val = val[part]
else:
return m.group(0)
return val if isinstance(val, str) else json.dumps(val, ensure_ascii=False)
return re.sub(r"\{([\w.]+)\}", repl, text)
def _resolve_locator(page: Page, item_locator, step: dict, vars_: dict):
scope = step.get("scope", "page")
base = item_locator if (scope == "item" and item_locator is not None) else page
by = step.get("by", "css")
value = step.get("value", "")
nth = int(step.get("nth", 0))
if by == "text":
loc = base.get_by_text(re.compile(re.escape(_interpolate(value, vars_)), re.I))
elif by == "role":
role = step.get("role", "button")
loc = base.get_by_role(role, name=re.compile(re.escape(_interpolate(value, vars_)), re.I))
elif by == "css":
loc = base.locator(value)
else:
raise FlowError(f"Okänt 'by'-värde: {by}")
return loc.nth(nth)
def _describe_selector(step: dict, vars_: dict) -> str:
"""Loggvänlig beskrivning av en selector, med {var} redan ersatt."""
by = step.get("by", "css")
value = step.get("value", "")
if by in ("text", "role"):
value = _interpolate(value, vars_)
return f"{by}={value!r}"
def _describe_target(loc) -> str:
"""Kort beskrivning av det faktiska elementet en locator pekar på.
Ger t.ex. ' -> <a.ids-link "Öppna ärende" href=/case/123>' så loggen visar
vad som faktiskt klickades - särskilt värdefullt för value=":scope" och
scope=item där selectorn i sig inte säger något.
"""
try:
info = loc.first.evaluate(
"el => ({t: el.tagName.toLowerCase(),"
" id: el.id || '',"
" cls: (el.getAttribute && el.getAttribute('class') || '').split(/\\s+/).filter(Boolean).slice(0,2).join('.'),"
" href: (el.getAttribute && el.getAttribute('href')) || '',"
" txt: (el.innerText || el.textContent || '').trim().replace(/\\s+/g,' ').slice(0,60)})",
timeout=2000,
)
except Exception: # noqa: BLE001
return ""
head = "<" + info["t"]
if info["id"]:
head += "#" + info["id"]
elif info["cls"]:
head += "." + info["cls"]
extra = ""
if info["href"]:
extra += f" href={info['href']}"
if info["txt"]:
extra += f' "{info["txt"]}"'
return f" → {head}>{extra}"
def _selector_or_literal_text(page: Page, item_locator, step: dict, vars_: dict) -> str:
"""Text från ett element (om 'value'/selector är satt) annars step['text']."""
if step.get("value"):
loc = _resolve_locator(page, item_locator, step, vars_)
return loc.first.inner_text()
return _interpolate(step.get("text", ""), vars_)
def _evaluate_condition(page: Page, item_locator, vars_: dict, condition: dict) -> bool:
kind = condition.get("kind", "element_exists")
if kind in ("element_exists", "element_not_exists"):
if page is None:
raise FlowError(
f"Villkoret '{kind}' kräver webbläsaren, men flödet körs utan den "
"(skip_browser)."
)
loc = _resolve_locator(page, item_locator, condition, vars_)
exists = loc.count() > 0
return exists if kind == "element_exists" else not exists
if kind in ("var_equals", "var_not_equals", "var_contains"):
actual = vars_.get(condition.get("var", ""), "")
actual_str = actual if isinstance(actual, str) else json.dumps(actual, ensure_ascii=False)
expected = condition.get("text", "")
if kind == "var_equals":
return actual_str == expected
if kind == "var_not_equals":
return actual_str != expected
return expected in actual_str
raise FlowError(f"Okänt villkor: {kind!r}")
def _truthy(v) -> bool:
return v is True or str(v).strip().lower() == "true"
def _execute_desktop_step(step, t, log, vars_, desktop_holder) -> None:
"""Kör ett desktop_*-steg. Skapar en DesktopSession lat vid första
desktop-steget så rena webbflöden aldrig rör pywinauto."""
if desktop_holder["session"] is None:
desktop_holder["session"] = DesktopSession(log)
d = desktop_holder["session"]
def v(key: str, default: str = "") -> str:
return _interpolate(str(step.get(key, default) or ""), vars_)
def selector():
return (
step.get("by", "name"),
v("value"),
str(step.get("control_type", "") or ""),
v("window") or None,
int(step.get("nth", 0) or 0),
)
if t == "desktop_launch":
log(f"Startar app: {step.get('path', '')}")
d.launch(v("path"), v("args"), v("cwd") or None, float(step.get("timeout", 20)), v("title") or None)
elif t == "desktop_connect":
log(f"Ansluter till fönster: {v('title')}")
d.connect(v("title"), float(step.get("timeout", 20)))
elif t == "desktop_wait_for_window":
log(f"Väntar på fönster: {v('title')}")
d.wait_for_window(v("title"), float(step.get("timeout", 20)))
elif t == "desktop_focus_window":
log("Fokuserar fönster")
d.focus(v("title") or None, float(step.get("timeout", 10)))
elif t == "desktop_click":
by, value, ct, win, nth = selector()
log(f"Desktop-klick: {value or ct}")
d.click(by, value, ct, win, nth, double=_truthy(step.get("double")))
elif t == "desktop_type":
by, value, ct, win, nth = selector()
log(f"Skriver i kontroll: {value or ct}")
d.set_text(by, value, ct, win, nth, v("text"))
elif t == "desktop_press_key":
log(f"Tangenter: {v('keys')}")
d.press_key(v("keys"), v("window") or None)
elif t == "desktop_read_text":
by, value, ct, win, nth = selector()
var_name = step.get("var")
if not var_name:
raise FlowError("desktop_read_text kräver 'var'.")
text = d.read_text(by, value, ct, win, nth)
vars_[var_name] = text
log(f"Läste kontrolltext -> variabel '{var_name}': {text[:60]!r}")
elif t == "desktop_screenshot":
d.screenshot(v("name") or "desktop", v("window") or None)
elif t == "desktop_click_image":
log(f"Klickar på bild: {step.get('image', '')}")
d.click_image(
v("image"),
float(step.get("confidence", 0.9)),
float(step.get("timeout", 20)),
double=_truthy(step.get("double")),
button=step.get("button", "left") or "left",
)
elif t == "desktop_wait_for_image":
log(f"Väntar på bild: {step.get('image', '')}")
d.wait_for_image(v("image"), float(step.get("confidence", 0.9)), float(step.get("timeout", 20)))
elif t == "desktop_click_xy":
x, y = int(step.get("x", 0)), int(step.get("y", 0))
log(f"Klickar @ ({x},{y})")
d.click_xy(x, y, double=_truthy(step.get("double")), button=step.get("button", "left") or "left")
else: # pragma: no cover
raise FlowError(f"Ohanterad desktop-stegtyp: {t!r}")
def _execute_step(step, t, page, log, vars_, item_locator, download_dir, desktop_holder) -> None:
"""Kör ett enskilt steg (icke-behållar). loop/if hanteras av grafmotorn.
Undantag propagerar - anroparen avgör (via stegets "on_error") om körningen
ska stanna eller fortsätta."""
if t in _DESKTOP_STEPS:
_execute_desktop_step(step, t, log, vars_, desktop_holder)
return
if page is None and t not in _NO_BROWSER_STEPS:
raise FlowError(
f"Steget '{t}' kräver webbläsaren, men flödet körs utan den "
"(skip_browser är på i Inställningar, eller flödet innehåller inga "
"andra webbläsarsteg)."
)
if t == "goto":
log(f"Går till {step['url']}")
page.goto(step["url"], wait_until="domcontentloaded")
elif t == "wait":
seconds = float(step.get("seconds", 1))
log(f"Väntar {seconds}s")
time.sleep(seconds)
elif t == "wait_for_load":
log("Väntar på att sidan laddas klart")
page.wait_for_load_state("domcontentloaded")
elif t == "go_back":
log("Går tillbaka i webbläsarhistoriken")
page.go_back(wait_until="domcontentloaded")
elif t == "wait_for_login":
timeout_s = int(step.get("timeout", 180))
raw_indicators = step.get("indicators") or ""
indicators = [s.strip() for s in raw_indicators.split(",") if s.strip()] or None
wait_for_login(page, timeout_s, log=log, indicators=indicators)
elif t == "wait_for_selector":
loc = _resolve_locator(page, item_locator, step, vars_)
timeout_s = float(step.get("timeout", 15))
log(f"Väntar på element: {_describe_selector(step, vars_)}")
loc.first.wait_for(timeout=timeout_s * 1000)
elif t == "click":
loc = _resolve_locator(page, item_locator, step, vars_)
log(f"Klickar: {_describe_selector(step, vars_)}{_describe_target(loc)}")
loc.first.click()
page.wait_for_load_state("domcontentloaded")
elif t == "type":
loc = _resolve_locator(page, item_locator, step, vars_)
text = _interpolate(step.get("text", ""), vars_)
log(f"Skriver text i: {_describe_selector(step, vars_)}{_describe_target(loc)}")
loc.first.fill(text)
elif t == "press_key":
key = step.get("key", "Enter")
if step.get("value"):
loc = _resolve_locator(page, item_locator, step, vars_)
log(f"Trycker '{key}' på: {_describe_selector(step, vars_)}{_describe_target(loc)}")
loc.first.press(key)
else:
log(f"Trycker '{key}'")
page.keyboard.press(key)
elif t == "select_option":
loc = _resolve_locator(page, item_locator, step, vars_)
option = step.get("option", "")
log(f"Väljer '{option}' i: {_describe_selector(step, vars_)}{_describe_target(loc)}")
loc.first.select_option(label=option)
elif t == "check":
loc = _resolve_locator(page, item_locator, step, vars_)
checked_raw = step.get("checked", True)
checked = checked_raw if isinstance(checked_raw, bool) else str(checked_raw).lower() == "true"
log(f"{'Kryssar i' if checked else 'Kryssar ur'}: {_describe_selector(step, vars_)}{_describe_target(loc)}")
if checked:
loc.first.check()
else:
loc.first.uncheck()
elif t == "extract_text":
loc = _resolve_locator(page, item_locator, step, vars_)
var_name = step["var"]
text = loc.first.inner_text()
vars_[var_name] = text
log(f"Läste text -> variabel '{var_name}': {text[:60]!r}")
elif t == "extract_row":
var_name = step["var"]
scope = step.get("scope", "item")
row = {}
for field in step.get("fields", []):
name = field.get("name", "")
if not name:
continue
sub_step = {
"by": field.get("by", "css"),
"value": field.get("value", ""),
"role": field.get("role", ""),
"scope": scope,
}
try:
loc = _resolve_locator(page, item_locator, sub_step, vars_)
row[name] = loc.first.inner_text()
except Exception as exc: # noqa: BLE001
row[name] = ""
log(f" Fältet '{name}' kunde inte läsas: {exc}")
lst = vars_.setdefault(var_name, [])
if not isinstance(lst, list):
raise FlowError(f"Variabel '{var_name}' är inte en lista (innehåller {type(lst).__name__}).")
lst.append(row)
log(f"Extraherade rad -> lista '{var_name}' ({len(lst)} poster): {row}")
elif t == "set_var":
var_name = step["var"]
value = _interpolate(step.get("text", ""), vars_)
vars_[var_name] = value
log(f"Sätter variabel '{var_name}' = {value[:60]!r}")
elif t == "increment":
var_name = step["var"]
try:
by = int(step.get("by", 1))
except (TypeError, ValueError):
by = 1
try:
current = int(vars_.get(var_name, 0))
except (TypeError, ValueError):
current = 0
vars_[var_name] = current + by
log(f"Ökar variabel '{var_name}' med {by} -> {vars_[var_name]}")
elif t == "clear_var":
var_name = step["var"]
existed = vars_.pop(var_name, None) is not None
log(f"Nollställer variabel '{var_name}'" + ("" if existed else " (fanns inte)"))
elif t == "list_append":
var_name = step["var"]
value = _selector_or_literal_text(page, item_locator, step, vars_)
lst = vars_.setdefault(var_name, [])
if not isinstance(lst, list):
raise FlowError(f"Variabel '{var_name}' är inte en lista (innehåller {type(lst).__name__}).")
lst.append(value)
log(f"Lägger till i lista '{var_name}' ({len(lst)} poster): {value[:60]!r}")
elif t == "save_var":
var_name = step["var"]
if var_name not in vars_:
raise FlowError(f"Okänd variabel '{var_name}' i save_var.")
value = vars_[var_name]
filename = step.get("filename") or f"{var_name}.json"
if Path(filename).name != filename:
raise FlowError(f"Ogiltigt filnamn: {filename!r} (inga sökvägar tillåtna).")
fmt = step.get("format") or ("csv" if filename.lower().endswith(".csv") else "json")
append = step.get("mode") == "append"
config.EXPORT_DIR.mkdir(parents=True, exist_ok=True)
target = config.EXPORT_DIR / filename
if fmt == "csv":
items = value if isinstance(value, list) else [value]
rows = [item if isinstance(item, dict) else {"value": item} for item in items]
fieldnames = list(rows[0].keys()) if rows else ["value"]
write_header = not (append and target.exists())
with open(target, "a" if append else "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
if write_header:
writer.writeheader()
writer.writerows(rows)
else:
to_write = value
if append and target.exists():
if not isinstance(value, list):
raise FlowError("save_var i append-läge (JSON) kräver att variabeln är en lista.")
try:
existing = json.loads(target.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
existing = []
to_write = (existing if isinstance(existing, list) else [existing]) + value
target.write_text(json.dumps(to_write, ensure_ascii=False, indent=2), encoding="utf-8")
log(f"Sparade '{var_name}' -> {target.name} ({fmt}{', append' if append else ''})")
elif t == "load_var":
var_name = step["var"]
filename = step["filename"]
if Path(filename).name != filename:
raise FlowError(f"Ogiltigt filnamn: {filename!r} (inga sökvägar tillåtna).")
source = config.EXPORT_DIR / filename
if not source.exists():
raise FlowError(f"Filen finns inte: {source}")
if source.suffix.lower() == ".csv":
with open(source, newline="", encoding="utf-8") as f:
rows = list(csv.DictReader(f))
if rows and set(rows[0].keys()) == {"value"}:
vars_[var_name] = [row["value"] for row in rows]
else:
vars_[var_name] = rows
else:
vars_[var_name] = json.loads(source.read_text(encoding="utf-8"))
log(f"Läste in '{var_name}' <- {source.name}")
elif t == "run_script":
script = step.get("path", "")
if not script:
raise FlowError("run_script kräver 'path'.")
script_path = Path(script)
if not script_path.is_absolute():
script_path = config.ROOT_DIR / script_path
if not script_path.exists():
raise FlowError(f"Skriptet finns inte: {script_path}")
args = shlex.split(step.get("args", ""))
timeout_s = float(step.get("timeout", 60))
log(f"Kör script: {script_path.name} {' '.join(args)}")
result = subprocess.run(
[sys.executable, str(script_path), *args],
cwd=str(config.ROOT_DIR),
capture_output=True,
text=True,
timeout=timeout_s,
)
for line in result.stdout.splitlines():
log(f" stdout: {line}")
for line in result.stderr.splitlines():
log(f" stderr: {line}")
if result.returncode != 0:
raise FlowError(f"Script avslutade med felkod {result.returncode}: {script_path.name}")
var_name = step.get("var")
if var_name:
try:
vars_[var_name] = json.loads(result.stdout)
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)
if table is None and "." in from_var: # cfg.elvis.translationTable
root, *rest = from_var.split(".")
table = vars_.get(root)
for k in rest:
table = table.get(k) if isinstance(table, dict) else None
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 == "stop":
raise FlowError(_interpolate(step.get("message") or "Flödet stoppades.", vars_))
elif t == "read_url":
var_name = step.get("var")
if not var_name:
raise FlowError("read_url kräver 'var'.")
url = page.url
pattern = step.get("pattern")
if pattern:
from . import ocr
url = ocr.apply_pattern(url, pattern)
vars_[var_name] = url
log(f"read_url -> '{var_name}': {url}")
elif t == "screenshot":
name = step.get("name") or "screenshot"
target = config.SCREENSHOT_DIR / f"{name}.png"
page.screenshot(path=str(target))
log(f"Skärmdump sparad: {target.name}")
elif t == "list":
selector = step["value"]
var_name = step["var"]
loc = page.locator(selector)
count = loc.count()
preview = []
for preview_idx in range(min(count, 3)):
try:
preview.append(loc.nth(preview_idx).inner_text(timeout=2000)[:200])
except Exception: # noqa: BLE001
preview.append("<kunde inte läsa text>")
vars_[var_name] = {"selector": selector, "count": count, "preview": preview}
log(f"Hittade {count} post(er) -> variabel '{var_name}'")
elif t == "download":
loc = _resolve_locator(page, item_locator, step, vars_)
log(f"Laddar ner via: {_describe_selector(step, vars_)}{_describe_target(loc)}")
try:
with page.expect_download(timeout=15000) as dl_info:
loc.first.click()
download = dl_info.value
target = download_dir / download.suggested_filename
download.save_as(str(target))
log(f"Nedladdad: {target.name}")
except Exception as exc: # noqa: BLE001
log(f"Nedladdning misslyckades: {exc}")
elif t == "save_attachment":
loc = _resolve_locator(page, item_locator, step, vars_)
timeout_s = float(step.get("timeout", 15))
download_dir.mkdir(parents=True, exist_ok=True)
log(f"Hämtar bilaga via: {_describe_selector(step, vars_)}{_describe_target(loc)}")
el = loc.first
href = None
try:
href = el.get_attribute("href")
except Exception: # noqa: BLE001
href = None
if not href:
try:
inner = el.locator("a[href]").first
if inner.count():
href = inner.get_attribute("href")
except Exception: # noqa: BLE001
pass
explicit_name = _interpolate(step.get("filename", ""), vars_).strip()
saved_path = None
is_real_link = bool(href) and not href.strip().lower().startswith("javascript:")
if is_real_link:
url = urljoin(page.url, href)
try:
r = page.context.request.get(url, timeout=timeout_s * 1000)
if r.ok:
name = explicit_name or unquote(urlsplit(url).path.rsplit("/", 1)[-1]) or "bilaga"
saved_path = _unique_path(download_dir, _safe_filename(name, "bilaga"))
saved_path.write_bytes(r.body())
log(f" Hämtade via URL: {url}")
else:
log(f" URL svarade {r.status}: {url}")
except Exception as exc: # noqa: BLE001
log(f" Kunde inte hämta via URL: {exc}")
if saved_path is None:
# javascript:-länk eller misslyckad URL-hämtning: klicka och fånga
# antingen en nedladdning direkt eller en popup med filen/länken.
popup_holder = []
download_holder = []
def _on_popup(p):
popup_holder.append(p)
def _on_download(d):
download_holder.append(d)
page.on("popup", _on_popup)
page.on("download", _on_download)
try:
try:
el.click()
except Exception as exc: # noqa: BLE001
log(f" Klick misslyckades: {exc}")
waited = 0.0
while waited < timeout_s and not popup_holder and not download_holder:
page.wait_for_timeout(200)
waited += 0.2
finally:
for evt, fn in (("popup", _on_popup), ("download", _on_download)):
try:
page.remove_listener(evt, fn)
except Exception: # noqa: BLE001
pass
download = download_holder[0] if download_holder else None
if download is not None:
name = explicit_name or download.suggested_filename or "bilaga"
saved_path = _unique_path(download_dir, _safe_filename(name, "bilaga"))
download.save_as(str(saved_path))
log(f" Nedladdning fångad: {saved_path.name}")
elif popup_holder:
pp = popup_holder[0]
try:
data, popup_name = _grab_file_from_popup(pp, page.context, timeout_s * 1000, log)
if data is not None:
name = explicit_name or popup_name or "bilaga"
saved_path = _unique_path(download_dir, _safe_filename(name, "bilaga"))
saved_path.write_bytes(data)
log(f" Hämtade via popup: {saved_path.name}")
else:
log(" Hittade ingen fil i popupen.")
except Exception as exc: # noqa: BLE001
log(f" Kunde inte hämta via popup: {exc}")
finally:
try:
pp.close()
except Exception: # noqa: BLE001
pass
if saved_path is None:
log(" Ingen fil kunde hämtas för bilagan.")
else:
log(f"Bilaga sparad: {saved_path.name}")
var_name = step.get("var")
if var_name:
lst = vars_.setdefault(var_name, [])
if not isinstance(lst, list):
raise FlowError(
f"Variabel '{var_name}' är inte en lista (innehåller {type(lst).__name__})."
)
lst.append(str(saved_path))
elif t == "capture_pdf":
filename = _interpolate(step.get("filename") or "utskrift.pdf", vars_)
if Path(filename).name != filename:
raise FlowError(f"Ogiltigt filnamn: {filename!r} (inga sökvägar tillåtna).")
if not filename.lower().endswith(".pdf"):
filename += ".pdf"
timeout_s = float(step.get("timeout", 15))
download_dir.mkdir(parents=True, exist_ok=True)
target = download_dir / filename
has_trigger = bool(step.get("value"))
# Hindra webbläsarens skrivardialog från att blockera körningen -
# både på sidan just nu och på ev. sidor som laddas av klicket.
neutralize = "window.print = () => {}"
try:
page.add_init_script(neutralize)
except Exception: # noqa: BLE001
pass
try:
page.evaluate(neutralize)
except Exception: # noqa: BLE001
pass
def _looks_like_pdf(resp) -> bool:
try:
ct = (resp.headers or {}).get("content-type", "").lower()
except Exception: # noqa: BLE001
ct = ""
return "application/pdf" in ct or resp.url.lower().split("?")[0].endswith(".pdf")
pdf_bytes = None
# (1) Fånga ett PDF-nätverkssvar (öppnas ofta i en flik / <embed>).
if has_trigger:
loc = _resolve_locator(page, item_locator, step, vars_)
log(f"Fångar PDF via: {_describe_selector(step, vars_)}")
try:
with page.expect_response(_looks_like_pdf, timeout=timeout_s * 1000) as resp_info:
loc.first.click()
pdf_bytes = resp_info.value.body()
log(f" PDF-svar fångat: {resp_info.value.url}")
except Exception as exc: # noqa: BLE001
log(f" Inget direkt PDF-svar ({exc}); provar andra metoder")
page.wait_for_load_state("domcontentloaded")
# (2) Inbäddad PDF i sidan (eller någon iframe).
if pdf_bytes is None:
pdf_src = None
for frame in page.frames:
try:
pdf_src = frame.evaluate(
"""() => {
const el = document.querySelector(
'embed[type=\\"application/pdf\\"], embed[src*=\\".pdf\\"], '
+ 'iframe[src*=\\".pdf\\"], object[data*=\\".pdf\\"], a[href*=\\".pdf\\"]'
);
return el ? (el.src || el.data || el.href || null) : null;
}"""
)
except Exception: # noqa: BLE001
pdf_src = None
if pdf_src:
break
if pdf_src:
try:
r = page.context.request.get(pdf_src)
if r.ok:
pdf_bytes = r.body()
log(f" Hämtade inbäddad PDF: {pdf_src}")
else:
log(f" Inbäddad PDF svarade {r.status}: {pdf_src}")
except Exception as exc: # noqa: BLE001
log(f" Kunde inte hämta inbäddad PDF: {exc}")
# (3) Rendera sidan som den ser ut till PDF via Chrome (fungerar
# även i icke-headless läge, till skillnad från page.pdf()).
if pdf_bytes is None:
try:
client = page.context.new_cdp_session(page)
result = client.send(
"Page.printToPDF", {"printBackground": True, "preferCSSPageSize": True}
)
pdf_bytes = base64.b64decode(result["data"])
log(" Genererade PDF från sidan via Chrome (printToPDF)")
except Exception as exc: # noqa: BLE001
log(f" printToPDF misslyckades: {exc}")
if pdf_bytes is None:
log("Ingen PDF kunde fångas.")
else:
target.write_bytes(pdf_bytes)
log(f"PDF sparad: {target.name} ({len(pdf_bytes)} byte)")
var_name = step.get("var")
if var_name:
vars_[var_name] = str(target)
else:
raise FlowError(f"Okänd stegtyp: {t!r}")
def _notify_run_complete(error: Exception | None) -> None:
try:
from plyer import notification
if error is None:
title, message = "Flöde klart", "Körningen är klar."
else:
title, message = "Flöde misslyckades", f"Fel: {error}"
notification.notify(title=title, message=message, timeout=10)
except Exception: # noqa: BLE001
pass
def run_flow(
flow,
log: LogFn = print,
download_dir: Path = None,
close_mode: str = "always",
close_event=None,
on_finished: Callable[[Exception | None, dict], None] | None = None,
trace_path: Path | None = None,
settings: dict | None = None,
initial_vars: dict | None = None,
entry_subflow: str | None = None,
) -> None:
"""Kör ett flöde. `flow` kan vara nod/kant-grafen ({nodes, edges, start}),
ett helt flöde ({steps, ...} eller {nodes, ...}), eller en rå steglista
(gammalt format) - gamla format migreras internt till grafen.
initial_vars seedar variabellagret innan första steget - används för
indata-variabler (flow["inputs"] / värde-set) som matas in vid körning.
close_mode styr vad som händer med webbläsaren när stegen är klara:
"always" -- stäng alltid direkt (tidigare/standardbeteende)
"on_success" -- stäng bara om inget steg gav fel, håll öppen vid fel
"never" -- håll alltid öppen tills close_event sätts
on_finished(error, vars_) anropas direkt när stegen är klara (innan ev.
väntan på close_event), så anroparen kan uppdatera körningens status och
visa variablernas innehåll utan att behöva vänta på att webbläsaren
faktiskt stängs.
Om trace_path anges spelas en Playwright-trace (skärmdumpar, DOM-snapshots,
nätverk) in under hela körningen och sparas dit. Öppna den efteråt med
`python -m playwright show-trace <trace_path>` (eller `python -m rpa.show_trace`)
för att steg-för-steg se exakt vad som klickades och när det gick fel.
settings vidarebefordras till browser.launch_context() (headless,
viewport_width/height, locale, slow_mo) - se DEFAULT_SETTINGS i
rpa/browser.py. settings["notify_on_complete"] = true visar en
skrivbordsnotis när körningen är klar (via plyer, samma som main.py).
"""
from . import flows as _flows, graph_engine, migrate
settings = settings or {}
download_dir = download_dir or config.DOWNLOAD_DIR
# Normalisera indata till nod/kant-grafen.
if isinstance(flow, list):
graph = migrate.steps_to_graph(flow)
elif isinstance(flow, dict) and flow.get("nodes") is None and "steps" in flow:
graph = migrate.steps_to_graph(flow.get("steps", []), flow.get("layout"))
elif isinstance(flow, dict) and flow.get("nodes") is not None:
graph = {"nodes": flow["nodes"], "edges": flow.get("edges", []), "start": flow.get("start")}
settings = settings or flow.get("settings", {})
else:
graph = {"nodes": [], "edges": [], "start": None}
nodes, edges, start = graph["nodes"], graph["edges"], graph.get("start")
# Lokala subgrafer (call_flow target "#<id>") + valfri "kör bara subgrafen".
subflows = flow.get("subflows", []) if isinstance(flow, dict) else []
by_id = {s["id"]: s for s in subflows if isinstance(s, dict) and s.get("id")}
if entry_subflow:
sf = by_id.get(entry_subflow)
if sf is None:
log(f"FEL: okänd subgraf '{entry_subflow}'.")
if on_finished is not None:
on_finished(FlowError(f"Okänd subgraf: {entry_subflow}"), {})
return
nodes, edges, start = sf.get("nodes", []), sf.get("edges", []), sf.get("start")
log(f"Kör subgrafen '{sf.get('name') or entry_subflow}'.")
def resolve_call(target: str):
if target.startswith("#"):
s = by_id.get(target[1:])
if s is None:
raise FlowError(f"Okänd lokal subgraf: {target}")
return s.get("nodes", []), s.get("edges", []), s.get("start"), s.get("inputs", [])
f = _flows.load_flow(target)
return f["nodes"], f["edges"], f.get("start"), f.get("inputs", [])
# Rena desktop-/variabelflöden startar ingen webbläsare. Sätts explicit via
# settings["skip_browser"] eller upptäcks automatiskt (inga webbläsarnoder).
skip_browser = bool(settings.get("skip_browser")) or not graph_engine.graph_needs_browser(nodes, resolve_call)
if skip_browser:
log("Kör utan webbläsare (skip_browser / inga webbläsarnoder).")
_run_and_finish(
None, None, nodes, edges, start, log, download_dir, close_mode, close_event,
on_finished, None, settings, initial_vars, resolve_call,
)
return
reason = graph_engine.first_browser_node(nodes)
if reason:
log(f"Startar webbläsare (krävs av: {reason}).")
with sync_playwright() as p:
context = launch_context(p, settings)
if trace_path is not None:
context.tracing.start(screenshots=True, snapshots=True, sources=True)
page = get_page(context)
_run_and_finish(
context, page, nodes, edges, start, log, download_dir, close_mode, close_event,
on_finished, trace_path, settings, initial_vars, resolve_call,
)
def _run_and_finish(
context, page, nodes, edges, start, log, download_dir, close_mode, close_event, on_finished,
trace_path, settings, initial_vars=None, resolve_call=None,
) -> None:
from . import graph_engine
error: Exception | None = None
# Indata-variabler är strängar (samma som övriga variabler i motorn).
vars_: dict = {str(k): (v if isinstance(v, str) else str(v)) for k, v in (initial_vars or {}).items()}
if vars_:
log(f"Indata-variabler: {', '.join(sorted(vars_))}")
desktop_holder: dict = {"session": None}
try:
graph_engine.run_graph(
nodes, edges, log, page=page, download_dir=download_dir,
vars_=vars_, desktop_holder=desktop_holder, start=start,
resolve_call=resolve_call,
)
except Exception as exc: # noqa: BLE001
error = exc
log(f"FEL: {exc}")
if settings.get("notify_on_complete"):
_notify_run_complete(error)
if on_finished is not None:
on_finished(error, vars_)
should_close_now = close_mode == "always" or (close_mode == "on_success" and error is None)
# Desktop-appar som flödet startat städas enligt samma close-mode som
# webbläsaren (kill vid "always", eller "on_success" utan fel).
if desktop_holder["session"] is not None:
try:
desktop_holder["session"].close(kill_launched=should_close_now)
except Exception as exc: # noqa: BLE001
log(f"Kunde inte städa desktop-sessionen: {exc}")
if context is not None:
if not should_close_now:
log('Webbläsaren hålls öppen. Klicka "Stäng webbläsare" i UI:t när du är klar.')
if close_event is not None:
close_event.wait()
if trace_path is not None:
trace_path.parent.mkdir(parents=True, exist_ok=True)
context.tracing.stop(path=str(trace_path))
log(f"Trace sparad: {trace_path.name} (öppna med \"Visa trace\")")
context.close()