dtoro/micronomicon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: composer improvements

Browse Source
This commit is contained in:
authentik Default Admin
2026-04-05 09:53:01 +02:00
parent 7838760ca4
commit e1db06104e
17 changed files with 1301 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
backend/uframe/codegen.py
View File

@@ -29,14 +29,56 @@ def _indent(code: str, level: int = 1) -> str:
def _resolve_vars(text: str) -> str:
"""Convert $var references to Python f-string expressions."""
"""Convert $var references to Python f-string expressions.
First escapes literal braces (e.g. @color{ff0}{text} → @color{{ff0}}{{text}})
so they survive f-string evaluation, then replaces $var → {var}.
"""
import re
# Use a unique placeholder for $var refs, escape all braces, then restore
_PH = "\x00VAR"
counter = [0]
placeholders: dict[str, str] = {}
def stash_var(m: re.Match) -> str:
var = m.group(1)
if "." in var:
parts = var.split(".")
base = parts[0]
chain = "".join(f"['{p}']" for p in parts[1:])
expr = "{" + base + chain + "}"
else:
expr = "{" + var + "}"
key = f"{_PH}{counter[0]}\x00"
counter[0] += 1
placeholders[key] = expr
return key
# 1. Stash $var references with placeholders
result = re.sub(r'\$([a-zA-Z_][\w.]*)', stash_var, text)
# 2. Escape all remaining braces for f-string safety
result = result.replace("{", "{{").replace("}", "}}")
# 3. Restore $var placeholders (unescaped)
for key, expr in placeholders.items():
result = result.replace(key, expr)
return result
def _resolve_vars_code(text: str) -> str:
"""Convert $var references to bare Python identifiers.
Simple vars: $name → name
Dotted paths: $item.name → item['name'] (dict access)
"""
import re
# Replace $var.attr.attr with {var_attr_attr} and simple $var with {var}
def replace_var(m: re.Match) -> str:
var = m.group(1)
# Replace dots with underscores for Python variable names
py_var = var.replace(".", "_")
return "{" + py_var + "}"
if "." in var:
parts = var.split(".")
base = parts[0]
chain = "".join(f"['{p}']" for p in parts[1:])
return base + chain
return var
return re.sub(r'\$([a-zA-Z_][\w.]*)', replace_var, text)
@@ -67,14 +109,28 @@ def _emit_node(node: IRNode, indent_level: int = 0) -> list[str]:
elif node.source_type == SourceType.JSON:
lines.append(f"{ind}{var} = _read_json({node.command!r})")
elif node.source_type == SourceType.PYTHON:
# Restricted eval — only datetime/secrets modules available
lines.append(f"{ind}{var} = eval({node.command!r}, {{'datetime': datetime, 'secrets': secrets}})")
lines.append(f"{ind}{var} = eval({node.command!r}, {{'datetime': _dt_cls, 'timedelta': timedelta, 'secrets': secrets, 'os': os, 'json': json}})")
elif node.source_type == SourceType.PARAM:
lines.append(f"{ind}{var} = _get_param({node.command!r})")
elif node.source_type == SourceType.RNS:
import shlex as _shlex
safe_cmd = _shlex.quote(node.command)
lines.append(f"{ind}{var} = _shell('rnstatus ' + shlex.quote({safe_cmd!r}), timeout={node.timeout})")
elif node.source_type == SourceType.HTTP:
method = node.http_method or "GET"
url = _resolve_vars(node.command)
hdrs = _resolve_vars(node.http_headers) if node.http_headers else ""
body = _resolve_vars(node.http_body) if node.http_body else ""
if body:
lines.append(f"""{ind}{var} = _http(f'''{url}''', method={method!r}, body=f'''{body}''', headers=f'''{hdrs}''', timeout={node.timeout})""")
elif hdrs:
lines.append(f"""{ind}{var} = _http(f'''{url}''', method={method!r}, headers=f'''{hdrs}''', timeout={node.timeout})""")
else:
lines.append(f"{ind}{var} = _http({node.command!r}, method={method!r}, timeout={node.timeout})")
elif node.source_type == SourceType.SQLITE:
lines.append(f"{ind}{var} = _sqlite({node.command!r}, {node.query!r})")
elif node.source_type == SourceType.ENV:
lines.append(f"{ind}{var} = os.environ.get({node.command!r}, '')")
elif isinstance(node, CacheControl):
lines.append(f"{ind}_cache_seconds = {node.seconds}")
@@ -83,8 +139,7 @@ def _emit_node(node: IRNode, indent_level: int = 0) -> list[str]:
lines.append(f"{ind}{node.state_name} = _load_state({node.path!r})")
elif isinstance(node, IfBlock):
cond = _resolve_vars(node.condition)
# Convert simple comparisons
cond = _resolve_vars_code(node.condition)
cond = cond.replace("&&", " and ").replace("||", " or ")
lines.append(f"{ind}if {cond}:")
if node.children:
@@ -94,7 +149,7 @@ def _emit_node(node: IRNode, indent_level: int = 0) -> list[str]:
lines.append(f"{ind} pass")
for elif_cond, elif_children in node.elif_branches:
ec = _resolve_vars(elif_cond).replace("&&", " and ").replace("||", " or ")
ec = _resolve_vars_code(elif_cond).replace("&&", " and ").replace("||", " or ")
lines.append(f"{ind}elif {ec}:")
if elif_children:
for child in elif_children:
@@ -108,7 +163,7 @@ def _emit_node(node: IRNode, indent_level: int = 0) -> list[str]:
lines.extend(_emit_node(child, indent_level + 1))
elif isinstance(node, ForLoop):
iterable = _resolve_vars(node.iterable)
iterable = _resolve_vars_code(node.iterable)
lines.append(f"{ind}for {node.var_name} in _iter({iterable}):")
if node.children:
for child in node.children:
@@ -146,13 +201,16 @@ def _emit_node(node: IRNode, indent_level: int = 0) -> list[str]:
elif isinstance(node, Gauge):
label = _resolve_vars(node.label)
value = _resolve_vars(str(node.value)) if "$" in str(node.value) else str(node.value)
raw_val = getattr(node, "_raw_value", None)
value = _resolve_vars(raw_val) if raw_val and "$" in raw_val else str(node.value)
raw_max = getattr(node, "_raw_max_val", None)
max_val = _resolve_vars(raw_max) if raw_max and "$" in raw_max else str(node.max_val)
extra = ""
if node.warn is not None:
extra += f" warn={node.warn}"
if node.crit is not None:
extra += f" crit={node.crit}"
lines.append(f"{ind}_uf_source_parts.append(f'gauge \"{label}\" {value} {node.max_val} {node.bar_width}{extra}')")
lines.append(f"{ind}_uf_source_parts.append(f'gauge \"{label}\" {value} {max_val} {node.bar_width}{extra}')")
elif isinstance(node, Status):
label = _resolve_vars(node.label)
@@ -221,6 +279,7 @@ def _build_script(uframe_import: str, page_logic: str, page_title: str, page_wid
"# Do not edit — regenerate with: uframe compile <source>.uf",
"",
"import os, sys, json, subprocess, datetime, secrets, shlex",
"from datetime import datetime as _dt_cls, timedelta",
"",
"# ─── Runtime Helpers ─────────────────────────────────────────",
"",
@@ -279,6 +338,42 @@ def _build_script(uframe_import: str, page_logic: str, page_title: str, page_wid
' return val.strip().splitlines()',
' return []',
"",
'def _http(url, method="GET", body="", headers="", timeout=10):',
' """HTTP request, return response body (JSON parsed if possible)."""',
' import urllib.request, urllib.error',
' try:',
' data = body.encode("utf-8") if body else None',
' req = urllib.request.Request(url, data=data, method=method)',
' req.add_header("User-Agent", "uframe/1.0")',
' if body and not headers:',
' req.add_header("Content-Type", "application/json")',
' if headers:',
' for pair in headers.split(";"):',
' if ":" in pair:',
' k, v = pair.split(":", 1)',
' req.add_header(k.strip(), v.strip())',
' with urllib.request.urlopen(req, timeout=timeout) as resp:',
' raw = resp.read().decode("utf-8")',
' try:',
' return json.loads(raw)',
' except (json.JSONDecodeError, ValueError):',
' return raw.strip()',
' except Exception as e:',
' return {"error": str(e)}',
"",
'def _sqlite(db_path, query):',
' """Run a SQLite query, return list of dicts."""',
' import sqlite3',
' try:',
' conn = sqlite3.connect(db_path)',
' conn.row_factory = sqlite3.Row',
' cur = conn.execute(query)',
' rows = [dict(r) for r in cur.fetchall()]',
' conn.close()',
' return rows if len(rows) != 1 else rows[0]',
' except Exception as e:',
' return {"error": str(e)}',
"",
f"# ─── µFrame Compile ──────────────────────────────────────────",
"",
uframe_import,

5
backend/uframe/emit_micron.py
View File

@@ -78,8 +78,9 @@ def emit_micron(grid: CharGrid, page_title: str = "") -> str:
if cur_style != _EMPTY_STYLE:
line_parts.append(_emit_style_close(cur_style))
cur_style = _EMPTY_STYLE
# Open new link
line_parts.append("[")
# Open new link — backtick enters formatting mode
# where the parser recognizes `[` as link start
line_parts.append("`[")
in_link = link
# Handle style transitions (not inside links — links handle their own style)

9
backend/uframe/ir.py
View File

@@ -302,6 +302,9 @@ class SourceType(Enum):
PYTHON = auto()
RNS = auto()
PARAM = auto()
HTTP = auto()
SQLITE = auto()
ENV = auto()
@dataclass
@@ -317,6 +320,12 @@ class Source(IRNode):
var_name: str = ""
source_type: SourceType = SourceType.SHELL
command: str = ""
# HTTP-specific
http_method: str = "GET"
http_body: str = ""
http_headers: str = ""
# SQLite-specific: command = db path, query = SQL
query: str = ""
timeout: int = 5

44
backend/uframe/parser.py
View File

@@ -138,7 +138,7 @@ def _parse_list_style(s: str) -> ListStyle:
}.get(s.lower(), ListStyle.BULLET)
def _parse_line(keyword: str, args: list[str], line_num: int) -> IRNode:
def _parse_line(keyword: str, args: list[str], line_num: int, raw_args: str = "") -> IRNode:
"""Parse a single line into an IR node based on the keyword."""
if keyword == "page":
@@ -364,7 +364,7 @@ def _parse_line(keyword: str, args: list[str], line_num: int) -> IRNode:
# Dynamic features
elif keyword == "let":
# let name = "value" or let name = 1,2,3
raw = " ".join(args)
raw = raw_args
eq = raw.find("=")
if eq != -1:
var_name = raw[:eq].strip()
@@ -377,11 +377,10 @@ def _parse_line(keyword: str, args: list[str], line_num: int) -> IRNode:
elif keyword == "source":
# source cpu : shell "grep 'cpu' /proc/stat"
# source name : type "command"
raw = " ".join(args)
colon = raw.find(":")
colon = raw_args.find(":")
if colon != -1:
var_name = raw[:colon].strip()
rest = raw[colon + 1:].strip()
var_name = raw_args[:colon].strip()
rest = raw_args[colon + 1:].strip()
parts = _split_args(rest)
src_type_str = parts[0] if parts else "shell"
command = parts[1] if len(parts) > 1 else ""
@@ -392,17 +391,38 @@ def _parse_line(keyword: str, args: list[str], line_num: int) -> IRNode:
"python": SourceType.PYTHON,
"rns": SourceType.RNS,
"param": SourceType.PARAM,
"http": SourceType.HTTP,
"sqlite": SourceType.SQLITE,
"env": SourceType.ENV,
}.get(src_type_str.lower(), SourceType.SHELL)
# Parse optional timeout
# Parse optional params from remaining parts
timeout = 5
for p in parts[2:]:
if p.startswith("timeout"):
http_method = "GET"
http_body = ""
http_headers = ""
query = ""
extra = parts[2:]
if src_type == SourceType.SQLITE and len(parts) > 2:
# sqlite "/path/db" "SELECT ..."
query = parts[2]
extra = parts[3:]
for p in extra:
if p.startswith("timeout="):
try:
timeout = int(p.split("=")[1]) if "=" in p else int(parts[parts.index(p) + 1])
timeout = int(p.split("=", 1)[1])
except (ValueError, IndexError):
pass
elif p.startswith("method="):
http_method = p.split("=", 1)[1].upper()
elif p.startswith("body="):
http_body = p.split("=", 1)[1]
elif p.startswith("headers="):
http_headers = p.split("=", 1)[1]
return Source(var_name=var_name, source_type=src_type,
command=command, timeout=timeout, source_line=line_num)
command=command, timeout=timeout,
http_method=http_method, http_body=http_body,
http_headers=http_headers, query=query,
source_line=line_num)
else:
return Source(var_name=args[0] if args else "", source_line=line_num)
@@ -817,7 +837,7 @@ def parse(source: str, components: dict[str, ComponentDef] | None = None) -> Pag
args = _split_args(arg_str)
# Parse this line into a node
node = _parse_line(keyword, args, line_num)
node = _parse_line(keyword, args, line_num, raw_args=arg_str)
# Pop stack back to find the parent (parent indent < this indent)
while stack and stack[-1][0] >= indent:

6
backend/uframe/tests/test_dynamic.py
View File

@@ -36,7 +36,7 @@ def test_if_block():
text "Critical"'''
result = uframe.compile(source)
assert result.is_dynamic
assert "if {val} > 90:" in result.script
assert "if val > 90:" in result.script
def test_for_loop():
@@ -46,7 +46,7 @@ def test_for_loop():
text "$item"'''
result = uframe.compile(source)
assert result.is_dynamic
assert "for item in _iter({items}):" in result.script
assert "for item in _iter(items):" in result.script
def test_let_variable():
@@ -120,5 +120,5 @@ def test_codegen_complete_dashboard():
assert "#!/usr/bin/env python3" in script
assert "_cache_seconds = 0" in script
assert "_shell" in script
assert "if {cpu} > 90:" in script
assert "if cpu > 90:" in script
assert "uframe.compile" in script