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.
272 lines
11 KiB
Python
272 lines
11 KiB
Python
#!/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())
|