wagtail localize translate page programmatically with deepl

[tchenevi]

I'm trying to automate the translation of a Wagtail website with hundreds of pages. My goal is to create a Django command to translate all the pages with deepl. I can't find any documentation about it. Anyone as ever encountered the same issue ?

I'm using Wagtail 5.2 and wagtail-localize 1.7

[Nigel2392]

Yeah sure - this should work with any wagtail-localize backend to translate all .po files in your project. It does need some editing however - I recommend using the Translation.export_po method to get the PO files for easy translation from each page.

I'll walk you through the basics -

First we must get the wagtail_localize.models.TranslationSource instance belonging to that page.

Now we can fetch the translations for that source - TranslationSource.translations.all() should give you these.

Loop over the translations
Get the po file for each translation
Run it through the translator
Use translation.import_po to re-import the PO file.

a rough sketch, the fields to filter should be correct; not paying attentions to methods as I'm explaining this.
To get more details, I recommend you look at the source defined in wagtail_localize/models.py

for p in page.objects.all.specific:
    source = TranslationSource.objects.get_for_instance(p)
    for translatable in source.translations.all():
            po_file = translatable.export_po
            # ... translating here
            translatable.import_po

Edit

This seems like something I'd like to use myself in the future so I built the main bit of functionality.
The following should work.

     def translate_pages(self, data = None, force_translate = False):
        if data:
            page_models = [apps.get_model(model) for model in data]
        else:
            page_models = [Page]

        for PageClass in page_models:
            for page in PageClass.objects.all():
                self.stdout.write(f"Translating {page}")
                try:
                    source: TranslationSource = TranslationSource.objects.get_for_instance(page)
                    if not source:
                        self.stderr.write(f"No translation source found for {page}")
                        continue
                except (TranslationSource.DoesNotExist, TranslatableObject.DoesNotExist):
                    self.stderr.write(f"No translation source found for {page}")
                    continue

                source_locale: Locale = source.locale
                translations: list[Translation] = source.translations.all()
                for translation in translations:
                    po_file = translation.export_po()
                    if not po_file:
                        self.stderr.write(f"No PO file found for {translation}")
                        continue

                    self.stdout.write(f"Translating {po_file} from {source_locale} to {translation.target_locale}")
                    
                    self.translate_po(
                        source_locale=source_locale, 
                        target_locale=translation.target_locale, 
                        po=po_file,
                        force_translate=force_translate,
                    )

                    self.stdout.write(f"Saving PO file to page {page}")
                    warnings = translation.import_po(
                        po=po_file,
                        delete=False,
                        user=None,
                        translation_type="machine",
                        tool_name=self.translator.display_name,
                    )

                    if warnings:
                        for warning in warnings:
                            self.stderr.write(f"Warning received importing {po_file}: {warning}")

You should also edit your command's handle method.

    def handle(self, *args, **options):
        self.translator: BaseMachineTranslator = get_machine_translator()

        
        if options["project"]:
            self.stdout.write("Translating all PO files in project")
            return self.translate_project(options=options)
        
        
        if options["pages"]:
            self.stdout.write("Translating all pages in project")
            return self.translate_pages(options["data"], options["force"])

        self.data = options["data"]
        if not self.data:
            self.stderr.write("No files specified")
            return

        .......

Command code

Here is the code I used for the command - again, you should edit this to fit your use-case.

The method you should be most interested in from the command-code is probably translate_po

from django.core.management.base import BaseCommand, CommandParser
from django.conf import settings
from django.utils import timezone
from wagtail.models import Locale
from wagtail_localize.machine_translators import get_machine_translator
from wagtail_localize.models import Translation
import polib, os, glob

from wagtail_localize.machine_translators.base import BaseMachineTranslator
from wagtail_localize.strings import StringValue

def get_translation_locale(po: polib.POFile, options: dict = None) -> tuple[Translation, Locale, Locale]:
    translation = None
    # Retrieve source and target locale from PO file metadata
    # Optionally retrieve the translation object
    if po and ("X-WagtailLocalize-TranslationID" in po.metadata):
        translation = po.metadata["X-WagtailLocalize-TranslationID"]
        translation = Translation.objects.get(uuid=translation)
        source_locale = Locale.objects.get(language_code=translation.source.locale.language_code)
        target_locale = Locale.objects.get(language_code=translation.target_locale.language_code)

    # Source and target provided by custom attributes in PO file metadata
    elif po and ("X-Src-Lang" in po.metadata and "X-Dst-Lang" in po.metadata):
        source_locale = Locale.objects.get(language_code=po.metadata["X-Src-Lang"])
        target_locale = Locale.objects.get(language_code=po.metadata["X-Dst-Lang"])

    elif po and ("X-Machine-Translator-Source" in po.metadata and "X-Machine-Translator-Target" in po.metadata):
        source_locale = Locale.objects.get(language_code=po.metadata["X-Machine-Translator-Source"])
        target_locale = Locale.objects.get(language_code=po.metadata["X-Machine-Translator-Target"])

    # Source and target provided by command line argument
    elif options and ("source" in options and "target" in options) and (options["source"] and options["target"]):
        source_locale = Locale(language_code=options["source"])
        target_locale = Locale(language_code=options["target"])

    elif options and ("target" in options) and (options["target"]):
        source = settings.LANGUAGE_CODE
        target = options["target"]

        source_locale = Locale(language_code=source)
        target_locale = Locale(language_code=target)

    # Source and target does not exist
    else:
        raise RuntimeError(f"Could not determine source and target locale for {po.fpath}, please specify --source and --target")

    return translation, source_locale, target_locale


class POTranslationEntry:
    def __init__(self, index: int, value: StringValue, plural: bool = False):
        self.index = index
        self.value = value
        self.plural = plural

    def scan(self, po: polib.POFile, translated: StringValue):
        if self.plural:
            po[self.index].msgstr = translated.data
        else:
            po[self.index].msgstr = translated.render_text()

    def __hash__(self) -> int:
        return hash(self.value)
    

class Command(BaseCommand):
    help = "Run a wagtail_localize machine translator on a PO file based on settings defined in settings.py"

    def add_arguments(self, parser: CommandParser) -> None:
        parser.add_argument("-d", "--data", type=str, default=None, nargs="+")
        parser.add_argument("-s", "--source", type=str, default=None, help="Source locale")
        parser.add_argument("-t", "--target", type=str, default=None, help="Target locale")
        parser.add_argument("-f", "--force", type=bool, default=False, help="Force translate all entries, even if already translated")
        parser.add_argument("-p", "--project", type=bool, default=False, help="Translate all PO files in project directory")
        return super().add_arguments(parser)

    def handle(self, *args, **options):
        self.translator: BaseMachineTranslator = get_machine_translator()

        
        if options["project"]:
            self.stdout.write("Translating all PO files in project")
            return self.translate_project(options=options)
        
        self.data = options["data"]
        if not self.data:
            self.stderr.write("No files specified")
            return

        for file in self.data:
            try:
                po = polib.pofile(file)

                filename = os.path.splitext(os.path.basename(file))[0]
                file_dir = os.path.dirname(file)

                translation, source_locale, target_locale = get_translation_locale(
                    po, options=options,
                )

                self.stdout.write(f"Translating {file} from {source_locale} to {target_locale}")

                # Check if translator supports source and target locale
                if not self.translator.can_translate(source_locale, target_locale):
                    self.stderr.write(f"Translator {self.translator.display_name} does not support {source_locale} to {target_locale}: {file}")
                    continue

                # Range over all entries in PO file, if set a check if any entries are translated
                any_traslated = self.translate_po(
                    source_locale=source_locale, 
                    target_locale=target_locale, 
                    po=po,
                    force_translate=options["force"],
                )
                if not any_traslated:
                    self.stderr.write(f"No entries translated in {file}")
                    continue

                # Save PO file
                self.stdout.write(f"Saving {file}")
                po.save(
                    os.path.join(
                        file_dir, 
                        f"{filename}_machine_translated.po",
                    ),
                )

                # wagtail_localize translation exists, import PO file
                if translation:
                    
                    self.stdout.write(f"Importing {file} into {translation}")
                    warnings = translation.import_po(
                        po, 
                        delete=False, 
                        user=None, 
                        translation_type="machine", 
                        tool_name=self.translator.display_name,
                    )

                    if warnings:
                        for warning in warnings:
                            self.stderr.write(f"Warning received importing {file}: {warning}")
                else:
                    self.stdout.write(f"No wagtail_localize translation found for {file}, skipping PO import")

            # Catch exceptions and log errors to user.
            except OSError as e:
                self.stderr.write(f"Could not open: {file} ({e})")
                continue
            except Locale.DoesNotExist as e:
                self.stderr.write(f"Could not find locale {source_locale} or {target_locale}: {file} ({e})")
                continue
            except Translation.DoesNotExist as e:
                self.stderr.write(f"Could not find translation {translation}: {file} ({e})")
                continue


    def translate_project(self, options: dict):

        _, source_locale, target_locale = get_translation_locale(
            po=None, 
            options=options,
        )

        if not source_locale or not target_locale:
            self.stderr.write("Please specify --source and --target")
            return
        
        fpath = os.path.join(
            settings.BASE_DIR,
            f"**/locale/{target_locale.language_code}/LC_MESSAGES/*.po"
        )

        fpath = fpath.replace("\\", os.sep)
        fpath = fpath.replace("/", os.sep)

        self.stdout.write(f"Translating all PO files in project from {source_locale} to {target_locale}, looking in {fpath}")

        globbed = glob.glob(fpath, recursive=True)
        if not globbed:
            self.stderr.write(f"No PO files found in {fpath}")
            return
        
        self.stdout.write(f"Found {len(globbed)} PO files in {fpath}")
        for file in globbed:
            self.stdout.write(f"\tTranslating {file}")

        import threading, math
        def _translate_po(cmd, globbed, source_locale, target_locale, force_translate=False):
            for file in globbed:
                po = polib.pofile(file)

                translated = cmd.translate_po(
                    source_locale=source_locale, 
                    target_locale=target_locale, 
                    po=po,
                    force_translate=force_translate,
                )

                if translated:
                    cmd.stdout.write(f"Saving translations for {file}")
                    po.save(file)

                else:
                    cmd.stderr.write(f"No entries translated in {file}")
        
        threads = []
        max_threads = 5
        batch_size = math.ceil(len(globbed) / max_threads)
        if batch_size < 1:
            batch_size = 1

        for i in range(0, len(globbed), batch_size):
            batch = globbed[i:i+batch_size]
            self.stdout.write(f"Starting thread for batch {i + 1} of {len(globbed)}")
            t = threading.Thread(target=_translate_po, args=(
                self, 
                batch, 
                source_locale, 
                target_locale,
                options["force"],
            ))
            t.start()
            threads.append(t)

        for t in threads:
            t.join()

        self.stdout.write("Done translating project")
        



    def translate_po(self, source_locale: Locale, target_locale: Locale, po: polib.POFile, add_metadata: bool = True, force_translate: bool = False) -> bool:
        any_traslated = False

        strings: dict[POTranslationEntry, StringValue] = {}

        for index in range(len(po)):
            entry = po[index]
            
            # Skip already translated entries
            if not force_translate and entry.translated():
                # if settings.DEBUG:
                #     self.stdout.write(f"Skipping entry {index + 1} in {file}")
                continue
                
            v = StringValue(entry.msgid)
            strings[POTranslationEntry(index, v)] = v
            if entry.msgid_plural:
                v = StringValue(entry.msgid_plural)
                strings[POTranslationEntry(index, v, plural=True)] = v


        self.stdout.write(f"Translating {len(strings)} strings from {source_locale} to {target_locale} in {po.fpath}")
        # Translate all strings
        translations: dict = self.translator.translate(
            source_locale=source_locale, 
            target_locale=target_locale, 
            strings=list(strings.values()),
        )

        for entry, string in strings.items():
            try:
                translated = translations[string]
            except KeyError:
                raise ValueError("You must use the source string as dictionary keys when returning from backend.translate()")

            entry.scan(po, translated)
            any_traslated = True

        # Set PO file metadata
        if add_metadata:
            po.metadata["X-Machine-Translator"] = self.translator.display_name
            po.metadata["X-Machine-Translator-Creation"] = str(timezone.now().isoformat())
            po.metadata["X-Machine-Translator-Source"] = source_locale.language_code
            po.metadata["X-Machine-Translator-Target"] = target_locale.language_code

        return any_traslated

[tchenevi]

Thanks for your answer ! It gave me some ideas.
I'm still struggling a bit because I can't get my changes to get published but I think I'm on the right trackss.
Here is the code I wrote :

from django.core.management.base import BaseCommand
import json

from collections import defaultdict
from django.db import models, transaction
from django.shortcuts import get_object_or_404, redirect, render

from wagtail_localize.machine_translators import get_machine_translator
from wagtail_localize.models import (
    StringTranslation,
    Translation,
)
from wagtail_localize.segments import StringSegmentValue


def machine_translate(translation_id):
    translation = get_object_or_404(Translation, id=translation_id)
    translator = get_machine_translator()

    # Get segments
    segments = defaultdict(list)
    for string_segment in translation.source.stringsegment_set.all().select_related(
        "context", "string"
    ):
        segment = StringSegmentValue(
            string_segment.context.path, string_segment.string.as_value()
        ).with_order(string_segment.order)
        if string_segment.attrs:
            segment.attrs = json.loads(string_segment.attrs)

        # Don't translate if there already is a translation
        if StringTranslation.objects.filter(
            translation_of_id=string_segment.string_id,
            locale=translation.target_locale,
            context_id=string_segment.context_id,
        ).exists():
            continue

        segments[segment.string].append(
            (string_segment.string_id, string_segment.context_id)
        )

    if segments:
        translations = translator.translate(
            translation.source.locale, translation.target_locale, segments.keys()
        )

        with transaction.atomic():
            for string, contexts in segments.items():
                for string_id, context_id in contexts:
                    StringTranslation.objects.get_or_create(
                        translation_of_id=string_id,
                        locale=translation.target_locale,
                        context_id=context_id,
                        defaults={
                            "data": translations[string].data,
                            "translation_type": StringTranslation.TRANSLATION_TYPE_MACHINE,
                            "tool_name": translator.display_name,
                            # "last_translated_by": User.objects.get(username="admin"),
                            "has_error": False,
                            "field_error": "",
                        },
                    )
        print("Successfully translated with {}.".format(translator.display_name))


    else:
        print("There isn't anything left to translate.")


class Command(BaseCommand):
    help = "Translate all pages if not already translated."

    # def add_arguments(self, parser):
    #     parser.add_argument("translation_id", nargs="+", type=int)
    def publish_translated_page(self, translation):
        # Récupérer la page traduite
        translated_page = translation.get_target_instance()

        # Vérifier si la page peut être publiée
        if translated_page.has_unpublished_changes:
            translated_page.save_revision().publish()
            self.stdout.write(
                self.style.SUCCESS(f'Page "{translated_page}" published in locale {translation.target_locale}'))

    def translate_page(self, translation):
        # translation = get_object_or_404(Translation, id=kwargs["translation_id"][0])
        translator = get_machine_translator()

        # Get segments
        segments = defaultdict(list)
        for string_segment in translation.source.stringsegment_set.all().select_related(
            "context", "string"
        ):
            segment = StringSegmentValue(
                string_segment.context.path, string_segment.string.as_value()
            ).with_order(string_segment.order)
            if string_segment.attrs:
                segment.attrs = json.loads(string_segment.attrs)

            # Don't translate if there already is a translation
            if StringTranslation.objects.filter(
                translation_of_id=string_segment.string_id,
                locale=translation.target_locale,
                context_id=string_segment.context_id,
            ).exists():
                continue

            segments[segment.string].append(
                (string_segment.string_id, string_segment.context_id)
            )

        if segments:
            translations = translator.translate(
                translation.source.locale, translation.target_locale, segments.keys()
            )

            with transaction.atomic():
                for string, contexts in segments.items():
                    for string_id, context_id in contexts:
                        StringTranslation.objects.get_or_create(
                            translation_of_id=string_id,
                            locale=translation.target_locale,
                            context_id=context_id,
                            defaults={
                                "data": translations[string].data,
                                "translation_type": StringTranslation.TRANSLATION_TYPE_MACHINE,
                                "tool_name": translator.display_name,
                                # "last_translated_by": User.objects.get(username="admin"),
                                "has_error": False,
                                "field_error": "",
                            },
                        )
            self.stdout.write(
                self.style.SUCCESS(
                    f"{translation.get_target_instance().title} - {translation.get_target_instance().pk} - Successfully translated with {translator.display_name}.")
            )

            self.publish_translated_page(translation)

        else:
            self.stdout.write(
                self.style.WARNING(
                    f"{translation.get_target_instance().title} - {translation.get_target_instance().pk}- There isn't anything left to translate.")
            )

    def handle(self, *args, **options):
        # translation = get_object_or_404(Translation, id=kwargs["translation_id"][0])
        for translation in Translation.objects.all():
            self.translate_page(translation)

[Nigel2392]

Hey! looks great!!! You sure did your research on how this package works, great! :)
I'll look at it later - got a lot going on right now.
I'd still recommend using the Translation.import_po and Translation.export_po methods, and letting wagtail_localize do all the import/export/publishing logic, saves you a lot of hassle. It's very easy!

Here is an example of a standalone script for translating pages, you can literally just import this into your command and use it, very easy to follow too! We'd want to pass as much as it off as we can to wagtail_localize itself. What if there's an update for example?

Anyways, here you go:

for translation in Translation.objects.all():
    po_file = translation.export_po()
    if not po_file:
        self.stderr.write(f"No PO file found for {translation}")
        continue
    self.stdout.write(f"Translating {po_file} from {translation.source.locale} to {translation.target_locale}")
    
    entries = {}
    # Pages will never have pluralized strings, we can skip any checks for them.
    for index in range(len(po_file)):
        entry = po_file[index]
        
        # Skip already translated entries
        if not force_translate and entry.translated():
            # if settings.DEBUG:
            #     self.stdout.write(f"Skipping entry {index + 1} in {file}")
            continue
            
        entries[entry.msgid] = StringValue(entry.msgid)
    translator: BaseMachineTranslator = get_machine_translator()
    translated: dict[StringValue, StringValue] = translator.translate(
        source_locale=translation.source.locale, 
        target_locale=translation.target_locale, 
        strings=list(entries.values()),
    )
    for index in range(len(po_file)):
        entry = po_file[index]
        string = entries[entry.msgid]
        translated = translated[string]
        entry.msgstr = translated.data
    self.stdout.write(f"Saving PO file to object {translation.source.specific_content_type} ({translation.source.object_id})
    warnings = translation.import_po(
        po=po_file,
        delete=False,
        user=None,
        translation_type="machine",
        tool_name=self.translator.display_name,
    )
    if warnings:
        for warning in warnings:
            self.stderr.write(f"Warning received importing {po_file}: {warning}")