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.
This commit is contained in:
@@ -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"(</(?:p|div|h\d|li|blockquote|tr|table|pre)>|<br\s*/?>|<hr\s*/?>)", 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"<del>{escape(''.join(old_words[i1:i2]))}</del>")
|
||||
if j2 > j1:
|
||||
out.append(f"<ins>{escape(''.join(new_words[j1:j2]))}</ins>")
|
||||
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'<p class="diff-eq">{escape(text)}</p>' for text in head)
|
||||
if skipped > 0:
|
||||
parts.append(f'<p class="diff-skip">{skipped} unchanged paragraph{"s" if skipped != 1 else ""}</p>')
|
||||
parts.extend(f'<p class="diff-eq">{escape(text)}</p>' for text in tail)
|
||||
elif op == "replace":
|
||||
pairs = min(i2 - i1, j2 - j1)
|
||||
for k in range(pairs):
|
||||
parts.append(f'<p class="diff-chg">{_word_diff(old[i1 + k], new[j1 + k])}</p>')
|
||||
changed += pairs
|
||||
for text in old[i1 + pairs : i2]:
|
||||
parts.append(f'<p class="diff-del">{escape(text)}</p>')
|
||||
removed += 1
|
||||
for text in new[j1 + pairs : j2]:
|
||||
parts.append(f'<p class="diff-ins">{escape(text)}</p>')
|
||||
added += 1
|
||||
elif op == "delete":
|
||||
parts.extend(f'<p class="diff-del">{escape(text)}</p>' for text in old[i1:i2])
|
||||
removed += i2 - i1
|
||||
elif op == "insert":
|
||||
parts.extend(f'<p class="diff-ins">{escape(text)}</p>' 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)
|
||||
@@ -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 :]
|
||||
@@ -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 <p> is split into several <p>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 <br> tags split inside a block into one paragraph per piece.
|
||||
|
||||
Only a <br> that is a direct child of a block counts, so a <br> inside
|
||||
inline markup such as <em> stays. Empty pieces are dropped, so a run of
|
||||
<br> tags makes one paragraph break. A <p> becomes several <p>s with the
|
||||
same attributes. In other blocks, the pieces go into new <p>s, and child
|
||||
blocks stay as they are.
|
||||
"""
|
||||
if "<br" not in html.lower():
|
||||
return html
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
for block in soup.find_all(CONTAINERS):
|
||||
if not any(isinstance(c, Tag) and c.name == "br" for c in block.children):
|
||||
continue
|
||||
parts: list[Any] = []
|
||||
run: list[Any] = []
|
||||
for child in list(block.children):
|
||||
child.extract()
|
||||
if isinstance(child, Tag) and child.name == "br":
|
||||
parts.append(run)
|
||||
run = []
|
||||
elif isinstance(child, Tag) and child.name in BLOCK_TAGS:
|
||||
parts.extend([run, child])
|
||||
run = []
|
||||
else:
|
||||
run.append(child)
|
||||
parts.append(run)
|
||||
nodes = []
|
||||
for part in parts:
|
||||
if isinstance(part, Tag):
|
||||
nodes.append(part)
|
||||
elif not _is_empty(part):
|
||||
attrs = dict(block.attrs) if block.name == "p" else {}
|
||||
if any(n.name == "p" for n in nodes if isinstance(n, Tag)):
|
||||
attrs.pop("id", None) # an id must stay unique
|
||||
paragraph = soup.new_tag("p", attrs=attrs)
|
||||
paragraph.extend(part)
|
||||
nodes.append(paragraph)
|
||||
if block.name == "p":
|
||||
for node in nodes:
|
||||
block.insert_before(node)
|
||||
block.decompose()
|
||||
else:
|
||||
block.extend(nodes)
|
||||
return str(soup)
|
||||
@@ -0,0 +1,441 @@
|
||||
"""SQLite storage for tracked stories, chapter revisions, images and replacement rules.
|
||||
|
||||
The store, not the epub, is the source of truth for a story. Every epub is built
|
||||
from the stored chapters, so rules can change and chapters can disappear from the
|
||||
site without losing text.
|
||||
"""
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
from collections.abc import Iterator
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS stories (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url TEXT NOT NULL UNIQUE,
|
||||
site TEXT NOT NULL,
|
||||
title TEXT NOT NULL DEFAULT '',
|
||||
author TEXT NOT NULL DEFAULT '',
|
||||
calibre_id INTEGER,
|
||||
calibre_uuid TEXT,
|
||||
metahtml TEXT,
|
||||
locked_fields TEXT NOT NULL DEFAULT '[]',
|
||||
pinned TEXT NOT NULL DEFAULT '{}',
|
||||
site_values TEXT NOT NULL DEFAULT '{}',
|
||||
auto_update INTEGER NOT NULL DEFAULT 1,
|
||||
status TEXT NOT NULL DEFAULT '',
|
||||
message TEXT NOT NULL DEFAULT '',
|
||||
ingested_at TEXT,
|
||||
last_checked TEXT,
|
||||
last_changed TEXT,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS chapters (
|
||||
id INTEGER PRIMARY KEY,
|
||||
story_id INTEGER NOT NULL REFERENCES stories(id) ON DELETE CASCADE,
|
||||
position INTEGER NOT NULL,
|
||||
url TEXT NOT NULL,
|
||||
title TEXT NOT NULL,
|
||||
on_site INTEGER NOT NULL DEFAULT 1,
|
||||
included INTEGER NOT NULL DEFAULT 1,
|
||||
current_rev INTEGER NOT NULL DEFAULT 1,
|
||||
added_at TEXT NOT NULL,
|
||||
changed_at TEXT,
|
||||
UNIQUE (story_id, url)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS revisions (
|
||||
chapter_id INTEGER NOT NULL REFERENCES chapters(id) ON DELETE CASCADE,
|
||||
rev INTEGER NOT NULL,
|
||||
html TEXT NOT NULL,
|
||||
text_hash TEXT NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
fetched_at TEXT NOT NULL,
|
||||
PRIMARY KEY (chapter_id, rev)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS images (
|
||||
story_id INTEGER NOT NULL REFERENCES stories(id) ON DELETE CASCADE,
|
||||
url TEXT NOT NULL,
|
||||
ext TEXT NOT NULL,
|
||||
mime TEXT NOT NULL,
|
||||
data BLOB NOT NULL,
|
||||
is_cover INTEGER NOT NULL DEFAULT 0,
|
||||
PRIMARY KEY (story_id, url, is_cover)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS rules (
|
||||
id INTEGER PRIMARY KEY,
|
||||
story_id INTEGER REFERENCES stories(id) ON DELETE CASCADE,
|
||||
position INTEGER NOT NULL,
|
||||
pattern TEXT NOT NULL,
|
||||
replacement TEXT NOT NULL DEFAULT '',
|
||||
flags TEXT NOT NULL DEFAULT '',
|
||||
enabled INTEGER NOT NULL DEFAULT 1,
|
||||
note TEXT NOT NULL DEFAULT ''
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS settings (
|
||||
story_id INTEGER NOT NULL, -- 0 for the global settings
|
||||
name TEXT NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
PRIMARY KEY (story_id, name)
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS chapters_story ON chapters (story_id, position);
|
||||
CREATE INDEX IF NOT EXISTS rules_story ON rules (story_id, position);
|
||||
"""
|
||||
|
||||
_JSON_COLUMNS = ("locked_fields", "pinned", "site_values")
|
||||
_STORY_COLUMNS = {
|
||||
"url",
|
||||
"site",
|
||||
"title",
|
||||
"author",
|
||||
"calibre_id",
|
||||
"calibre_uuid",
|
||||
"metahtml",
|
||||
"locked_fields",
|
||||
"pinned",
|
||||
"site_values",
|
||||
"auto_update",
|
||||
"status",
|
||||
"message",
|
||||
"ingested_at",
|
||||
"last_checked",
|
||||
"last_changed",
|
||||
}
|
||||
|
||||
|
||||
def now() -> 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)),
|
||||
)
|
||||
@@ -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)
|
||||
@@ -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 = '<div><p class="cnMwZG">Hello <em>there</em>.</p><p class="x1">Second</p></div>'
|
||||
b = '<div><p class="zzQ99">Hello <em>there</em>.</p><p class="y2">Second</p></div>'
|
||||
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("<p>one</p>") != text_hash("<p>two</p>")
|
||||
|
||||
|
||||
def test_diff_counts_and_word_level_changes():
|
||||
old = "<p>Same.</p><p>The cat sat.</p><p>Gone.</p>"
|
||||
new = "<p>Same.</p><p>The dog sat.</p><p>Gone.</p><p>Added.</p>"
|
||||
result = diff_html(old, new)
|
||||
assert (result["changed"], result["added"], result["removed"]) == (1, 1, 0)
|
||||
assert "<del>cat</del><ins>dog</ins>" in result["html"]
|
||||
assert not result["identical"]
|
||||
|
||||
|
||||
def test_diff_output_is_escaped():
|
||||
result = diff_html("<p>a</p>", "<p><script>alert(1)</script></p>")
|
||||
assert "<script>" not in result["html"]
|
||||
assert "<script>" in result["html"]
|
||||
|
||||
|
||||
def test_source_diff_sees_markup_changes():
|
||||
old, new = "<p>Hi</p>", '<p class="x">Hi</p>'
|
||||
assert diff_html(old, new)["identical"]
|
||||
assert not diff_html(old, new, source=True)["identical"]
|
||||
@@ -0,0 +1,55 @@
|
||||
from bindery.merge import merge_chapter_lists
|
||||
|
||||
|
||||
def urls(*numbers):
|
||||
return [f"c{n}" for n in numbers]
|
||||
|
||||
|
||||
def test_stubbed_chapters_stay_and_new_chapter_goes_last():
|
||||
existing = urls(*range(1, 61))
|
||||
site = urls(1, 2, 3, 4, *range(46, 62))
|
||||
assert merge_chapter_lists(existing, site) == urls(*range(1, 62))
|
||||
|
||||
|
||||
def test_new_chapter_goes_after_its_site_predecessor():
|
||||
assert merge_chapter_lists(urls(1, 2, 3), ["c1", "c2", "x", "c3"]) == ["c1", "c2", "x", "c3"]
|
||||
|
||||
|
||||
def test_new_chapter_after_a_gap_goes_before_removed_chapters():
|
||||
# The author replaced 2-3 with a new chapter x.
|
||||
assert merge_chapter_lists(urls(1, 2, 3, 4), ["c1", "x", "c4"]) == ["c1", "x", "c2", "c3", "c4"]
|
||||
|
||||
|
||||
def test_new_chapter_before_everything_goes_first():
|
||||
assert merge_chapter_lists(urls(1, 2), ["p", "c1", "c2"]) == ["p", "c1", "c2"]
|
||||
|
||||
|
||||
def test_failed_chapter_later_lands_in_order():
|
||||
# c61 failed last time, c62 was added. The next run adds c61 before c62.
|
||||
assert merge_chapter_lists(urls(59, 60, 62), urls(59, 60, 61, 62)) == urls(59, 60, 61, 62)
|
||||
|
||||
|
||||
def test_existing_order_wins_and_duplicates_are_dropped():
|
||||
# The site moved b before a. Our order stays, and c is newer than every chapter we have.
|
||||
assert merge_chapter_lists(["a", "b", "b"], ["b", "a", "a", "c"]) == ["a", "b", "c"]
|
||||
|
||||
|
||||
def test_new_chapter_after_a_stubbed_tail_goes_last():
|
||||
# The site kept 1-5, removed 6-10 and added 11 and 12.
|
||||
assert merge_chapter_lists(urls(*range(1, 11)), urls(1, 2, 3, 4, 5, 11, 12)) == urls(*range(1, 13))
|
||||
|
||||
|
||||
def test_new_chapters_after_every_old_chapter_is_gone_go_last():
|
||||
assert merge_chapter_lists(urls(1, 2, 3), urls(7, 8)) == urls(1, 2, 3, 7, 8)
|
||||
|
||||
|
||||
def test_empty_lists():
|
||||
assert merge_chapter_lists([], urls(1, 2)) == urls(1, 2)
|
||||
assert merge_chapter_lists(urls(1), []) == urls(1)
|
||||
|
||||
|
||||
def test_older_copy_fills_the_gap_before_newer_chapters():
|
||||
existing = urls(*range(1, 6), *range(50, 61))
|
||||
epub = urls(*range(1, 46))
|
||||
assert merge_chapter_lists(existing, epub, newest_last=False) == urls(*range(1, 46), *range(50, 61))
|
||||
assert merge_chapter_lists(urls(1, 2), urls(1, 2, 3), newest_last=False) == urls(1, 2, 3)
|
||||
@@ -0,0 +1,90 @@
|
||||
import pytest
|
||||
|
||||
from bindery import rules as rules_module
|
||||
from bindery.rules import RuleError, apply_rules, breaks_to_paragraphs, compile_rules, normalize_rules
|
||||
|
||||
|
||||
def test_rules_apply_in_order_and_skip_disabled():
|
||||
rules = normalize_rules(
|
||||
[
|
||||
{"pattern": "cat", "replacement": "dog"},
|
||||
{"pattern": "dog", "replacement": "wolf", "enabled": False},
|
||||
{"pattern": r"(\w+) house", "replacement": r"\1 home", "flags": "I"},
|
||||
]
|
||||
)
|
||||
assert apply_rules("The cat is in the HOUSE", compile_rules(rules)) == "The dog is in the home"
|
||||
assert apply_rules("a cat House", compile_rules(rules)) == "a dog home"
|
||||
|
||||
|
||||
def test_global_rules_run_before_story_rules():
|
||||
global_rules = normalize_rules([{"pattern": "a", "replacement": "b"}])
|
||||
story_rules = normalize_rules([{"pattern": "b", "replacement": "c"}])
|
||||
assert apply_rules("a", compile_rules(global_rules, story_rules)) == "c"
|
||||
|
||||
|
||||
def test_flags_are_cleaned():
|
||||
assert normalize_rules([{"pattern": "x", "flags": "smi"}])[0]["flags"] == "ims"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"rule, message",
|
||||
[
|
||||
({"pattern": ""}, "empty"),
|
||||
({"pattern": "(open"}, "missing"),
|
||||
({"pattern": "a", "replacement": r"\3"}, "group"),
|
||||
({"pattern": "a", "flags": "q"}, "flag"),
|
||||
],
|
||||
)
|
||||
def test_bad_rules_name_the_rule(rule, message):
|
||||
with pytest.raises(RuleError) as info:
|
||||
normalize_rules([{"pattern": "fine"}, rule])
|
||||
assert info.value.index == 1
|
||||
assert message in str(info.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"rule",
|
||||
[
|
||||
{"pattern": r"(a)(?P<b>b)", "replacement": r"\2 \g<b> \1"},
|
||||
{"pattern": r"(?i)(a)", "replacement": r"\1"},
|
||||
{"pattern": "(a) # a comment", "replacement": r"\1", "flags": "x"},
|
||||
{"pattern": "a", "replacement": r"x\\y\n"},
|
||||
],
|
||||
)
|
||||
def test_good_replacements_pass(rule):
|
||||
assert normalize_rules([rule])[0]["replacement"] == rule["replacement"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("replacement", [r"\g<nope>", "\\", r"\q"])
|
||||
def test_bad_replacements_fail_before_a_match(replacement):
|
||||
with pytest.raises(RuleError):
|
||||
normalize_rules([{"pattern": "(a)", "replacement": replacement}])
|
||||
|
||||
|
||||
def test_a_slow_rule_stops_at_the_timeout(monkeypatch):
|
||||
monkeypatch.setattr(rules_module, "TIMEOUT", 0.1)
|
||||
steps = compile_rules(normalize_rules([{"pattern": "(a|aa)+$"}]))
|
||||
with pytest.raises(ValueError, match=r"takes longer than 0.1 seconds"):
|
||||
apply_rules("a" * 60 + "b", steps)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("html", "expected"),
|
||||
[
|
||||
("<div>One<br>Two<br/><br/>Three</div>", "<div><p>One</p><p>Two</p><p>Three</p></div>"),
|
||||
('<div><p class="x" id="p1">A<br>B</p></div>', '<div><p class="x" id="p1">A</p><p class="x">B</p></div>'),
|
||||
("<div>Loose<br><p>Block</p>After</div>", "<div><p>Loose</p><p>Block</p><p>After</p></div>"),
|
||||
("<div><p><em>a<br/>b</em></p></div>", "<div><p><em>a<br/>b</em></p></div>"),
|
||||
('<div>Text<br> <br><img src="i.png"/><br> </div>', '<div><p>Text</p><p><img src="i.png"/></p></div>'),
|
||||
("<div><p>No breaks</p></div>", "<div><p>No breaks</p></div>"),
|
||||
],
|
||||
)
|
||||
def test_breaks_become_paragraphs(html, expected):
|
||||
assert breaks_to_paragraphs(html) == expected
|
||||
|
||||
|
||||
def test_paragraphs_come_before_the_rules():
|
||||
rules = normalize_rules([{"pattern": "<p>", "replacement": '<p class="x">'}])
|
||||
steps = compile_rules(rules, paragraphs=True)
|
||||
assert apply_rules("<div>a<br>b</div>", steps) == '<div><p class="x">a</p><p class="x">b</p></div>'
|
||||
assert apply_rules("<div>a<br>b</div>", compile_rules(rules)) == "<div>a<br>b</div>"
|
||||
@@ -0,0 +1,71 @@
|
||||
from bindery.store import Store
|
||||
|
||||
|
||||
def test_chapters_revisions_and_rules(tmp_path):
|
||||
store = Store(tmp_path / "db.sqlite")
|
||||
sid = store.create_story("https://example/s/1", "example", "T", "A", ["title"])
|
||||
c1 = store.add_chapter(sid, "u1", "One", "<p>1</p>", "h1", "new", 0)
|
||||
c2 = store.add_chapter(sid, "u2", "Two", "<p>2</p>", "h2", "new", 1)
|
||||
store.set_positions(sid, [c2, c1])
|
||||
assert [c["url"] for c in store.chapters_for_build(sid)] == ["u2", "u1"]
|
||||
|
||||
assert store.add_revision(c1, "<p>1b</p>", "h1b", "recheck") == 2
|
||||
assert store.current_revisions(sid)[c1] == {"rev": 2, "text_hash": "h1b", "source": "recheck"}
|
||||
store.replace_revision(c1, 2, "<p>1c</p>", "h1c")
|
||||
assert store.current_revisions(sid)[c1]["text_hash"] == "h1c"
|
||||
assert store.revision_html(c1, 1) == "<p>1</p>"
|
||||
store.update_chapter(c1, current_rev=1, included=0)
|
||||
assert [c["url"] for c in store.chapters_for_build(sid)] == ["u2"]
|
||||
|
||||
store.set_rules(None, [{"pattern": "g"}])
|
||||
store.set_rules(sid, [{"pattern": "s", "flags": "i"}])
|
||||
assert [r["pattern"] for r in store.rules(None)] == ["g"]
|
||||
assert store.rules(sid)[0]["flags"] == "i"
|
||||
|
||||
story = store.get_story(sid)
|
||||
assert story["locked_fields"] == ["title"] and story["pinned"] == {}
|
||||
store.update_story(sid, pinned={"title": "Kept"})
|
||||
assert store.get_story_by_url("https://example/s/1")["pinned"] == {"title": "Kept"}
|
||||
|
||||
store.delete_story(sid)
|
||||
assert store.get_chapter(c1) is None
|
||||
assert store.rules(sid) == []
|
||||
assert store.rules(None) != []
|
||||
|
||||
|
||||
def test_cover_replaces_previous_cover(tmp_path):
|
||||
store = Store(tmp_path / "db.sqlite")
|
||||
sid = store.create_story("u", "s", "T", "A", [])
|
||||
store.put_image(sid, "c1", "jpg", "image/jpeg", b"1", is_cover=True)
|
||||
store.put_image(sid, "c2", "jpg", "image/jpeg", b"2", is_cover=True)
|
||||
store.put_image(sid, "i1", "png", "image/png", b"3")
|
||||
assert store.cover(sid)["url"] == "c2"
|
||||
assert [i["url"] for i in store.images(sid)] == ["i1"]
|
||||
|
||||
|
||||
def test_add_chapters_places_them_in_one_step(tmp_path):
|
||||
store = Store(tmp_path / "db.sqlite")
|
||||
sid = store.create_story("u", "s", "T", "A", [])
|
||||
store.add_chapter(sid, "u1", "One", "<p>1</p>", "h1", "new", 0)
|
||||
store.add_chapter(sid, "u3", "Three", "<p>3</p>", "h3", "new", 1)
|
||||
store.add_chapter(sid, "u9", "Nine", "<p>9</p>", "h9", "new", 2)
|
||||
new = [{"url": u, "title": u, "html": "<p/>", "text_hash": "h"} for u in ("u2", "u4")]
|
||||
ids = store.add_chapters(sid, new, "import", ["u1", "u2", "u3", "u4"])
|
||||
chapters = store.list_chapters(sid)
|
||||
assert [c["url"] for c in chapters] == ["u1", "u2", "u3", "u4", "u9"]
|
||||
assert [c["position"] for c in chapters] == [0, 1, 2, 3, 4]
|
||||
assert [c["id"] for c in chapters if c["url"] in ("u2", "u4")] == ids
|
||||
|
||||
|
||||
def test_settings(tmp_path):
|
||||
store = Store(tmp_path / "db.sqlite")
|
||||
sid = store.create_story("u", "s", "T", "A", [])
|
||||
assert store.get_setting(None, "paragraphs") is None
|
||||
store.set_setting(None, "paragraphs", True)
|
||||
store.set_setting(sid, "paragraphs", False)
|
||||
assert (store.get_setting(None, "paragraphs"), store.get_setting(sid, "paragraphs")) == (True, False)
|
||||
store.set_setting(sid, "paragraphs", None)
|
||||
assert store.get_setting(sid, "paragraphs") is None
|
||||
store.set_setting(sid, "paragraphs", True)
|
||||
store.delete_story(sid)
|
||||
assert store.get_setting(sid, "paragraphs") is None
|
||||
Reference in New Issue
Block a user