From 6c2574cf0ff88263a89485dfe325c3fd069f8afe Mon Sep 17 00:00:00 2001 From: Timothy Allen Date: Sat, 26 Sep 2026 08:06:02 +0200 Subject: [PATCH] Store chapters, merge chapter lists, diff and apply rules The SQLite store keeps every chapter revision, so a chapter that the site removes stays in the book. The merge puts new chapters after the chapter that comes before them on the site. Diffs compare text, not HTML, and replacement rules run on chapter HTML when a book is built. --- bindery/__init__.py | 0 bindery/diffing.py | 129 ++++++++++++ bindery/merge.py | 34 ++++ bindery/rules.py | 192 ++++++++++++++++++ bindery/store.py | 441 ++++++++++++++++++++++++++++++++++++++++++ tests/conftest.py | 9 + tests/test_diffing.py | 33 ++++ tests/test_merge.py | 55 ++++++ tests/test_rules.py | 90 +++++++++ tests/test_store.py | 71 +++++++ 10 files changed, 1054 insertions(+) create mode 100644 bindery/__init__.py create mode 100644 bindery/diffing.py create mode 100644 bindery/merge.py create mode 100644 bindery/rules.py create mode 100644 bindery/store.py create mode 100644 tests/conftest.py create mode 100644 tests/test_diffing.py create mode 100644 tests/test_merge.py create mode 100644 tests/test_rules.py create mode 100644 tests/test_store.py 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 "