commits
tags
"""Bygger skelett-flödet data/flows/salma-elvis.json och delar config.json i
laddbara delar i data/exports/ (salma_config.json + salma_avsandare.json).
python scripts/build_salma_flow.py
Kör från projektroten (där config.json ligger). Se docs/salma-elvis.md.
Idempotent – kör om efter att du ändrat config.json. OBS: skriver över
data/flows/salma-elvis.json (dina egna selector-ändringar går förlorade –
kör bara om när du medvetet vill regenerera skelettet).
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from rpa import config, flows # noqa: E402
def _write_config_parts() -> None:
cfg = json.loads((ROOT / "config.json").read_text(encoding="utf-8"))
elvis = cfg["elvis"]
config.EXPORT_DIR.mkdir(parents=True, exist_ok=True)
(config.EXPORT_DIR / "salma_config.json").write_text(
json.dumps({
"translationTable": elvis["translationTable"],
"groupLevelRules": elvis["groupLevelRules"],
"specialWorkflows": elvis["specialWorkflows"],
"unsupportedGroups": elvis["unsupportedGroups"],
"fixedValues": elvis["fixedValues"],
"skipFields": elvis["skipFields"],
"fieldOrder": elvis["fieldOrder"],
"betalarkodNote": elvis["betalarkodNote"],
}, ensure_ascii=False, indent=2), encoding="utf-8")
avs = {k: v for k, v in elvis["avsandarregister"].items()
if not k.startswith("_") and k != "fallbackNumber"}
avs["_fallback"] = elvis["avsandarregister"]["fallbackNumber"]
(config.EXPORT_DIR / "salma_avsandare.json").write_text(
json.dumps(avs, ensure_ascii=False, indent=2), encoding="utf-8")
class G:
"""Liten nod/kant-byggare."""
def __init__(self) -> None:
self.nodes: list[dict] = []
self.edges: list[dict] = []
self._n = 0
def add(self, typ, _from=None, _handle="out", label="", **fields):
self._n += 1
nid = f"{typ[:4]}{self._n}"
node = {"id": nid, "type": typ, **fields}
if label:
node["label"] = label
self.nodes.append(node)
if _from:
self.edge(_from, nid, _handle)
return nid
def edge(self, src, tgt, handle="out"):
self.edges.append({"id": f"e{len(self.edges)}", "source": src,
"sourceHandle": handle, "target": tgt, "targetHandle": "in"})
def result(self, start):
return {"nodes": self.nodes, "edges": self.edges, "start": start}
def _build_main() -> dict:
m = G()
start = m.add("prompt", mode="confirm",
message="Stå på ärendets HUVUDVY i Sälma (inte Redigera-läge). Fortsätt?",
label="Startkontroll")
c = m.add("read_url", start, var="arendenr", pattern=r"\d{5,8}", label="Ärendenummer ur adressen")
c = m.add("extract_text", c, by="css", value="TODO: selector för Grupp", scope="page", var="grupp",
label="TODO: läs Grupp i Sälma")
c = m.add("extract_text", c, by="css", value="TODO: selector för Specialitet", scope="page", var="spec",
label="TODO: läs Specialitet i Sälma")
c = m.add("load_var", c, var="cfg", filename="salma_config.json")
c = m.add("load_var", c, var="avs", filename="salma_avsandare.json")
c = m.add("lookup", c, from_var="cfg.specialWorkflows", key="{grupp}|{spec}", var="special",
ci=True, on_missing="empty", label="Specialflöde? (ej ELVIS)")
ifsp = m.add("if", c, branches=[{"condition": {"kind": "var_not_equals", "var": "special", "text": ""}}],
label="Special vs normal")
m.add("call_flow", ifsp, "case0", target="#special",
**{"in": [{"name": "arendenr", "value": "{arendenr}"},
{"name": "instruktion", "value": "{special}"}], "out": []},
label="Specialhantering")
c = m.add("lookup", ifsp, "else", from_var="cfg.unsupportedGroups", key="{grupp}", var="unsup",
ci=True, on_missing="empty", label="Ohanterad grupp?")
ifun = m.add("if", c, branches=[{"condition": {"kind": "var_not_equals", "var": "unsup", "text": ""}}],
label="Stopp om ohanterad")
m.add("stop", ifun, "case0", message="Gruppen '{grupp}' hanteras inte av flödet: {unsup}",
label="Stanna – manuell hantering")
c = m.add("click", ifun, "else", by="text", value="Filer", scope="page", on_error="continue",
label="TODO: öppna Filer-fliken")
c = m.add("save_attachment", c, by="css", value="TODO: selector för remiss-länken", scope="page",
filename="remiss_{arendenr}.pdf", var="pdf_paths", label="TODO: ladda ner remiss-PDF")
for var, pat, near, lbl in [
("personnummer", r"(19|20)?\d{6}[-+]?\d{4}", None, "OCR: personnummer"),
("remissdatum", r"\d{4}-\d{2}-\d{2}", None, "OCR: remissdatum"),
("lakarnamn", None, "Läkare", "OCR: läkare"),
("avsandarnamn", None, "Avsändare", "OCR: avsändare"),
]:
step: dict = {"source": "pdf", "file": "remiss_{arendenr}.pdf", "var": var, "lang": "sv-SE"}
if pat:
step["pattern"] = pat
if near:
step["near"] = near
step["near_mode"] = "same_line"
c = m.add("ocr_read", c, label=lbl, **step)
c = m.add("lookup", c, from_var="cfg.translationTable", key="{grupp}|{spec}", var="trans",
ci=True, on_missing="error", label="Översättningstabell")
for var, text in [
("team", "{trans.team}"), ("orsak", "{trans.orsak}"), ("vardniva", "{trans.vardniva}"),
("typAvVardbegaran", "{cfg.fixedValues.typAvVardbegaran}"),
("avvikandeAvtal", "{cfg.fixedValues.avvikandeAvtal}"),
("betalarkod", "{cfg.fixedValues.betalarkod}"),
("internNotering", "{arendenr}"), ("beslutsdatum", "{remissdatum}"),
]:
c = m.add("set_var", c, var=var, text=text)
c = m.add("lookup", c, from_var="avs", key="{avsandarnamn}", var="fran", ci=True, on_missing="empty",
label="Slå upp avsändarnummer")
ifav = m.add("if", c, branches=[{"condition": {"kind": "var_equals", "var": "fran", "text": ""}}],
label="Nytt avsändarnummer?")
p = m.add("prompt", ifav, "case0", mode="input", var="fran",
message="Ingen träff i avsändarregistret för '{avsandarnamn}'.\nSlå upp numret i ELVIS "
"(förstoringsglaset vid Från) och skriv in det – lämna tomt för fallback.",
label="Fråga efter nummer")
p = m.add("merge_json", p, file="salma_avsandare.json", key="{avsandarnamn}", value="{fran}",
label="Kom ihåg numret")
review = m.add("prompt", mode="fields",
message="Kontrollera alla värden innan de skrivs in i ELVIS. Rätta ev. OCR-fel.",
fields="personnummer, remissdatum, beslutsdatum, team, orsak, typAvVardbegaran, "
"fran, vardniva, avvikandeAvtal, betalarkod, internNotering",
label="Granska innan ELVIS")
m.edge(p, review)
m.edge(ifav, review, "else")
c = m.add("desktop_connect", review, title="ELVIS SU", label="Anslut till ELVIS")
c = m.add("desktop_press_key", c, keys="{personnummer}{ENTER}{ENTER}", label="Personnummer + Enter x2")
c = m.add("desktop_press_key", c, keys="!nb", label="Ny vårdbegäran (Alt+N,B)")
fields = [
("beslutsdatum", "{beslutsdatum}"), ("(vårdgaranti – hoppas över)", ""),
("team", "{team}"), ("orsak", "{orsak}"), ("typAvVardbegaran", "{typAvVardbegaran}"),
("(accepterar ej hänvisning)", ""), ("fran", "{fran}"),
("(vårdenhet)", ""), ("(gatuadress)", ""), ("(postadress)", ""), ("(avsändande resurs)", ""),
("internNotering", "{internNotering}"), ("(betalningsförbindelse – hoppas över)", ""),
("vardniva", "{vardniva}"), ("avvikandeAvtal", "{avvikandeAvtal}"), ("betalarkod", "{betalarkod}"),
]
for name, val in fields:
c = m.add("desktop_press_key", c, keys="{TAB}", label=f"Tab -> {name}")
if val:
c = m.add("desktop_press_key", c, keys=val, label=f"Skriv {name}")
c = m.add("prompt", c, mode="confirm",
message="Granska det ifyllda ELVIS-formuläret på skärmen. OK = spara (Ctrl+S), Avbryt = stoppa.",
label="Granska ELVIS")
c = m.add("desktop_press_key", c, keys="^s", label="Spara ELVIS (Ctrl+S)")
c = m.add("desktop_press_key", c, keys="%{LEFT}", label="Tillbaka (Alt+Vänster)")
m.add("call_flow", c, target="#uppdatera_salma",
**{"in": [{"name": "personnummer", "value": "{personnummer}"},
{"name": "remissdatum", "value": "{remissdatum}"},
{"name": "status", "value": "Tilldelat"}], "out": []},
label="Uppdatera Sälma")
return m.result(start)
def _build_uppdatera_salma() -> dict:
s = G()
first = s.add("desktop_connect", title="ahk_exe msedge.exe", label="TODO: fönster för Sälma (Edge)")
c = s.add("desktop_click", first, by="name", value="Redigera", label="TODO: klicka Redigera")
c = s.add("prompt", c, mode="confirm",
message="Klicka penn-ikonen bredvid Personnummer i Sälma (fältet är låst), klicka sedan OK.",
label="Lås upp personnummer (manuellt tills selector finns)")
c = s.add("desktop_type", c, by="name", value="TODO: personnummer-fältet", text="{personnummer}",
label="TODO: skriv personnummer")
c = s.add("desktop_click", c, by="name", value="Kontrollera personnummer", on_error="continue",
label="TODO: Kontrollera personnummer")
c = s.add("prompt", c, mode="confirm",
message="Bekräfta 'Kontrollera personnummer'-popupen i Sälma, klicka sedan OK.",
label="Popup: kontrollera personnummer")
c = s.add("desktop_type", c, by="name", value="TODO: remissdatum-fältet", text="{remissdatum}",
label="TODO: skriv remissdatum")
c = s.add("desktop_type", c, by="name", value="TODO: status-fältet", text="{status}",
label="TODO: sätt Status")
c = s.add("prompt", c, mode="confirm",
message="Kontrollera Personnummer, Remissdatum och Status i Sälma. OK = Spara, Avbryt = stoppa.",
label="Granska Sälma")
s.add("desktop_click", c, by="name", value="Spara", label="Spara Sälma")
return s.result(first)
def _build_special() -> dict:
sp = G()
first = sp.add("prompt", mode="confirm", message="{instruktion}", label="Instruktioner för specialhantering")
c = sp.add("click", first, by="text", value="Filer", scope="page", on_error="continue", label="TODO: Filer-fliken")
c = sp.add("save_attachment", c, by="css", value="TODO: remiss-länken", scope="page",
filename="remiss_{arendenr}.pdf", var="pdf_paths", label="TODO: ladda ner remiss-PDF")
c = sp.add("ocr_read", c, source="pdf", file="remiss_{arendenr}.pdf", var="personnummer",
pattern=r"(19|20)?\d{6}[-+]?\d{4}", label="OCR: personnummer")
c = sp.add("ocr_read", c, source="pdf", file="remiss_{arendenr}.pdf", var="remissdatum",
pattern=r"\d{4}-\d{2}-\d{2}", label="OCR: remissdatum")
sp.add("call_flow", c, target="#uppdatera_salma",
**{"in": [{"name": "personnummer", "value": "{personnummer}"},
{"name": "remissdatum", "value": "{remissdatum}"},
{"name": "status", "value": "Stängt"}], "out": []},
label="Uppdatera Sälma (stäng)")
return sp.result(first)
def main() -> None:
_write_config_parts()
main_g = _build_main()
flow = {
"format": "graph",
"nodes": main_g["nodes"], "edges": main_g["edges"], "start": main_g["start"],
"settings": {}, "inputs": [],
"subflows": [
{"id": "uppdatera_salma", "name": "Uppdatera Sälma", **_build_uppdatera_salma(),
"inputs": [{"name": "personnummer"}, {"name": "remissdatum"},
{"name": "status", "default": "Tilldelat"}]},
{"id": "special", "name": "Specialhantering", **_build_special(),
"inputs": [{"name": "arendenr"}, {"name": "instruktion"}]},
],
}
flows.save_flow("salma-elvis", flow, strict=False)
print(f"data/flows/salma-elvis.json: {len(main_g['nodes'])} main-noder, 2 subgrafer")
print("data/exports/salma_config.json + salma_avsandare.json")
if __name__ == "__main__":
main()