# -*- coding: utf-8 -*- """ Template-Formatter mit Platzhalter-System und Ausdruecken. Verarbeitungs-Pipeline: 1. {placeholder}-Variablen aufloesen (date, sys, env, ctx, app) 2. (:math)-Ausdruecke auswerten 3. (condition ? true : false) Logik-Ausdruecke auswerten 4. -basierte String-/Zahlenformatierung Unterstuetzte Namespaces fuer {placeholder}: - {date.now.*} Aktuelles Datum/Uhrzeit - {date.tomorrow.*} Morgen - {date.yesterday.*} Gestern - {sys.*} Systeminformationen - {env.*} Umgebungsvariablen - {ctx.*} Werte aus dem uebergebenen Kontext-Dict - {app.*} Werte aus der Server-Applikation (magische Konstanten) Mathematische Ausdruecke: (:10 + 5) → 15 (:2 ^ 10) → 1024 3 * 4 → 12 Logische Ausdruecke: (5 > 3 ? ja : nein) → ja (x == 1 ? eins :: x == 2 ? zwei : andere) String-Operationen: text → TEXT TEXT → text text → text hello → 5 ab → ababab hello → ell hello → 2 alt → neu Zahlenformatierung: 1234567 → 1.234.567 3.7 → 4 3.1 → 3.10 1073741824 → 1 GB """ from __future__ import annotations import locale import os import platform import re import socket from datetime import datetime, timedelta from typing import Any # Wochentage und Monate auf Deutsch _WEEKDAY_NAMES = [ "Montag", "Dienstag", "Mittwoch", "Donnerstag", "Freitag", "Samstag", "Sonntag", ] _MONTH_NAMES = [ "", "Januar", "Februar", "Maerz", "April", "Mai", "Juni", "Juli", "August", "September", "Oktober", "November", "Dezember", ] # Pattern fuer {placeholder} (keine verschachtelten Klammern) _PLACEHOLDER_RE = re.compile(r"\{([^{}]+)\}") # Pattern fuer (:math-expression) _MATH_INLINE_RE = re.compile(r"\(:([^)]+)\)") # Pattern fuer ... und ... _MATH_TAG_RE = re.compile(r"<(calc|math)>(.*?)", re.DOTALL | re.IGNORECASE) # Pattern fuer (condition ? true : false) — muss ? enthalten _LOGIC_RE = re.compile(r"\(([^()]*?\?[^()]*?)\)") # String-Tags _UPPER_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) _LOWER_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) _TRIM_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) _LEN_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) _REPEAT_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) _SUBSTR_RE = re.compile( r"(.*?)", re.DOTALL | re.IGNORECASE ) _STRPOS_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) _REPLACE_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) # Zahlen-Tags _NUMBER_RE = re.compile(r"]*))?>(.*?)", re.DOTALL | re.IGNORECASE) _INT_RE = re.compile(r"]*))?>(.*?)", re.DOTALL | re.IGNORECASE) _DECIMAL_RE = re.compile( r"]*))?>(.*?)", re.DOTALL | re.IGNORECASE ) _SIZE_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) # ===================================================================== # Haupt-API # ===================================================================== def format_template( template: str, context: dict[str, Any] | None = None, now: datetime | None = None, application: Any | None = None, ) -> str: """ Verarbeitet ein Template mit der vollstaendigen Pipeline. Args: template: Text mit Platzhaltern und Ausdruecken. context: Optionaler Kontext fuer {ctx.*}-Variablen. now: Optionaler Zeitpunkt (default: datetime.now()). application: Optionale Referenz zur Applikation fuer {app.*}. Returns: Der formatierte Text. """ if not template: return template dt_now = now or datetime.now() ctx = context or {} text = template # Phase 1: {placeholder}-Variablen aufloesen text = _resolve_placeholders(text, dt_now, ctx, application) # Phase 2+3: Math und Logik iterativ aufloesen (max 32 Durchlaeufe) for _ in range(32): prev = text text = _process_math(text) text = _process_logic(text) if text == prev: break # Phase 4: String-Operationen text = _process_string_ops(text) # Phase 5: Zahlenformatierung text = _process_number_format(text) return text # ===================================================================== # Phase 1: Platzhalter-Auflosung # ===================================================================== def _resolve_placeholders( text: str, now: datetime, ctx: dict[str, Any], app: Any | None ) -> str: """Ersetzt alle {placeholder}-Variablen.""" if "{" not in text: return text def _replace(match: re.Match) -> str: key = match.group(1).strip() try: value = _resolve(key, now, ctx, app) if value is not None: return str(value) except Exception: pass return match.group(0) return _PLACEHOLDER_RE.sub(_replace, text) def _resolve(key: str, now: datetime, ctx: dict[str, Any], app: Any | None) -> Any: """Loest einen einzelnen Platzhalter-Schluessel auf.""" parts = key.split(".") if len(parts) < 2: return None namespace = parts[0].lower() if namespace == "date": return _resolve_date(parts[1:], now) if namespace == "sys": return _resolve_sys(parts[1:]) if namespace == "env": return _resolve_env(parts[1:]) if namespace == "ctx": return _resolve_ctx(parts[1:], ctx) if namespace == "app" and app is not None: return _resolve_app(parts[1:], app) return None # --- Datum/Uhrzeit --- def _resolve_date(parts: list[str], now: datetime) -> Any: """Loest date.*-Platzhalter auf.""" if not parts: return None base_key = parts[0].lower() if base_key == "now": dt = now elif base_key == "tomorrow": dt = now + timedelta(days=1) elif base_key == "yesterday": dt = now - timedelta(days=1) else: return None if len(parts) < 2: return dt.strftime("%Y-%m-%d %H:%M:%S") field = parts[1].lower() return _date_field(dt, field) def _date_field(dt: datetime, field: str) -> Any: """Gibt ein einzelnes Datumsfeld zurueck.""" mapping = { "year": dt.year, "month": dt.month, "month_name": _MONTH_NAMES[dt.month], "day": dt.day, "hour": dt.hour, "minute": dt.minute, "second": dt.second, "weekday": _WEEKDAY_NAMES[dt.weekday()], "weekday_num": dt.isoweekday(), "date": dt.strftime("%Y-%m-%d"), "time": dt.strftime("%H:%M:%S"), "iso": dt.isoformat(), "timestamp": int(dt.timestamp()), "week": dt.isocalendar()[1], } return mapping.get(field) # --- System --- def _resolve_sys(parts: list[str]) -> Any: """Loest sys.*-Platzhalter auf.""" if not parts: return None key = ".".join(parts).lower() mapping = { "hostname": socket.gethostname, "os.name": lambda: platform.system(), "os.version": lambda: platform.version(), "os.release": lambda: platform.release(), "os.platform": lambda: platform.platform(), "os.arch": lambda: platform.machine(), "python.version": lambda: platform.python_version(), "user": lambda: os.getenv("USER") or os.getenv("USERNAME") or "unknown", "pid": lambda: os.getpid(), "cpu_count": lambda: os.cpu_count(), } resolver = mapping.get(key) if resolver: return resolver() return None # --- Umgebungsvariablen --- def _resolve_env(parts: list[str]) -> Any: """Loest env.*-Platzhalter auf.""" if not parts: return None var_name = ".".join(parts) return os.getenv(var_name, "") # --- Kontext --- def _resolve_ctx(parts: list[str], ctx: dict[str, Any]) -> Any: """Loest ctx.*-Platzhalter auf (verschachtelte Dict-Keys).""" if not parts or not ctx: return None current: Any = ctx for part in parts: if isinstance(current, dict): current = current.get(part) else: return None if current is None: return None return current # --- Applikation (Magische Klassenkonstanten) --- def _resolve_app(parts: list[str], app: Any) -> Any: """ Loest app.*-Platzhalter gegen die Applikation auf. Navigiert ueber Properties und Attribute des Applikationsobjekts. Beispiele: {app.satellites.count} → Anzahl registrierter Satellites {app.satellites.connected} → Anzahl verbundener Satellites {app.scheduler.job_count} → Anzahl Scheduler-Jobs {app.scheduler.running} → Ob Scheduler laeuft {app.plugins.count} → Anzahl geladener Plugins {app.config.network.command_port} → Config-Wert {app.uptime} → Laufzeit in Sekunden """ if not parts: return None current: Any = app for part in parts: if current is None: return None # Spezielle Zugriffe fuer bekannte Trixy-Typen result = _resolve_app_special(current, part) if result is not _UNRESOLVED: current = result continue # Generischer Attribut-Zugriff if hasattr(current, part): current = getattr(current, part) elif hasattr(current, f"_{part}"): current = getattr(current, f"_{part}") elif isinstance(current, dict): current = current.get(part) else: return None # Callables nicht zurueckgeben (Properties sind ok, Methoden nicht) if callable(current) and not isinstance(current, (int, float, str, bool, list, dict)): try: current = current() except Exception: return None return current # Sentinel fuer nicht-aufgeloeste Spezialzugriffe _UNRESOLVED = object() def _resolve_app_special(obj: Any, key: str) -> Any: """ Behandelt spezielle Zugriffe auf bekannte Trixy-Objekte. Gibt _UNRESOLVED zurueck wenn nicht anwendbar. """ cls_name = type(obj).__name__ # SatelliteManager if cls_name == "SatelliteManager": if key == "count": return len(obj) if hasattr(obj, "__len__") else 0 if key == "connected": return len(obj.get_connected()) if hasattr(obj, "get_connected") else 0 if key == "disconnected": return len(obj.get_disconnected()) if hasattr(obj, "get_disconnected") else 0 # Zugriff auf einzelnen Satellite per ID if hasattr(obj, "get"): sat = obj.get(key) if sat is not None: return sat # Satellite if cls_name == "Satellite": if key == "connected": return obj.is_connected if hasattr(obj, "is_connected") else False if key == "room": return obj.room_id if hasattr(obj, "room_id") else "" if key == "alias": return obj.alias if hasattr(obj, "alias") else "" # SchedulerService / Scheduler if cls_name in ("SchedulerService", "Scheduler"): if key == "running": return obj.is_running if hasattr(obj, "is_running") else False # ServiceContainer if cls_name == "ServiceContainer": if key == "count": return len(obj.services) if hasattr(obj, "services") else 0 # Zugriff auf einzelnen Service if hasattr(obj, "get_service"): svc = obj.get_service(key) if svc is not None: return svc # PluginManager if cls_name == "PluginManager": if key == "count": return obj.plugin_count if hasattr(obj, "plugin_count") else 0 if key == "loaded": return obj.loaded_plugins if hasattr(obj, "loaded_plugins") else [] # ConfigManager — config-Werte auslesen if cls_name == "ConfigManager": return _UNRESOLVED # Standard-Attributzugriff nutzen return _UNRESOLVED # ===================================================================== # Phase 2: Mathematische Ausdruecke # ===================================================================== def _process_math(text: str) -> str: """Wertet alle (:expr) und / Ausdruecke aus.""" from trixy_core.utils.math_eval import eval_math # (:expression) def _inline_replace(match: re.Match) -> str: result = eval_math(match.group(1)) return result if result is not None else match.group(0) text = _MATH_INLINE_RE.sub(_inline_replace, text) # expression / expression def _tag_replace(match: re.Match) -> str: result = eval_math(match.group(2)) return result if result is not None else match.group(0) text = _MATH_TAG_RE.sub(_tag_replace, text) return text # ===================================================================== # Phase 3: Logische Ausdruecke # ===================================================================== def _process_logic(text: str) -> str: """Wertet alle (cond ? true : false) Ausdruecke aus.""" from trixy_core.utils.logic_eval import eval_logic def _replace(match: re.Match) -> str: result = eval_logic(match.group(1)) return result if result is not None else match.group(0) return _LOGIC_RE.sub(_replace, text) # ===================================================================== # Phase 4: String-Operationen # ===================================================================== def _process_string_ops(text: str) -> str: """Verarbeitet alle String-Tags.""" # Reihenfolge wie in PrintText: substr/len/strpos zuerst, dann extra # text def _substr(match: re.Match) -> str: start = int(match.group(1)) length = int(match.group(2)) if match.group(2) else None content = match.group(3) if length is not None: return content[start:start + length] return content[start:] text = _SUBSTR_RE.sub(_substr, text) # text text = _LEN_RE.sub(lambda m: str(len(m.group(1))), text) # text def _strpos(match: re.Match) -> str: needle = match.group(1) haystack = match.group(2) return str(haystack.find(needle)) text = _STRPOS_RE.sub(_strpos, text) # text text = _UPPER_RE.sub(lambda m: m.group(1).upper(), text) # text text = _LOWER_RE.sub(lambda m: m.group(1).lower(), text) # text text = _TRIM_RE.sub(lambda m: m.group(1).strip(), text) # text def _repeat(match: re.Match) -> str: n = min(int(match.group(1)), 1000) return match.group(2) * n text = _REPEAT_RE.sub(_repeat, text) # text def _replace_tag(match: re.Match) -> str: params = match.group(1) content = match.group(2) # Komma splitten, aber \, escapen parts = re.split(r"(? str: """Verarbeitet Zahlen-Tags.""" # value def _fmt_number(match: re.Match) -> str: fmt = match.group(1) or "" raw = match.group(2).strip() try: num = float(raw) except ValueError: return raw if not fmt: # Standard: Tausender-Trennung mit Punkt if num == int(num): return f"{int(num):,}".replace(",", ".") return f"{num:,.2f}".replace(",", "X").replace(".", ",").replace("X", ".") # N0-N9: Anzahl Dezimalstellen mit Tausender-Trennung m = re.match(r"[Nn](\d)", fmt) if m: decimals = int(m.group(1)) formatted = f"{num:,.{decimals}f}" return formatted.replace(",", "X").replace(".", ",").replace("X", ".") return raw text = _NUMBER_RE.sub(_fmt_number, text) # value def _fmt_int(match: re.Match) -> str: raw = match.group(2).strip() try: return str(round(float(raw))) except ValueError: return raw text = _INT_RE.sub(_fmt_int, text) # value def _fmt_decimal(match: re.Match) -> str: fmt = match.group(1) or "2" raw = match.group(2).strip() try: num = float(raw) decimals = int(fmt) if fmt.isdigit() else 2 return f"{num:.{decimals}f}" except ValueError: return raw text = _DECIMAL_RE.sub(_fmt_decimal, text) # bytes → menschenlesbare Groesse def _fmt_size(match: re.Match) -> str: raw = match.group(1).strip() try: size = float(raw) except ValueError: return raw return _format_byte_size(size) text = _SIZE_RE.sub(_fmt_size, text) return text def _format_byte_size(size: float) -> str: """Formatiert Bytes als menschenlesbare Groesse.""" units = ["Bytes", "KB", "MB", "GB", "TB", "PB", "EB"] unit_index = 0 value = abs(size) while value >= 1000 and unit_index < len(units) - 1: value /= 1024 unit_index += 1 sign = "-" if size < 0 else "" if unit_index == 0: return f"{sign}{int(value)} {units[0]}" if value < 10: return f"{sign}{value:.2f} {units[unit_index]}" if value < 100: return f"{sign}{value:.1f} {units[unit_index]}" return f"{sign}{value:.0f} {units[unit_index]}"