Coverage for src/lilbee/crawler/save.py: 100%
270 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Page addresses and saved names, and the per-page save to disk under the crawl lock."""
3from __future__ import annotations
5import hashlib
6import ipaddress
7import logging
8import os
9import re
10import string
11import tempfile
12import unicodedata
13from collections.abc import Callable
14from dataclasses import dataclass, field
15from datetime import UTC, datetime
16from pathlib import Path, PurePosixPath
17from urllib.parse import unquote
19from lilbee.core.config import cfg
20from lilbee.core.security import validate_path_within
21from lilbee.core.web_names import INDEX_STEM, PAGE_HASH_MARK, PAGE_SUFFIX, WEB_DIR
22from lilbee.crawler.models import CrawlResult
23from lilbee.crawler.url_filter import parsed_address, parsed_url
24from lilbee.data.store import CrawlPage, CrawlPages, Store
26log = logging.getLogger(__name__)
28_NOT_SAVED = "Page %s is not saved: %s holds a file or folder of its own; move or delete it"
30# The characters RFC 3986 section 2.3 calls unreserved: an escape of one names the character.
31_UNRESERVED = frozenset(string.ascii_letters + string.digits + "-._~")
32_PERCENT_ESCAPE = re.compile(r"%([0-9A-Fa-f]{2})")
33_MALFORMED_ESCAPE = re.compile(r"%(?![0-9A-Fa-f]{2})")
34_HEX = 16
35_QUERY_MARK = "?"
36_QUERY_SEPARATOR = "&"
37_QUERY_VALUE_MARK = "="
38# Hex digits of the sha256 of a page's canonical address in its file name: 64 bits.
39_HASH_HEX_DIGITS = 16
40_MAX_SLUG_BYTES = 80
41_MAX_PATH_BYTES = 200
42# The bytes a readable name may take: the path less a full host folder, "/", "~", hash and suffix.
43_MAX_NAME_BYTES = (
44 _MAX_PATH_BYTES
45 - _MAX_SLUG_BYTES
46 - len("/")
47 - len(PAGE_HASH_MARK)
48 - _HASH_HEX_DIGITS
49 - len(PAGE_SUFFIX)
50)
51# A slug keeps letters, digits and these; every other run of characters becomes one fill.
52_WORD_BREAKS = frozenset("-_.")
53_SLUG_KEPT = _WORD_BREAKS | {"@"}
54_SLUG_FILL = "-"
55_SLUG_FILL_RUN = re.compile(f"{_SLUG_FILL}{{2,}}")
56_SLUG_ENDS = ".- "
57_EMPTY_SLUG = "_"
58_DEVICE_MARK = "_"
59_EXTENSION_MARK = "."
60_PORT_MARK = ":"
61# An IPv6 host is written in brackets, its groups apart by colons.
62_IPV6_OPEN = "["
63_IPV6_CLOSE = "]"
64_IPV6_GROUP_MARK = ":"
65_UNKNOWN_HOST = "unknown"
66# Names Windows gives to devices, with or without an extension, in any letter case.
67_DEVICE_NAMES = frozenset(
68 {"CON", "PRN", "AUX", "NUL", "CONIN$", "CONOUT$"}
69 | {f"{port}{n}" for port in ("COM", "LPT") for n in [*string.digits, "¹", "²", "³"]}
70)
73def _normalized_escape(escape: re.Match[str]) -> str:
74 """The unreserved character *escape* names, or *escape* with upper-case hex digits."""
75 char = chr(int(escape.group(1), _HEX))
76 return char if char in _UNRESERVED else escape.group().upper()
79def _parameter_name(parameter: str) -> str:
80 """The name of one query parameter: the text before its first ``=``."""
81 return parameter.partition(_QUERY_VALUE_MARK)[0]
84def _sorted_query(address: str) -> str:
85 """*address* with its query parameters sorted by name, stable, each written as it was."""
86 before, mark, query = address.partition(_QUERY_MARK)
87 if not mark:
88 return address
89 parameters = sorted(query.split(_QUERY_SEPARATOR), key=_parameter_name)
90 return f"{before}{_QUERY_MARK}{_QUERY_SEPARATOR.join(parameters)}"
93def canonical_address(url: str) -> str:
94 """The key crawlberg 1.12.3's ``normalize_url`` gives the page at *url*, or *url* unparsed.
96 An address with a malformed escape keeps all its escapes as written: decoding the digits
97 after a bad ``%`` could make a valid escape the next time the key is made.
98 """
99 address = parsed_address(url)
100 if address is None:
101 return url
102 if _MALFORMED_ESCAPE.search(address):
103 return _sorted_query(address)
104 return _sorted_query(_PERCENT_ESCAPE.sub(_normalized_escape, address))
107def _cut(text: str, limit: int) -> str:
108 """*text* cut to at most *limit* UTF-8 bytes, on a character boundary."""
109 return text.encode("utf-8")[:limit].decode("utf-8", errors="ignore")
112def _filled(part: str) -> str:
113 """One escaped part of an address, decoded, NFC, with every other run of characters as ``-``."""
114 text = unicodedata.normalize("NFC", unquote(part, errors="replace"))
115 kept = "".join(c if c.isalnum() or c in _SLUG_KEPT else _SLUG_FILL for c in text)
116 return _SLUG_FILL_RUN.sub(_SLUG_FILL, kept)
119def _legal(text: str) -> str:
120 """*text* with legal ends, never empty, and a mark before a Windows device name."""
121 slug = text.strip(_SLUG_ENDS) or _EMPTY_SLUG
122 if slug.split(_EXTENSION_MARK)[0].upper() in _DEVICE_NAMES:
123 return _DEVICE_MARK + slug
124 return slug
127def _slug(part: str) -> str:
128 """A legal, readable file name for one escaped part of an address; lossy by design."""
129 return _legal(_cut(_filled(part), _MAX_SLUG_BYTES))
132def _without_suffix(text: str) -> str:
133 """*text* without its last ``.`` and the text after it, when text stands on both sides."""
134 stem, mark, suffix = text.rpartition(_EXTENSION_MARK)
135 return stem if stem and mark and suffix else text
138def _cut_at_word(text: str, limit: int) -> str:
139 """*text* cut to at most *limit* UTF-8 bytes, at the last separator when a word would break."""
140 head = _cut(text, limit)
141 if len(head) == len(text) or text[len(head)] in _WORD_BREAKS:
142 return head
143 end = max(head.rfind(c) for c in _WORD_BREAKS)
144 return head[:end] if end > 0 else head
147def _page_slug(segment: str) -> str:
148 """The readable name of the page whose last path segment is *segment*, without its suffix."""
149 marked = _legal(_without_suffix(_filled(segment)))
150 return _legal(_cut_at_word(marked, _MAX_NAME_BYTES))
153def _folder_slug(part: str) -> str:
154 """The slug of a folder; one ending in the page suffix ends in ``-md``, never a file's name."""
155 slug = _slug(part)
156 if not slug.lower().endswith(PAGE_SUFFIX):
157 return slug
158 mark = len(slug) - len(PAGE_SUFFIX)
159 return f"{slug[:mark]}{_SLUG_FILL}{slug[mark + 1 :]}"
162def _readable_host(hostname: str) -> str:
163 """*hostname* in Unicode, or an IPv6 address as its eight groups joined by ``-``."""
164 import ada_url
166 if not hostname.startswith(_IPV6_OPEN):
167 return ada_url.idna_to_unicode(hostname)
168 address = ipaddress.IPv6Address(hostname.removeprefix(_IPV6_OPEN).removesuffix(_IPV6_CLOSE))
169 groups = address.exploded.split(_IPV6_GROUP_MARK)
170 return _SLUG_FILL.join(f"{int(group, _HEX):x}" for group in groups)
173def _address_parts(address: str) -> tuple[str, list[str]]:
174 """The host, readable and with any port, and the path segments of *address*."""
175 parsed = parsed_url(address)
176 if parsed is None:
177 return _UNKNOWN_HOST, [""]
178 host = _readable_host(parsed.hostname)
179 if parsed.port:
180 host = f"{host}{_PORT_MARK}{parsed.port}"
181 return host, parsed.pathname.removeprefix("/").split("/")
184def page_name(address: str) -> str:
185 """The path under ``_web`` of canonical *address*: readable folders and name, ``~``, hash.
187 The name comes from the last non-empty segment of the path, less its suffix.
188 """
189 host, segments = _address_parts(address)
190 named = [segment for segment in segments if segment]
191 *folders, last = named or [""]
192 digest = hashlib.sha256(address.encode("utf-8")).hexdigest()[:_HASH_HEX_DIGITS]
193 readable = _page_slug(last) if last else INDEX_STEM
194 file_name = f"{readable}{PAGE_HASH_MARK}{digest}{PAGE_SUFFIX}"
195 head = [_slug(host), *map(_folder_slug, folders)]
196 while len("/".join([*head, file_name]).encode("utf-8")) > _MAX_PATH_BYTES:
197 head.pop()
198 return "/".join([*head, file_name])
201def url_to_filename(url: str) -> str:
202 """The path under ``_web`` that the page at *url* is saved under."""
203 return page_name(canonical_address(url))
206def web_dir() -> Path:
207 """The folder under the documents directory that holds every crawled page."""
208 return cfg.documents_dir / WEB_DIR
211def _replace_atomically(path: Path, text: str) -> None:
212 """Write *text* to a temporary file beside *path*, sync it, then put it in place of *path*."""
213 tmp_name: str | None = None
214 try:
215 with tempfile.NamedTemporaryFile(dir=path.parent, suffix=".tmp", delete=False) as tmp:
216 tmp_name = tmp.name
217 tmp.write(text.encode("utf-8"))
218 tmp.flush()
219 os.fsync(tmp.fileno())
220 Path(tmp_name).replace(path)
221 except BaseException:
222 if tmp_name is not None:
223 Path(tmp_name).unlink(missing_ok=True)
224 raise
227def content_hash(text: str) -> str:
228 """SHA-256 hex digest of text content."""
229 return hashlib.sha256(text.encode()).hexdigest()
232# Reference-style nested-bracket links, e.g. Wikipedia footnote markers like
233# ``[[1]](https://en.wikipedia.org/wiki/Foo#cite_note-1)``. The inner brackets
234# make this a normal Markdown link with the text ``[1]``, but readers that
235# treat ``[[...]]`` as a wikilink (Obsidian) mis-parse it as a broken wikilink
236# followed by the literal URL.
237_REFERENCE_LINK_RE = re.compile(r"\[\[([^\]]*)\]\]\(([^)]*)\)")
238# The same link with its inner brackets escaped: ``[\[1\]](url)``.
239_ESCAPED_REFERENCE_LINK = r"[\\[\1\\]](\2)"
242def normalize_crawled_markdown(markdown: str) -> str:
243 """Give the page LF line ends and escape the inner brackets of ``[[N]](url)`` links.
245 LF line ends keep the saved bytes equal to the text a universal-newline read
246 gives back. ``[[N]](url)`` becomes ``[\\[N\\]](url)``: no ``[[`` is left for a
247 wikilink reader, and the link text still reads ``[N]``, so "word[1]" does not
248 index as "word1". Ordinary single-bracket links stay.
249 """
250 lf = markdown.replace("\r\n", "\n").replace("\r", "\n")
251 return _REFERENCE_LINK_RE.sub(_ESCAPED_REFERENCE_LINK, lf)
254@dataclass
255class CrawlRecord:
256 """The crawl page table one crawl saves into, and the files it warned about."""
258 pages: CrawlPages
259 # Files standing where a folder of a page belongs, each warned about once per crawl.
260 blocked: set[Path] = field(default_factory=set)
263def load_crawl_record(store: Store) -> CrawlRecord:
264 """The crawl page table of *store*, each earlier copy keyed by its canonical address."""
265 pages = store.crawl_pages
266 with pages.held():
267 pages.rekey(canonical_address, _holds_text)
268 return CrawlRecord(pages)
271def finish_crawl(record: CrawlRecord) -> None:
272 """Fold the commits the crawl's saves made into one version of the table."""
273 with record.pages.held():
274 record.pages.compact()
277def _same_file(one: str, other: str) -> bool:
278 """Whether two names under ``_web`` are one file as the disk resolves them."""
279 try:
280 return os.path.samefile(web_dir() / one, web_dir() / other)
281 except OSError:
282 return False
285def _holds_text(copy: CrawlPage) -> bool:
286 """Whether the file of the earlier *copy* is under ``_web`` and holds the copy's text."""
287 path = web_dir() / copy.file
288 try:
289 validate_path_within(path, web_dir().resolve())
290 except ValueError:
291 return False
292 if not path.is_file():
293 return False
294 return content_hash(path.read_text(encoding="utf-8", errors="replace")) == copy.content_hash
297def _needed_by_twin(copy: CrawlPage, pages: CrawlPages) -> bool:
298 """Whether another page, not yet saved at its own name, has the file of *copy* as its copy."""
299 return any(
300 twin.address != copy.address
301 and pages.current(twin.address) is None
302 and _same_file(twin.file, copy.file)
303 for twin in pages.copies_of(copy.content_hash)
304 )
307def _remove_empty_folders(folder: Path) -> None:
308 """Remove *folder* and each parent that is left empty, up to ``_web``."""
309 root = web_dir()
310 while folder != root and folder.is_relative_to(root) and not any(folder.iterdir()):
311 folder.rmdir()
312 folder = folder.parent
315def _disk_name(name: str) -> str:
316 """*name* as the disk spells it, which a case-folding disk may spell differently."""
317 folder, parts = web_dir(), []
318 for part in PurePosixPath(name).parts:
319 target = folder / part
320 found = (e.name for e in os.scandir(folder) if os.path.samefile(e.path, target))
321 parts.append(next(found, part))
322 folder = folder / parts[-1]
323 return "/".join(parts)
326def _retire_earlier(
327 address: str, pages: CrawlPages, remove_sources: Callable[[list[str]], object]
328) -> None:
329 """Delete each earlier copy of the page at *address* that holds its text and no other's.
331 The caller holds the crawl lock. The file goes before its index source, so a sync
332 flush that runs between the two finds no file and adds no source. The row of a copy
333 goes with its file, or at once when its file is gone, changed, or another page's own.
334 """
335 for copy in pages.earlier(address):
336 if pages.owners(copy.file) or not _holds_text(copy):
337 pages.forget(copy)
338 continue
339 if _needed_by_twin(copy, pages):
340 continue
341 source = PurePosixPath(WEB_DIR, _disk_name(copy.file)).as_posix()
342 (web_dir() / copy.file).unlink()
343 remove_sources([source])
344 _remove_empty_folders((web_dir() / copy.file).parent)
345 pages.forget(copy)
348def _blocking_file(path: Path) -> Path | None:
349 """The file that stands where a folder of *path* under ``_web`` belongs, or None."""
350 root = web_dir()
351 return next(
352 (
353 folder
354 for folder in path.parents
355 if folder.is_relative_to(root) and os.path.lexists(folder) and not folder.is_dir()
356 ),
357 None,
358 )
361def _folders_are_free(path: Path, record: CrawlRecord) -> bool:
362 """Whether the folders of *path* can be made; a file in their way is warned about once."""
363 blocker = _blocking_file(path)
364 if blocker is None:
365 return True
366 if blocker not in record.blocked:
367 record.blocked.add(blocker)
368 log.warning("Pages under %s are not saved: it is a file; move or delete it", blocker)
369 return False
372def _page_path(result: CrawlResult, address: str, record: CrawlRecord) -> Path | None:
373 """Where the page at *address* is saved, or None when its name is not free for it."""
374 root = web_dir()
375 path = root / page_name(address)
376 try:
377 validate_path_within(path, root.resolve())
378 except ValueError:
379 log.warning("Path traversal blocked: %s -> %s", result.url, path)
380 return None
381 if not _folders_are_free(path, record):
382 return None
383 if os.path.lexists(path) and (path.is_symlink() or not path.is_file()):
384 log.warning(_NOT_SAVED, result.url, path)
385 return None
386 return path
389def _held_text(path: Path) -> str | None:
390 """The hash of the text the file at *path* holds, or None when there is no file.
392 The file is read with universal newlines: an older lilbee wrote pages in text
393 mode, so on Windows a page it saved holds CRLF line ends its record never hashed.
394 """
395 if not path.is_file():
396 return None
397 return content_hash(path.read_text(encoding="utf-8", errors="replace"))
400def _may_write(prev: CrawlPage | None, held: str | None, new_hash: str) -> bool:
401 """Whether the file at the page's name holds nothing, the new text, or a text it claimed."""
402 owned = {new_hash} if prev is None else {new_hash, prev.content_hash, prev.replaces}
403 return held is None or held in owned
406def _unchanged(prev: CrawlPage | None, held: str | None, new_hash: str, title: str) -> bool:
407 """Whether the row and the file already hold this text, and the row this title."""
408 if prev is None or held != new_hash or prev.content_hash != new_hash:
409 return False
410 return prev.title == title
413def _kept_title(result: CrawlResult, prev: CrawlPage | None) -> str:
414 """The title the crawl reports, else the one the row holds; no title never clears one."""
415 return result.title or (prev.title if prev is not None else "")
418def _write(
419 result: CrawlResult,
420 address: str,
421 path: Path,
422 markdown: str,
423 held: str | None,
424 prev: CrawlPage | None,
425 pages: CrawlPages,
426) -> Path | None:
427 """Claim *markdown* for the page, then write it at *path*; None when the file holds it.
429 The claim commits before the file is written, so a file exists only after its claim.
430 """
431 new_hash = content_hash(markdown)
432 title = _kept_title(result, prev)
433 if _unchanged(prev, held, new_hash, title):
434 log.info("Content unchanged, skipping save: %s", result.url)
435 return None
436 now = datetime.now(UTC).isoformat()
437 replaces = "" if held is None else held
438 pages.claim(CrawlPage(address, page_name(address), new_hash, now, replaces, title))
439 if held == new_hash:
440 log.info("Recorded the page %s at %s, which already holds its text", result.url, path)
441 return None
442 path.parent.mkdir(parents=True, exist_ok=True)
443 _replace_atomically(path, markdown)
444 return path
447def save_page(
448 result: CrawlResult, record: CrawlRecord, remove_sources: Callable[[list[str]], object]
449) -> Path | None:
450 """Write one crawled page at its name; the path, or None when nothing was written.
452 The check, the claim, the file write and the retire of earlier copies run under one
453 hold of the crawl lock, so the last claim of a page always describes its file. The
454 save waits for the lock without a deadline: every holder's work is bounded, and a
455 page skipped for a lock timeout would be a page lost.
456 """
457 if result.failure_reason() is not None:
458 return None
459 address = canonical_address(result.url)
460 with record.pages.held(timeout=None):
461 path = _page_path(result, address, record)
462 if path is None:
463 return None
464 markdown = normalize_crawled_markdown(result.markdown)
465 held = _held_text(path)
466 prev = record.pages.current(address)
467 if not _may_write(prev, held, content_hash(markdown)):
468 log.warning(_NOT_SAVED, result.url, path)
469 return None
470 written = _write(result, address, path, markdown, held, prev, record.pages)
471 _retire_earlier(address, record.pages, remove_sources)
472 record.pages.compact_when_due()
473 return written