Merge pull request #90 from openzim/dupt

Added duplicate failsafe feature to Creator

Author: rgaudin

CHANGELOG.md

@@ -5,6 +5,16 @@ All notable changes to this project are documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html) (as of version 1.5.0).
 
+## [1.7.0] - 2022-08-02
+
+### Added
+
+- `zim.Creator` now supports `ignore_duplicates: bool` parameter to
+  prevent duplicates from raising exceptions
+- `zim.Creator.add_item`, `zim.Creator.add_redirect` and `zim.Creator.add_item_for`
+  now supports a `duplicate_ok: bool` parameter to prevent an exception
+  should this item/redirect be a duplicate
+
 ## [1.6.3] - 2022-08-02
 
 ### Added

src/zimscraperlib/VERSION

@@ -1 +1 @@
-1.6.3
+1.7.0

src/zimscraperlib/zim/creator.py

@@ -20,6 +20,7 @@
 
 import datetime
 import pathlib
+import re
 import weakref
 from typing import Any, Callable, Dict, Optional, Tuple, Union
 
@@ -30,6 +31,13 @@
 from ..types import get_mime_for_name
 from .items import StaticItem
 
+DUPLICATE_EXC_STR = re.compile(
+    r"^Impossible to add(.+)"
+    r"dirent\'s title to add is(.+)"
+    r"existing dirent's title is(.+)",
+    re.MULTILINE | re.DOTALL,
+)
+
 
 def mimetype_for(
     path: str,
@@ -76,6 +84,7 @@ def __init__(
         language: Optional[str] = "eng",
         compression: Optional[str] = None,
         workaround_nocancel: Optional[bool] = True,
+        ignore_duplicates: Optional[bool] = False,
         **metadata: Dict[str, Union[str, datetime.date, datetime.datetime]]
     ):
         super().__init__(filename=filename)
@@ -103,6 +112,7 @@ def __init__(
             self.metadata = metadata
 
         self.workaround_nocancel = workaround_nocancel
+        self.ignore_duplicates = ignore_duplicates
 
     def start(self):
         super().__enter__()
@@ -129,6 +139,7 @@ def add_item_for(
         is_front: Optional[bool] = None,
         should_compress: Optional[bool] = None,
         delete_fpath: Optional[bool] = False,
+        duplicate_ok: Optional[bool] = None,
         callback: Optional[Union[callable, Tuple[callable, Any]]] = None,
     ):
         """Add a File or content at a specified path and get its path
@@ -178,12 +189,15 @@ def add_item_for(
                 cb += list(callback)
             callback = tuple(cb)
 
-        self.add_item(StaticItem(**kwargs), callback)
+        self.add_item(
+            StaticItem(**kwargs), callback=callback, duplicate_ok=duplicate_ok
+        )
         return path
 
     def add_item(
         self,
         item: libzim.writer.Item,
+        duplicate_ok: Optional[bool] = None,
         callback: Optional[Union[Callable, Tuple[Callable, Any]]] = None,
     ):
         """Add a libzim.writer.Item
@@ -196,8 +210,14 @@ def add_item(
                 weakref.finalize(item, callback)
             else:
                 weakref.finalize(item, *callback)
+
+        duplicate_ok = duplicate_ok or self.ignore_duplicates
         try:
-            super().add_item(item)
+            try:
+                super().add_item(item)
+            except RuntimeError as exc:
+                if not DUPLICATE_EXC_STR.match(str(exc)) or not duplicate_ok:
+                    raise exc
         except Exception:
             if self.workaround_nocancel:
                 self.can_finish = False  # pragma: no cover
@@ -209,6 +229,7 @@ def add_redirect(
         target_path: str,
         title: Optional[str] = "",
         is_front: Optional[bool] = None,
+        duplicate_ok: Optional[bool] = None,
     ):
         """Add a redirect from path to target_path
 
@@ -217,7 +238,19 @@ def add_redirect(
         hints = {}
         if is_front is not None:
             hints[libzim.writer.Hint.FRONT_ARTICLE] = bool(is_front)
-        super().add_redirection(path, title, target_path, hints)
+
+        duplicate_ok = duplicate_ok or self.ignore_duplicates
+
+        try:
+            try:
+                super().add_redirection(path, title, target_path, hints)
+            except RuntimeError as exc:
+                if not DUPLICATE_EXC_STR.match(str(exc)) or not duplicate_ok:
+                    raise exc
+        except Exception:
+            if self.workaround_nocancel:
+                self.can_finish = False  # pragma: no cover
+            raise
 
     def add_default_illustration(self, content: bytes):
         self.add_illustration(48, content)

tests/zim/test_zim_creator.py

@@ -458,3 +458,22 @@ def cb(*args):
     assert not html_file.exists()
     assert Store.called
     assert Store.called == 2
+
+
+def test_duplicates(tmp_path):
+    with Creator(tmp_path / "test.zim") as creator:
+        creator.add_item_for(path="A", content="A")
+        creator.add_item_for(path="C", content="C")
+        creator.add_redirect(path="B", target_path="A")
+        with pytest.raises(RuntimeError, match="existing dirent's title"):
+            creator.add_item_for(path="A", content="test2")
+        with pytest.raises(RuntimeError, match="existing dirent's title"):
+            creator.add_redirect(path="B", target_path="C")
+
+
+def test_ignore_duplicates(tmp_path):
+    with Creator(tmp_path / "test.zim", ignore_duplicates=True) as creator:
+        creator.add_item_for(path="A", content="A")
+        creator.add_item_for(path="A", content="A2")
+        creator.add_redirect(path="B", target_path="A")
+        creator.add_redirect(path="B", target_path="C")