commits
tags
import json
import re
import subprocess
import sys
from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, HTMLResponse, PlainTextResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from rpa import assist, blocks, config, flows, schema
from rpa.blocks import BlockNotFound, InvalidBlock, InvalidBlockName
from rpa.export import generate_standalone_script
from rpa.flows import FlowNotFound, InvalidFlow, InvalidFlowName
from rpa.runs import close_browser, get_run, get_trace_path, start_run
app = FastAPI(title="Egenremiss RPA - flödesbyggare")
STATIC_DIR = Path(__file__).parent / "static"
WIKI_PATH = config.ROOT_DIR / "docs" / "wiki.html"
CLOSE_MODES = set(schema.CLOSE_MODES)
class Flow(BaseModel):
# Grafform. `steps` accepteras fortfarande (migreras vid inläsning) tills
# frontend är helt övergången.
nodes: list | None = None
edges: list | None = None
start: str | None = None
steps: list | None = None
settings: dict = {}
layout: dict = {}
inputs: list = []
input_sets: dict = {}
subflows: list | None = None
class RunOptions(BaseModel):
close_mode: str = "always"
initial_vars: dict = {}
subflow: str | None = None
class AssistRequest(BaseModel):
instruction: str
nodes: list = []
edges: list = []
start: str | None = None
steps: list = []
settings: dict = {}
subflows: list = []
class CaptureRequest(BaseModel):
name: str = ""
_IMAGE_NAME_RE = re.compile(r"^[a-zA-Z0-9_\-åäöÅÄÖ ]{1,80}\.png$")
def _image_path(name: str) -> Path:
if not _IMAGE_NAME_RE.match(name or ""):
raise HTTPException(400, "Ogiltigt bildnamn")
return config.DESKTOP_IMAGE_DIR / name
def _require_name(name: str) -> str:
try:
flows.flow_path(name)
except InvalidFlowName as exc:
raise HTTPException(400, str(exc)) from exc
return name
@app.get("/wiki", response_class=HTMLResponse)
def wiki() -> str:
if not WIKI_PATH.exists():
raise HTTPException(404, "Wiki-sidan hittas inte (docs/wiki.html saknas)")
return WIKI_PATH.read_text(encoding="utf-8")
@app.get("/api/step-schema")
def step_schema() -> dict:
"""Auktoritativ stegschema (rpa/schema.py) - samma som MCP-servern och
NL-till-flöde använder."""
return {
"step_types": schema.STEP_TYPES,
"condition_kinds": schema.CONDITION_KINDS,
"settings": schema.SETTINGS_SCHEMA,
"input_fields": schema.INPUT_FIELDS,
"close_modes": schema.CLOSE_MODES,
}
# --- Referensbilder för desktop_click_image / desktop_wait_for_image ---
@app.get("/api/desktop-images")
def list_desktop_images() -> list[str]:
return sorted(p.name for p in config.DESKTOP_IMAGE_DIR.glob("*.png"))
@app.get("/api/desktop-images/{name}")
def get_desktop_image(name: str) -> FileResponse:
path = _image_path(name)
if not path.exists():
raise HTTPException(404, "Bilden finns inte")
return FileResponse(str(path), media_type="image/png")
@app.delete("/api/desktop-images/{name}")
def delete_desktop_image(name: str) -> dict:
path = _image_path(name)
if path.exists():
path.unlink()
return {"ok": True}
class InspectRequest(BaseModel):
title: str
@app.get("/api/desktop/windows")
def desktop_windows() -> list[str]:
from rpa import desktop_inspect
try:
return desktop_inspect.list_windows()
except desktop_inspect.InspectUnavailable as exc:
raise HTTPException(501, str(exc)) from exc
@app.post("/api/desktop/inspect")
def desktop_inspect_window(req: InspectRequest) -> list[dict]:
from rpa import desktop_inspect
try:
return desktop_inspect.inspect_window(req.title)
except desktop_inspect.InspectUnavailable as exc:
raise HTTPException(501, str(exc)) from exc
except ValueError as exc:
raise HTTPException(404, str(exc)) from exc
@app.post("/api/desktop-images/capture")
def capture_desktop_image(req: CaptureRequest) -> dict:
"""Kör rpa/desktop_capture i en egen process (öppnar ett overlay-fönster på
servermaskinen). Blockerar tills användaren dragit klart eller avbrutit."""
try:
proc = subprocess.run(
[sys.executable, "-m", "rpa.desktop_capture", req.name or ""],
cwd=str(config.ROOT_DIR),
capture_output=True,
text=True,
timeout=180,
)
except subprocess.TimeoutExpired as exc:
raise HTTPException(408, "Bildfångsten tog för lång tid (avbröts)") from exc
last_line = (proc.stdout or "").strip().splitlines()[-1:] or [""]
try:
result = json.loads(last_line[0])
except json.JSONDecodeError:
raise HTTPException(500, f"Oväntat svar från bildfångaren: {proc.stdout} {proc.stderr}") from None
if not result.get("ok"):
raise HTTPException(422, result.get("reason", "avbruten"))
return result
# --- Block-bibliotek (återanvändbara flödesdelar) ---
class BlockBody(BaseModel):
name: str = ""
description: str = ""
nodes: list | None = None
edges: list | None = None
start: str | None = None
steps: list | None = None
settings: dict = {}
@app.get("/api/blocks")
def list_blocks() -> list[dict]:
return blocks.list_blocks()
@app.get("/api/blocks/{name}")
def get_block(name: str) -> dict:
try:
return blocks.load_block(name)
except BlockNotFound as exc:
raise HTTPException(404, "Blocket finns inte") from exc
except InvalidBlockName as exc:
raise HTTPException(400, str(exc)) from exc
@app.put("/api/blocks/{name}")
def save_block(name: str, body: BlockBody) -> dict:
try:
blocks.save_block(name, body.model_dump())
except InvalidBlockName as exc:
raise HTTPException(400, str(exc)) from exc
except InvalidBlock as exc:
raise HTTPException(422, {"errors": exc.errors}) from exc
return {"ok": True}
@app.delete("/api/blocks/{name}")
def delete_block(name: str) -> dict:
try:
blocks.delete_block(name)
except InvalidBlockName as exc:
raise HTTPException(400, str(exc)) from exc
return {"ok": True}
@app.get("/api/flows")
def list_flows() -> list[str]:
return flows.list_flows()
@app.get("/api/flows/{name}")
def get_flow(name: str) -> dict:
_require_name(name)
try:
return flows.load_flow(name)
except FlowNotFound:
return {"format": "graph", "nodes": [], "edges": [], "start": None,
"settings": {}, "layout": {}, "inputs": [], "input_sets": {}}
@app.put("/api/flows/{name}")
def save_flow(name: str, flow: Flow) -> dict:
_require_name(name)
try:
flows.save_flow(name, flow.model_dump())
except InvalidFlow as exc:
raise HTTPException(422, {"errors": exc.errors}) from exc
return {"ok": True}
@app.delete("/api/flows/{name}")
def delete_flow(name: str) -> dict:
_require_name(name)
flows.delete_flow(name)
return {"ok": True}
@app.post("/api/flows/{name}/assist")
def assist_flow(name: str, req: AssistRequest) -> dict:
"""NL → flöde. Genererar/ändrar ett flöde utifrån en instruktion men SPARAR
INTE - klienten visar förslaget och användaren trycker Spara/Kör själv."""
_require_name(name)
from rpa.migrate import flow_to_graph
current = flow_to_graph({
"nodes": req.nodes or None,
"edges": req.edges,
"start": req.start,
"steps": req.steps,
"settings": req.settings,
"subflows": req.subflows,
})
try:
result = assist.generate_flow(req.instruction, current_flow=current)
except assist.AssistUnavailable as exc:
raise HTTPException(501, str(exc)) from exc
except assist.AssistError as exc:
raise HTTPException(422, str(exc)) from exc
return result
@app.post("/api/flows/{name}/run")
def run_flow_endpoint(name: str, options: RunOptions = RunOptions()) -> dict:
_require_name(name)
try:
flow = flows.load_flow(name)
except FlowNotFound as exc:
raise HTTPException(404, "Flödet finns inte") from exc
close_mode = options.close_mode if options.close_mode in CLOSE_MODES else "always"
initial_vars = {str(k): str(v) for k, v in (options.initial_vars or {}).items()}
run_id = start_run(
flow,
close_mode=close_mode,
settings=flow.get("settings", {}),
initial_vars=initial_vars,
entry_subflow=options.subflow,
)
return {"run_id": run_id}
@app.get("/api/flows/{name}/export")
def export_flow(name: str) -> PlainTextResponse:
_require_name(name)
try:
flow = flows.load_flow(name)
except FlowNotFound as exc:
raise HTTPException(404, "Flödet finns inte") from exc
script = generate_standalone_script(name, flow)
return PlainTextResponse(
script,
media_type="text/x-python",
headers={"Content-Disposition": f'attachment; filename="{name}.py"'},
)
@app.get("/api/runs/{run_id}")
def run_status(run_id: str) -> dict:
run = get_run(run_id)
if run is None:
raise HTTPException(404, "Okänt run_id")
return run
@app.post("/api/runs/{run_id}/close-browser")
def close_browser_endpoint(run_id: str) -> dict:
if not close_browser(run_id):
raise HTTPException(404, "Okänt run_id")
return {"ok": True}
@app.post("/api/runs/{run_id}/show-trace")
def show_trace_endpoint(run_id: str) -> dict:
trace_path = get_trace_path(run_id)
if trace_path is None:
raise HTTPException(404, "Ingen trace sparad för denna körning (än)")
# Öppnar Playwrights Trace Viewer i ett eget fönster på samma maskin
# som servern körs på - startas som en fristående process så webui:t
# inte blockeras.
subprocess.Popen([sys.executable, "-m", "playwright", "show-trace", str(trace_path)])
return {"ok": True}
# Frontend (byggd av webui/frontend, incheckad). Monteras SIST så att alla
# /api- och /wiki-rutter matchas först. Kör `npm run build` i webui/frontend
# efter UI-ändringar.
if STATIC_DIR.exists():
app.mount("/", StaticFiles(directory=str(STATIC_DIR), html=True), name="static")
else: # pragma: no cover
@app.get("/", response_class=HTMLResponse)
def _no_build() -> str:
return (
"<h1>Frontend inte byggd</h1><p>Kör <code>cd webui/frontend && "
"npm install && npm run build</code>.</p>"
)