Add a script that cleans up smart punctuation in chapter revisions

Some revisions differ from others only by curly quotes, curly
apostrophes and the ellipsis. The script deletes those revisions and
keeps the one with the least smart punctuation. In revisions with other
text changes, it makes the punctuation straight and stores the new text
hash. It does not change the newest revision if it came from the site.

The script uses only the Python standard library, so you can run it on
the host against bindery.db. It is a dry run unless you give --apply,
and --apply makes a backup first.

scripts/remove_smart_quote_revisions.py

@@ -0,0 +1,271 @@
+#!/usr/bin/env python3
+"""Clean up smart punctuation in chapter revisions: curly quotes, curly apostrophes and the ellipsis.
+
+Usage:
+
+    python3 scripts/remove_smart_quote_revisions.py /path/to/config/bindery.db
+    python3 scripts/remove_smart_quote_revisions.py /path/to/config/bindery.db --apply
+
+The first command only lists what it would do. With --apply, the script copies
+the database to a backup file next to it, and then changes the revisions. Stop
+Bindery before you use --apply.
+
+The script compares the text of the revisions of each chapter, as Bindery does,
+so differences in the markup do not count. Smart punctuation is ‘ ’ “ ” and …,
+and its straight form is ' " and ...
+
+- If two or more revisions of a chapter differ only by smart punctuation, the
+  revision with the least smart punctuation stays, and the others go. If a
+  deleted revision was in the book, the revision that stays goes in the book.
+- If a revision has text that no other revision of its chapter has, the script
+  keeps it, and changes its smart punctuation to straight. Only the text
+  changes, not the tags or comments. The script stores the new text hash, so
+  Bindery compares the next site text with the new text.
+
+The script does not delete or change the newest revision of a chapter if it came
+from the site. Each recheck compares the site text with the newest revision, so
+the next recheck would add the site text again.
+
+Rebuild the stories that the script names, because the text in their books
+changed. The script uses only the Python standard library.
+"""
+
+import argparse
+import hashlib
+import re
+import sqlite3
+import sys
+from collections import defaultdict
+from datetime import datetime
+from html import unescape
+from html.parser import HTMLParser
+from pathlib import Path
+
+SMART = {"‘": "'", "’": "'", "“": '"', "”": '"', "…": "..."}
+STRAIGHT = str.maketrans(SMART)
+ENTITY = re.compile(r"&(?:#\d+|#[xX][0-9a-fA-F]+|[a-zA-Z]+);")
+MARKUP = re.compile(r"(<!--.*?-->|<[^>]*>)", re.S)
+IMPORT_SOURCES = {"import", "epub-import"}
+# The same tags as BLOCK_TAGS in bindery/diffing.py.
+BLOCK_TAGS = {
+    "p", "div", "h1", "h2", "h3", "h4", "h5", "h6", "li", "blockquote", "pre",
+    "tr", "table", "hr", "section", "article", "ul", "ol", "center",
+}  # fmt: skip
+
+
+class _TextParser(HTMLParser):
+    def __init__(self) -> None:
+        super().__init__(convert_charrefs=True)
+        self.parts: list[str] = []
+        self.hidden = 0
+
+    def handle_starttag(self, tag: str, attrs: list) -> None:
+        if tag in ("script", "style"):
+            self.hidden += 1
+        elif tag == "br" or tag in BLOCK_TAGS:
+            self.parts.append("\n")
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag in ("script", "style"):
+            self.hidden = max(0, self.hidden - 1)
+        elif tag in BLOCK_TAGS:
+            self.parts.append("\n")
+
+    def handle_data(self, data: str) -> None:
+        if not self.hidden:
+            self.parts.append(data)
+
+
+def text_blocks(markup: str) -> list[str]:
+    """The text of a chapter, one string per paragraph, as html_to_blocks in bindery/diffing.py makes it."""
+    parser = _TextParser()
+    parser.feed(markup or "")
+    parser.close()
+    lines = (re.sub(r"\s+", " ", line).strip() for line in "".join(parser.parts).split("\n"))
+    return [line for line in lines if line]
+
+
+def text_hash(markup: str) -> str:
+    """The hash of the text of a chapter, as text_hash in bindery/diffing.py makes it."""
+    return hashlib.sha256("\n".join(text_blocks(markup)).encode("utf-8")).hexdigest()
+
+
+def smart_count(text: str) -> int:
+    return sum(text.count(c) for c in SMART)
+
+
+def straighten_html(markup: str) -> str:
+    """Change the smart punctuation in the text of a chapter to straight, also where the HTML has it as an entity.
+
+    Tags and comments do not change, so a straight quote cannot end an attribute value.
+    """
+
+    def entity(match: re.Match) -> str:
+        return SMART.get(unescape(match.group(0)), match.group(0))
+
+    # With a group in the pattern, split() puts the tags and comments at the odd places.
+    parts = MARKUP.split(markup)
+    for i in range(0, len(parts), 2):
+        parts[i] = ENTITY.sub(entity, parts[i]).translate(STRAIGHT)
+    return "".join(parts)
+
+
+def plan(conn: sqlite3.Connection) -> list[dict]:
+    """For each chapter with revisions to change: what goes, what stays, what gets straight, and what is in the book."""
+    chapters = conn.execute(
+        """
+        SELECT c.id, c.title, c.current_rev, s.id AS story_id, s.title AS story_title
+        FROM chapters c JOIN stories s ON s.id = c.story_id
+        WHERE (SELECT COUNT(*) FROM revisions r WHERE r.chapter_id = c.id) > 1
+        ORDER BY s.title COLLATE NOCASE, c.position"""
+    ).fetchall()
+    actions = []
+    for chapter in chapters:
+        revisions = conn.execute(
+            "SELECT rev, source, html FROM revisions WHERE chapter_id = ? ORDER BY rev", (chapter["id"],)
+        ).fetchall()
+        newest = revisions[-1]["rev"]
+        sources = {r["rev"]: r["source"] for r in revisions}
+        markup = {r["rev"]: r["html"] for r in revisions}
+
+        def from_site(rev: int, newest: int = newest, sources: dict = sources) -> bool:
+            return rev == newest and sources[rev] not in IMPORT_SOURCES
+
+        groups = defaultdict(list)
+        smart = {}
+        for r in revisions:
+            text = "\n".join(text_blocks(r["html"]))
+            smart[r["rev"]] = smart_count(text)
+            groups[text.translate(STRAIGHT)].append(r["rev"])
+        delete, keep, skipped = [], [], []
+        for group in groups.values():
+            least = min(smart[rev] for rev in group)
+            if all(smart[rev] == least for rev in group):
+                continue
+            keep += [rev for rev in group if smart[rev] == least]
+            for rev in group:
+                if smart[rev] > least:
+                    (skipped if from_site(rev) else delete).append(rev)
+        straighten = []
+        if len(groups) > 1:
+            for group in groups.values():
+                left = [rev for rev in group if rev not in delete]
+                if len(left) == 1 and smart[left[0]] and left[0] not in skipped:
+                    (skipped if from_site(left[0]) else straighten).append(left[0])
+        if not delete and not straighten and not skipped:
+            continue
+        current = chapter["current_rev"]
+        if current in delete:
+            # The revision that stays in the group of the deleted one.
+            same = next(g for g in groups.values() if current in g)
+            current = max(rev for rev in same if smart[rev] == min(smart[x] for x in same))
+        straight = {rev: straighten_html(markup[rev]) for rev in straighten}
+        actions.append(
+            {
+                "chapter_id": chapter["id"],
+                "chapter": unescape(chapter["title"]),
+                "story": chapter["story_title"],
+                "delete": sorted(delete),
+                "keep": sorted(keep),
+                "straighten": sorted(straighten),
+                "straight": {rev: (html, text_hash(html)) for rev, html in straight.items()},
+                "skipped": sorted(skipped),
+                "sources": sources,
+                "old_current": chapter["current_rev"],
+                "new_current": current,
+            }
+        )
+    return actions
+
+
+def book_changed(action: dict) -> bool:
+    return action["new_current"] != action["old_current"] or action["new_current"] in action["straighten"]
+
+
+def apply(conn: sqlite3.Connection, actions: list[dict]) -> None:
+    now = datetime.now().isoformat(timespec="seconds")
+    with conn:
+        for a in actions:
+            if a["new_current"] != a["old_current"]:
+                conn.execute(
+                    "UPDATE chapters SET current_rev = ?, changed_at = ? WHERE id = ?",
+                    (a["new_current"], now, a["chapter_id"]),
+                )
+            conn.executemany(
+                "UPDATE revisions SET html = ?, text_hash = ? WHERE chapter_id = ? AND rev = ?",
+                [(html, digest, a["chapter_id"], rev) for rev, (html, digest) in a["straight"].items()],
+            )
+            conn.executemany(
+                "DELETE FROM revisions WHERE chapter_id = ? AND rev = ?",
+                [(a["chapter_id"], rev) for rev in a["delete"]],
+            )
+
+
+def backup(conn: sqlite3.Connection, db: Path) -> Path:
+    path = db.with_name(f"{db.stem}.before-quote-cleanup-{datetime.now():%Y%m%d-%H%M%S}{db.suffix}")
+    target = sqlite3.connect(path)
+    try:
+        conn.backup(target)
+    finally:
+        target.close()
+    return path
+
+
+def _revs(revs: list[int], sources: dict[int, str]) -> str:
+    return ", ".join(f"{rev} ({sources[rev]})" for rev in revs)
+
+
+def report(actions: list[dict]) -> None:
+    for a in actions:
+        print(f"{a['story']} / {a['chapter']}")
+        if a["delete"]:
+            print(f"  delete revision {_revs(a['delete'], a['sources'])}, keep {_revs(a['keep'], a['sources'])}")
+        if a["new_current"] != a["old_current"]:
+            print(f"  revision {a['new_current']} goes in the book instead of revision {a['old_current']}")
+        if a["straighten"]:
+            print(f"  make the punctuation straight in revision {_revs(a['straighten'], a['sources'])}")
+        if a["skipped"]:
+            print(
+                f"  not changed: revision {_revs(a['skipped'], a['sources'])} is the newest text from the site, "
+                "so the next recheck would add it again"
+            )
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("db", type=Path, help="Path to bindery.db")
+    parser.add_argument("--apply", action="store_true", help="Back up the database, then change the revisions")
+    args = parser.parse_args(argv)
+    if not args.db.is_file():
+        print(f"Error: no such file: {args.db}", file=sys.stderr)
+        return 1
+
+    conn = sqlite3.connect(args.db, timeout=30)
+    conn.row_factory = sqlite3.Row
+    try:
+        actions = plan(conn)
+        report(actions)
+        deletions = sum(len(a["delete"]) for a in actions)
+        straightened = sum(len(a["straighten"]) for a in actions)
+        if not deletions and not straightened:
+            print("No revisions to change.")
+            return 0
+        print(f"\n{deletions} revisions to delete, and {straightened} revisions to make straight.")
+        if not args.apply:
+            print("Nothing changed. Run again with --apply to change them.")
+            return 0
+        print(f"Backup: {backup(conn, args.db)}")
+        apply(conn, actions)
+        print(f"Deleted {deletions} revisions, and made {straightened} revisions straight.")
+        rebuild = sorted({a["story"] for a in actions if book_changed(a)})
+        if rebuild:
+            print("The text in the book changed for these stories. Rebuild them:")
+            for title in rebuild:
+                print(f"  - {title}")
+    finally:
+        conn.close()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_remove_smart_quote_revisions.py

@@ -0,0 +1,137 @@
+"""Tests for scripts/remove_smart_quote_revisions.py, a manual cleanup tool."""
+
+import pytest
+
+from bindery import diffing
+from bindery.store import Store
+from scripts.remove_smart_quote_revisions import main, straighten_html, text_blocks, text_hash
+
+CURLY = "<div><p class='a1'>“Hi,” he said.</p><p>It’s ‘fine’…</p></div>"
+STRAIGHT = "<div><p class='b2'>\"Hi,\" he said.</p><p>It's 'fine'...</p></div>"
+# The same text as CURLY, with the smart punctuation as entities.
+ENTITIES = "<div><p>&ldquo;Hi,&#8221; he said.</p><p>It&rsquo;s &#x2018;fine&#x2019;&hellip;</p></div>"
+
+
+@pytest.mark.parametrize(
+    "html",
+    [
+        CURLY,
+        ENTITIES,
+        "<p>One<br/>Two</p><script>x()</script><style>p{}</style>",
+        "<div>Loose <em>text</em><ul><li>a</li><li>b &amp; c</li></ul>  tail </div><hr/>",
+        "",
+    ],
+)
+def test_text_and_hash_match_bindery(html):
+    assert text_blocks(html) == diffing.html_to_blocks(html)
+    assert text_hash(html) == diffing.text_hash(html)
+
+
+def test_straighten_html_changes_only_the_text():
+    html = '<p title="“x”">“Hi,” it’s ‘fine’… &amp; &rsquo;<!-- “note” --></p>'
+    assert straighten_html(html) == "<p title=\"“x”\">\"Hi,\" it's 'fine'... &amp; '<!-- “note” --></p>"
+    assert diffing.html_to_blocks(straighten_html(ENTITIES)) == diffing.html_to_blocks(STRAIGHT)
+
+
+@pytest.fixture
+def db(tmp_path):
+    path = tmp_path / "bindery.db"
+    store = Store(path)
+    story_id = store.create_story("https://example.com/s/1", "example.com", "Story", "Writer", [])
+    return path, store, story_id
+
+
+def chapter(store, story_id, n, revisions, current=None):
+    """A chapter with revisions given as (html, source). The last one is in the book, unless `current` says."""
+    html, source = revisions[0]
+    cid = store.add_chapter(story_id, f"u{n}", f"Chapter {n}", html, diffing.text_hash(html), source, n)
+    for html, source in revisions[1:]:
+        store.add_revision(cid, html, diffing.text_hash(html), source)
+    if current:
+        store.update_chapter(cid, current_rev=current)
+    return cid
+
+
+def revs(store, cid):
+    return [r["rev"] for r in store.list_revisions(cid)]
+
+
+def test_dry_run_changes_nothing(db, capsys):
+    path, store, story_id = db
+    cid = chapter(store, story_id, 1, [(CURLY, "import"), (STRAIGHT, "recheck")])
+    assert main([str(path)]) == 0
+    assert revs(store, cid) == [1, 2]
+    out = capsys.readouterr().out
+    assert "delete revision 1 (import), keep 2 (recheck)" in out
+    assert "Run again with --apply" in out
+    assert not list(path.parent.glob("*before-quote-cleanup*"))
+
+
+def test_apply_deletes_revisions_that_differ_only_by_smart_punctuation(db, capsys):
+    path, store, story_id = db
+    # The smart import, then the straight site text.
+    older = chapter(store, story_id, 1, [(CURLY, "import"), (STRAIGHT, "recheck")])
+    # The same, but the user put the smart revision in the book.
+    chosen = chapter(store, story_id, 2, [(CURLY, "import"), (STRAIGHT, "recheck")], current=1)
+    # A smart import after the site text. It is the newest, but not from the site.
+    imported = chapter(store, story_id, 3, [(STRAIGHT, "new"), (ENTITIES, "import")])
+    # The site sends smart punctuation now. The next recheck would add it again.
+    site = chapter(store, story_id, 4, [(STRAIGHT, "new"), (CURLY, "recheck")])
+
+    assert main([str(path), "--apply"]) == 0
+    assert revs(store, older) == [2]
+    assert revs(store, chosen) == [2]
+    assert store.get_chapter(chosen)["current_rev"] == 2
+    assert revs(store, imported) == [1]
+    assert store.get_chapter(imported)["current_rev"] == 1
+    assert revs(store, site) == [1, 2]
+    out = capsys.readouterr().out
+    assert "3 revisions to delete, and 0 revisions to make straight" in out
+    assert "not changed: revision 2 (recheck) is the newest text from the site" in out
+    assert "Rebuild them:\n  - Story" in out
+    (backup,) = path.parent.glob("bindery.before-quote-cleanup-*.db")
+    assert revs(Store(backup), older) == [1, 2]
+
+
+def test_apply_makes_revisions_with_other_changes_straight(db, capsys):
+    path, store, story_id = db
+    changed = CURLY.replace("fine", "good")
+    # Three different texts. The newest is not from the site, so all of them get straight.
+    both = chapter(
+        store, story_id, 1, [(CURLY, "import"), (changed, "recheck"), (ENTITIES.replace("Hi", "Yo"), "import")]
+    )
+    # The newest is from the site, so only the older one gets straight.
+    older = chapter(store, story_id, 2, [(CURLY, "import"), (changed, "recheck")])
+    # A smart revision that also matches a straight one goes, and the other text gets straight.
+    mixed = chapter(store, story_id, 3, [(CURLY, "import"), (STRAIGHT, "new"), (changed, "import")], current=1)
+
+    assert main([str(path), "--apply"]) == 0
+
+    def smart(cid, rev):
+        text = "\n".join(diffing.html_to_blocks(store.revision_html(cid, rev)))
+        return any(c in text for c in "‘’“”…")
+
+    assert [smart(both, rev) for rev in (1, 2, 3)] == [False, False, False]
+    assert [smart(older, rev) for rev in (1, 2)] == [False, True]
+    assert revs(store, mixed) == [2, 3]
+    assert [smart(mixed, rev) for rev in (2, 3)] == [False, False]
+    assert store.get_chapter(mixed)["current_rev"] == 2
+    # Bindery compares the next site text with the stored hash, so it must be the hash of the new text.
+    for cid in (both, older, mixed):
+        for r in store.list_revisions(cid):
+            assert r["text_hash"] == diffing.text_hash(store.revision_html(cid, r["rev"]))
+    assert "class='a1'" in store.revision_html(both, 1)
+    out = capsys.readouterr().out
+    assert "make the punctuation straight in revision 1 (import), 2 (recheck), 3 (import)" in out
+    assert "1 revisions to delete, and 5 revisions to make straight" in out
+    # A second run finds nothing to do.
+    assert main([str(path)]) == 0
+    assert "not changed: revision 2 (recheck)" in capsys.readouterr().out
+
+
+def test_no_work_and_a_missing_file(db, tmp_path, capsys):
+    path, store, story_id = db
+    chapter(store, story_id, 1, [(STRAIGHT, "new")])
+    assert main([str(path), "--apply"]) == 0
+    assert "No revisions to change." in capsys.readouterr().out
+    assert main([str(tmp_path / "missing.db")]) == 1