foxygit / RPA-Remisser Log in
commits tags

/rpa/export.py · 3.78 KB

raw
"""Genererar ett fristående Python-script från ett sparat flöde.

Det exporterade scriptet importerar rpa-paketet och anropar samma
run_flow()-motor som webui:t använder - ingen logik dupliceras, så
beteendet blir identiskt med att köra flödet via "Kör flöde".
"""

from pprint import pformat

from .migrate import flow_to_graph

SCRIPT_TEMPLATE = '''#!/usr/bin/env python
"""Fristående script, exporterat från flödet "{name}" (RPA-verktyg).

Kör med:
    python {filename}.py
    python {filename}.py --headless
    python {filename}.py --close-mode never
    python {filename}.py --set "Test"            # använd ett sparat värde-set
    python {filename}.py --var org_id=9          # sätt en indata-variabel (upprepbar)

Kräver att rpa-paketet finns tillgängligt (kör från projektroten där
rpa/ och data/ ligger, eller kopiera med hela rpa/-mappen och låt en
data/-mapp skapas bredvid scriptet vid första körningen).
"""

import argparse

from rpa import config
from rpa.flow_engine import run_flow

NODES = {nodes_json}

EDGES = {edges_json}

START = {start_json}

SUBFLOWS = {subflows_json}

SETTINGS = {settings_json}

INPUTS = {inputs_json}

INPUT_SETS = {input_sets_json}


def build_initial_vars(chosen_set, var_overrides):
    resolved = {{}}
    for inp in INPUTS:
        if inp.get("name") and inp.get("default") not in (None, ""):
            resolved[str(inp["name"])] = str(inp["default"])
    if chosen_set:
        if chosen_set not in INPUT_SETS:
            raise SystemExit(f"Okänt värde-set: {{chosen_set!r}}. Tillgängliga: {{list(INPUT_SETS)}}")
        resolved.update({{str(k): str(v) for k, v in INPUT_SETS[chosen_set].items()}})
    for item in var_overrides or []:
        if "=" not in item:
            raise SystemExit(f"--var förväntar NAMN=VÄRDE, fick: {{item!r}}")
        k, v = item.split("=", 1)
        resolved[k.strip()] = v
    return resolved


def main() -> None:
    parser = argparse.ArgumentParser(description="Kör flödet: {name}")
    parser.add_argument(
        "--headless", action="store_true",
        help="Kör utan synligt webbläsarfönster (kräver en redan giltig inloggningssession)",
    )
    parser.add_argument(
        "--close-mode", choices=["always", "on_success", "never"], default="always",
        help="Vad som ska hända med webbläsaren när flödet är klart",
    )
    parser.add_argument("--set", dest="value_set", default=None, help="Namn på ett sparat värde-set")
    parser.add_argument("--var", action="append", default=[], metavar="NAMN=VÄRDE",
                        help="Sätt en indata-variabel (kan upprepas)")
    args = parser.parse_args()

    settings = dict(SETTINGS)
    if args.headless:
        settings["headless"] = True

    run_flow(
        {{"format": "graph", "nodes": NODES, "edges": EDGES, "start": START, "subflows": SUBFLOWS}},
        log=print,
        download_dir=config.DOWNLOAD_DIR,
        close_mode=args.close_mode,
        settings=settings,
        initial_vars=build_initial_vars(args.value_set, args.var),
    )


if __name__ == "__main__":
    main()
'''


def generate_standalone_script(name: str, flow: dict) -> str:
    # pformat (inte json.dumps) - genererad kod är Python, inte JSON: JSON:s
    # true/false/null är inte giltig Python-syntax (True/False/None krävs).
    def src(value):
        return pformat(value, indent=2, width=100, sort_dicts=False)

    g = flow_to_graph(flow)
    return SCRIPT_TEMPLATE.format(
        name=name,
        filename=name,
        nodes_json=src(g.get("nodes", [])),
        edges_json=src(g.get("edges", [])),
        start_json=src(g.get("start")),
        subflows_json=src(flow.get("subflows", [])),
        settings_json=src(flow.get("settings", {})),
        inputs_json=src(flow.get("inputs", [])),
        input_sets_json=src(flow.get("input_sets", {})),
    )