From 49cb8133ed277d86a7171e3b1fd2f5b63b7f2219 Mon Sep 17 00:00:00 2001 From: Vaso73 Date: Tue, 15 Sep 2026 14:40:50 +0200 Subject: [PATCH 01/21] test: cover backup notification integrity on v1.2.6 --- .../scripts/tests/test_vzdump_ai_integrity.py | 335 ++++++++++++++++++ 1 file changed, 335 insertions(+) create mode 100644 AppImage/scripts/tests/test_vzdump_ai_integrity.py diff --git a/AppImage/scripts/tests/test_vzdump_ai_integrity.py b/AppImage/scripts/tests/test_vzdump_ai_integrity.py new file mode 100644 index 00000000..faf79382 --- /dev/null +++ b/AppImage/scripts/tests/test_vzdump_ai_integrity.py @@ -0,0 +1,335 @@ +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + + +SCRIPTS_DIR = Path(__file__).resolve().parents[1] +if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) + +import notification_manager +import notification_templates +from notification_channels import EmailChannel, TelegramChannel +from notification_manager import NotificationManager +from notification_templates import render_template + + +AI_CONFIG = { + "ai_enabled": "true", + "ai_provider": "ollama", + "ai_ollama_url": "http://localhost:11434", + "ai_model": "test-model", + "ai_language": "en", +} + + +def _vzdump_report(count=49, failed=False): + header = "{:<8}{:<22}{:<10}{:<10}{:<14}{}".format( + "VMID", "Name", "Status", "Time", "Size", "Filename" + ) + rows = [] + for index in range(count): + vmid = 100 + index + status = "ERROR" if failed and index == count - 1 else "OK" + rows.append( + "{:<8}{:<22}{:<10}{:<10}{:<14}{}".format( + str(vmid), + f"guest-{vmid}", + status, + f"00:00:{index + 10:02d}", + f"{index + 1}.25 GiB", + f"/mnt/pve/archive/dump/vzdump-lxc-{vmid}-2026_09_15-01_00_00.tar.zst", + ) + ) + return ( + "Proxmox vzdump report\n\n" + + header + + "\n" + + "\n".join(rows) + + "\nTotal running time: 00:49:00\nTotal size: 1225.25 GiB\n" + ) + + +def _render(event_type, failed=False): + return render_template( + event_type, + { + "hostname": "pve-a", + "storage": "archive", + "vmname": "49 guests", + "vmid": "batch", + "size": "1225.25 GiB", + "reason": "last guest failed" if failed else "", + "pve_title": "pve-a: vzdump backup status", + "pve_message": _vzdump_report(failed=failed), + }, + ) + + +class _CapturingChannel: + def __init__(self): + self.calls = [] + + def send(self, title, body, severity, data): + self.calls.append((title, body, severity, data)) + return {"success": True, "error": None} + + +class _ReplacingEnhancer: + calls = 0 + + def __init__(self, config): + self.config = config + + def enhance(self, *args, **kwargs): + type(self).calls += 1 + return {"title": "AI shortened title", "body": "AI kept only one guest."} + + +class _FailingEnhancer: + def __init__(self, config): + self.config = config + + def enhance(self, *args, **kwargs): + return None + + +class VzdumpAIIntegrityTests(unittest.TestCase): + def setUp(self): + with notification_templates._AI_CACHE_LOCK: + notification_templates._AI_CACHE.clear() + _ReplacingEnhancer.calls = 0 + + def _dispatch(self, event_type, rendered, enhancer): + manager = NotificationManager() + channel = _CapturingChannel() + manager._channels = {"telegram": channel} + with ( + patch.object(manager, "_build_ai_config", return_value=dict(AI_CONFIG)), + patch.object(manager, "_record_history"), + patch.object(notification_manager, "enrich_context_for_ai", return_value=""), + patch.object(notification_templates, "AIEnhancer", enhancer), + ): + delivered = manager._dispatch_to_channels( + rendered["title"], rendered["body"], + "CRITICAL" if event_type == "backup_fail" else "INFO", + event_type, {}, "test", + ) + self.assertTrue(delivered) + self.assertEqual(len(channel.calls), 1) + return channel.calls[0] + + def _send_public(self, event_type, title, body): + manager = NotificationManager() + channel = _CapturingChannel() + manager._channels = {"telegram": channel} + manager._config = {"telegram.rich_format": "false"} + with ( + patch.object(manager, "_build_ai_config", return_value=dict(AI_CONFIG)), + patch.object(manager, "_record_history"), + patch.object(notification_templates, "AIEnhancer", _ReplacingEnhancer), + ): + result = manager.send_notification( + event_type, + "CRITICAL" if event_type == "backup_fail" else "INFO", + title, + body, + source="test", + skip_toggle_check=True, + ) + self.assertTrue(result["success"]) + self.assertEqual(len(channel.calls), 1) + return channel.calls[0] + + def test_public_send_backup_complete_skips_ai_and_keeps_all_49_items(self): + rendered = _render("backup_complete") + + title, body, _, _ = self._send_public( + "backup_complete", rendered["title"], rendered["body"] + ) + + self.assertEqual(title, rendered["title"]) + self.assertEqual(body, rendered["body"]) + self.assertEqual(_ReplacingEnhancer.calls, 0) + self.assertEqual(body.count("✅ CT guest-"), 49) + self.assertIn("✅ CT guest-100 (100)", body) + self.assertIn("✅ CT guest-148 (148)", body) + + def test_public_send_backup_fail_skips_ai_and_keeps_all_49_items(self): + rendered = _render("backup_fail", failed=True) + + title, body, _, _ = self._send_public( + "backup_fail", rendered["title"], rendered["body"] + ) + + self.assertEqual(title, rendered["title"]) + self.assertEqual(body, rendered["body"]) + self.assertEqual(_ReplacingEnhancer.calls, 0) + self.assertEqual(body.count("✅ CT guest-") + body.count("❌ CT guest-"), 49) + self.assertIn("✅ CT guest-100 (100)", body) + self.assertIn("❌ CT guest-148 (148)", body) + + def test_public_send_non_backup_event_still_uses_ai(self): + title, body, _, _ = self._send_public( + "cpu_high", "pve-a: CPU high", "CPU reached 95%." + ) + + self.assertEqual(title, "AI shortened title") + self.assertEqual(body, "AI kept only one guest.") + self.assertEqual(_ReplacingEnhancer.calls, 1) + + def test_backup_complete_skips_ai_and_keeps_full_49_item_inventory(self): + rendered = _render("backup_complete") + + title, body, _, _ = self._dispatch( + "backup_complete", rendered, _ReplacingEnhancer + ) + + self.assertEqual(title, rendered["title"]) + self.assertEqual(body, rendered["body"]) + self.assertEqual(_ReplacingEnhancer.calls, 0) + self.assertIn("✅ CT guest-100 (100)", body) + self.assertIn("📏 Size: 1.25 GiB | ⏱️ Duration: 00:00:10", body) + self.assertIn("✅ CT guest-148 (148)", body) + self.assertIn("📏 Size: 49.25 GiB | ⏱️ Duration: 00:00:58", body) + self.assertIn("📊 49 backups", body) + self.assertEqual(body.count("✅ CT guest-"), 49) + + def test_backup_fail_skips_ai_and_keeps_inventory_and_failure_details(self): + rendered = _render("backup_fail", failed=True) + + title, body, _, _ = self._dispatch( + "backup_fail", rendered, _ReplacingEnhancer + ) + + self.assertEqual(title, rendered["title"]) + self.assertEqual(body, rendered["body"]) + self.assertEqual(_ReplacingEnhancer.calls, 0) + self.assertIn("✅ CT guest-100 (100)", body) + self.assertIn("❌ CT guest-148 (148)", body) + self.assertIn("📏 Size: 49.25 GiB | ⏱️ Duration: 00:00:58", body) + self.assertIn("📊 49 backups | ❌ 1 failed", body) + self.assertEqual(body.count("✅ CT guest-"), 48) + + def test_backup_ai_failure_sends_original_title_and_body_unchanged(self): + rendered = _render("backup_complete") + + title, body, _, _ = self._dispatch( + "backup_complete", rendered, _FailingEnhancer + ) + + self.assertEqual(title, rendered["title"]) + self.assertEqual(body, rendered["body"]) + + def test_non_backup_event_retains_existing_ai_rewrite(self): + rendered = {"title": "pve-a: CPU high", "body": "CPU reached 95%."} + + title, body, _, _ = self._dispatch( + "cpu_high", rendered, _ReplacingEnhancer + ) + + self.assertEqual(title, "AI shortened title") + self.assertEqual(body, "AI kept only one guest.") + + def test_backup_complete_email_html_contains_each_inventory_edge_and_summary_once(self): + rendered = _render("backup_complete") + channel = EmailChannel({}) + data = { + "_event_type": "backup_complete", + "_group": "backup", + "hostname": "pve-a", + "storage": "archive", + "vmname": "49 guests", + "vmid": "batch", + "size": "1225.25 GiB", + } + + html = channel._format_html( + "[ProxMenux] [INFO] " + rendered["title"], + rendered["body"], "INFO", data, + ) + + self.assertEqual(html.count("guest-100 (100)"), 1) + self.assertEqual(html.count("guest-148 (148)"), 1) + self.assertEqual(html.count("49 backups"), 1) + + def test_backup_fail_email_html_keeps_inventory_and_localized_status_once(self): + rendered = _render("backup_fail", failed=True) + channel = EmailChannel({}) + data = { + "_event_type": "backup_fail", + "_group": "backup", + "_notification_language": "sk", + "hostname": "pve-a", + "storage": "archive", + "vmname": "49 guests", + "vmid": "batch", + "status": "failed", + "size": "1225.25 GiB", + "reason": "last guest failed", + } + + html = channel._format_html( + "[ProxMenux] [CRITICAL] " + rendered["title"], + rendered["body"], "CRITICAL", data, + ) + + self.assertEqual(html.count("guest-100 (100)"), 1) + self.assertEqual(html.count("guest-148 (148)"), 1) + self.assertEqual(html.count("49 backups"), 1) + self.assertEqual(html.count("1 failed"), 1) + self.assertEqual(html.count(">Zlyhalo<"), 1) + self.assertNotIn(">Failed<", html) + + def test_telegram_chunks_preserve_complete_49_item_message(self): + rendered = _render("backup_complete") + body = rendered["body"] + channel = TelegramChannel("123:token", "456") + html_message = ( + f"🔵 {channel._escape_html(rendered['title'])}\n\n" + f"{channel._escape_html(body)}" + ) + + chunks = channel._split_message(html_message) + + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(chunk) <= 4096 for chunk in chunks)) + joined = "".join(chunks) + self.assertIn("guest-100 (100)", joined) + self.assertIn("guest-148 (148)", joined) + self.assertIn("49.25 GiB", joined) + + def test_telegram_chunks_do_not_split_entities_or_open_tags(self): + from html.parser import HTMLParser + + class _BalancedParser(HTMLParser): + def __init__(self): + super().__init__(convert_charrefs=False) + self.stack = [] + + def handle_starttag(self, tag, attrs): + self.stack.append(tag) + + def handle_endtag(self, tag): + if not self.stack or self.stack.pop() != tag: + raise AssertionError(f"unbalanced closing tag: {tag}") + + channel = TelegramChannel("123:token", "456") + html_message = "" + ("A & B " * 900) + "" + + chunks = channel._split_message(html_message) + + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(chunk) <= 4096 for chunk in chunks)) + for chunk in chunks: + parser = _BalancedParser() + parser.feed(chunk) + parser.close() + self.assertEqual(parser.stack, []) + self.assertNotRegex(chunk, r"&(?:amp)?$") + self.assertNotRegex(chunk, r"^amp;") + + +if __name__ == "__main__": + unittest.main() From c83c68b4ee005648b50e69ac023ca0811b5039ef Mon Sep 17 00:00:00 2001 From: Vaso73 Date: Tue, 15 Sep 2026 08:29:12 +0200 Subject: [PATCH 02/21] Preserve vzdump inventory when AI is enabled --- AppImage/scripts/notification_channels.py | 61 +++++++++++++++++++---- AppImage/scripts/notification_manager.py | 45 +++++++++-------- 2 files changed, 77 insertions(+), 29 deletions(-) diff --git a/AppImage/scripts/notification_channels.py b/AppImage/scripts/notification_channels.py index 932879fd..adee2743 100644 --- a/AppImage/scripts/notification_channels.py +++ b/AppImage/scripts/notification_channels.py @@ -308,18 +308,51 @@ class TelegramChannel(NotificationChannel): return self._http_request(url, payload, {'Content-Type': 'application/json'}) def _split_message(self, text: str) -> list: + """Split Telegram HTML without cutting entities or formatting tags. + + Open formatting tags are closed at the end of a chunk and reopened in + the next one, so every API request is valid HTML on its own. + """ if len(text) <= self.MAX_LENGTH: return [text] + + token_re = re.compile( + r'&(?:#[0-9]+|#x[0-9A-Fa-f]+|[A-Za-z][A-Za-z0-9]+);|<[^<>]+>|.', + re.DOTALL, + ) + tag_re = re.compile(r'<\s*(/?)\s*([A-Za-z0-9-]+)(?:\s[^<>]*)?>') + void_tags = {'br'} + + def _advance(stack, token): + match = tag_re.fullmatch(token) + if not match: + return list(stack) + closing, name = match.groups() + name = name.lower() + next_stack = list(stack) + if closing: + if next_stack and next_stack[-1][0] == name: + next_stack.pop() + elif not token.rstrip().endswith('/>') and name not in void_tags: + next_stack.append((name, token)) + return next_stack + + def _closers(stack): + return ''.join(f'' for name, _ in reversed(stack)) + chunks = [] - while text: - if len(text) <= self.MAX_LENGTH: - chunks.append(text) - break - split_at = text.rfind('\n', 0, self.MAX_LENGTH) - if split_at == -1: - split_at = self.MAX_LENGTH - chunks.append(text[:split_at]) - text = text[split_at:].lstrip('\n') + current = '' + open_tags = [] + for token in token_re.findall(text): + next_tags = _advance(open_tags, token) + if current and len(current) + len(token) + len(_closers(next_tags)) > self.MAX_LENGTH: + chunks.append(current + _closers(open_tags)) + current = ''.join(opener for _, opener in open_tags) + current += token + open_tags = _advance(open_tags, token) + + if current: + chunks.append(current + _closers(open_tags)) return chunks @staticmethod @@ -949,6 +982,16 @@ class EmailChannel(NotificationChannel): # ── Build structured detail rows from known data fields ── detail_rows = self._build_detail_rows(data, event_type, group, html_mod) + # Vzdump bodies are authoritative multi-item inventories. Structured + # backup metadata is only a summary and previously replaced the body in + # the HTML alternative, hiding all guest rows. Render every body line + # exactly once for these events instead of mixing both representations. + if event_type in {'backup_complete', 'backup_fail'}: + detail_rows = [ + ('', html_mod.escape(line.strip())) + for line in body.split('\n') if line.strip() + ] + # ── Fallback: if no structured rows, render body text lines ── if not detail_rows: for line in body.split('\n'): diff --git a/AppImage/scripts/notification_manager.py b/AppImage/scripts/notification_manager.py index 0a14ea05..c7d56338 100644 --- a/AppImage/scripts/notification_manager.py +++ b/AppImage/scripts/notification_manager.py @@ -1376,27 +1376,32 @@ class NotificationManager: # raw template-formatted notification. Audit Tier 6 — # `_dispatch_to_channels`: AI failure dropped the notification. try: - enriched_context = enrich_context_for_ai( - title=ch_title, - body=ch_body, - event_type=event_type, - data=data, - journal_context=raw_journal_context, - detail_level=detail_level - ) + # Backup reports are authoritative inventories. A model can + # neither be trusted to avoid repeating all guest rows nor to + # preserve every value, so these two events bypass AI entirely. + ai_result = None + if event_type not in {'backup_complete', 'backup_fail'}: + enriched_context = enrich_context_for_ai( + title=ch_title, + body=ch_body, + event_type=event_type, + data=data, + journal_context=raw_journal_context, + detail_level=detail_level + ) - # Wrap the AI rewrite with a hard timeout so a slow Ollama - # call (90-120 s on slow CPUs) doesn't stall the dispatch - # thread and delay every other queued event. On timeout we - # ship the non-AI title/body — the user still gets the - # notification, just without LLM polish. Audit Tier 3.2 #2. - ai_result = _format_with_ai_bounded( - format_with_ai_full, - ch_title, ch_body, severity, channel_ai_config, - detail_level=detail_level, - journal_context=enriched_context, - use_emojis=use_rich_format, - ) + # Wrap the AI rewrite with a hard timeout so a slow Ollama + # call (90-120 s on slow CPUs) doesn't stall the dispatch + # thread and delay every other queued event. On timeout we + # ship the non-AI title/body — the user still gets the + # notification, just without LLM polish. Audit Tier 3.2 #2. + ai_result = _format_with_ai_bounded( + format_with_ai_full, + ch_title, ch_body, severity, channel_ai_config, + detail_level=detail_level, + journal_context=enriched_context, + use_emojis=use_rich_format, + ) if ai_result is not None: ch_title = ai_result.get('title', ch_title) ch_body = ai_result.get('body', ch_body) From 7587b2ad6a3015aa7fa23f8b2b588b5ddaa46a2d Mon Sep 17 00:00:00 2001 From: Vaso73 Date: Tue, 15 Sep 2026 09:55:20 +0200 Subject: [PATCH 03/21] Preserve localized backup status with vzdump inventory --- AppImage/scripts/notification_channels.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/AppImage/scripts/notification_channels.py b/AppImage/scripts/notification_channels.py index adee2743..f120d37b 100644 --- a/AppImage/scripts/notification_channels.py +++ b/AppImage/scripts/notification_channels.py @@ -982,15 +982,18 @@ class EmailChannel(NotificationChannel): # ── Build structured detail rows from known data fields ── detail_rows = self._build_detail_rows(data, event_type, group, html_mod) - # Vzdump bodies are authoritative multi-item inventories. Structured - # backup metadata is only a summary and previously replaced the body in - # the HTML alternative, hiding all guest rows. Render every body line - # exactly once for these events instead of mixing both representations. + # Vzdump bodies are authoritative multi-item inventories. Keep their + # lines exactly once, but retain the localized structured status row; + # the remaining structured backup metadata only duplicates the report. if event_type in {'backup_complete', 'backup_fail'}: - detail_rows = [ + status_label = html_mod.escape( + _runtime_text('email.fields.status', data) + ) + detail_rows = [row for row in detail_rows if row[0] == status_label] + detail_rows.extend( ('', html_mod.escape(line.strip())) for line in body.split('\n') if line.strip() - ] + ) # ── Fallback: if no structured rows, render body text lines ── if not detail_rows: From ca163cd2bb274a50f91e0265c39decbc90635989 Mon Sep 17 00:00:00 2001 From: Vaso73 Date: Tue, 15 Sep 2026 14:41:41 +0200 Subject: [PATCH 04/21] test: cover runtime notification localization on v1.2.6 --- .../tests/test_notification_runtime_i18n.py | 433 ++++++++++++++++++ 1 file changed, 433 insertions(+) create mode 100644 AppImage/scripts/tests/test_notification_runtime_i18n.py diff --git a/AppImage/scripts/tests/test_notification_runtime_i18n.py b/AppImage/scripts/tests/test_notification_runtime_i18n.py new file mode 100644 index 00000000..db3e4d5a --- /dev/null +++ b/AppImage/scripts/tests/test_notification_runtime_i18n.py @@ -0,0 +1,433 @@ +import json +import re +import sqlite3 +import string +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +SCRIPTS_DIR = Path(__file__).resolve().parents[1] +APPIMAGE_DIR = SCRIPTS_DIR.parent +if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) + +import notification_manager +import notification_templates +import notification_channels + + +def _placeholders(value): + return { + field_name + for _literal, field_name, _format_spec, _conversion + in string.Formatter().parse(value) + if field_name + } + + +class RuntimeCatalogTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.catalogs = {} + for language in ("en", "sk"): + path = APPIMAGE_DIR / "messages" / language / "common.json" + cls.catalogs[language] = json.loads(path.read_text(encoding="utf-8"))["runtime"]["notifications"] + + def test_runtime_catalog_covers_every_template_dynamically(self): + expected = set(notification_templates.TEMPLATES) + for language, catalog in self.catalogs.items(): + templates = catalog["templates"] + self.assertEqual(set(templates), expected, language) + for event_type, source in notification_templates.TEMPLATES.items(): + self.assertEqual(set(templates[event_type]), {"title", "body", "label"}) + for field in ("title", "body", "label"): + self.assertIsInstance(templates[event_type][field], str) + self.assertTrue(templates[event_type][field]) + if field in source: + self.assertEqual( + _placeholders(templates[event_type][field]), + _placeholders(source[field]), + f"{language}:{event_type}:{field}", + ) + + def test_runtime_catalog_keys_and_placeholders_match(self): + def flatten(value, prefix=""): + result = {} + for key, child in value.items(): + dotted = f"{prefix}.{key}" if prefix else key + if isinstance(child, dict): + result.update(flatten(child, dotted)) + else: + result[dotted] = child + return result + + en = flatten(self.catalogs["en"]) + sk = flatten(self.catalogs["sk"]) + self.assertEqual(set(sk), set(en)) + for key in en: + self.assertEqual(_placeholders(sk[key]), _placeholders(en[key]), key) + + def test_slovak_catalog_preserves_placeholders_and_translates_static_text(self): + en = self.catalogs["en"]["templates"] + sk = self.catalogs["sk"]["templates"] + for event_type, source in notification_templates.TEMPLATES.items(): + for field in ("title", "body", "label"): + static_text = source.get(field, "") + for placeholder in _placeholders(static_text): + static_text = static_text.replace("{" + placeholder + "}", "") + if any(ch.isalpha() for ch in static_text): + self.assertNotEqual(sk[event_type][field], en[event_type][field], f"{event_type}:{field}") + + def test_every_template_renders_in_slovak_and_preserves_dynamic_values(self): + values = { + name: f"DYNAMIC_{name.upper()}" + for template in notification_templates.TEMPLATES.values() + for field in ("title", "body") + for name in _placeholders(template.get(field, "")) + } + values.update({"hostname": "HOST-ŽILINA", "severity": "WARNING"}) + with mock.patch.object(notification_templates, "_get_hostname", return_value="HOST-ŽILINA"): + for event_type in notification_templates.TEMPLATES: + rendered = notification_templates.render_template(event_type, values, language="sk") + combined = rendered["title"] + "\n" + rendered["body"] + if notification_templates.TEMPLATES[event_type].get("formatter"): + continue + for name in _placeholders( + notification_templates.TEMPLATES[event_type]["title"] + + notification_templates.TEMPLATES[event_type]["body"] + ): + if name not in {"entity_suffix", "title_or_default"}: + self.assertIn(str(values[name]), combined, f"{event_type}:{name}") + + def test_missing_slovak_key_falls_back_to_english(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "en").mkdir() + (root / "sk").mkdir() + (root / "en" / "common.json").write_text( + json.dumps({"runtime": {"notifications": {"fallback": {"unknownTitle": "{hostname}: {event_type}"}}}}), + encoding="utf-8", + ) + (root / "sk" / "common.json").write_text( + json.dumps({"runtime": {"notifications": {}}}), encoding="utf-8" + ) + with mock.patch.object(notification_templates, "RUNTIME_CATALOG_DIR", root): + notification_templates._load_runtime_catalog.cache_clear() + self.assertEqual( + notification_templates.runtime_message( + "fallback.unknownTitle", "sk", hostname="pve01", event_type="vendor_event" + ), + "pve01: vendor_event", + ) + notification_templates._load_runtime_catalog.cache_clear() + + def test_special_formatters_digest_and_test_message_are_slovak(self): + startup = notification_templates.render_template( + "system_startup", + {"hostname": "pve01", "has_issues": False, "vms_started": [{"name": "alpha", "vmid": 100}]}, + language="sk", + ) + self.assertIn("Spustenie systému", startup["title"]) + self.assertIn("Všetky systémy sú funkčné", startup["body"]) + self.assertIn("alpha", startup["body"]) + + app = notification_templates.render_template( + "app_update_available", + {"hostname": "pve01", "app_name": "Redis", "vmid": 115, "ct_name": "cache", "installed": "7.0", "latest": "8.1"}, + language="sk", + ) + self.assertIn("dostupná aktualizácia", app["title"]) + self.assertIn("Redis", app["body"]) + self.assertIn("7.0 → 8.1", app["body"]) + + backup = notification_templates.render_template( + "backup_complete", + { + "hostname": "pve01", "storage": "pbs-main", "vmname": "alpha", "vmid": "100", + "pve_title": "Backup job finished", + "pve_message": ( + "INFO: Starting Backup of VM 100 (qemu)\n" + "INFO: VM Name: alpha\n" + "INFO: transferred 1.5 GiB in 10 seconds\n" + "INFO: Finished Backup of VM 100 (00:00:10)" + ), + }, + language="sk", + ) + self.assertIn("Záloha dokončená", backup["title"]) + self.assertNotIn("Backup job finished", backup["title"]) + self.assertIn("Veľkosť: 1.5 GiB", backup["body"]) + self.assertIn("Trvanie: 00:00:10", backup["body"]) + + manager = notification_manager.NotificationManager() + manager._config = {"notification_language": "sk"} + rows = [(1, "cpu_high", "resources", 0, "pve01: Vysoké využitie CPU", "body")] + digest = manager._compose_digest_body(rows) + self.assertIn("udalostí INFO zoskupených podľa kategórie", digest) + self.assertIn("Zdroje", digest) + title, body, caption = manager._build_test_message(False, False, "groq / sk") + self.assertEqual(title, "Test ProxMenux") + self.assertIn("Vitajte", body) + self.assertIn("profilovú fotografiu", caption) + + def test_ai_disabled_telegram_receives_deterministic_slovak(self): + class RecordingTelegram: + def __init__(self): + self.payload = None + + def send(self, title, body, severity, data=None): + self.payload = (title, body, severity, data) + return {"success": True, "error": ""} + + channel = RecordingTelegram() + manager = notification_manager.NotificationManager() + manager._channels = {"telegram": channel} + manager._config = { + "notification_language": "sk", + "ai_enabled": "false", + "telegram.rich_format": "false", + } + with mock.patch.object(manager, "_record_history"): + result = manager.send_notification( + "vm_start", "INFO", "", "", + data={"hostname": "pve01", "vmname": "účtovníctvo", "vmid": "123"}, + skip_toggle_check=True, + ) + self.assertTrue(result["success"]) + self.assertIn("spustený", channel.payload[0]) + self.assertIn("účtovníctvo", channel.payload[1]) + self.assertNotIn("is now running", channel.payload[1]) + + def test_email_channel_chrome_uses_the_runtime_catalog(self): + channel = object.__new__(notification_channels.EmailChannel) + channel.subject_prefix = "[ProxMenux]" + html = channel._format_html( + "[ProxMenux] pve01: Vysoké využitie CPU", + "Využitie CPU dosiahlo 95 %.", + "WARNING", + { + "_notification_language": "sk", "_event_type": "cpu_high", + "_group": "resources", "hostname": "pve01", "value": "95", + "threshold": "90", "cores": "8", + }, + ) + self.assertIn("Systémové zdroje", html) + self.assertIn("UPOZORNENIE", html) + self.assertIn("Hostiteľ:", html) + self.assertIn("Aktuálna hodnota", html) + self.assertNotIn(">Details<", html) + self.assertNotIn("System Resources Report", html) + + vm_html = channel._format_html( + "pve01: VM účtovníctvo (123) spustený", + "Virtuálny stroj účtovníctvo (ID: 123) je spustený.", + "INFO", + { + "_notification_language": "sk", "_event_type": "vm_start", + "_group": "vm_ct", "hostname": "pve01", "vmid": "123", + "vmname": "účtovníctvo", + }, + ) + self.assertIn("VM bola spustená", vm_html) + + def test_notification_language_precedence_and_roundtrip(self): + manager = notification_manager.NotificationManager() + manager._config = {"notification_language": "sk", "ai_language": "de"} + self.assertEqual(manager._notification_language(), "sk") + manager._config = {"ai_language": "sk"} + self.assertEqual(manager._notification_language(), "sk") + manager._config = {} + self.assertEqual(manager._notification_language(), "en") + + with tempfile.TemporaryDirectory() as directory: + db_path = Path(directory) / "settings.db" + conn = sqlite3.connect(db_path) + conn.execute("CREATE TABLE user_settings (setting_key TEXT PRIMARY KEY, setting_value TEXT, updated_at TEXT)") + conn.commit() + conn.close() + with mock.patch.object(notification_manager, "DB_PATH", db_path): + saved = manager.save_settings({"notification_language": "sk", "ai_language": "de"}) + self.assertTrue(saved["success"]) + loaded = notification_manager.NotificationManager() + loaded._load_config() + self.assertEqual(loaded.get_settings()["config"]["notification_language"], "sk") + self.assertEqual(loaded.get_settings()["config"]["ai_language"], "de") + + def test_ai_language_is_independent_from_runtime_notification_language(self): + manager = notification_manager.NotificationManager() + manager._config = { + "notification_language": "sk", + "ai_language": "de", + "ai_provider": "groq", + } + self.assertEqual(manager._notification_language(), "sk") + self.assertEqual(manager._build_ai_config()["ai_language"], "de") + + def test_legacy_non_runtime_ai_language_roundtrips_as_english_runtime(self): + manager = notification_manager.NotificationManager() + manager._config = {"ai_language": "de"} + self.assertEqual(manager._notification_language(), "en") + self.assertEqual(manager.get_settings()["config"]["notification_language"], "en") + + manager._config = {"notification_language": "invalid", "ai_language": "sk"} + self.assertEqual(manager._notification_language(), "sk") + manager._config = {"notification_language": "invalid", "ai_language": "de"} + self.assertEqual(manager._notification_language(), "en") + + with tempfile.TemporaryDirectory() as directory: + db_path = Path(directory) / "settings.db" + conn = sqlite3.connect(db_path) + conn.execute("CREATE TABLE user_settings (setting_key TEXT PRIMARY KEY, setting_value TEXT, updated_at TEXT)") + conn.commit() + conn.close() + with mock.patch.object(notification_manager, "DB_PATH", db_path): + result = manager.save_settings({ + "notification_language": manager.get_settings()["config"]["notification_language"], + "ai_language": "de", + }) + self.assertTrue(result["success"], result) + + def test_backup_email_localizes_status_and_important_packages(self): + channel = object.__new__(notification_channels.EmailChannel) + channel.subject_prefix = "[ProxMenux]" + for event_type, localized_status, english_status in ( + ("backup_fail", "Zlyhalo", "Failed"), + ("backup_complete", "Dokončené", "Completed"), + ("backup_start", "Spustené", "Started"), + ): + backup_html = channel._format_html( + "pve01: Záloha", "Stav zálohy VM 100.", "WARNING", + { + "_notification_language": "sk", "_event_type": event_type, + "_group": "backup", "hostname": "pve01", "vmid": "100", + "vmname": "alpha", "storage": "pbs-main", + }, + ) + self.assertIn(f">{localized_status}<", backup_html) + self.assertNotIn(f">{english_status}<", backup_html) + + updates_html = channel._format_html( + "pve01: Aktualizácie", "Dostupné aktualizácie.", "INFO", + { + "_notification_language": "sk", "_event_type": "system_updates", + "_group": "updates", "hostname": "pve01", + "important_list": "pve-manager\nproxmox-kernel", + }, + ) + self.assertIn("Dôležité balíky", updates_html) + self.assertNotIn("Important Packages", updates_html) + + def test_metric_and_system_email_values_are_localized(self): + channel = object.__new__(notification_channels.EmailChannel) + channel.subject_prefix = "[ProxMenux]" + metric_html = channel._format_html( + "pve01: Vysoké využitie CPU", "CPU dosiahlo 95 %.", "WARNING", + { + "_notification_language": "sk", "_event_type": "cpu_high", + "_group": "resources", "hostname": "pve01", "value": "95", + }, + ) + self.assertIn("Vysoké využitie CPU", metric_html) + self.assertNotIn("Cpu High", metric_html) + + system_html = channel._format_html( + "pve01: Systémový problém", "Zistil sa problém.", "WARNING", + { + "_notification_language": "sk", "_event_type": "system_problem", + "_group": "system", "hostname": "pve01", "reason": "chyba", + }, + ) + self.assertIn("Prehľad: Systém", system_html) + self.assertNotIn("Prehľad:

", system_html) + + def test_every_ui_ai_language_is_accepted_by_backend(self): + source = (APPIMAGE_DIR / "components" / "notification-settings.tsx").read_text(encoding="utf-8") + block = re.search(r"const AI_LANGUAGES = \[(.*?)\n\]", source, re.DOTALL) + self.assertIsNotNone(block) + ui_languages = set(re.findall(r'value: "([a-z]+)"', block.group(1))) + self.assertIn("sv", ui_languages) + self.assertIn("no", ui_languages) + self.assertEqual(ui_languages - set(notification_manager.ALLOWED_AI_LANGUAGES), set()) + + def test_quiet_hours_digest_localizes_system_group(self): + class RecordingChannel: + def __init__(self): + self.payload = None + + def send(self, title, body, severity, data=None): + self.payload = (title, body, severity, data) + return {"success": True} + + manager = notification_manager.NotificationManager() + manager._config = {"notification_language": "sk", "hostname": "pve01"} + channel = RecordingChannel() + with tempfile.TemporaryDirectory() as directory: + db_path = Path(directory) / "settings.db" + conn = sqlite3.connect(db_path) + conn.execute( + "CREATE TABLE quiet_pending (" + "id INTEGER PRIMARY KEY, channel TEXT, event_type TEXT, event_group TEXT, " + "ts REAL, title TEXT, body TEXT)" + ) + conn.execute( + "INSERT INTO quiet_pending " + "(channel, event_type, event_group, ts, title, body) VALUES (?, ?, ?, ?, ?, ?)", + ("telegram", "ai_model_migrated", "system", 0, "pve01: AI model updated", "body"), + ) + conn.commit() + conn.close() + with mock.patch.object(notification_manager, "DB_PATH", db_path), mock.patch.object( + manager, "_record_history" + ): + manager._flush_quiet_for_channel("telegram", channel) + + self.assertIsNotNone(channel.payload) + self.assertIn("Systém: 1", channel.payload[1]) + self.assertNotIn("System", channel.payload[1]) + self.assertTrue(channel.payload[3]["_quiet_hours_summary"]) + + def test_visible_templates_use_known_backend_and_frontend_groups(self): + visible_groups = { + template.get("group", "other") + for template in notification_templates.TEMPLATES.values() + if not template.get("hidden", False) + } + backend_groups = set(notification_templates.EVENT_GROUPS) + + source = (APPIMAGE_DIR / "components" / "notification-settings.tsx").read_text(encoding="utf-8") + match = re.search(r"const EVENT_CATEGORIES = \[(.*?)\]\.map", source) + self.assertIsNotNone(match) + frontend_groups = set(re.findall(r'"([a-z_]+)"', match.group(1))) + + self.assertEqual(visible_groups - backend_groups, set()) + self.assertEqual(visible_groups - frontend_groups, set()) + self.assertEqual(frontend_groups, backend_groups) + for language in ("en", "sk"): + catalog = json.loads( + (APPIMAGE_DIR / "messages" / language / "common.json").read_text(encoding="utf-8") + ) + self.assertEqual( + set(catalog["settings"]["notifications"]["categories"]), + frontend_groups, + language, + ) + + def test_build_bundles_runtime_catalogs(self): + build = (SCRIPTS_DIR / "build_appimage.sh").read_text(encoding="utf-8") + self.assertIn('messages/en/common.json', build) + self.assertIn('messages/sk/common.json', build) + self.assertIn('$APP_DIR/usr/share/proxmenux/messages', build) + + def test_missing_event_type_names_exist_in_both_ui_catalogs(self): + for language in ("en", "sk"): + path = APPIMAGE_DIR / "messages" / language / "common.json" + event_types = json.loads(path.read_text(encoding="utf-8"))["settings"]["notifications"]["eventTypes"] + self.assertIn("lxc_update_applied", event_types) + self.assertIn("docker_stack_update_available", event_types) + + +if __name__ == "__main__": + unittest.main() From 584c4c31a3fb469972c51cc761be658621dadd80 Mon Sep 17 00:00:00 2001 From: Vaso73 Date: Tue, 15 Sep 2026 09:05:08 +0200 Subject: [PATCH 05/21] Localize runtime notifications in Slovak --- AppImage/components/notification-settings.tsx | 34 +++ AppImage/scripts/build_appimage.sh | 4 + AppImage/scripts/notification_channels.py | 67 +++++- AppImage/scripts/notification_templates.py | 222 +++++++++++------- 4 files changed, 229 insertions(+), 98 deletions(-) diff --git a/AppImage/components/notification-settings.tsx b/AppImage/components/notification-settings.tsx index ecebac0a..af00650d 100644 --- a/AppImage/components/notification-settings.tsx +++ b/AppImage/components/notification-settings.tsx @@ -77,6 +77,7 @@ interface NotificationConfig { ai_api_keys: Record // Per-provider API keys ai_models: Record // Per-provider selected models ai_model: string // Current active model (for the selected provider) + notification_language: string ai_language: string ai_ollama_url: string ai_openai_base_url: string @@ -199,6 +200,11 @@ const AI_PROVIDERS = [ }, ] +const NOTIFICATION_LANGUAGES = [ + { value: "en", label: "English" }, + { value: "sk", label: "Slovenčina" }, +] + const AI_LANGUAGES = [ { value: "en", label: "English" }, { value: "sk", label: "Slovenčina" }, @@ -284,6 +290,7 @@ const DEFAULT_CONFIG: NotificationConfig = { openrouter: "", }, ai_model: "", + notification_language: "en", ai_language: "en", ai_ollama_url: "http://localhost:11434", ai_openai_base_url: "", @@ -401,6 +408,7 @@ export function NotificationSettings() { ai_prompt_mode: data.config.ai_prompt_mode || "default", ai_custom_prompt: data.config.ai_custom_prompt || "", ai_allow_suggestions: data.config.ai_allow_suggestions || "false", + notification_language: data.config.notification_language || data.config.ai_language || "en", } // If ai_model exists but ai_models doesn't have it, save it if (configWithDefaults.ai_model && !configWithDefaults.ai_models[configWithDefaults.ai_provider]) { @@ -834,6 +842,7 @@ export function NotificationSettings() { ai_enabled: String(cfg.ai_enabled), ai_provider: cfg.ai_provider, ai_model: cfg.ai_model, + notification_language: cfg.notification_language, ai_language: cfg.ai_language, ai_ollama_url: cfg.ai_ollama_url, ai_openai_base_url: cfg.ai_openai_base_url, @@ -2176,6 +2185,31 @@ export function NotificationSettings() {

+ {/* ── Runtime notification language (independent of AI) ── */} +
+
+ + +
+ +

+ {t("settings.notifications.ui.notificationLanguageHint")} +

+
+ {/* ── Advanced: AI Enhancement ── */}
diff --git a/AppImage/scripts/build_appimage.sh b/AppImage/scripts/build_appimage.sh index 85b031f4..79935210 100755 --- a/AppImage/scripts/build_appimage.sh +++ b/AppImage/scripts/build_appimage.sh @@ -163,6 +163,10 @@ cp "$SCRIPT_DIR/proxmox_known_errors.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo cp "$SCRIPT_DIR/ai_context_enrichment.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ ai_context_enrichment.py not found" cp "$SCRIPT_DIR/startup_grace.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ startup_grace.py not found" cp "$SCRIPT_DIR/flask_notification_routes.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ flask_notification_routes.py not found" +# Backend notification rendering reuses the canonical Monitor catalogs. +mkdir -p "$APP_DIR/usr/share/proxmenux/messages/en" "$APP_DIR/usr/share/proxmenux/messages/sk" +cp "$APPIMAGE_ROOT/messages/en/common.json" "$APP_DIR/usr/share/proxmenux/messages/en/common.json" +cp "$APPIMAGE_ROOT/messages/sk/common.json" "$APP_DIR/usr/share/proxmenux/messages/sk/common.json" cp "$SCRIPT_DIR/oci_manager.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ oci_manager.py not found" cp "$SCRIPT_DIR/flask_oci_routes.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ flask_oci_routes.py not found" cp "$SCRIPT_DIR/flask_audit_routes.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ flask_audit_routes.py not found" diff --git a/AppImage/scripts/notification_channels.py b/AppImage/scripts/notification_channels.py index f120d37b..74490dbd 100644 --- a/AppImage/scripts/notification_channels.py +++ b/AppImage/scripts/notification_channels.py @@ -36,6 +36,18 @@ _KNOWN_SSRF_TARGETS = { _BLOCKED_LOOPBACK_PORTS = {'8006', '8007'} # PVE API HTTPS / HTTPS-alt +def _runtime_notification_text(key: str, data: Optional[Dict] = None, + **values: Any) -> str: + """Resolve runtime text lazily to avoid the manager/channel import cycle.""" + from notification_templates import runtime_message + language = str((data or {}).get('_notification_language', 'en')) + return runtime_message(key, language, **values) + + +def _runtime_text(key: str, data: Optional[Dict] = None, **values: Any) -> str: + return _runtime_notification_text(f'channels.{key}', data, **values) + + def _validate_user_webhook_url(url: str) -> Tuple[bool, str]: """Lightweight SSRF guard for Gotify-style channels. @@ -645,11 +657,11 @@ class DiscordChannel(NotificationChannel): ] elif data: if data.get('category'): - fields.append({'name': 'Category', 'value': data['category'], 'inline': True}) + fields.append({'name': _runtime_text('discord.category', data), 'value': data['category'], 'inline': True}) if data.get('hostname'): - fields.append({'name': 'Host', 'value': data['hostname'], 'inline': True}) + fields.append({'name': _runtime_text('discord.host', data), 'value': data['hostname'], 'inline': True}) if data.get('severity'): - fields.append({'name': 'Severity', 'value': data['severity'], 'inline': True}) + fields.append({'name': _runtime_text('discord.severity', data), 'value': data['severity'], 'inline': True}) embeds: List[Dict[str, Any]] = [] for idx, chunk in enumerate(chunks): @@ -969,12 +981,17 @@ class EmailChannel(NotificationChannel): import time as _time data = data or {} - sev = self._SEV_STYLE.get(severity, self._SEV_DEFAULT) + sev = dict(self._SEV_STYLE.get(severity, self._SEV_DEFAULT)) + severity_key = severity.lower() if severity in self._SEV_STYLE else 'default' + sev['label'] = _runtime_text(f'email.severity.{severity_key}', data) # Determine group for section header event_type = data.get('_event_type', '') group = data.get('_group', 'other') - section_label = self._GROUP_LABELS.get(group, 'System Notification') + section_label = _runtime_text(f'email.groups.{group}', data) + report_label = _runtime_text('email.report', data, group=section_label) + host_label = _runtime_text('email.host', data) + footer_label = _runtime_text('email.footer', data) # Timestamp ts = data.get('timestamp', '') or _time.strftime('%Y-%m-%d %H:%M:%S UTC', _time.gmtime()) @@ -1029,7 +1046,7 @@ class EmailChannel(NotificationChannel): if reason and len(reason) > 80: reason_html = f'''
-

Details

+

{_runtime_text('email.details', data)}

{html_mod.escape(reason)}

''' @@ -1039,7 +1056,7 @@ class EmailChannel(NotificationChannel): display_title = display_title.replace(prefix, '').strip() return f''' - +
@@ -1050,7 +1067,7 @@ class EmailChannel(NotificationChannel):

ProxMenux Monitor

-

{html_mod.escape(section_label)} Report

+

{html_mod.escape(report_label)}

{sev['label'].upper()} @@ -1070,7 +1087,7 @@ class EmailChannel(NotificationChannel):
- Host: {html_mod.escape(data.get('hostname', ''))} + {html_mod.escape(host_label)}: {html_mod.escape(data.get('hostname', ''))} {html_mod.escape(ts)} @@ -1090,7 +1107,7 @@ class EmailChannel(NotificationChannel):
- +
ProxMenux Notification Service{html_mod.escape(footer_label)} proxmenux.com
@@ -1110,11 +1127,34 @@ class EmailChannel(NotificationChannel): """ esc = html_mod.escape rows = [] + field_keys = { + 'VM/CT ID': 'vmCtId', 'Name': 'name', 'Action': 'action', + 'Target Node': 'targetNode', 'Reason': 'reason', 'Storage': 'storage', + 'Status': 'status', 'Size': 'size', 'Duration': 'duration', + 'Snapshot': 'snapshot', 'Metric': 'metric', 'Current Value': 'currentValue', + 'Threshold': 'threshold', 'CPU Cores': 'cpuCores', 'Memory': 'memory', + 'Temperature': 'temperature', 'Mount Point': 'mountPoint', 'Usage': 'usage', + 'Available': 'available', 'Device': 'device', 'Severity': 'severity', + 'Storage Name': 'storageName', 'Type': 'type', 'Interface': 'interface', + 'Latency': 'latency', 'Event': 'event', 'Source IP': 'sourceIp', + 'Username': 'username', 'Service': 'service', 'Jail': 'jail', + 'Failures': 'failures', 'Change': 'change', 'Node': 'node', + 'Quorum': 'quorum', 'Nodes Affected': 'nodesAffected', 'Process': 'process', + 'Details': 'reason', 'Category': 'category', + 'Previous Severity': 'previousSeverity', 'Active Issues': 'activeIssues', + 'Total Updates': 'totalUpdates', 'Security Updates': 'securityUpdates', + 'Proxmox Updates': 'proxmoxUpdates', 'Kernel Updates': 'kernelUpdates', + 'Important Packages': 'importantPackages', 'Current Version': 'currentVersion', + 'New Version': 'newVersion', + } + language_data = data def _add(label: str, value, fmt: str = ''): - """Add a row if value is truthy.""" + """Add a localized row if value is truthy.""" + original_label = label + label = _runtime_text(f"email.fields.{field_keys[label]}", language_data) v = str(value).strip() if value else '' - if not v or v == '0' and label not in ('Failures',): + if not v or v == '0' and original_label not in ('Failures',): return if fmt == 'severity': sev_colors = { @@ -1136,7 +1176,8 @@ class EmailChannel(NotificationChannel): if group == 'vm_ct': _add('VM/CT ID', data.get('vmid'), 'code') _add('Name', data.get('vmname'), 'bold') - _add('Action', event_type.replace('_', ' ').replace('vm ', 'VM ').replace('ct ', 'CT ').title()) + action = _runtime_notification_text(f'templates.{event_type}.label', data) + _add('Action', action) _add('Target Node', data.get('target_node')) _add('Reason', data.get('reason')) diff --git a/AppImage/scripts/notification_templates.py b/AppImage/scripts/notification_templates.py index 2befce56..7b4b1ed7 100644 --- a/AppImage/scripts/notification_templates.py +++ b/AppImage/scripts/notification_templates.py @@ -17,9 +17,62 @@ import socket import time import urllib.request import urllib.error +from functools import lru_cache +from pathlib import Path from typing import Dict, Any, Optional, List, Tuple +_SCRIPT_ROOT = Path(__file__).resolve().parents[1] +_BUNDLED_CATALOG_DIR = _SCRIPT_ROOT / 'share' / 'proxmenux' / 'messages' +_SOURCE_CATALOG_DIR = _SCRIPT_ROOT / 'messages' +RUNTIME_CATALOG_DIR = ( + _BUNDLED_CATALOG_DIR + if _BUNDLED_CATALOG_DIR.is_dir() + else _SOURCE_CATALOG_DIR + if _SOURCE_CATALOG_DIR.is_dir() + else Path('/usr/share/proxmenux/messages') +) + + +class _SafeFormatDict(dict): + def __missing__(self, key): + return '' + + +@lru_cache(maxsize=8) +def _load_runtime_catalog(language: str) -> Dict[str, Any]: + """Load one existing Monitor catalog's runtime notification namespace.""" + path = RUNTIME_CATALOG_DIR / language / 'common.json' + try: + with path.open(encoding='utf-8') as handle: + return json.load(handle).get('runtime', {}).get('notifications', {}) + except (OSError, ValueError, TypeError): + return {} + + +def _catalog_value(catalog: Dict[str, Any], dotted_key: str) -> Optional[str]: + value: Any = catalog + for part in dotted_key.split('.'): + if not isinstance(value, dict) or part not in value: + return None + value = value[part] + return value if isinstance(value, str) and value else None + + +def runtime_message(key: str, language: str = 'en', **values: Any) -> str: + """Resolve runtime text with per-key English fallback and safe placeholders.""" + requested = (language or 'en').split('-', 1)[0].lower() + value = _catalog_value(_load_runtime_catalog(requested), key) + if value is None: + value = _catalog_value(_load_runtime_catalog('en'), key) + if value is None: + return '' + try: + return value.format_map(_SafeFormatDict(values)) + except (ValueError, IndexError): + return value + + # ─── vzdump message parser ─────────────────────────────────────── def _parse_vzdump_message(message: str) -> Optional[Dict[str, Any]]: @@ -248,7 +301,8 @@ def _parse_vzdump_message(message: str) -> Optional[Dict[str, Any]]: } -def _format_vzdump_body(parsed: Dict[str, Any], is_success: bool) -> str: +def _format_vzdump_body(parsed: Dict[str, Any], is_success: bool, + language: str = 'en') -> str: """Format parsed vzdump data into a clean Telegram-friendly message.""" parts = [] @@ -285,9 +339,9 @@ def _format_vzdump_body(parsed: Dict[str, Any], is_success: bool) -> str: # Size and Duration on same line with icons detail_line = [] if vm.get('size'): - detail_line.append(f"\U0001F4CF Size: {vm['size']}") + detail_line.append(f"\U0001F4CF {runtime_message('vzdump.size', language, value=vm['size'])}") if vm.get('time'): - detail_line.append(f"\u23F1\uFE0F Duration: {vm['time']}") + detail_line.append(f"\u23F1\uFE0F {runtime_message('vzdump.duration', language, value=vm['time'])}") if detail_line: parts.append(' | '.join(detail_line)) @@ -302,7 +356,7 @@ def _format_vzdump_body(parsed: Dict[str, Any], is_success: bool) -> str: label = storage_name if storage_name else 'PBS' parts.append(f"\U0001F5C4\uFE0F {label}: {fname}") else: - label = storage_name if storage_name else 'File' + label = storage_name if storage_name else runtime_message('vzdump.file', language) parts.append(f"\U0001F4C1 {label}: {fname}") # Error reason if failed @@ -320,13 +374,13 @@ def _format_vzdump_body(parsed: Dict[str, Any], is_success: bool) -> str: summary_parts = [] if vm_count: - summary_parts.append(f"\U0001F4CA {vm_count} backups") + summary_parts.append(f"\U0001F4CA {runtime_message('vzdump.backups', language, count=vm_count)}") if fail_count: - summary_parts.append(f"\u274C {fail_count} failed") + summary_parts.append(f"\u274C {runtime_message('vzdump.failed', language, count=fail_count)}") if parsed.get('total_size'): - summary_parts.append(f"\U0001F4E6 Total: {parsed['total_size']}") + summary_parts.append(f"\U0001F4E6 {runtime_message('vzdump.total', language, value=parsed['total_size'])}") if parsed.get('total_time'): - summary_parts.append(f"\u23F1\uFE0F Time: {parsed['total_time']}") + summary_parts.append(f"\u23F1\uFE0F {runtime_message('vzdump.time', language, value=parsed['total_time'])}") if summary_parts: parts.append(' | '.join(summary_parts)) @@ -334,88 +388,67 @@ def _format_vzdump_body(parsed: Dict[str, Any], is_success: bool) -> str: return '\n'.join(parts) -def _format_system_startup(data: Dict[str, Any]) -> Tuple[str, str]: - """ - Format comprehensive system startup report. - - Returns (title, body) tuple for the notification. - Handles both simple startups (all OK) and those with issues. - """ +def _format_system_startup(data: Dict[str, Any], language: str = 'en') -> Tuple[str, str]: + """Format the comprehensive startup report using runtime catalogs.""" hostname = data.get('hostname', 'unknown') has_issues = data.get('has_issues', False) - - # Build title if has_issues: total_issues = ( - data.get('total_failed', 0) + - len(data.get('services_failed', [])) + - len(data.get('storage_unavailable', [])) + data.get('total_failed', 0) + + len(data.get('services_failed', [])) + + len(data.get('storage_unavailable', [])) ) - title = f"{hostname}: System startup - {total_issues} issue(s) detected" + title = runtime_message('startup.issuesTitle', language, hostname=hostname, count=total_issues) else: - title = f"{hostname}: System startup completed" - - # Build body + title = runtime_message('startup.completeTitle', language, hostname=hostname) + parts = [] - - # Overall status if not has_issues: - parts.append("All systems operational.") - - # VMs/CTs started + parts.append(runtime_message('startup.operational', language)) + vms_ok = len(data.get('vms_started', [])) cts_ok = len(data.get('cts_started', [])) if vms_ok or cts_ok: - count_parts = [] + counts = [] if vms_ok: - count_parts.append(f"{vms_ok} VM{'s' if vms_ok > 1 else ''}") + key = 'startup.vmCountOne' if vms_ok == 1 else 'startup.vmCountMany' + counts.append(runtime_message(key, language, count=vms_ok)) if cts_ok: - count_parts.append(f"{cts_ok} CT{'s' if cts_ok > 1 else ''}") - - # List names (up to 5) - names = [] - for vm in data.get('vms_started', [])[:3]: - names.append(f"{vm['name']} ({vm['vmid']})") - for ct in data.get('cts_started', [])[:3]: - names.append(f"{ct['name']} ({ct['vmid']})") - - line = f"\u2705 {' and '.join(count_parts)} started" + key = 'startup.ctCountOne' if cts_ok == 1 else 'startup.ctCountMany' + counts.append(runtime_message(key, language, count=cts_ok)) + names = [ + f"{item['name']} ({item['vmid']})" + for item in (data.get('vms_started', [])[:3] + data.get('cts_started', [])[:3]) + ] + line = runtime_message('startup.started', language, counts=', '.join(counts)) if names: - if len(names) <= 5: - line += f": {', '.join(names)}" - else: - line += f": {', '.join(names[:5])}..." + line += f": {', '.join(names[:5])}" + if len(names) > 5: + line += '…' parts.append(line) - - # Failed VMs/CTs + + unknown_error = runtime_message('startup.unknownError', language) for vm in data.get('vms_failed', []): - reason = vm.get('reason', 'unknown error') - parts.append(f"\u274C VM failed: {vm['name']} - {reason}") - + parts.append(runtime_message('startup.vmFailed', language, name=vm['name'], reason=vm.get('reason', unknown_error))) for ct in data.get('cts_failed', []): - reason = ct.get('reason', 'unknown error') - parts.append(f"\u274C CT failed: {ct['name']} - {reason}") - - # Storage issues + parts.append(runtime_message('startup.ctFailed', language, name=ct['name'], reason=ct.get('reason', unknown_error))) + storage_unavailable = data.get('storage_unavailable', []) if storage_unavailable: - names = [s['name'] for s in storage_unavailable[:3]] - parts.append(f"\u26A0\uFE0F Storage: {len(storage_unavailable)} unavailable ({', '.join(names)})") - - # Service issues + parts.append(runtime_message( + 'startup.storageUnavailable', language, count=len(storage_unavailable), + names=', '.join(item['name'] for item in storage_unavailable[:3]), + )) services_failed = data.get('services_failed', []) if services_failed: - names = [s['name'] for s in services_failed[:3]] - parts.append(f"\u26A0\uFE0F Services: {len(services_failed)} failed ({', '.join(names)})") - - # Startup duration + parts.append(runtime_message( + 'startup.servicesFailed', language, count=len(services_failed), + names=', '.join(item['name'] for item in services_failed[:3]), + )) duration = data.get('startup_duration_seconds', 0) if duration: - minutes = int(duration // 60) - parts.append(f"\u23F1\uFE0F Startup completed in {minutes} min") - - body = '\n'.join(parts) - return title, body + parts.append(runtime_message('startup.duration', language, minutes=int(duration // 60))) + return title, '\n'.join(parts) def _format_app_update_available(data: Dict[str, Any]) -> Tuple[str, str]: @@ -1568,7 +1601,8 @@ def _format_bytes_human(n: Any) -> str: return f'{size:.1f} {units[i]}' -def render_template(event_type: str, data: Dict[str, Any]) -> Dict[str, Any]: +def render_template(event_type: str, data: Dict[str, Any], + language: str = 'en') -> Dict[str, Any]: """Render a template into a structured notification object. Returns structured output usable by all channels: @@ -1576,19 +1610,35 @@ def render_template(event_type: str, data: Dict[str, Any]) -> Dict[str, Any]: """ import html as html_mod - template = TEMPLATES.get(event_type) - if not template: + source_template = TEMPLATES.get(event_type) + if not source_template: # Catch-all: unknown event types always get delivered (group 'other') # so no Proxmox notification is ever silently dropped. fallback_body = data.get('message', data.get('reason', str(data))) severity = data.get('severity', 'INFO') return { - 'title': f"{_get_hostname()}: {event_type}", + 'title': runtime_message( + 'fallback.unknownTitle', language, + hostname=_get_hostname(), event_type=event_type, + ), 'body': fallback_body, 'body_text': fallback_body, 'body_html': f'

{html_mod.escape(str(fallback_body))}

', 'fields': [], 'tags': [severity, 'other', event_type], 'severity': severity, 'group': 'other', } + + template = dict(source_template) + requested_language = (language or 'en').split('-', 1)[0].lower() + requested_catalog = _load_runtime_catalog(requested_language) + english_catalog = _load_runtime_catalog('en') + for field in ('title', 'body', 'label'): + key = f'templates.{event_type}.{field}' + localized = ( + _catalog_value(requested_catalog, key) + or _catalog_value(english_catalog, key) + ) + if localized: + template[field] = localized # Ensure hostname is always available variables = { @@ -1610,7 +1660,7 @@ def render_template(event_type: str, data: Dict[str, Any]) -> Dict[str, Any]: 'packages': '', 'pve_packages': '', 'version': '', 'issue_list': '', 'error_key': '', 'storage_name': '', 'storage_type': '', - 'important_list': 'none', + 'important_list': runtime_message('fallback.none', language), # Host Backup specifics (run_scheduled_backup.sh + backup_host.sh). 'job_id': '', 'backend': '', 'backend_label': '', 'destination': '', 'profile_mode': '', @@ -1626,9 +1676,9 @@ def render_template(event_type: str, data: Dict[str, Any]) -> Dict[str, Any]: if _byte_key in data: variables[f'{_byte_key}_human'] = _format_bytes_human(data[_byte_key]) - # Ensure important_list is never blank (fallback to 'none') + # Ensure important_list is never blank (fallback to localized "none") if not variables.get('important_list', '').strip(): - variables['important_list'] = 'none' + variables['important_list'] = runtime_message('fallback.none', language) # Derive the affected object's display name for titles that use it. # Priority: caller-supplied `entity` (health_monitor.emit_event) → @@ -1657,7 +1707,8 @@ def render_template(event_type: str, data: Dict[str, Any]) -> Dict[str, Any]: _caller_title = str(variables.get('title', '')).strip() if not _caller_title: hn = variables.get('hostname', '') - _caller_title = f'{hn}: Health check degraded' if hn else 'Health check degraded' + degraded = runtime_message('fallback.healthCheckDegraded', language) + _caller_title = f'{hn}: {degraded}' if hn else degraded variables['title_or_default'] = _caller_title # `format_map` with a SafeDict avoids the KeyError → "show raw template @@ -1685,7 +1736,7 @@ def render_template(event_type: str, data: Dict[str, Any]) -> Dict[str, Any]: if formatter_name and formatter_name in globals(): formatter_func = globals()[formatter_name] try: - title, body_text = formatter_func(data) + title, body_text = formatter_func(data, language=language) except Exception: # Fallback to standard formatting if formatter fails try: @@ -1696,9 +1747,10 @@ def render_template(event_type: str, data: Dict[str, Any]) -> Dict[str, Any]: parsed = _parse_vzdump_message(pve_message) if parsed: is_success = (event_type == 'backup_complete') - body_text = _format_vzdump_body(parsed, is_success) - # Use PVE's own title if available (contains hostname and status) - if pve_title: + body_text = _format_vzdump_body(parsed, is_success, language=language) + # Preserve PVE's source title for English, but never leak it into a + # deterministic localized notification. + if pve_title and requested_language == 'en': title = pve_title else: # Couldn't parse -- use PVE raw message as body @@ -1722,15 +1774,15 @@ def render_template(event_type: str, data: Dict[str, Any]) -> Dict[str, Any]: # Build structured fields for Discord embeds / rich notifications fields = [] field_map = [ - ('vmid', 'VM/CT'), ('vmname', 'Name'), ('device', 'Device'), - ('source_ip', 'Source IP'), ('node_name', 'Node'), ('category', 'Category'), - ('service_name', 'Service'), ('jail', 'Jail'), ('username', 'User'), - ('count', 'Count'), ('window', 'Window'), ('entity_list', 'Affected'), + ('vmid', 'fields.vmid'), ('vmname', 'fields.name'), ('device', 'fields.device'), + ('source_ip', 'fields.sourceIp'), ('node_name', 'fields.node'), ('category', 'fields.category'), + ('service_name', 'fields.service'), ('jail', 'fields.jail'), ('username', 'fields.user'), + ('count', 'fields.count'), ('window', 'fields.window'), ('entity_list', 'fields.affected'), ] - for key, label in field_map: + for key, label_key in field_map: val = variables.get(key, '') if val: - fields.append((label, str(val))) + fields.append((runtime_message(label_key, language), str(val))) # Build HTML body with escaped content body_html_parts = [] From 376b9504f8a721cd2a187e92732908bf91a8c3b5 Mon Sep 17 00:00:00 2001 From: Vaso73 Date: Tue, 15 Sep 2026 09:29:58 +0200 Subject: [PATCH 06/21] Fix runtime language migration and email localization --- AppImage/components/notification-settings.tsx | 15 +++++++++--- AppImage/scripts/notification_channels.py | 23 ++++++++++++++----- 2 files changed, 29 insertions(+), 9 deletions(-) diff --git a/AppImage/components/notification-settings.tsx b/AppImage/components/notification-settings.tsx index af00650d..9dc70a73 100644 --- a/AppImage/components/notification-settings.tsx +++ b/AppImage/components/notification-settings.tsx @@ -205,6 +205,12 @@ const NOTIFICATION_LANGUAGES = [ { value: "sk", label: "Slovenčina" }, ] +function normalizeNotificationLanguage(notificationLanguage?: string, legacyAiLanguage?: string): string { + if (notificationLanguage === "en" || notificationLanguage === "sk") return notificationLanguage + if (legacyAiLanguage === "en" || legacyAiLanguage === "sk") return legacyAiLanguage + return "en" +} + const AI_LANGUAGES = [ { value: "en", label: "English" }, { value: "sk", label: "Slovenčina" }, @@ -408,7 +414,10 @@ export function NotificationSettings() { ai_prompt_mode: data.config.ai_prompt_mode || "default", ai_custom_prompt: data.config.ai_custom_prompt || "", ai_allow_suggestions: data.config.ai_allow_suggestions || "false", - notification_language: data.config.notification_language || data.config.ai_language || "en", + notification_language: normalizeNotificationLanguage( + data.config.notification_language, + data.config.ai_language, + ), } // If ai_model exists but ai_models doesn't have it, save it if (configWithDefaults.ai_model && !configWithDefaults.ai_models[configWithDefaults.ai_provider]) { @@ -842,7 +851,7 @@ export function NotificationSettings() { ai_enabled: String(cfg.ai_enabled), ai_provider: cfg.ai_provider, ai_model: cfg.ai_model, - notification_language: cfg.notification_language, + notification_language: normalizeNotificationLanguage(cfg.notification_language, cfg.ai_language), ai_language: cfg.ai_language, ai_ollama_url: cfg.ai_ollama_url, ai_openai_base_url: cfg.ai_openai_base_url, @@ -2192,7 +2201,7 @@ export function NotificationSettings() {