666 lines
24 KiB
Python
666 lines
24 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Gateway i18n registry: t(), @i18nModel, boot-sync, in-memory cache.
|
|
|
|
All UI-visible texts in the gateway (HTTPException details, model labels,
|
|
API messages) are tagged with t() and registered at import time.
|
|
At boot, the registry is synced to the xx base set in the DB.
|
|
At runtime, t() returns the cached translation for the current request language.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from contextvars import ContextVar
|
|
from dataclasses import dataclass, field as dataclass_field
|
|
from typing import Any, Dict, List, Optional, Type
|
|
|
|
from pydantic import BaseModel
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry (populated at import time by t() and @i18nModel)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class _I18nRegistryEntry:
|
|
context: str
|
|
value: str
|
|
|
|
|
|
_REGISTRY: Dict[str, _I18nRegistryEntry] = {}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Translation cache (populated at boot by _loadCache)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_CACHE: Dict[str, Dict[str, str]] = {}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-request language (set by middleware)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_CURRENT_LANGUAGE: ContextVar[str] = ContextVar("i18n_lang", default="de")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Model labels (backwards-compatible with getModelLabels / getModelLabel)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
MODEL_LABELS: Dict[str, Dict[str, Any]] = {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# t() -- tag and translate
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def t(key: str, context: str = "api", value: str = "") -> str:
|
|
"""Tag a UI-visible text for i18n and return the translation.
|
|
|
|
At import time: registers the key with context and AI description.
|
|
At runtime: returns the cached translation for _CURRENT_LANGUAGE.
|
|
Falls back to the key itself (German base text) if no translation found.
|
|
"""
|
|
if key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context=context, value=value)
|
|
lang = _CURRENT_LANGUAGE.get()
|
|
if lang == "de":
|
|
return key
|
|
return _CACHE.get(lang, {}).get(key, key)
|
|
|
|
|
|
def apiRouteContext(routeModuleName: str):
|
|
"""Return a callable that registers + translates HTTPException details.
|
|
|
|
The key is registered eagerly in ``_REGISTRY`` the moment ``_apiMsg(key)``
|
|
is evaluated (module-level ``detail=routeApiMsg("…")`` runs at import time).
|
|
At runtime ``t()`` returns the cached translation for the current language.
|
|
"""
|
|
_ctx = f"api.{routeModuleName}"
|
|
|
|
def _apiMsg(key: str, value: str = "") -> str:
|
|
if key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context=_ctx, value=value)
|
|
return t(key, _ctx, value)
|
|
return _apiMsg
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# @i18nModel -- class decorator for Pydantic models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def i18nModel(modelLabel: str, aiContext: str = ""):
|
|
"""Class decorator: registers model and field labels for i18n.
|
|
|
|
1. Registers t(modelLabel, "table.<ClassName>", aiContext or docstring)
|
|
2. For each Field with json_schema_extra["label"]:
|
|
Registers t(label, "table.<ClassName>.<fieldName>", field.description)
|
|
3. Populates MODEL_LABELS for getModelLabels()/getModelLabel() in attributeUtils
|
|
"""
|
|
def _decorator(cls: Type[BaseModel]) -> Type[BaseModel]:
|
|
className = cls.__name__
|
|
ctx = aiContext or _extractDocstringFirstLine(cls)
|
|
t(modelLabel, f"table.{className}", ctx)
|
|
|
|
attributes: Dict[str, str] = {}
|
|
for fieldName, fieldInfo in cls.model_fields.items():
|
|
extra = fieldInfo.json_schema_extra
|
|
if not isinstance(extra, dict):
|
|
continue
|
|
label = extra.get("label")
|
|
if label:
|
|
desc = fieldInfo.description or ""
|
|
t(label, f"table.{className}.{fieldName}", desc)
|
|
attributes[fieldName] = label
|
|
else:
|
|
attributes[fieldName] = fieldName
|
|
|
|
MODEL_LABELS[className] = {
|
|
"model": modelLabel,
|
|
"attributes": attributes,
|
|
}
|
|
return cls
|
|
return _decorator
|
|
|
|
|
|
def _extractDocstringFirstLine(cls: type) -> str:
|
|
doc = cls.__doc__
|
|
if not doc:
|
|
return ""
|
|
return doc.strip().split("\n")[0].strip()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Language setter (called by middleware)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _setLanguage(lang: str):
|
|
"""Set the language for the current request context."""
|
|
_CURRENT_LANGUAGE.set(lang)
|
|
|
|
|
|
def _getLanguage() -> str:
|
|
"""Get the language for the current request context."""
|
|
return _CURRENT_LANGUAGE.get()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Boot: scan route files for routeApiMsg("…") calls → register eagerly
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_ROUTE_API_MSG_RE = None # compiled lazily
|
|
|
|
def _scanRouteApiMsgKeys():
|
|
"""Scan all gateway route/feature Python files for routeApiMsg("…") calls
|
|
and register the keys in _REGISTRY so they appear in the boot DB sync.
|
|
"""
|
|
import re
|
|
from pathlib import Path
|
|
|
|
global _ROUTE_API_MSG_RE
|
|
if _ROUTE_API_MSG_RE is None:
|
|
_ROUTE_API_MSG_RE = re.compile(
|
|
r"""routeApiMsg\(\s*(['"])((?:\\.|(?!\1).)+)\1""",
|
|
)
|
|
|
|
gatewayRoot = Path(__file__).resolve().parents[1]
|
|
scanDirs = [gatewayRoot / "routes", gatewayRoot / "features"]
|
|
|
|
_ctxRe = re.compile(r'''apiRouteContext\(\s*['"]([^'"]+)['"]\s*\)''')
|
|
|
|
for scanDir in scanDirs:
|
|
if not scanDir.is_dir():
|
|
continue
|
|
for pyFile in scanDir.rglob("*.py"):
|
|
try:
|
|
src = pyFile.read_text(encoding="utf-8", errors="replace")
|
|
except OSError:
|
|
continue
|
|
ctxMatch = _ctxRe.search(src)
|
|
if not ctxMatch:
|
|
continue
|
|
ctx = f"api.{ctxMatch.group(1)}"
|
|
for m in _ROUTE_API_MSG_RE.finditer(src):
|
|
key = m.group(2).replace("\\'", "'").replace('\\"', '"')
|
|
if key and key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context=ctx, value="")
|
|
|
|
logger.info("i18n route scan: %d api.* keys in registry after scan",
|
|
sum(1 for e in _REGISTRY.values() if e.context.startswith("api.")))
|
|
|
|
|
|
def _registerNavLabels():
|
|
"""Register all navigation labels from NAVIGATION_SECTIONS as i18n keys.
|
|
|
|
Called at boot before DB sync so that nav labels appear in the xx base set
|
|
and can be translated via the Admin UI.
|
|
"""
|
|
try:
|
|
from modules.system.mainSystem import NAVIGATION_SECTIONS
|
|
except ImportError:
|
|
logger.warning("i18n: could not import NAVIGATION_SECTIONS for nav label registration")
|
|
return
|
|
|
|
count = 0
|
|
for section in NAVIGATION_SECTIONS:
|
|
title = section.get("title", "")
|
|
if title and title not in _REGISTRY:
|
|
_REGISTRY[title] = _I18nRegistryEntry(context="nav", value="")
|
|
count += 1
|
|
|
|
for item in section.get("items", []):
|
|
label = item.get("label", "")
|
|
if label and label not in _REGISTRY:
|
|
_REGISTRY[label] = _I18nRegistryEntry(context="nav", value="")
|
|
count += 1
|
|
|
|
for subgroup in section.get("subgroups", []):
|
|
sgTitle = subgroup.get("title", "")
|
|
if sgTitle and sgTitle not in _REGISTRY:
|
|
_REGISTRY[sgTitle] = _I18nRegistryEntry(context="nav", value="")
|
|
count += 1
|
|
for item in subgroup.get("items", []):
|
|
label = item.get("label", "")
|
|
if label and label not in _REGISTRY:
|
|
_REGISTRY[label] = _I18nRegistryEntry(context="nav", value="")
|
|
count += 1
|
|
|
|
logger.info("i18n nav labels: registered %d nav keys", count)
|
|
|
|
|
|
def _registerFeatureUiLabels():
|
|
"""Register FEATURE_LABEL and UI_OBJECTS labels from all feature modules (German i18n keys)."""
|
|
try:
|
|
from modules.system import mainSystem as _mainSystem
|
|
_fl = getattr(_mainSystem, "FEATURE_LABEL", None)
|
|
if isinstance(_fl, str) and _fl and _fl not in _REGISTRY:
|
|
_REGISTRY[_fl] = _I18nRegistryEntry(context="nav", value="")
|
|
except ImportError:
|
|
pass
|
|
|
|
_featureModulePaths = (
|
|
"modules.features.trustee.mainTrustee",
|
|
"modules.features.graphicalEditor.mainGraphicalEditor",
|
|
"modules.features.commcoach.mainCommcoach",
|
|
"modules.features.teamsbot.mainTeamsbot",
|
|
"modules.features.workspace.mainWorkspace",
|
|
"modules.features.realEstate.mainRealEstate",
|
|
"modules.features.neutralization.mainNeutralization",
|
|
"modules.features.chatbot.mainChatbot",
|
|
)
|
|
added = 0
|
|
for modPath in _featureModulePaths:
|
|
try:
|
|
mod = __import__(modPath, fromlist=["FEATURE_LABEL", "UI_OBJECTS"])
|
|
except ImportError:
|
|
continue
|
|
fl = getattr(mod, "FEATURE_LABEL", None)
|
|
if isinstance(fl, str) and fl and fl not in _REGISTRY:
|
|
_REGISTRY[fl] = _I18nRegistryEntry(context="nav", value="")
|
|
added += 1
|
|
for uiObj in getattr(mod, "UI_OBJECTS", []) or []:
|
|
lab = uiObj.get("label")
|
|
if isinstance(lab, str) and lab and lab not in _REGISTRY:
|
|
_REGISTRY[lab] = _I18nRegistryEntry(context="nav", value="")
|
|
added += 1
|
|
elif isinstance(lab, dict):
|
|
base = lab.get("de") or lab.get("en")
|
|
if base and base not in _REGISTRY:
|
|
_REGISTRY[base] = _I18nRegistryEntry(context="nav", value="")
|
|
added += 1
|
|
logger.info("i18n feature UI labels: %d new keys (nav context)", added)
|
|
|
|
|
|
def _registerRbacLabels():
|
|
"""Register DATA_OBJECTS, RESOURCE_OBJECTS labels and TEMPLATE_ROLES descriptions
|
|
from all feature modules and system module as i18n keys.
|
|
|
|
context mapping:
|
|
- DATA_OBJECTS → rbac.data
|
|
- RESOURCE_OBJECTS → rbac.resource
|
|
- TEMPLATE_ROLES[].description (de) → rbac.role
|
|
- QUICK_ACTIONS[].label/description (de) → rbac.quickaction
|
|
- QUICK_ACTION_CATEGORIES[].label (de) → rbac.quickaction
|
|
"""
|
|
_systemModule = "modules.system.mainSystem"
|
|
_featureModulePaths = (
|
|
_systemModule,
|
|
"modules.features.trustee.mainTrustee",
|
|
"modules.features.graphicalEditor.mainGraphicalEditor",
|
|
"modules.features.commcoach.mainCommcoach",
|
|
"modules.features.teamsbot.mainTeamsbot",
|
|
"modules.features.workspace.mainWorkspace",
|
|
"modules.features.realEstate.mainRealEstate",
|
|
"modules.features.neutralization.mainNeutralization",
|
|
"modules.features.chatbot.mainChatbot",
|
|
)
|
|
|
|
def _extractDe(obj) -> str:
|
|
if isinstance(obj, str):
|
|
return obj
|
|
if isinstance(obj, dict):
|
|
return obj.get("de") or obj.get("en") or ""
|
|
return ""
|
|
|
|
added = 0
|
|
for modPath in _featureModulePaths:
|
|
try:
|
|
mod = __import__(modPath, fromlist=[
|
|
"DATA_OBJECTS", "RESOURCE_OBJECTS", "TEMPLATE_ROLES",
|
|
"QUICK_ACTIONS", "QUICK_ACTION_CATEGORIES",
|
|
])
|
|
except ImportError:
|
|
continue
|
|
|
|
for dataObj in getattr(mod, "DATA_OBJECTS", []) or []:
|
|
key = _extractDe(dataObj.get("label"))
|
|
if key and key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context="rbac.data", value="")
|
|
added += 1
|
|
|
|
for resObj in getattr(mod, "RESOURCE_OBJECTS", []) or []:
|
|
key = _extractDe(resObj.get("label"))
|
|
if key and key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context="rbac.resource", value="")
|
|
added += 1
|
|
|
|
for role in getattr(mod, "TEMPLATE_ROLES", []) or []:
|
|
key = _extractDe(role.get("description"))
|
|
if key and key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context="rbac.role", value="")
|
|
added += 1
|
|
|
|
for qa in getattr(mod, "QUICK_ACTIONS", []) or []:
|
|
for field in ("label", "description"):
|
|
key = _extractDe(qa.get(field))
|
|
if key and key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context="rbac.quickaction", value="")
|
|
added += 1
|
|
|
|
for cat in getattr(mod, "QUICK_ACTION_CATEGORIES", []) or []:
|
|
key = _extractDe(cat.get("label"))
|
|
if key and key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context="rbac.quickaction", value="")
|
|
added += 1
|
|
|
|
logger.info("i18n rbac labels: %d new keys (rbac.* context)", added)
|
|
|
|
|
|
def _registerServiceCenterLabels():
|
|
"""Register service-center category labels and bootstrap role descriptions."""
|
|
added = 0
|
|
|
|
def _extractDe(obj) -> str:
|
|
if isinstance(obj, str):
|
|
return obj
|
|
if isinstance(obj, dict):
|
|
return obj.get("de") or obj.get("en") or ""
|
|
return ""
|
|
|
|
try:
|
|
from modules.serviceCenter.registry import IMPORTABLE_SERVICES
|
|
for svc in IMPORTABLE_SERVICES.values():
|
|
key = _extractDe(svc.get("label"))
|
|
if key and key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context="service", value="")
|
|
added += 1
|
|
except ImportError:
|
|
pass
|
|
|
|
_bootstrapRoleDescriptions = [
|
|
"Administrator - Benutzer und Ressourcen im Mandanten verwalten",
|
|
"Benutzer - Standard-Benutzer mit Zugriff auf eigene Datensätze",
|
|
"Betrachter - Nur-Lese-Zugriff auf Gruppen-Datensätze",
|
|
"System-Administrator - Vollständiger administrativer Zugriff über alle Mandanten",
|
|
]
|
|
for desc in _bootstrapRoleDescriptions:
|
|
if desc not in _REGISTRY:
|
|
_REGISTRY[desc] = _I18nRegistryEntry(context="rbac.role", value="")
|
|
added += 1
|
|
|
|
logger.info("i18n service/bootstrap labels: %d new keys", added)
|
|
|
|
|
|
def _registerNodeLabels():
|
|
"""Register all graph-editor node labels, descriptions, parameter descriptions,
|
|
output labels, port descriptions, category labels, and entry-point titles."""
|
|
added = 0
|
|
|
|
def _extractDe(obj) -> str:
|
|
if isinstance(obj, str):
|
|
return obj
|
|
if isinstance(obj, dict):
|
|
return obj.get("de") or obj.get("en") or ""
|
|
return ""
|
|
|
|
def _reg(key: str, ctx: str):
|
|
nonlocal added
|
|
if key and key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context=ctx, value="")
|
|
added += 1
|
|
|
|
try:
|
|
from modules.features.graphicalEditor.nodeDefinitions import STATIC_NODE_TYPES
|
|
for nd in STATIC_NODE_TYPES:
|
|
_reg(_extractDe(nd.get("label")), "node.label")
|
|
_reg(_extractDe(nd.get("description")), "node.desc")
|
|
|
|
for param in nd.get("parameters", []) or []:
|
|
_reg(_extractDe(param.get("description")), "node.param")
|
|
_reg(_extractDe(param.get("label")), "node.param")
|
|
|
|
outLabels = nd.get("outputLabels")
|
|
if isinstance(outLabels, dict):
|
|
deList = outLabels.get("de") or outLabels.get("en") or []
|
|
for lbl in deList:
|
|
_reg(lbl, "node.output")
|
|
elif isinstance(outLabels, list):
|
|
for lbl in outLabels:
|
|
_reg(lbl, "node.output")
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
from modules.features.graphicalEditor.portTypes import PORT_TYPE_CATALOG
|
|
for schema in PORT_TYPE_CATALOG.values():
|
|
for field in getattr(schema, "fields", []) or []:
|
|
desc = getattr(field, "description", None)
|
|
if desc:
|
|
_reg(_extractDe(desc if isinstance(desc, (str, dict)) else None), "port.desc")
|
|
except ImportError:
|
|
pass
|
|
|
|
_nodeCategoryLabels = [
|
|
"Trigger", "Eingabe/Mensch", "Ablauf", "Daten", "KI",
|
|
"Datei", "E-Mail", "SharePoint", "ClickUp", "Treuhand",
|
|
]
|
|
for lbl in _nodeCategoryLabels:
|
|
_reg(lbl, "node.category")
|
|
|
|
_entryPointTitles = ["Jetzt ausführen", "Start"]
|
|
for lbl in _entryPointTitles:
|
|
_reg(lbl, "node.entry")
|
|
|
|
logger.info("i18n node labels: %d new keys (node.*/port.* context)", added)
|
|
|
|
|
|
def _registerDatamodelOptionLabels():
|
|
"""Register all frontend_options labels from Pydantic datamodels and subscription plans."""
|
|
added = 0
|
|
|
|
def _extractDe(obj) -> str:
|
|
if isinstance(obj, str):
|
|
return obj
|
|
if isinstance(obj, dict):
|
|
return obj.get("de") or obj.get("en") or ""
|
|
return ""
|
|
|
|
def _reg(key: str, ctx: str):
|
|
nonlocal added
|
|
if key and key not in _REGISTRY:
|
|
_REGISTRY[key] = _I18nRegistryEntry(context=ctx, value="")
|
|
added += 1
|
|
|
|
_datamodelModules = (
|
|
"modules.datamodels.datamodelRbac",
|
|
"modules.datamodels.datamodelChat",
|
|
"modules.datamodels.datamodelMessaging",
|
|
"modules.datamodels.datamodelNotification",
|
|
"modules.datamodels.datamodelUam",
|
|
"modules.datamodels.datamodelFiles",
|
|
"modules.datamodels.datamodelDataSource",
|
|
"modules.datamodels.datamodelFeatureDataSource",
|
|
"modules.datamodels.datamodelUiLanguage",
|
|
"modules.features.trustee.datamodelFeatureTrustee",
|
|
"modules.features.neutralization.datamodelFeatureNeutralizer",
|
|
)
|
|
|
|
for modPath in _datamodelModules:
|
|
try:
|
|
mod = __import__(modPath, fromlist=["__all__"])
|
|
except ImportError:
|
|
continue
|
|
for attrName in dir(mod):
|
|
cls = getattr(mod, attrName, None)
|
|
if not isinstance(cls, type) or not issubclass(cls, BaseModel):
|
|
continue
|
|
for fieldName, fieldInfo in cls.model_fields.items():
|
|
extra = (fieldInfo.json_schema_extra or {}) if hasattr(fieldInfo, "json_schema_extra") else {}
|
|
if not isinstance(extra, dict):
|
|
continue
|
|
options = extra.get("frontend_options")
|
|
if not isinstance(options, list):
|
|
continue
|
|
ctx = f"option.{cls.__name__}.{fieldName}"
|
|
for opt in options:
|
|
if isinstance(opt, dict):
|
|
_reg(_extractDe(opt.get("label")), ctx)
|
|
|
|
try:
|
|
from modules.datamodels.datamodelSubscription import BUILTIN_PLANS
|
|
for plan in BUILTIN_PLANS.values():
|
|
_reg(_extractDe(getattr(plan, "title", None)), "subscription.title")
|
|
_reg(_extractDe(getattr(plan, "description", None)), "subscription.desc")
|
|
except (ImportError, AttributeError):
|
|
pass
|
|
|
|
logger.info("i18n datamodel option labels: %d new keys", added)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Boot: sync registry to DB
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _syncRegistryToDb():
|
|
"""Boot hook: write all registered keys into UiLanguageSet(xx).
|
|
|
|
1. Scans route files for routeApiMsg("…") to eagerly register api.* keys.
|
|
2. Registers navigation labels as nav.* keys.
|
|
3. Registers feature UI labels (FEATURE_LABEL, UI_OBJECTS).
|
|
4. Registers RBAC labels (DATA/RESOURCE/ROLE/QuickAction).
|
|
5. Merges with existing UI keys (context="ui"), only touches gateway keys.
|
|
"""
|
|
_scanRouteApiMsgKeys()
|
|
_registerNavLabels()
|
|
_registerFeatureUiLabels()
|
|
_registerRbacLabels()
|
|
_registerServiceCenterLabels()
|
|
_registerNodeLabels()
|
|
_registerDatamodelOptionLabels()
|
|
|
|
if not _REGISTRY:
|
|
logger.info("i18n registry: no keys to sync (empty registry)")
|
|
return
|
|
|
|
from modules.datamodels.datamodelUiLanguage import UiLanguageSet
|
|
from modules.shared.configuration import APP_CONFIG
|
|
from modules.connectors.connectorDbPostgre import _get_cached_connector
|
|
from modules.shared.timeUtils import getUtcTimestamp
|
|
|
|
db = _get_cached_connector(
|
|
dbHost=APP_CONFIG.get("DB_HOST", "localhost"),
|
|
dbDatabase="poweron_management",
|
|
dbUser=APP_CONFIG.get("DB_USER"),
|
|
dbPassword=APP_CONFIG.get("DB_PASSWORD_SECRET"),
|
|
dbPort=int(APP_CONFIG.get("DB_PORT", 5432)),
|
|
userId="__i18n_boot__",
|
|
)
|
|
|
|
rows = db.getRecordset(UiLanguageSet, recordFilter={"id": "xx"})
|
|
|
|
gatewayEntries = [
|
|
{"context": entry.context, "key": key, "value": entry.value}
|
|
for key, entry in _REGISTRY.items()
|
|
]
|
|
gatewayKeys = set(_REGISTRY.keys())
|
|
|
|
if not rows:
|
|
now = getUtcTimestamp()
|
|
rec = {
|
|
"id": "xx",
|
|
"label": "Basisset (Meta)",
|
|
"entries": gatewayEntries,
|
|
"status": "complete",
|
|
"isDefault": True,
|
|
"sysCreatedAt": now,
|
|
"sysCreatedBy": "__i18n_boot__",
|
|
"sysModifiedAt": now,
|
|
"sysModifiedBy": "__i18n_boot__",
|
|
}
|
|
db.recordCreate(UiLanguageSet, rec)
|
|
logger.info("i18n boot-sync: created xx set with %d gateway keys", len(gatewayEntries))
|
|
return
|
|
|
|
row = dict(rows[0])
|
|
existingEntries: List[dict] = row.get("entries") or []
|
|
if not isinstance(existingEntries, list):
|
|
existingEntries = []
|
|
|
|
uiEntries = [e for e in existingEntries if e.get("context", "") == "ui"]
|
|
|
|
oldGatewayEntries = [
|
|
e for e in existingEntries
|
|
if e.get("context", "") != "ui"
|
|
]
|
|
oldGatewayByKey = {e["key"]: e for e in oldGatewayEntries}
|
|
|
|
added = 0
|
|
updated = 0
|
|
removed = 0
|
|
|
|
newGatewayEntries: List[dict] = []
|
|
for key, entry in _REGISTRY.items():
|
|
newEntry = {"context": entry.context, "key": key, "value": entry.value}
|
|
old = oldGatewayByKey.get(key)
|
|
if old is None:
|
|
added += 1
|
|
elif old.get("context") != entry.context or old.get("value") != entry.value:
|
|
updated += 1
|
|
newGatewayEntries.append(newEntry)
|
|
|
|
removed = len(set(oldGatewayByKey.keys()) - gatewayKeys)
|
|
|
|
mergedEntries = uiEntries + newGatewayEntries
|
|
|
|
if added == 0 and updated == 0 and removed == 0:
|
|
logger.info("i18n boot-sync: xx set up-to-date (%d gateway + %d ui keys)", len(newGatewayEntries), len(uiEntries))
|
|
return
|
|
|
|
now = getUtcTimestamp()
|
|
row["entries"] = mergedEntries
|
|
if "keys" in row:
|
|
del row["keys"]
|
|
row["sysModifiedAt"] = now
|
|
row["sysModifiedBy"] = "__i18n_boot__"
|
|
db.recordModify(UiLanguageSet, "xx", row)
|
|
|
|
logger.info(
|
|
"i18n boot-sync: xx updated (+%d added, ~%d updated, -%d removed, total=%d gateway + %d ui)",
|
|
added, updated, removed, len(newGatewayEntries), len(uiEntries),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Boot: load translation cache
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _loadCache():
|
|
"""Boot hook: load all UiLanguageSets into the in-memory cache.
|
|
|
|
After this, t() lookups are O(1) dict access with no DB calls.
|
|
"""
|
|
from modules.datamodels.datamodelUiLanguage import UiLanguageSet
|
|
from modules.shared.configuration import APP_CONFIG
|
|
from modules.connectors.connectorDbPostgre import _get_cached_connector
|
|
|
|
db = _get_cached_connector(
|
|
dbHost=APP_CONFIG.get("DB_HOST", "localhost"),
|
|
dbDatabase="poweron_management",
|
|
dbUser=APP_CONFIG.get("DB_USER"),
|
|
dbPassword=APP_CONFIG.get("DB_PASSWORD_SECRET"),
|
|
dbPort=int(APP_CONFIG.get("DB_PORT", 5432)),
|
|
userId="__i18n_cache__",
|
|
)
|
|
|
|
rows = db.getRecordset(UiLanguageSet)
|
|
_CACHE.clear()
|
|
|
|
for row in rows:
|
|
code = row.get("id", "")
|
|
if code == "xx":
|
|
continue
|
|
entries = row.get("entries")
|
|
if not isinstance(entries, list):
|
|
continue
|
|
langDict: Dict[str, str] = {}
|
|
for e in entries:
|
|
key = e.get("key", "")
|
|
val = e.get("value", "")
|
|
if key and val:
|
|
langDict[key] = val
|
|
if langDict:
|
|
_CACHE[code] = langDict
|
|
|
|
logger.info("i18n cache loaded: %d languages, %d total keys",
|
|
len(_CACHE), sum(len(v) for v in _CACHE.values()))
|