diff --git a/bindery/__init__.py b/bindery/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bindery/diffing.py b/bindery/diffing.py new file mode 100644 index 0000000..6f107b0 --- /dev/null +++ b/bindery/diffing.py @@ -0,0 +1,129 @@ +"""Text extraction, change detection and HTML diffs for chapters.""" + +import difflib +import hashlib +import re +from html import escape +from typing import Any + +from bs4 import BeautifulSoup + +BLOCK_TAGS = [ + "p", + "div", + "h1", + "h2", + "h3", + "h4", + "h5", + "h6", + "li", + "blockquote", + "pre", + "tr", + "table", + "hr", + "section", + "article", + "ul", + "ol", + "center", +] + + +def html_to_blocks(html: str) -> list[str]: + """Return the readable text of a chapter as one string per paragraph. + + Attributes and markup are ignored. Some sites (RoyalRoad, for example) put + random class names on paragraphs on every request, so comparing raw HTML + would report a change every time. + """ + soup = BeautifulSoup(html or "", "html.parser") + for tag in soup.find_all(["script", "style"]): + tag.decompose() + for br in soup.find_all("br"): + br.replace_with("\n") + for tag in soup.find_all(BLOCK_TAGS): + tag.insert_before("\n") + tag.insert_after("\n") + blocks = [] + for line in soup.get_text().split("\n"): + line = re.sub(r"\s+", " ", line).strip() + if line: + blocks.append(line) + return blocks + + +def html_source_lines(html: str) -> list[str]: + """Split HTML source at block boundaries, for diffs where markup matters.""" + html = re.sub(r"(||)", r"\1\n", html or "") + return [line.strip() for line in html.split("\n") if line.strip()] + + +def text_hash(html: str) -> str: + return hashlib.sha256("\n".join(html_to_blocks(html)).encode("utf-8")).hexdigest() + + +def _word_diff(old: str, new: str) -> str: + old_words, new_words = re.findall(r"\S+|\s+", old), re.findall(r"\S+|\s+", new) + out = [] + matcher = difflib.SequenceMatcher(None, old_words, new_words, autojunk=False) + for op, i1, i2, j1, j2 in matcher.get_opcodes(): + if op == "equal": + out.append(escape("".join(old_words[i1:i2]))) + continue + if i2 > i1: + out.append(f"{escape(''.join(old_words[i1:i2]))}") + if j2 > j1: + out.append(f"{escape(''.join(new_words[j1:j2]))}") + return "".join(out) + + +def diff_blocks(old: list[str], new: list[str], context: int = 2) -> dict[str, Any]: + """Render a paragraph diff as HTML. All text is escaped.""" + matcher = difflib.SequenceMatcher(None, old, new, autojunk=False) + parts: list[str] = [] + added = removed = changed = 0 + opcodes = matcher.get_opcodes() + for n, (op, i1, i2, j1, j2) in enumerate(opcodes): + if op == "equal": + lines = old[i1:i2] + head = lines[:context] if n > 0 else [] + tail = lines[-context:] if n < len(opcodes) - 1 else [] + if len(lines) <= len(head) + len(tail) + 1: + head, tail = lines, [] + skipped = len(lines) - len(head) - len(tail) + parts.extend(f'

{escape(text)}

' for text in head) + if skipped > 0: + parts.append(f'

{skipped} unchanged paragraph{"s" if skipped != 1 else ""}

') + parts.extend(f'

{escape(text)}

' for text in tail) + elif op == "replace": + pairs = min(i2 - i1, j2 - j1) + for k in range(pairs): + parts.append(f'

{_word_diff(old[i1 + k], new[j1 + k])}

') + changed += pairs + for text in old[i1 + pairs : i2]: + parts.append(f'

{escape(text)}

') + removed += 1 + for text in new[j1 + pairs : j2]: + parts.append(f'

{escape(text)}

') + added += 1 + elif op == "delete": + parts.extend(f'

{escape(text)}

' for text in old[i1:i2]) + removed += i2 - i1 + elif op == "insert": + parts.extend(f'

{escape(text)}

' for text in new[j1:j2]) + added += j2 - j1 + return { + "html": "\n".join(parts), + "added": added, + "removed": removed, + "changed": changed, + "identical": added == removed == changed == 0, + } + + +def diff_html(old_html: str, new_html: str, source: bool = False, context: int = 2) -> dict[str, Any]: + """Diff two chapter versions by text, or by HTML source when source is True.""" + split = html_source_lines if source else html_to_blocks + return diff_blocks(split(old_html), split(new_html), context=context) diff --git a/bindery/merge.py b/bindery/merge.py new file mode 100644 index 0000000..2acfdc5 --- /dev/null +++ b/bindery/merge.py @@ -0,0 +1,34 @@ +"""Merge the site's chapter list into the chapter list we already have.""" + +from collections.abc import Sequence + + +def merge_chapter_lists(existing: Sequence[str], site: Sequence[str], newest_last: bool = True) -> list[str]: + """Return the new book order of chapter URLs. + + Every chapter in `existing` stays, in its current order, even when the site + no longer lists it. Each chapter that is only on the site goes directly after + the chapter that comes before it on the site. A new chapter before any known + chapter goes to the start. New chapters after the last known chapter on the + site go to the end of the book, after chapters that the site removed. + + Example: we have 1-10, the site now lists 1-5 and 11. The result is 1-11. + + An older copy of the story, such as an epub, passes newest_last=False. Its + last chapters can be older than the chapters we have, so they go after their + predecessor too. Example: we have 1-5 and 50-60, the epub has 1-45. The + result is 1-45 and 50-60. + """ + result = list(dict.fromkeys(existing)) + known = set(result) + site = list(dict.fromkeys(site)) + last_known = max((i for i, url in enumerate(site) if url in known), default=-1) if newest_last else len(site) + insert_at = 0 + for url in site[: last_known + 1]: + if url in known: + insert_at = result.index(url) + 1 + else: + result.insert(insert_at, url) + known.add(url) + insert_at += 1 + return result + site[last_known + 1 :] diff --git a/bindery/rules.py b/bindery/rules.py new file mode 100644 index 0000000..ee2b39e --- /dev/null +++ b/bindery/rules.py @@ -0,0 +1,192 @@ +"""Regex search-and-replace rules applied to chapter HTML when a book is built. + +Rules never change the stored chapters. They run on a copy while the epub is +written, so a rule can be edited or removed and the book rebuilt without a new +download. An optional first step turns line breaks into paragraphs. +""" + +import functools +from collections.abc import Callable +from typing import Any + +import regex +from bs4 import BeautifulSoup, NavigableString, Tag + +FLAGS = {"i": regex.IGNORECASE, "s": regex.DOTALL, "m": regex.MULTILINE, "x": regex.VERBOSE} +# The longest time in seconds that one rule can take on one chapter. Some patterns backtrack for hours. +TIMEOUT = 10 + +# One step of the work on a chapter: HTML in, HTML out. +Step = Callable[[str], str] + +BLOCK_TAGS = { + "address", + "article", + "aside", + "blockquote", + "center", + "dd", + "details", + "div", + "dl", + "dt", + "fieldset", + "figcaption", + "figure", + "footer", + "form", + "h1", + "h2", + "h3", + "h4", + "h5", + "h6", + "header", + "hr", + "li", + "main", + "nav", + "ol", + "p", + "pre", + "section", + "table", + "ul", +} +# Blocks whose loose text can go into paragraphs. A

is split into several

s instead. +CONTAINERS = ["article", "aside", "blockquote", "dd", "div", "li", "main", "p", "section", "td"] + + +class RuleError(ValueError): + def __init__(self, index: int, message: str) -> None: + super().__init__(f"Rule {index + 1}: {message}") + self.index = index + + +def normalize_rules(rules: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Check a list of rules from the UI and return clean copies. + + Raises RuleError for the first rule with a bad pattern, replacement or flag. + """ + clean = [] + for index, rule in enumerate(rules): + pattern = str(rule.get("pattern", "")) + if not pattern: + raise RuleError(index, "the pattern is empty") + flags = "".join(sorted(set(str(rule.get("flags", "")).lower()))) + bad = [f for f in flags if f not in FLAGS] + if bad: + raise RuleError(index, f"unknown flag {''.join(bad)!r}, use i, s, m or x") + replacement = str(rule.get("replacement", "")) + try: + regex.compile(pattern, _flag_bits(flags)) + # regex reads a replacement only at a match. An empty match with the same groups finds bad group + # references now. The newline ends a comment of the x flag. + regex.compile(f"(?:{pattern}\n)|", _flag_bits(flags)).match("").expand(replacement) + except (regex.error, IndexError) as e: + raise RuleError(index, str(e)) from e + clean.append( + { + "pattern": pattern, + "replacement": replacement, + "flags": flags, + "enabled": bool(rule.get("enabled", True)), + "note": str(rule.get("note", "")), + } + ) + return clean + + +def _flag_bits(flags: str) -> int: + bits = 0 + for f in flags: + bits |= FLAGS[f] + return bits + + +def compile_rules(*rule_lists: list[dict[str, Any]], paragraphs: bool = False) -> list[Step]: + """Return the steps for the enabled rules of each list, in order. + + With `paragraphs`, line breaks become paragraphs before the first rule runs. + """ + steps: list[Step] = [breaks_to_paragraphs] if paragraphs else [] + for rules in rule_lists: + for rule in rules: + if rule.get("enabled", True): + pattern = regex.compile(rule["pattern"], _flag_bits(rule.get("flags", ""))) + steps.append(functools.partial(_substitute, pattern, rule.get("replacement", ""))) + return steps + + +def _substitute(pattern: regex.Pattern, replacement: str, html: str) -> str: + try: + return pattern.sub(replacement, html, timeout=TIMEOUT) + except TimeoutError as e: + raise ValueError( + f"The rule {pattern.pattern!r} takes longer than {TIMEOUT} seconds on one chapter. " + "Change the rule or turn it off." + ) from e + + +def apply_rules(html: str, steps: list[Step]) -> str: + for step in steps: + html = step(html) + return html + + +def _is_empty(nodes: list[Any]) -> bool: + for node in nodes: + if isinstance(node, Tag): + if node.name == "img" or node.find("img") or node.get_text().strip(): + return False + elif isinstance(node, NavigableString) and node.strip(): + return False + return True + + +def breaks_to_paragraphs(html: str) -> str: + """Turn text that
tags split inside a block into one paragraph per piece. + + Only a
that is a direct child of a block counts, so a
inside + inline markup such as stays. Empty pieces are dropped, so a run of +
tags makes one paragraph break. A

becomes several

s with the + same attributes. In other blocks, the pieces go into new

s, and child + blocks stay as they are. + """ + if " str: + return datetime.now().isoformat(timespec="seconds") + + +def _story(row: sqlite3.Row | None) -> dict[str, Any] | None: + if row is None: + return None + d = dict(row) + for col in _JSON_COLUMNS: + if col in d: + d[col] = json.loads(d[col] or ("[]" if col == "locked_fields" else "{}")) + return d + + +class Store: + def __init__(self, path: Path) -> None: + self.path = Path(path) + self.path.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as conn: + conn.execute("PRAGMA journal_mode=WAL") + conn.executescript(SCHEMA) + + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: + conn = sqlite3.connect(self.path, timeout=30) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys=ON") + try: + with conn: + yield conn + finally: + conn.close() + + def backup(self, path: Path) -> None: + """Write a consistent copy of the database, also while jobs write to it.""" + with self._connect() as conn: + target = sqlite3.connect(path) + try: + conn.backup(target) + finally: + target.close() + + # Stories + + def create_story( + self, + url: str, + site: str, + title: str, + author: str, + locked_fields: list[str], + calibre_id: int | None = None, + calibre_uuid: str | None = None, + ) -> int: + with self._connect() as conn: + cur = conn.execute( + "INSERT INTO stories (url, site, title, author, calibre_id, calibre_uuid, locked_fields, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + (url, site, title, author, calibre_id, calibre_uuid, json.dumps(locked_fields), now()), + ) + return int(cur.lastrowid) + + def get_story(self, story_id: int) -> dict[str, Any] | None: + with self._connect() as conn: + return _story(conn.execute("SELECT * FROM stories WHERE id = ?", (story_id,)).fetchone()) + + def get_story_by_url(self, url: str) -> dict[str, Any] | None: + with self._connect() as conn: + return _story(conn.execute("SELECT * FROM stories WHERE url = ?", (url,)).fetchone()) + + def get_story_by_calibre_id(self, calibre_id: int) -> dict[str, Any] | None: + with self._connect() as conn: + return _story(conn.execute("SELECT * FROM stories WHERE calibre_id = ?", (calibre_id,)).fetchone()) + + def list_stories(self) -> list[dict[str, Any]]: + with self._connect() as conn: + rows = conn.execute(""" + SELECT s.id, s.url, s.site, s.title, s.author, s.calibre_id, s.auto_update, s.status, + s.message, s.last_checked, s.last_changed, s.ingested_at, s.locked_fields, + (SELECT COUNT(*) FROM chapters c WHERE c.story_id = s.id AND c.included = 1) AS chapter_count, + (SELECT COUNT(*) FROM chapters c WHERE c.story_id = s.id AND c.included = 1 + AND c.on_site = 0) AS offsite_count + FROM stories s ORDER BY s.title COLLATE NOCASE""").fetchall() + return [_story(r) for r in rows] + + def update_story(self, story_id: int, **fields: Any) -> None: + if not fields: + return + unknown = set(fields) - _STORY_COLUMNS + if unknown: + raise ValueError(f"Unknown story columns: {sorted(unknown)}") + values = [json.dumps(v) if k in _JSON_COLUMNS else v for k, v in fields.items()] + assignments = ", ".join(f"{k} = ?" for k in fields) + with self._connect() as conn: + conn.execute(f"UPDATE stories SET {assignments} WHERE id = ?", (*values, story_id)) # noqa: S608 (checked columns) + + def delete_story(self, story_id: int) -> None: + with self._connect() as conn: + conn.execute("DELETE FROM stories WHERE id = ?", (story_id,)) + conn.execute("DELETE FROM settings WHERE story_id = ?", (story_id,)) + + # Chapters + + def list_chapters(self, story_id: int) -> list[dict[str, Any]]: + with self._connect() as conn: + rows = conn.execute( + """ + SELECT c.*, (SELECT COUNT(*) FROM revisions r WHERE r.chapter_id = c.id) AS revisions + FROM chapters c WHERE c.story_id = ? ORDER BY c.position""", + (story_id,), + ).fetchall() + return [dict(r) for r in rows] + + def chapters_for_build(self, story_id: int) -> list[dict[str, Any]]: + """Included chapters in book order, with the HTML of their current revision.""" + with self._connect() as conn: + rows = conn.execute( + """ + SELECT c.id, c.url, c.title, c.position, r.html + FROM chapters c JOIN revisions r ON r.chapter_id = c.id AND r.rev = c.current_rev + WHERE c.story_id = ? AND c.included = 1 ORDER BY c.position""", + (story_id,), + ).fetchall() + return [dict(r) for r in rows] + + def current_revisions(self, story_id: int) -> dict[int, dict[str, Any]]: + """The rev, text_hash and source of the current revision of each chapter, by chapter id.""" + with self._connect() as conn: + rows = conn.execute( + """ + SELECT c.id, r.rev, r.text_hash, r.source FROM chapters c + JOIN revisions r ON r.chapter_id = c.id AND r.rev = c.current_rev + WHERE c.story_id = ?""", + (story_id,), + ).fetchall() + return {r["id"]: {"rev": r["rev"], "text_hash": r["text_hash"], "source": r["source"]} for r in rows} + + def replace_revision(self, chapter_id: int, rev: int, html: str, text_hash: str) -> None: + with self._connect() as conn: + conn.execute( + "UPDATE revisions SET html = ?, text_hash = ? WHERE chapter_id = ? AND rev = ?", + (html, text_hash, chapter_id, rev), + ) + + def get_chapter(self, chapter_id: int) -> dict[str, Any] | None: + with self._connect() as conn: + row = conn.execute("SELECT * FROM chapters WHERE id = ?", (chapter_id,)).fetchone() + return dict(row) if row else None + + def add_chapter( + self, story_id: int, url: str, title: str, html: str, text_hash: str, source: str, position: int + ) -> int: + with self._connect() as conn: + cur = conn.execute( + "INSERT INTO chapters (story_id, position, url, title, added_at) VALUES (?, ?, ?, ?, ?)", + (story_id, position, url, title, now()), + ) + chapter_id = int(cur.lastrowid) + conn.execute( + "INSERT INTO revisions (chapter_id, rev, html, text_hash, source, fetched_at) " + "VALUES (?, 1, ?, ?, ?, ?)", + (chapter_id, html, text_hash, source, now()), + ) + return chapter_id + + def add_chapters(self, story_id: int, chapters: list[dict[str, str]], source: str, order: list[str]) -> list[int]: + """Add chapters and put the story in the order of `order`, a list of chapter URLs, in one transaction. + + Each chapter needs url, title, html and text_hash. Chapters that `order` + leaves out go after the others and keep their order. + """ + with self._connect() as conn: + ids = [] + for c in chapters: + cur = conn.execute( + "INSERT INTO chapters (story_id, position, url, title, added_at) VALUES (?, 0, ?, ?, ?)", + (story_id, c["url"], c["title"], now()), + ) + ids.append(int(cur.lastrowid)) + conn.execute( + "INSERT INTO revisions (chapter_id, rev, html, text_hash, source, fetched_at) " + "VALUES (?, 1, ?, ?, ?, ?)", + (ids[-1], c["html"], c["text_hash"], source, now()), + ) + rows = conn.execute( + "SELECT id, url FROM chapters WHERE story_id = ? ORDER BY position, id", (story_id,) + ).fetchall() + by_url = {r["url"]: r["id"] for r in rows} + placed = [by_url[u] for u in dict.fromkeys(order) if u in by_url] + seen = set(placed) + rest = [r["id"] for r in rows if r["id"] not in seen] + conn.executemany("UPDATE chapters SET position = ? WHERE id = ?", list(enumerate(placed + rest))) + return ids + + def add_revision(self, chapter_id: int, html: str, text_hash: str, source: str) -> int: + """Store a new revision and make it the current one.""" + with self._connect() as conn: + rev = conn.execute( + "SELECT COALESCE(MAX(rev), 0) + 1 FROM revisions WHERE chapter_id = ?", (chapter_id,) + ).fetchone()[0] + conn.execute( + "INSERT INTO revisions (chapter_id, rev, html, text_hash, source, fetched_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (chapter_id, rev, html, text_hash, source, now()), + ) + conn.execute("UPDATE chapters SET current_rev = ?, changed_at = ? WHERE id = ?", (rev, now(), chapter_id)) + return int(rev) + + def update_chapter(self, chapter_id: int, **fields: Any) -> None: + if not fields: + return + unknown = set(fields) - {"position", "url", "title", "on_site", "included", "current_rev", "changed_at"} + if unknown: + raise ValueError(f"Unknown chapter columns: {sorted(unknown)}") + assignments = ", ".join(f"{k} = ?" for k in fields) + with self._connect() as conn: + conn.execute( + f"UPDATE chapters SET {assignments} WHERE id = ?", # noqa: S608 (checked columns) + (*fields.values(), chapter_id), + ) + + def set_positions(self, story_id: int, chapter_ids: list[int]) -> None: + with self._connect() as conn: + conn.executemany( + "UPDATE chapters SET position = ? WHERE id = ? AND story_id = ?", + [(i, cid, story_id) for i, cid in enumerate(chapter_ids)], + ) + + def list_revisions(self, chapter_id: int) -> list[dict[str, Any]]: + with self._connect() as conn: + rows = conn.execute( + "SELECT rev, text_hash, source, fetched_at, LENGTH(html) AS size FROM revisions " + "WHERE chapter_id = ? ORDER BY rev", + (chapter_id,), + ).fetchall() + return [dict(r) for r in rows] + + def delete_revision(self, chapter_id: int, rev: int) -> None: + with self._connect() as conn: + conn.execute("DELETE FROM revisions WHERE chapter_id = ? AND rev = ?", (chapter_id, rev)) + + def revision_html(self, chapter_id: int, rev: int) -> str | None: + with self._connect() as conn: + row = conn.execute( + "SELECT html FROM revisions WHERE chapter_id = ? AND rev = ?", (chapter_id, rev) + ).fetchone() + return row["html"] if row else None + + # Images + + def put_image(self, story_id: int, url: str, ext: str, mime: str, data: bytes, is_cover: bool = False) -> None: + with self._connect() as conn: + if is_cover: + conn.execute("DELETE FROM images WHERE story_id = ? AND is_cover = 1", (story_id,)) + conn.execute( + "INSERT OR REPLACE INTO images (story_id, url, ext, mime, data, is_cover) VALUES (?, ?, ?, ?, ?, ?)", + (story_id, url, ext, mime, data, int(is_cover)), + ) + + def image_urls(self, story_id: int) -> set: + with self._connect() as conn: + rows = conn.execute("SELECT url FROM images WHERE story_id = ? AND is_cover = 0", (story_id,)) + return {r["url"] for r in rows} + + def images(self, story_id: int) -> list[dict[str, Any]]: + with self._connect() as conn: + rows = conn.execute( + "SELECT url, ext, mime, data FROM images WHERE story_id = ? AND is_cover = 0", (story_id,) + ).fetchall() + return [dict(r) for r in rows] + + def cover(self, story_id: int) -> dict[str, Any] | None: + with self._connect() as conn: + row = conn.execute( + "SELECT url, ext, mime, data FROM images WHERE story_id = ? AND is_cover = 1", (story_id,) + ).fetchone() + return dict(row) if row else None + + # Replacement rules. story_id None holds the global rules. + + def rules(self, story_id: int | None) -> list[dict[str, Any]]: + with self._connect() as conn: + rows = conn.execute( + "SELECT pattern, replacement, flags, enabled, note FROM rules WHERE story_id IS ? ORDER BY position", + (story_id,), + ).fetchall() + return [dict(r, enabled=bool(r["enabled"])) for r in rows] + + def set_rules(self, story_id: int | None, rules: list[dict[str, Any]]) -> None: + with self._connect() as conn: + conn.execute("DELETE FROM rules WHERE story_id IS ?", (story_id,)) + conn.executemany( + "INSERT INTO rules (story_id, position, pattern, replacement, flags, enabled, note) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + [ + ( + story_id, + i, + r["pattern"], + r.get("replacement", ""), + r.get("flags", ""), + int(bool(r.get("enabled", True))), + r.get("note", ""), + ) + for i, r in enumerate(rules) + ], + ) + + # Settings. Story 0 holds the global settings. + + def get_setting(self, story_id: int | None, name: str) -> Any: + with self._connect() as conn: + row = conn.execute( + "SELECT value FROM settings WHERE story_id = ? AND name = ?", (story_id or 0, name) + ).fetchone() + return json.loads(row["value"]) if row else None + + def story_settings(self, name: str) -> dict[int, Any]: + """The value of one setting for each story that has it.""" + with self._connect() as conn: + rows = conn.execute("SELECT story_id, value FROM settings WHERE name = ? AND story_id != 0", (name,)) + return {r["story_id"]: json.loads(r["value"]) for r in rows} + + def set_setting(self, story_id: int | None, name: str, value: Any) -> None: + """Store a setting. None removes it.""" + with self._connect() as conn: + if value is None: + conn.execute("DELETE FROM settings WHERE story_id = ? AND name = ?", (story_id or 0, name)) + else: + conn.execute( + "INSERT OR REPLACE INTO settings (story_id, name, value) VALUES (?, ?, ?)", + (story_id or 0, name, json.dumps(value)), + ) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..7788b19 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,9 @@ +"""Point every directory at a temporary folder before the app modules load.""" + +import os +import tempfile +from pathlib import Path + +_tmp = Path(tempfile.mkdtemp(prefix="bindery-tests-")) +for name, sub in [("CONFIG_DIR", "config"), ("INGEST_DIR", "ingest"), ("TMP_DIR", "tmp")]: + os.environ[name] = str(_tmp / sub) diff --git a/tests/test_diffing.py b/tests/test_diffing.py new file mode 100644 index 0000000..0c313da --- /dev/null +++ b/tests/test_diffing.py @@ -0,0 +1,33 @@ +from bindery.diffing import diff_html, html_to_blocks, text_hash + + +def test_random_class_names_do_not_count_as_changes(): + a = '

Hello there.

Second

' + b = '

Hello there.

Second

' + assert text_hash(a) == text_hash(b) + assert html_to_blocks(a) == ["Hello there.", "Second"] + + +def test_text_changes_change_the_hash(): + assert text_hash("

one

") != text_hash("

two

") + + +def test_diff_counts_and_word_level_changes(): + old = "

Same.

The cat sat.

Gone.

" + new = "

Same.

The dog sat.

Gone.

Added.

" + result = diff_html(old, new) + assert (result["changed"], result["added"], result["removed"]) == (1, 1, 0) + assert "catdog" in result["html"] + assert not result["identical"] + + +def test_diff_output_is_escaped(): + result = diff_html("

a

", "

<script>alert(1)</script>

") + assert "