Coverage for src/lilbee/wiki/rerender.py: 100%
124 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Render the current path a wiki page or draft shows for each source it names by id."""
3from __future__ import annotations
5import json
6import logging
7from typing import TYPE_CHECKING, Any
9import yaml
11from lilbee.wiki.citations import fence_flags, format_source_ref
12from lilbee.wiki.grammar import FOOTNOTE_RE
13from lilbee.wiki.shared import WIKI_BUILD_LOCK, frontmatter_span, rewrite_keeping_time
15if TYPE_CHECKING:
16 from collections.abc import Iterable, Mapping
17 from pathlib import Path
19 from lilbee.core.config import Config
20 from lilbee.data.store import CitationRecord, Store
22log = logging.getLogger(__name__)
24_SOURCES_PREFIX = "sources: "
25_PROVENANCE_LINE = "provenance:"
26# What follows the path in a footnote reference that names more than its path.
27_AFTER_PATH = ","
30def _shown_before(ref: str, record: CitationRecord) -> str | None:
31 """The path footnote text *ref* shows when it is *record* under another path, else None."""
32 path = record["source_filename"]
33 suffix = format_source_ref(record)[len(path) :]
34 before = ref[: len(ref) - len(suffix)]
35 if not ref.endswith(suffix) or not before or before == path:
36 return None
37 return before
40def _rendered_footnotes(
41 lines: list[str], records: Mapping[str, CitationRecord]
42) -> tuple[list[str], dict[str, str]]:
43 """*lines* with each footnote that shows a stale path rendered again; each old path to the new.
45 Two footnotes of one source that render alike are written once.
46 """
47 renamed: dict[str, str] = {}
48 out: list[str] = []
49 for line, fenced in zip(lines, fence_flags(lines), strict=True):
50 match = None if fenced else FOOTNOTE_RE.match(line)
51 record = None if match is None else records.get(match.group(1))
52 before = None if match is None or record is None else _shown_before(match.group(2), record)
53 if match is None or record is None or before is None:
54 out.append(line)
55 continue
56 renamed[before] = record["source_filename"]
57 rendered = f"[^{match.group(1)}]: {format_source_ref(record)}"
58 if rendered not in out:
59 out.append(rendered)
60 return out, renamed
63def _sources_line(line: str, renamed: Mapping[str, str]) -> str:
64 """A ``sources:`` frontmatter line with each renamed path replaced in place."""
65 if not line.startswith(_SOURCES_PREFIX):
66 return line
67 try:
68 names = json.loads(line[len(_SOURCES_PREFIX) :])
69 except ValueError:
70 return line
71 # Frontmatter is untyped: a hand-edited page can hold anything here.
72 if not isinstance(names, list) or not all(isinstance(name, str) for name in names):
73 return line
74 if not any(name in renamed for name in names):
75 return line
76 shown = dict.fromkeys(renamed.get(name, name) for name in names)
77 return _SOURCES_PREFIX + json.dumps(list(shown))
80def _provenance(block: list[str], renamed: Mapping[str, str]) -> list[str]:
81 """The lines of a ``provenance:`` block with each renamed chunk source replaced."""
82 try:
83 parsed: Any = yaml.safe_load("\n".join(block))
84 chunks = parsed["provenance"]["chunks"]
85 if not any(chunk["source"] in renamed for chunk in chunks):
86 return block
87 for chunk in chunks:
88 chunk["source"] = renamed.get(chunk["source"], chunk["source"])
89 except (yaml.YAMLError, KeyError, TypeError, AttributeError):
90 return block # a hand-edited block is left as it is
91 # Two sources that now show one path name each of its chunks once.
92 parsed["provenance"]["chunks"] = [
93 chunk for index, chunk in enumerate(chunks) if chunk not in chunks[:index]
94 ]
95 return yaml.safe_dump(parsed, sort_keys=False).rstrip("\n").split("\n")
98def _frontmatter(lines: list[str], renamed: Mapping[str, str]) -> list[str]:
99 """Frontmatter *lines* with the renamed paths of the sources list and the provenance block."""
100 out: list[str] = []
101 index = 0
102 while index < len(lines):
103 if lines[index] != _PROVENANCE_LINE:
104 out.append(_sources_line(lines[index], renamed))
105 index += 1
106 continue
107 end = index + 1
108 while end < len(lines) and lines[end].startswith((" ", "-")):
109 end += 1
110 out.extend(_provenance(lines[index:end], renamed))
111 index = end
112 return out
115def _with_frontmatter(lines: list[str], renamed: Mapping[str, str]) -> str:
116 """Page *lines* with the renamed paths of the frontmatter replaced, joined."""
117 span = frontmatter_span(lines)
118 if span is not None:
119 start, end = span
120 lines[start + 1 : end] = _frontmatter(lines[start + 1 : end], renamed)
121 return "\n".join(lines)
124def rerendered_page(text: str, records: Iterable[CitationRecord]) -> str:
125 """Page *text* showing the current path of each source its citation *records* name."""
126 by_key = {record["citation_key"]: record for record in records}
127 lines, renamed = _rendered_footnotes(text.split("\n"), by_key)
128 if not renamed:
129 return text
130 return _with_frontmatter(lines, renamed)
133def _renamed_footnote(line: str, renamed: Mapping[str, str]) -> str:
134 """A footnote whose reference starts with a path of *renamed*, showing its new path."""
135 match = FOOTNOTE_RE.match(line)
136 if match is None:
137 return line
138 ref = match.group(2)
139 for before in sorted(renamed, key=len, reverse=True):
140 if ref == before or ref.startswith(f"{before}{_AFTER_PATH}"):
141 return line[: match.start(2)] + renamed[before] + ref[len(before) :]
142 return line
145def renamed_page(text: str, renamed: Mapping[str, str]) -> str:
146 """Page *text* showing each path of *renamed* as its new path in footnotes and frontmatter."""
147 if not renamed:
148 return text
149 lines = text.split("\n")
150 shown = [
151 line if fenced else _renamed_footnote(line, renamed)
152 for line, fenced in zip(lines, fence_flags(lines), strict=True)
153 ]
154 return _with_frontmatter(shown, renamed)
157def _rerender(store: Store, page: Path, wiki_source: str) -> Path | None:
158 """Write *page* again when a source it cites shows another path; the page, or None."""
159 try:
160 text = page.read_text(encoding="utf-8")
161 except (OSError, UnicodeDecodeError) as exc:
162 log.warning("Left %s as it is, it cannot be read: %s", page, exc)
163 return None
164 rendered = rerendered_page(text, store.get_citations_for_wiki(wiki_source))
165 if rendered == text:
166 return None
167 # The page keeps its time: the stale check of a pending draft compares the two.
168 rewrite_keeping_time(page, rendered)
169 return page
172def rerender_pages(store: Store, config: Config, wiki_sources: Iterable[str]) -> list[Path]:
173 """Render again the citation block and frontmatter of *wiki_sources*; the pages written."""
174 with WIKI_BUILD_LOCK:
175 written = [
176 _rerender(store, config.data_root / wiki_source, wiki_source)
177 for wiki_source in sorted(wiki_sources)
178 ]
179 return [page for page in written if page is not None]
182def _shows_other_paths(page: Path, records: Iterable[CitationRecord]) -> bool:
183 """Whether *page* shows a path other than the one its citation *records* name now."""
184 try:
185 text = page.read_text(encoding="utf-8")
186 except (OSError, UnicodeDecodeError):
187 return False # a page that is gone or cannot be read is left as it is
188 return rerendered_page(text, records) != text
191def rerender_stale_pages(
192 store: Store, config: Config, citations: Mapping[str, Iterable[CitationRecord]]
193) -> list[Path]:
194 """Render again each page of *citations* that shows a path its cited source no longer has."""
195 stale = [
196 wiki_source
197 for wiki_source, records in citations.items()
198 if _shows_other_paths(config.data_root / wiki_source, records)
199 ]
200 return rerender_pages(store, config, stale) if stale else []