translate.py

import logging
import os
import sys
from typing import Any

from openslides_backend.i18n.translator import Translator
from openslides_backend.models.models import Organization
from openslides_backend.services.database.extended_database import ExtendedDatabase
from openslides_backend.services.postgresql.db_connection_handling import (
    get_new_os_conn,
)
from openslides_backend.shared.env import Environment
from openslides_backend.shared.interfaces.write_request import (
    RequestUpdateEvent,
    WriteRequest,
)
from openslides_backend.shared.patterns import fqid_from_collection_and_id
from openslides_backend.shared.util import ONE_ORGANIZATION_FQID

env = Environment(os.environ)
collection_to_fields_map = {
    "organization": [
        "name",
        "login_text",
        "description",
    ],
    "meeting": [
        "name",
        "welcome_title",
        "welcome_text",
        "motion_preamble",
        "motions_export_title",
        "assignments_export_title",
        "users_pdf_welcometitle",
        "users_pdf_welcometext",
        "users_email_sender",
        "users_email_subject",
        "users_email_body",
    ],
    "group": ["name"],
    "motion_workflow": ["name"],
    "motion_state": ["name", "recommendation_label"],
    "projector_countdown": ["name"],
    "projector": ["name"],
    "motion": ["recommendation_label"],
}
possible_languages = Organization().default_language.constraints["enum"]


def read_collection(collection: str, fields: list[str]) -> Any:
    with get_new_os_conn() as conn:
        response = ExtendedDatabase(conn, logging, env).get_all(
            collection, ["id", *fields]
        )

    return response.items()


def check_language(language: str) -> None:
    if language not in possible_languages:
        print("language is not allowed.")
        print_help()
        sys.exit(2)


def check_organization_language() -> None:
    with get_new_os_conn() as conn:
        response = ExtendedDatabase(conn, logging, env).get(
            ONE_ORGANIZATION_FQID, ["default_language"]
        )

    if response["default_language"] != "en":
        print("Cannot translate from source languages other than `en`.")
        print_help()
        sys.exit(3)


def print_help() -> None:
    print("Usage:  python translate.py <language>")
    print("     Translates from 'en' to <language>.")
    print(f"""     language could be '{"', '".join(possible_languages)}'.""")


def main() -> None:
    if len(sys.argv) != 2:
        print_help()
        sys.exit(1)

    language = sys.argv[1]
    check_language(language)
    Translator.set_translation_language(language)

    check_organization_language()

    # translate and generate events
    events = []
    for collection in collection_to_fields_map:
        fields = collection_to_fields_map[collection]
        for id_, model in read_collection(collection, fields):
            changed_fields: dict[str, Any] = {}
            for field in fields:
                old_value = model[field]
                possible_new_value = Translator.translate(old_value)
                if possible_new_value != old_value and possible_new_value is not None:
                    changed_fields[field] = possible_new_value
            if collection == "organization":
                changed_fields["default_language"] = language
            elif collection == "meeting":
                changed_fields["language"] = language
            if changed_fields:
                events.append(
                    RequestUpdateEvent(
                        fqid_from_collection_and_id(collection, id_),
                        changed_fields,
                    )
                )

    # write events into the datastore
    if events:
        write_request = WriteRequest(events, None, 0, {})  # type: ignore
        with get_new_os_conn() as conn:
            ExtendedDatabase(conn, logging, env).write([write_request])


if __name__ == "__main__":
    main()