Coverage for baibu/localization/services.py: 100%
229 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-10-06 13:07 +0000
1"""The translation workflow: source strings, drafts, publishing and restoring.
3Files (templates, drafts, published catalogues) are stored through
4:mod:`baibu.core.storage`; the database keeps their keys and the audit trail.
5Storage keys are never overwritten, so saving a draft writes a new file and
6removes the previous one once the database change is committed.
7"""
9import hashlib
10import logging
11import uuid
12from dataclasses import dataclass
13from dataclasses import field
14from pathlib import Path
16import polib
17from django.conf import settings
18from django.db import transaction
19from django.db.models import Max
20from django.utils import timezone
21from django.utils.translation import to_locale
23from baibu.core import storage
25from . import catalogues
26from . import extraction
27from . import runtime
28from .models import TranslationCatalogue
29from .models import TranslationPublication
30from .models import TranslationSource
32logger = logging.getLogger(__name__)
34SOURCE_LANGUAGE = "en"
35AREA = "translations"
38class NoSourceError(Exception):
39 """Source strings have not been extracted yet."""
42class UnknownLanguageError(LookupError):
43 """The language is not one this deployment translates."""
46@dataclass
47class SourceUpdate:
48 source: TranslationSource
49 created: bool
50 merged: list[str] = field(default_factory=list)
53@dataclass
54class Edit:
55 """A translator's change to one entry."""
57 key: str
58 # One value for an ordinary entry; one per plural form otherwise.
59 values: list[str]
60 fuzzy: bool
61 # Fingerprint of the entry as the translator saw it (see ``fingerprint``).
62 seen: str
65@dataclass
66class SaveResult:
67 changed: int = 0
68 conflicts: list[str] = field(default_factory=list)
69 errors: dict[str, str] = field(default_factory=dict)
72def translatable_languages() -> list[tuple[str, str]]:
73 """The deployment's languages other than the source language (English)."""
74 return [(code, name) for code, name in settings.LANGUAGES if code != SOURCE_LANGUAGE]
77def check_language(language_code: str) -> None:
78 if language_code not in dict(translatable_languages()):
79 raise UnknownLanguageError(language_code)
82def latest_source() -> TranslationSource | None:
83 return TranslationSource.objects.order_by("-created_at").first()
86def _read_po(key: str) -> polib.POFile:
87 data = storage.read_bytes(key)
88 if data is None:
89 msg = f"Stored catalogue {key} is missing."
90 raise FileNotFoundError(msg)
91 return catalogues.parse(data)
94def read_template(source: TranslationSource) -> polib.POFile:
95 return _read_po(source.pot_key)
98def read_draft(catalogue: TranslationCatalogue) -> polib.POFile:
99 return _read_po(catalogue.draft_key)
102def _save_draft(catalogue: TranslationCatalogue, po: polib.POFile, *, user=None, source=None) -> None:
103 """Store ``po`` as the catalogue's draft and update the row (inside a transaction)."""
104 old_key = catalogue.draft_key
105 new_key = storage.save_bytes(
106 storage.build_key(f"{AREA}/{catalogue.language_code}/drafts", f"{uuid.uuid7()}.po"), catalogues.dump(po)
107 )
108 catalogue.draft_key = new_key
109 catalogue.updated_at = timezone.now()
110 catalogue.updated_by = user
111 if source is not None:
112 catalogue.source = source
113 try:
114 catalogue.save()
115 except Exception:
116 storage.delete(new_key)
117 raise
118 if old_key and old_key != new_key:
119 transaction.on_commit(lambda: storage.delete(old_key))
122# Source strings --------------------------------------------------------------------
125def update_sources(*, user=None, roots=None, force: bool = False) -> SourceUpdate:
126 """Extract the interface strings from the code and merge them into every draft.
128 An extraction identical to the current template is not stored again
129 (unless ``force``), but drafts that are behind are still brought up to date.
130 """
131 template = extraction.extract(roots)
132 digest = catalogues.content_hash(template)
133 current = latest_source()
134 created = force or current is None or current.content_hash != digest
135 with transaction.atomic():
136 if created:
137 source_id = uuid.uuid7()
138 key = storage.save_bytes(
139 storage.build_key(f"{AREA}/sources", f"{source_id}.pot"), catalogues.dump(template)
140 )
141 try:
142 current = TranslationSource.objects.create(
143 id=source_id,
144 pot_key=key,
145 content_hash=digest,
146 entry_count=len(template),
147 created_by=user,
148 )
149 except Exception:
150 storage.delete(key)
151 raise
152 merged = [
153 code
154 for code, _name in translatable_languages()
155 if _bring_up_to_date(code, current, template=template, user=user)
156 ]
157 logger.info("Source strings %s (%d entries); drafts updated: %s", current.pk, len(template), merged or "none")
158 return SourceUpdate(source=current, created=created, merged=merged)
161def _deployment_catalogue(language_code: str) -> polib.POFile | None:
162 """The deployment's own .po file for the language, if it has one."""
163 path = Path(settings.DEPLOYMENT_DIR) / "locale" / to_locale(language_code) / "LC_MESSAGES" / "django.po"
164 if not path.is_file():
165 return None
166 try:
167 return catalogues.parse(path.read_bytes())
168 except catalogues.CatalogueError:
169 logger.warning("Ignoring %s: not a valid PO file", path)
170 return None
173def _bring_up_to_date(language_code: str, source: TranslationSource, *, template=None, user=None) -> bool:
174 """Create the language's draft, or merge a newer template into it. Returns whether it changed."""
175 catalogue = TranslationCatalogue.objects.select_for_update().filter(language_code=language_code).first()
176 if catalogue is not None and catalogue.source_id == source.pk:
177 return False
178 template = template if template is not None else read_template(source)
179 if catalogue is None:
180 po = catalogues.new_catalogue(template, language_code=language_code, base=_deployment_catalogue(language_code))
181 catalogue = TranslationCatalogue(language_code=language_code)
182 else:
183 po = catalogues.merge_template(read_draft(catalogue), template)
184 _save_draft(catalogue, po, user=user, source=source)
185 return True
188def get_catalogue(language_code: str, *, user=None) -> TranslationCatalogue:
189 """The language's catalogue, created or brought up to date with the newest template."""
190 check_language(language_code)
191 source = latest_source()
192 if source is None:
193 raise NoSourceError
194 with transaction.atomic():
195 _bring_up_to_date(language_code, source, user=user)
196 return TranslationCatalogue.objects.get(language_code=language_code)
199# Editing ---------------------------------------------------------------------------
202def fingerprint(entry: polib.POEntry) -> str:
203 """A short digest of an entry's translation, to detect someone else's change."""
204 return hashlib.sha256(repr((_values(entry), entry.fuzzy)).encode()).hexdigest()[:16]
207def _values(entry: polib.POEntry) -> list[str]:
208 if entry.msgid_plural:
209 return [entry.msgstr_plural[index] for index in sorted(entry.msgstr_plural)]
210 return [entry.msgstr]
213def _apply(entry: polib.POEntry, edit: Edit) -> bool:
214 """Apply an edit to an entry. Returns whether anything changed."""
215 before = fingerprint(entry)
216 if entry.msgid_plural:
217 for index in sorted(entry.msgstr_plural):
218 entry.msgstr_plural[index] = edit.values[index] if index < len(edit.values) else ""
219 else:
220 entry.msgstr = edit.values[0] if edit.values else ""
221 entry.fuzzy = edit.fuzzy and bool(any(edit.values))
222 if not entry.fuzzy:
223 entry.previous_msgid = entry.previous_msgid_plural = entry.previous_msgctxt = None
224 return fingerprint(entry) != before
227def save_entries(language_code: str, edits: list[Edit], *, user) -> SaveResult:
228 """Apply translators' edits to the draft.
230 Entries someone else changed since the page was loaded are not
231 overwritten (reported in ``conflicts``); translations that would break a
232 format string are refused (reported in ``errors``).
233 """
234 check_language(language_code)
235 result = SaveResult()
236 with transaction.atomic():
237 catalogue = TranslationCatalogue.objects.select_for_update().get(language_code=language_code)
238 po = read_draft(catalogue)
239 for edit in edits:
240 entry = catalogues.find_entry(po, edit.key)
241 if entry is None:
242 result.conflicts.append(edit.key)
243 continue
244 values = [value.replace("\r\n", "\n") for value in edit.values]
245 if values == _values(entry) and edit.fuzzy == entry.fuzzy:
246 continue
247 if fingerprint(entry) != edit.seen:
248 result.conflicts.append(edit.key)
249 continue
250 if problem := catalogues.placeholder_problem(entry, *values):
251 result.errors[edit.key] = problem
252 continue
253 if _apply(entry, Edit(edit.key, values, edit.fuzzy, edit.seen)):
254 result.changed += 1
255 if result.changed:
256 _save_draft(catalogue, po, user=user)
257 return result
260def set_plural_forms(language_code: str, value: str, *, user) -> None:
261 catalogues.parse_plural_forms(value)
262 check_language(language_code)
263 with transaction.atomic():
264 catalogue = TranslationCatalogue.objects.select_for_update().get(language_code=language_code)
265 po = read_draft(catalogue)
266 catalogues.set_header(po, language_code=language_code, plural_forms_value=value.strip())
267 _save_draft(catalogue, po, user=user)
270def import_catalogue(language_code: str, data: bytes, *, user) -> SaveResult:
271 """Take translations from an uploaded .po file for messages the draft has.
273 Messages not in the draft are ignored; translations that would break a
274 format string are skipped.
275 """
276 check_language(language_code)
277 uploaded = catalogues.parse(data)
278 result = SaveResult()
279 with transaction.atomic():
280 catalogue = TranslationCatalogue.objects.select_for_update().get(language_code=language_code)
281 po = read_draft(catalogue)
282 count = catalogues.plural_count(po)
283 by_key = {catalogues.entry_key(entry): entry for entry in catalogues.active_entries(po)}
284 for incoming in catalogues.active_entries(uploaded):
285 key = catalogues.entry_key(incoming)
286 entry = by_key.get(key)
287 if entry is None or not (incoming.msgstr or any(incoming.msgstr_plural.values())):
288 continue
289 if entry.msgid_plural:
290 values = [incoming.msgstr_plural.get(index, "") for index in range(count)]
291 else:
292 values = [incoming.msgstr]
293 if problem := catalogues.placeholder_problem(entry, *values):
294 result.errors[key] = problem
295 continue
296 if _apply(entry, Edit(key, values, incoming.fuzzy, "")):
297 result.changed += 1
298 if result.changed:
299 _save_draft(catalogue, po, user=user)
300 return result
303# Publishing ------------------------------------------------------------------------
306def latest_publication(language_code: str) -> TranslationPublication | None:
307 return TranslationPublication.objects.filter(language_code=language_code).order_by("-version").first()
310def _next_version(language_code: str) -> int:
311 return (
312 TranslationPublication.objects.filter(language_code=language_code).aggregate(v=Max("version"))["v"] or 0
313 ) + 1
316def _after_publish(publication: TranslationPublication) -> None:
317 def announce_and_load():
318 runtime.announce(str(publication.pk))
319 if settings.LOCALIZATION_SYNC:
320 runtime.sync()
322 transaction.on_commit(announce_and_load)
325def publish(language_code: str, *, user) -> TranslationPublication:
326 """Compile the draft and make it the language's live translations."""
327 check_language(language_code)
328 with transaction.atomic():
329 catalogue = TranslationCatalogue.objects.select_for_update().get(language_code=language_code)
330 po = read_draft(catalogue)
331 counts = catalogues.stats(po)
332 publication_id = uuid.uuid7()
333 area = f"{AREA}/{language_code}/published"
334 po_key = storage.save_bytes(storage.build_key(area, f"{publication_id}.po"), catalogues.dump(po))
335 mo_key = storage.save_bytes(storage.build_key(area, f"{publication_id}.mo"), catalogues.compile_mo(po))
336 try:
337 publication = TranslationPublication.objects.create(
338 id=publication_id,
339 language_code=language_code,
340 version=_next_version(language_code),
341 po_key=po_key,
342 mo_key=mo_key,
343 total_count=counts.total,
344 translated_count=counts.translated,
345 fuzzy_count=counts.fuzzy,
346 published_by=user,
347 )
348 except Exception:
349 storage.delete(po_key)
350 storage.delete(mo_key)
351 raise
352 _after_publish(publication)
353 logger.info("Published %s by %s", publication, getattr(user, "pk", None))
354 return publication
357def restore(publication: TranslationPublication, *, user) -> TranslationPublication:
358 """Make an earlier publication live again, as a new version. The draft is not changed."""
359 check_language(publication.language_code)
360 with transaction.atomic():
361 # Serialise with other publications of the same language.
362 TranslationCatalogue.objects.select_for_update().filter(language_code=publication.language_code).first()
363 restored = TranslationPublication.objects.create(
364 language_code=publication.language_code,
365 version=_next_version(publication.language_code),
366 po_key=publication.po_key,
367 mo_key=publication.mo_key,
368 total_count=publication.total_count,
369 translated_count=publication.translated_count,
370 fuzzy_count=publication.fuzzy_count,
371 restored_from=publication,
372 published_by=user,
373 )
374 _after_publish(restored)
375 logger.info("Restored %s as %s by %s", publication, restored, getattr(user, "pk", None))
376 return restored
379def _translations(po: polib.POFile) -> set:
380 return {
381 (catalogues.entry_key(entry), tuple(_values(entry)), entry.fuzzy) for entry in catalogues.active_entries(po)
382 } | {("Plural-Forms", catalogues.plural_forms(po))}
385def has_unpublished_changes(draft: polib.POFile, publication: TranslationPublication | None) -> bool:
386 """Whether the draft's translations differ from what is live."""
387 if publication is None:
388 return True
389 data = storage.read_bytes(publication.po_key)
390 return data is None or _translations(draft) != _translations(catalogues.parse(data))