Kincses
425f598fa3
- Fixed AttributeError in User model (added region_code, preferred_language) - Fixed InvalidRequestError in AssetAssignment (added organization relationship) - Configured STATIC_DIR for translation sync - Applied Alembic migrations for user schema updates
105 lines
3.9 KiB
Python
Executable File
105 lines
3.9 KiB
Python
Executable File
import json
|
|
import os
|
|
import logging
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, update
|
|
from app.models.translation import Translation
|
|
from app.core.config import settings
|
|
from typing import Dict, Any, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class TranslationService:
|
|
# Memória-cache a szerveroldali hibaüzenetekhez és emailekhez
|
|
_published_cache: Dict[str, Dict[str, str]] = {}
|
|
|
|
@classmethod
|
|
async def load_cache(cls, db: AsyncSession):
|
|
"""Betölti a publikált szövegeket a memóriába az adatbázisból."""
|
|
result = await db.execute(
|
|
select(Translation).where(Translation.is_published == True)
|
|
)
|
|
translations = result.scalars().all()
|
|
|
|
cls._published_cache = {}
|
|
for t in translations:
|
|
if t.lang_code not in cls._published_cache:
|
|
cls._published_cache[t.lang_code] = {}
|
|
cls._published_cache[t.lang_code][t.key] = t.value
|
|
logger.info(f"🌍 i18n Cache: {len(translations)} szöveg betöltve.")
|
|
|
|
@classmethod
|
|
def get_text(cls, key: str, lang: str = "hu", variables: Optional[Dict[str, Any]] = None) -> str:
|
|
"""
|
|
Szerveroldali lekérés Fallback (EN) logikával és változó behelyettesítéssel.
|
|
Példa: get_text("AUTH.WELCOME", "hu", {"name": "Péter"})
|
|
"""
|
|
# 1. Kért nyelv lekérése
|
|
text = cls._published_cache.get(lang, {}).get(key)
|
|
|
|
# 2. Fallback angolra, ha nincs meg a kért nyelven
|
|
if not text and lang != "en":
|
|
text = cls._published_cache.get("en", {}).get(key)
|
|
|
|
# 3. Ha sehol nincs meg, adjuk vissza a kulcsot
|
|
if not text:
|
|
return f"[{key}]"
|
|
|
|
# 4. Változók behelyettesítése (pl. {{name}})
|
|
if variables:
|
|
for k, v in variables.items():
|
|
text = text.replace(f"{{{{{k}}}}}", str(v))
|
|
|
|
return text
|
|
|
|
@classmethod
|
|
async def publish_all(cls, db: AsyncSession):
|
|
"""Minden piszkozatot élesít, frissíti a memóriát és legenerálja a JSON-öket."""
|
|
await db.execute(
|
|
update(Translation).where(Translation.is_published == False).values(is_published=True)
|
|
)
|
|
await db.commit()
|
|
await cls.load_cache(db)
|
|
await cls.export_to_json(db)
|
|
|
|
@staticmethod
|
|
async def export_to_json(db: AsyncSession):
|
|
"""
|
|
Adatbázis -> Hierarchikus JSON export.
|
|
'AUTH.LOGIN.TITLE' -> { "AUTH": { "LOGIN": { "TITLE": "..." } } }
|
|
"""
|
|
stmt = select(Translation).where(Translation.is_published == True)
|
|
result = await db.execute(stmt)
|
|
translations = result.scalars().all()
|
|
|
|
languages: Dict[str, Any] = {}
|
|
for t in translations:
|
|
if t.lang_code not in languages:
|
|
languages[t.lang_code] = {}
|
|
|
|
# Hierarchikus struktúra felépítése
|
|
parts = t.key.split('.')
|
|
current_level = languages[t.lang_code]
|
|
for part in parts[:-1]:
|
|
if part not in current_level:
|
|
current_level[part] = {}
|
|
current_level = current_level[part]
|
|
|
|
current_level[parts[-1]] = t.value
|
|
|
|
# Fájlok mentése
|
|
locales_path = os.path.join(settings.STATIC_DIR, "locales")
|
|
os.makedirs(locales_path, exist_ok=True)
|
|
|
|
for lang, content in languages.items():
|
|
file_path = os.path.join(locales_path, f"{lang}.json")
|
|
try:
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
json.dump(content, f, ensure_ascii=False, indent=2)
|
|
logger.info(f"🚀 JSON legenerálva: {file_path}")
|
|
except Exception as e:
|
|
logger.error(f"Fájl hiba ({lang}): {str(e)}")
|
|
|
|
return True
|
|
|
|
translation_service = TranslationService() |