diff --git a/machine/corpora/place_markers_usfm_update_block_handler.py b/machine/corpora/place_markers_usfm_update_block_handler.py index 421d8b2b..1b8bcd55 100644 --- a/machine/corpora/place_markers_usfm_update_block_handler.py +++ b/machine/corpora/place_markers_usfm_update_block_handler.py @@ -1,6 +1,7 @@ from __future__ import annotations -from typing import List, TypedDict, cast +import logging +from typing import List, Optional, TypedDict, cast from ..translation.word_alignment_matrix import WordAlignmentMatrix from .segment_boundary_adjuster import SegmentBoundaryAdjuster @@ -10,6 +11,8 @@ from .usfm_update_block_element import UsfmUpdateBlockElement, UsfmUpdateBlockElementType from .usfm_update_block_handler import UsfmUpdateBlockHandler, UsfmUpdateBlockHandlerError +logger = logging.getLogger(__name__) + PLACE_MARKERS_ALIGNMENT_INFO_KEY = "alignment_info" @@ -21,6 +24,30 @@ class PlaceMarkersAlignmentInfo(TypedDict): style_behavior: UpdateUsfmMarkerBehavior +def _get_alignment_info(block: UsfmUpdateBlock) -> Optional[PlaceMarkersAlignmentInfo]: + if len(block.row_metadata) > 1: + # Verse ranges put all of their text on the first row + infos = [ + info + for info in ( + cast(Optional[PlaceMarkersAlignmentInfo], metadata.get(PLACE_MARKERS_ALIGNMENT_INFO_KEY)) + for metadata in block.row_metadata + ) + if info is not None and info["alignment"].row_count > 0 and info["alignment"].column_count > 0 + ] + if len(infos) > 1: + # Only the first row should have alignment info + logger.warning( + "Expected at most one row with alignment info for %s, but found %d. Markers may be misplaced.", + ", ".join(str(ref) for ref in block.refs), + len(infos), + ) + return infos[-1] if len(infos) > 0 else None + if PLACE_MARKERS_ALIGNMENT_INFO_KEY not in block.metadata: + return None + return cast(PlaceMarkersAlignmentInfo, block.metadata[PLACE_MARKERS_ALIGNMENT_INFO_KEY]) + + class PlaceMarkersUsfmUpdateBlockHandler(UsfmUpdateBlockHandler): def __init__(self, *args): super().__init__(*args) @@ -30,10 +57,10 @@ def process_block(self, block: UsfmUpdateBlock) -> UsfmUpdateBlock: elements = list(block.elements) # Nothing to do if there are no markers to place or no alignment to use - if PLACE_MARKERS_ALIGNMENT_INFO_KEY not in block.metadata: + alignment_info = _get_alignment_info(block) + if alignment_info is None: return block - alignment_info = cast(PlaceMarkersAlignmentInfo, block.metadata[PLACE_MARKERS_ALIGNMENT_INFO_KEY]) if ( len(elements) == 0 or alignment_info["alignment"].row_count == 0 diff --git a/machine/corpora/update_usfm_parser_handler.py b/machine/corpora/update_usfm_parser_handler.py index 87689d8a..7e384fc0 100644 --- a/machine/corpora/update_usfm_parser_handler.py +++ b/machine/corpora/update_usfm_parser_handler.py @@ -379,9 +379,9 @@ def get_usfm(self, stylesheet: Union[str, UsfmStylesheet] = "usfm.sty") -> str: tokens[index:index] = remark_tokens return tokenizer.detokenize(tokens) - def _advance_rows(self, seg_scr_refs: Sequence[ScriptureRef]) -> Tuple[List[str], Optional[dict[str, object]]]: + def _advance_rows(self, seg_scr_refs: Sequence[ScriptureRef]) -> Tuple[List[str], List[dict[str, object]]]: row_texts: List[str] = [] - row_metadata = None + row_metadata: List[dict[str, object]] = [] source_index: int = 0 # handle the special case of verse 0, which although first in the rows, @@ -407,7 +407,7 @@ def _advance_rows(self, seg_scr_refs: Sequence[ScriptureRef]) -> Tuple[List[str] # source and row match # grab the text - both source and row will be incremented in due time... row_texts.append(text) - row_metadata = metadata + row_metadata.append(metadata if metadata is not None else {}) break if compare <= 0: # source is ahead of row, increment row @@ -485,10 +485,8 @@ def _has_new_text(self) -> bool: return any(self._replace_stack) and self._replace_stack[-1] def _start_update_block(self, scripture_refs: Sequence[ScriptureRef]) -> None: - row_texts, metadata = self._advance_rows(scripture_refs) - self._update_block_stack.append( - UsfmUpdateBlock(scripture_refs, metadata=metadata if metadata is not None else {}) - ) + row_texts, row_metadata = self._advance_rows(scripture_refs) + self._update_block_stack.append(UsfmUpdateBlock(scripture_refs, row_metadata=row_metadata)) self._push_updated_text([UsfmToken(UsfmTokenType.TEXT, text=t + " ") for t in row_texts]) def _end_update_block(self, state: UsfmParserState, scripture_refs: Sequence[ScriptureRef]) -> None: diff --git a/machine/corpora/usfm_update_block.py b/machine/corpora/usfm_update_block.py index 977e82b2..094ad17e 100644 --- a/machine/corpora/usfm_update_block.py +++ b/machine/corpora/usfm_update_block.py @@ -12,11 +12,13 @@ def __init__( self, refs: Iterable[ScriptureRef] = [], elements: Iterable[UsfmUpdateBlockElement] = [], - metadata: dict[str, object] = {}, + row_metadata: Iterable[dict[str, object]] = [], ) -> None: self._refs: list[ScriptureRef] = list(refs) self._elements: list[UsfmUpdateBlockElement] = list(elements) - self._metadata: dict[str, object] = metadata + # One entry per row matched to this block, in order. A verse range can be matched by + # several rows, in which case this block's text is those rows' texts concatenated. + self._row_metadata: list[dict[str, object]] = list(row_metadata) @property def refs(self) -> Sequence[ScriptureRef]: @@ -28,7 +30,11 @@ def elements(self) -> Sequence[UsfmUpdateBlockElement]: @property def metadata(self) -> dict[str, object]: - return self._metadata + return self._row_metadata[-1] if len(self._row_metadata) > 0 else {} + + @property + def row_metadata(self) -> Sequence[dict[str, object]]: + return self._row_metadata def add_text(self, tokens: Iterable[UsfmToken]) -> None: self._elements.append(UsfmUpdateBlockElement(UsfmUpdateBlockElementType.TEXT, list(tokens))) @@ -68,7 +74,11 @@ def get_tokens(self) -> list[UsfmToken]: return [token for element in self._elements for token in element.get_tokens()] def __eq__(self, other: UsfmUpdateBlock) -> bool: - return self._refs == other._refs and self._elements == other._elements and self._metadata == other._metadata + return ( + self._refs == other._refs + and self._elements == other._elements + and self._row_metadata == other._row_metadata + ) def copy(self) -> UsfmUpdateBlock: - return UsfmUpdateBlock(self._refs, self._elements, self._metadata) + return UsfmUpdateBlock(self._refs, self._elements, self._row_metadata) diff --git a/tests/corpora/test_place_markers_usfm_update_block_handler.py b/tests/corpora/test_place_markers_usfm_update_block_handler.py index fe26a15d..aa2ba333 100644 --- a/tests/corpora/test_place_markers_usfm_update_block_handler.py +++ b/tests/corpora/test_place_markers_usfm_update_block_handler.py @@ -1,5 +1,8 @@ +import logging from typing import List, Optional, Sequence +import pytest + from machine.corpora import ( AlignedWordPair, PlaceMarkersAlignmentInfo, @@ -940,6 +943,85 @@ def test_anusvara_tokenization() -> None: assert_usfm_equals(target, result) +def test_verse_range_with_empty_trailing_row(caplog: pytest.LogCaptureFixture) -> None: + # Verse ranges consist of multiple rows, but only the first one is non-empty and has a non-empty alignment matrix. + # An empty alignment matrix must not clobber any non-empty matrices. + source = "This is the first part. This is the second part." + pretranslation = "Esta es la primera parte. Esta es la segunda parte." + align_info = PlaceMarkersAlignmentInfo( + source_tokens=[t for t in TOKENIZER.tokenize(source)], + translation_tokens=[t for t in TOKENIZER.tokenize(pretranslation)], + alignment=to_word_alignment_matrix("0-0 1-1 2-2 3-3 4-4 5-5 6-6 7-7 8-8 9-9 10-10 11-11"), + paragraph_behavior=UpdateUsfmMarkerBehavior.PRESERVE, + style_behavior=UpdateUsfmMarkerBehavior.STRIP, + ) + empty_align_info = PlaceMarkersAlignmentInfo( + source_tokens=[], + translation_tokens=[], + alignment=to_word_alignment_matrix(""), + paragraph_behavior=UpdateUsfmMarkerBehavior.PRESERVE, + style_behavior=UpdateUsfmMarkerBehavior.STRIP, + ) + usfm = r"""\id MAT +\c 1 +\v 1-2 This is the first part. +\p This is the second part. +""" + result = r"""\id MAT +\c 1 +\v 1-2 Esta es la primera parte. +\p Esta es la segunda parte. +""" + + # the result should be the same whether the second row has an empty alignment matrix or no alignment matrix at all + for empty_row in [ + UpdateUsfmRow(scr_ref("MAT 1:2"), "", metadata={"alignment_info": empty_align_info}), + UpdateUsfmRow(scr_ref("MAT 1:2"), ""), + ]: + rows = [ + UpdateUsfmRow(scr_ref("MAT 1:1"), str(pretranslation), metadata={"alignment_info": align_info}), + empty_row, + ] + with caplog.at_level(logging.WARNING): + target = update_usfm(rows, usfm, update_block_handlers=[PlaceMarkersUsfmUpdateBlockHandler()]) + assert_usfm_equals(target, result) + # a row that contributed no text is expected, so it must not be reported as unexpected + assert "Expected at most one row with alignment info" not in caplog.text + + +def test_multiple_rows_with_alignment_info_warns(caplog: pytest.LogCaptureFixture) -> None: + # Only one row's alignment info can be used, so a block matched by several rows that each have + # text of their own is outside what the handler can place markers for + first_align_info = PlaceMarkersAlignmentInfo( + source_tokens=[t for t in TOKENIZER.tokenize("This is the first part.")], + translation_tokens=[t for t in TOKENIZER.tokenize("Esta es la primera parte.")], + alignment=to_word_alignment_matrix("0-0 1-1 2-2 3-3 4-4 5-5"), + paragraph_behavior=UpdateUsfmMarkerBehavior.PRESERVE, + style_behavior=UpdateUsfmMarkerBehavior.STRIP, + ) + second_align_info = PlaceMarkersAlignmentInfo( + source_tokens=[t for t in TOKENIZER.tokenize("This is the second part.")], + translation_tokens=[t for t in TOKENIZER.tokenize("Esta es la segunda parte.")], + alignment=to_word_alignment_matrix("0-0 1-1 2-2 3-3 4-4 5-5"), + paragraph_behavior=UpdateUsfmMarkerBehavior.PRESERVE, + style_behavior=UpdateUsfmMarkerBehavior.STRIP, + ) + rows = [ + UpdateUsfmRow(scr_ref("MAT 1:1"), "Esta es la primera parte.", metadata={"alignment_info": first_align_info}), + UpdateUsfmRow(scr_ref("MAT 1:2"), "Esta es la segunda parte.", metadata={"alignment_info": second_align_info}), + ] + usfm = r"""\id MAT +\c 1 +\v 1-2 This is the first part. +\p This is the second part. +""" + + with caplog.at_level(logging.WARNING): + update_usfm(rows, usfm, update_block_handlers=[PlaceMarkersUsfmUpdateBlockHandler()]) + + assert "Expected at most one row with alignment info for MAT 1:1, MAT 1:2, but found 2" in caplog.text + + def scr_ref(*refs: str) -> List[ScriptureRef]: return [ScriptureRef.parse(ref) for ref in refs]