"""Merge translator edits from an older tree into a regenerated tree."""
from __future__ import annotations
import json
import shutil
from collections import Counter, defaultdict
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Literal
from .document import (
parse_sentence_text,
parse_translation_document,
replace_translation_text,
)
from .filesystem import reset_dir
from .manifest import TREE_MANIFEST_PATH, load_tree_manifest
from .models import TranslationTreeError
from .outline import refresh_outline_markdown
from .placeholders import (
contains_raw_jinja_in_translation,
extract_translator_placeholder_names,
translation_placeholder_counts,
)
MERGE_REPORT_PATH = Path(".translation-tree") / "merge-report.json"
ExistingTranslationPolicy = Literal["preserve", "replace"]
[docs]
@dataclass(frozen=True)
class TranslationCandidate:
"""One reusable translation from an older translation tree."""
source_file: str
unit_key: str
unit_source_hash: str
document_path: str
sentence_text: str
translation_text: str
[docs]
@dataclass(frozen=True)
class ReuseIndexes:
"""Lookup tables for translations that are safe to migrate."""
by_key: dict[tuple[str, str], TranslationCandidate]
by_hash: dict[str, TranslationCandidate]
by_sentence: dict[str, TranslationCandidate]
[docs]
@dataclass(frozen=True)
class CandidateMatch:
"""One old-tree candidate selected for one new-tree unit."""
candidate: TranslationCandidate
kind: str
[docs]
@dataclass(frozen=True)
class TranslationMergeReport:
"""Summary of one old-tree to new-tree translation merge."""
total_units: int
preserved_units: int
migrated_units: int
updated_units: int
untranslated_units: int
skipped_unsafe_old_units: int
exact_key_matches: int
source_hash_matches: int
sentence_matches: int
[docs]
def merge_translation_tree(
*,
old_tree_dir: Path,
new_tree_dir: Path,
output_dir: Path,
source_lang: str,
target_lang: str,
allow_sentence_matches: bool = False,
existing_translation_policy: ExistingTranslationPolicy = "preserve",
) -> TranslationMergeReport:
"""Copy a regenerated tree and reuse safe translations from an older tree.
Matching is intentionally conservative. Exact `(source_file, unit_key)` can
reuse a translation only when the unit source hash is unchanged, then unique
source hash can recover moved but byte-identical source units. Visible
sentence matches are intentionally disabled by default because they cannot
prove that the underlying Jinja/HTML structure is still equivalent. Existing
translations are preserved by default; cross-version synchronization may
explicitly replace them when the source unit is an exact structural match.
"""
old_tree_dir = Path(old_tree_dir).resolve()
new_tree_dir = Path(new_tree_dir).resolve()
output_dir = Path(output_dir).resolve()
if old_tree_dir == output_dir:
raise TranslationTreeError("--output must differ from --old-tree")
if existing_translation_policy not in {"preserve", "replace"}:
raise TranslationTreeError("existing_translation_policy must be 'preserve' or 'replace'")
if existing_translation_policy == "replace" and allow_sentence_matches:
raise TranslationTreeError("replace policy cannot be combined with sentence matches")
if new_tree_dir != output_dir:
reset_dir(output_dir)
shutil.copytree(new_tree_dir, output_dir, dirs_exist_ok=True)
old_candidates = _load_candidates(
tree_dir=old_tree_dir,
source_lang=source_lang,
target_lang=target_lang,
forgiving=True,
)
new_candidates = _load_candidates(
tree_dir=output_dir,
source_lang=source_lang,
target_lang=target_lang,
forgiving=False,
)
reusable_old_candidates = [
candidate for candidate in old_candidates if _is_reusable_translation(candidate)
]
translated_old_units = [
candidate for candidate in old_candidates if candidate.translation_text.strip()
]
skipped_unsafe_old_units = len(translated_old_units) - len(reusable_old_candidates)
reuse_indexes = _build_reuse_indexes(
reusable_old_candidates,
allow_sentence_matches=allow_sentence_matches,
)
preserved = 0
migrated = 0
updated = 0
untranslated = 0
match_counts: Counter[str] = Counter()
for candidate in new_candidates:
has_translation = bool(candidate.translation_text.strip())
if has_translation and existing_translation_policy == "preserve":
preserved += 1
continue
match = _find_reusable_match(
candidate,
reuse_indexes=reuse_indexes,
allow_sentence_matches=allow_sentence_matches,
)
if match is None:
if has_translation:
preserved += 1
else:
untranslated += 1
continue
replacement = match.candidate.translation_text
if has_translation and candidate.translation_text == replacement:
preserved += 1
continue
replace_translation_text(
document_path=output_dir / candidate.document_path,
target_lang=target_lang,
translation_text=replacement,
)
if has_translation:
updated += 1
else:
migrated += 1
match_counts[match.kind] += 1
report = TranslationMergeReport(
total_units=len(new_candidates),
preserved_units=preserved,
migrated_units=migrated,
updated_units=updated,
untranslated_units=untranslated,
skipped_unsafe_old_units=skipped_unsafe_old_units,
exact_key_matches=match_counts["exact-key"],
source_hash_matches=match_counts["source-hash"],
sentence_matches=match_counts["sentence"],
)
report_path = output_dir / MERGE_REPORT_PATH
report_path.parent.mkdir(parents=True, exist_ok=True)
report_path.write_text(
json.dumps(asdict(report), indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
refresh_outline_markdown(
tree_dir=output_dir,
source_lang=source_lang,
target_lang=target_lang,
)
return report
def _load_candidates(
*,
tree_dir: Path,
source_lang: str,
target_lang: str,
forgiving: bool,
) -> list[TranslationCandidate]:
try:
manifest = load_tree_manifest(tree_dir)
except TranslationTreeError:
if forgiving:
return []
raise
units = manifest.get("units")
if not isinstance(units, list):
raise TranslationTreeError(
f"Invalid translation-tree manifest at {tree_dir / TREE_MANIFEST_PATH}"
)
candidates: list[TranslationCandidate] = []
for unit in units:
try:
candidate = _candidate_from_manifest_unit(
tree_dir=tree_dir,
unit=unit,
source_lang=source_lang,
target_lang=target_lang,
)
except TranslationTreeError:
if forgiving:
continue
raise
candidates.append(candidate)
return candidates
def _build_reuse_indexes(
candidates: list[TranslationCandidate],
*,
allow_sentence_matches: bool,
) -> ReuseIndexes:
by_sentence: dict[str, TranslationCandidate] = {}
if allow_sentence_matches:
by_sentence = _unique_index(
candidates,
key=lambda candidate: _normalize_sentence(candidate.sentence_text),
)
return ReuseIndexes(
by_key={(candidate.source_file, candidate.unit_key): candidate for candidate in candidates},
by_hash=_unique_index(
candidates,
key=lambda candidate: candidate.unit_source_hash,
),
by_sentence=by_sentence,
)
def _find_reusable_match(
candidate: TranslationCandidate,
*,
reuse_indexes: ReuseIndexes,
allow_sentence_matches: bool,
) -> CandidateMatch | None:
key_match = reuse_indexes.by_key.get((candidate.source_file, candidate.unit_key))
if key_match is not None and key_match.unit_source_hash == candidate.unit_source_hash:
return CandidateMatch(candidate=key_match, kind="exact-key")
hash_match = reuse_indexes.by_hash.get(candidate.unit_source_hash)
if hash_match is not None:
return CandidateMatch(candidate=hash_match, kind="source-hash")
if not allow_sentence_matches:
return None
sentence_match = reuse_indexes.by_sentence.get(_normalize_sentence(candidate.sentence_text))
if sentence_match is None:
return None
return CandidateMatch(candidate=sentence_match, kind="sentence")
def _candidate_from_manifest_unit(
*,
tree_dir: Path,
unit: object,
source_lang: str,
target_lang: str,
) -> TranslationCandidate:
if not isinstance(unit, dict):
raise TranslationTreeError(
f"Invalid translation-tree manifest entry at {tree_dir / TREE_MANIFEST_PATH}"
)
source_file = unit.get("source_file")
unit_key = unit.get("unit_key")
unit_source_hash = unit.get("unit_source_hash")
document_path_raw = unit.get("document_path")
if (
not isinstance(source_file, str)
or not isinstance(unit_key, str)
or not isinstance(unit_source_hash, str)
or not isinstance(document_path_raw, str)
):
raise TranslationTreeError(
f"Invalid translation-tree manifest entry at {tree_dir / TREE_MANIFEST_PATH}"
)
tree_root = tree_dir.resolve()
relative_document_path = Path(document_path_raw)
document_path = (tree_root / relative_document_path).resolve()
if (
relative_document_path.is_absolute()
or ".." in relative_document_path.parts
or not document_path.is_relative_to(tree_root)
):
raise TranslationTreeError(
f"Translation document path escapes tree directory: {document_path_raw}"
)
if not document_path.is_file():
raise TranslationTreeError(f"Missing translation document at {document_path}")
return TranslationCandidate(
source_file=source_file,
unit_key=unit_key,
unit_source_hash=unit_source_hash,
document_path=str(document_path.relative_to(tree_root)),
sentence_text=parse_sentence_text(
document_path=document_path,
source_lang=source_lang,
),
translation_text=parse_translation_document(
document_path=document_path,
source_lang=source_lang,
target_lang=target_lang,
),
)
def _unique_index(
candidates: list[TranslationCandidate],
*,
key,
) -> dict[str, TranslationCandidate]:
grouped: dict[str, list[TranslationCandidate]] = defaultdict(list)
for candidate in candidates:
if not candidate.translation_text.strip():
continue
grouped[key(candidate)].append(candidate)
return {
index_key: grouped_candidates[0]
for index_key, grouped_candidates in grouped.items()
if index_key and len(grouped_candidates) == 1
}
def _is_reusable_translation(candidate: TranslationCandidate) -> bool:
"""Return whether an old translation can be migrated without repair."""
translation_text = candidate.translation_text
if not translation_text.strip():
return False
if contains_raw_jinja_in_translation(translation_text):
return False
required_placeholders = Counter(extract_translator_placeholder_names(candidate.sentence_text))
if required_placeholders != translation_placeholder_counts(translation_text):
return False
return True
def _normalize_sentence(sentence_text: str) -> str:
return " ".join(sentence_text.split())