Define an idempotent failed-block writing policy

bibtexparser/writer.py

@@ -12,6 +12,7 @@
 
 VAL_SEP = " = "
 PARSING_FAILED_COMMENT = "% WARNING Parsing failed for the following {n} lines."
+FAILED_BLOCK_POLICIES = ("preserve", "annotate", "raise")
 
 
 def _treat_entry(block: Entry, bibtex_format) -> list[str]:
@@ -63,9 +64,35 @@ def _treat_expl_comment(block: ExplicitComment, bibtex_format: "BibtexFormat") -
 def _treat_failed_block(block: ParsingFailedBlock, bibtex_format: "BibtexFormat") -> list[str]:
     if block.raw is None:
         raise ValueError(_failed_blocks_without_raw_error([block]))
+    if bibtex_format.failed_block_policy == "preserve":
+        return [block.raw]
+    if bibtex_format.failed_block_policy == "raise":
+        # The complete-library check in `write` normally reports all failures at
+        # once. Keep this guard so direct internal use cannot bypass the policy.
+        raise ValueError(_failed_blocks_forbidden_error([block]))
     lines = len(block.raw.splitlines())
-    parsing_failed_comment = PARSING_FAILED_COMMENT.format(n=lines)
-    return [parsing_failed_comment, "\n", block.raw, "\n"]
+    parsing_failed_comment = bibtex_format.parsing_failed_comment.format(n=lines)
+    rendered = [parsing_failed_comment, "\n", block.raw]
+    if not block.raw.endswith(("\n", "\r")):
+        rendered.append("\n")
+    return rendered
+
+
+def _is_existing_failed_block_annotation(
+    block: object,
+    next_block: object,
+    bibtex_format: "BibtexFormat",
+) -> bool:
+    """Recognize an annotation emitted for the immediately following failure."""
+    if bibtex_format.failed_block_policy != "annotate":
+        return False
+    if not isinstance(block, ImplicitComment) or not isinstance(next_block, ParsingFailedBlock):
+        return False
+    if next_block.raw is None:
+        return False
+    source_lines = len(next_block.raw.splitlines())
+    expected = bibtex_format.parsing_failed_comment.format(n=source_lines)
+    return block.comment == expected
 
 
 def _failed_blocks_without_raw_error(blocks: list[ParsingFailedBlock]) -> str:
@@ -79,12 +106,32 @@ def _failed_blocks_without_raw_error(blocks: list[ParsingFailedBlock]) -> str:
     )
 
 
+def _failed_blocks_forbidden_error(blocks: list[ParsingFailedBlock]) -> str:
+    descriptions = "\n".join(f"  - {type(b).__name__}: {b.error}" for b in blocks)
+    return (
+        "Cannot write library with failed_block_policy='raise':\n"
+        f"{descriptions}\n"
+        "Inspect `library.failed_blocks` and resolve or remove every failed block, "
+        "or select the 'preserve' or 'annotate' policy explicitly."
+    )
+
+
 def _raise_on_unwritable_blocks(library: Library) -> None:
     unwritable = [b for b in library.blocks if isinstance(b, ParsingFailedBlock) and b.raw is None]
     if unwritable:
         raise ValueError(_failed_blocks_without_raw_error(unwritable))
 
 
+def _raise_when_failed_blocks_are_forbidden(
+    library: Library, bibtex_format: "BibtexFormat"
+) -> None:
+    if bibtex_format.failed_block_policy != "raise":
+        return
+    failed_blocks = [b for b in library.blocks if isinstance(b, ParsingFailedBlock)]
+    if failed_blocks:
+        raise ValueError(_failed_blocks_forbidden_error(failed_blocks))
+
+
 def _calculate_auto_value_align(library: Library) -> int:
     max_key_len = 0
     for entry in library.entries:
@@ -105,29 +152,29 @@ def write(library: Library, bibtex_format: Optional["BibtexFormat"] = None) -> s
     :param bibtex_format: Customized BibTeX format to use (optional).
     :raises ValueError: If the library contains failed blocks without raw bibtex
         (e.g. duplicate-key blocks resulting from programmatically created entries)."""
-    _raise_on_unwritable_blocks(library)
-
     if bibtex_format is None:
         bibtex_format = BibtexFormat()
 
+    _raise_on_unwritable_blocks(library)
+    _raise_when_failed_blocks_are_forbidden(library, bibtex_format)
+
     if bibtex_format.value_column == "auto":
         auto_val: int = _calculate_auto_value_align(library)
         # Copy the format instance to avoid modifying the original
         # (which would be bad if the format is used for multiple libraries)
         bibtex_format = deepcopy(bibtex_format)
         bibtex_format.value_column = auto_val
 
-    string_pieces = []
-
-    for i, block in enumerate(library.blocks):
-        # Get string representation (as list of strings) of block
-        string_block_pieces = _treat_block(bibtex_format, block)
-        string_pieces.extend(string_block_pieces)
-        # Separate Blocks
-        if i < len(library.blocks) - 1:
-            string_pieces.append(bibtex_format.block_separator)
+    rendered_blocks: list[str] = []
+    for index, block in enumerate(library.blocks):
+        next_block = library.blocks[index + 1] if index + 1 < len(library.blocks) else None
+        if _is_existing_failed_block_annotation(block, next_block, bibtex_format):
+            # The following failed block recreates this exact annotation. Omitting
+            # the parsed copy prevents one new comment from appearing per cycle.
+            continue
+        rendered_blocks.append("".join(_treat_block(bibtex_format, block)))
 
-    return "".join(string_pieces)
+    return bibtex_format.block_separator.join(rendered_blocks)
 
 
 def _treat_block(bibtex_format, block) -> list[str]:
@@ -161,6 +208,7 @@ def __init__(self):
         self._block_separator: str = "\n\n"
         self._trailing_comma: bool = False
         self._parsing_failed_comment: str = PARSING_FAILED_COMMENT
+        self._failed_block_policy: str = "preserve"
 
     @property
     def indent(self) -> str:
@@ -232,3 +280,27 @@ def parsing_failed_comment(self) -> str:
     @parsing_failed_comment.setter
     def parsing_failed_comment(self, parsing_failed_comment: str):
         self._parsing_failed_comment = parsing_failed_comment
+
+    @property
+    def failed_block_policy(self) -> str:
+        """Control writing of blocks that could not be parsed.
+
+        ``"preserve"`` (the default) writes the retained raw block without
+        modifying it. This makes repeated parse/write cycles stable and avoids
+        accumulating generated warning comments.
+
+        ``"annotate"`` prepends ``parsing_failed_comment`` to the raw block.
+        This is the behavior used before the policy was introduced, but it
+        intentionally changes the document on every parse/write cycle.
+
+        ``"raise"`` refuses to write a library containing any failed block.
+        Use it when every parse failure must be resolved before export.
+        """
+        return self._failed_block_policy
+
+    @failed_block_policy.setter
+    def failed_block_policy(self, failed_block_policy: str):
+        if failed_block_policy not in FAILED_BLOCK_POLICIES:
+            choices = ", ".join(repr(policy) for policy in FAILED_BLOCK_POLICIES)
+            raise ValueError(f"failed_block_policy must be one of {choices}")
+        self._failed_block_policy = failed_block_policy

docs/source/customize.rst

@@ -296,3 +296,31 @@ Specifically, a user may pass a :class:`bibtexparser.BibtexFormatter` object to
 
 A few more options are provided and we refer to the docstrings of :class:`bibtexparser.BibtexFormat` for details.
 Note: Sorting of blocks and fields is done with the corresponding middleware, as described above.
+
+Failed-block Writing Policy
+---------------------------
+
+Blocks that fail to parse retain their raw BibTeX. They are preserved unchanged
+when writing by default so that saving a file does not alter the input merely
+because parsing failed. Strict applications can reject such output, while the
+former behavior of inserting a warning comment remains available explicitly:
+
+.. code-block:: python
+
+    bibtex_format = bibtexparser.BibtexFormat()
+    bibtex_format.failed_block_policy = "raise"     # "preserve" is the default
+    bib_str = bibtexparser.write_string(library, bibtex_format=bibtex_format)
+
+    bibtex_format.failed_block_policy = "annotate"
+    bibtex_format.parsing_failed_comment = "% REVIEW REQUIRED ({n} source lines)"
+
+The three policies are:
+
+* ``"preserve"``: write the retained raw block without modification;
+* ``"annotate"``: prepend ``parsing_failed_comment`` to the retained block;
+* ``"raise"``: refuse to write until all failed blocks are resolved or removed.
+
+This setting affects failed blocks only. Normal parsed blocks are serialized
+from their current model values, because their ``raw`` property may be stale
+after middleware or user edits. Consequently, the regular writer is not a
+byte-for-byte concrete-syntax editor for otherwise valid blocks.

docs/source/quickstart.rst

@@ -174,3 +174,10 @@ This can be quickly achieved using the following:
 
 As you can see, the content (besides some white-spacing and other layout) is identical to the original string.
 Naturally, the writer can be configured to your needs. For more information on that, see :ref:`the customization documentation <customizing>`.
+
+Failed blocks retain their raw source and are written unchanged by default. This
+prevents a parse/write cycle from silently dropping failed input or adding new
+content to it. Applications that require every failure to be resolved before
+export can select the ``"raise"`` failed-block policy; the legacy behavior of
+inserting a warning comment is available as ``"annotate"``. See
+:ref:`the formatting options <writing_formatting>` for an example.

tests/test_entrypoint.py

@@ -6,14 +6,57 @@
 
 import pytest
 
+from bibtexparser import BibtexFormat
 from bibtexparser import parse_file
+from bibtexparser import parse_string
 from bibtexparser import write_file
 from bibtexparser import write_string
 from bibtexparser.library import Library
 from bibtexparser.model import Entry
 from bibtexparser.model import Field
 
 
+def test_failed_block_roundtrip_is_idempotent_by_default():
+    """Repeated saves must not annotate or otherwise mutate retained failed input."""
+    source = "@article{broken,\n  title = {Still retained}\n"
+
+    first_write = write_string(parse_string(source))
+    second_write = write_string(parse_string(first_write))
+
+    assert first_write == source
+    assert second_write == first_write
+
+
+def test_duplicate_block_roundtrip_retains_both_sources_and_stabilizes():
+    """A duplicate remains visible and repeated serialization neither drops nor annotates it."""
+    source = "@article{same, title={First}}\n\n@article{same, title={Second}}"
+
+    first_write = write_string(parse_string(source))
+    reparsed = parse_string(first_write)
+    second_write = write_string(reparsed)
+
+    assert first_write.count("@article{same") == 2
+    assert "@article{same, title={Second}}" in first_write
+    assert len(reparsed.failed_blocks) == 1
+    assert second_write == first_write
+
+
+def test_failed_block_annotation_does_not_accumulate_across_roundtrips():
+    """Opt-in annotations remain a single diagnostic instead of growing on every save."""
+    source = "@article{broken,\n  title = {Still retained}\n"
+    bibtex_format = BibtexFormat()
+    bibtex_format.failed_block_policy = "annotate"
+
+    first_write = write_string(parse_string(source), bibtex_format=bibtex_format)
+    second_write = write_string(parse_string(first_write), bibtex_format=bibtex_format)
+    third_write = write_string(parse_string(second_write), bibtex_format=bibtex_format)
+
+    warning = "% WARNING Parsing failed for the following 2 lines."
+    assert first_write.count(warning) == 1
+    assert second_write == first_write
+    assert third_write == first_write
+
+
 def test_gbk():
     library = parse_file("tests/resources/gbk_test.bib", encoding="gbk")
     assert library.entries[0]["author"] == "凯撒"

tests/test_writer.py

@@ -164,19 +164,58 @@ def test_block_separator(block_separator):
         assert lines[1 + len(expected_lines)] == '@preamble{"myValue"}'
 
 
-def test_write_failed_block():
+def test_write_failed_block_preserves_raw_by_default():
+    """Writing a parsed failure must not modify its only authoritative representation."""
     raw = "@article{irrelevant-for-this-test,\nexcept = {that-there-need-to-be},\nother = {multiple-lines}\n}"
     block = ParsingFailedBlock(error=ValueError("Some error"), raw=raw, ignore_error_block=None)
     library = Library(blocks=[block])
     string = writer.write(library)
-    lines = string.splitlines()
 
-    assert len(lines) == 5
-    assert lines[0].startswith("% WARNING")
-    assert lines[1] == "@article{irrelevant-for-this-test,"
-    assert lines[2] == "except = {that-there-need-to-be},"
-    assert lines[3] == "other = {multiple-lines}"
-    assert lines[4] == "}"
+    assert string == raw
+
+
+def test_write_failed_block_can_retain_legacy_annotation():
+    """Annotation remains opt-in for callers that want failures embedded in output."""
+    raw = "@article{broken,\ntitle = {Unclosed entry}"
+    block = ParsingFailedBlock(error=ValueError("Some error"), raw=raw)
+    bibtex_format = BibtexFormat()
+    bibtex_format.failed_block_policy = "annotate"
+
+    string = writer.write(Library(blocks=[block]), bibtex_format)
+
+    assert string == "% WARNING Parsing failed for the following 2 lines.\n" + raw + "\n"
+
+
+def test_write_failed_block_uses_configured_comment():
+    """Failed-block output must honor the public format setting and line placeholder."""
+    raw = "@article{broken,\ntitle = {Unclosed entry}"
+    block = ParsingFailedBlock(error=ValueError("Some error"), raw=raw)
+    bibtex_format = BibtexFormat()
+    bibtex_format.failed_block_policy = "annotate"
+    bibtex_format.parsing_failed_comment = "% CUSTOM FAILURE ({n} source lines)"
+
+    string = writer.write(Library(blocks=[block]), bibtex_format)
+
+    assert string.startswith("% CUSTOM FAILURE (2 source lines)\n")
+
+
+def test_write_failed_block_can_fail_closed():
+    """Strict consumers can require resolution of every retained parse failure."""
+    first = ParsingFailedBlock(error=ValueError("first error"), raw="@article(first)")
+    second = ParsingFailedBlock(error=ValueError("second error"), raw="@article(second)")
+    bibtex_format = BibtexFormat()
+    bibtex_format.failed_block_policy = "raise"
+
+    with pytest.raises(ValueError, match="(?s)first error.*second error"):
+        writer.write(Library(blocks=[first, second]), bibtex_format)
+
+
+def test_failed_block_policy_rejects_unknown_value():
+    """A misspelled integrity policy must fail instead of selecting accidental behavior."""
+    bibtex_format = BibtexFormat()
+
+    with pytest.raises(ValueError, match="preserve.*annotate.*raise"):
+        bibtex_format.failed_block_policy = "discard"
 
 
 def test_write_failed_block_without_raw_raises():