Coverage for src/lilbee/crawler/save.py: 100%

1"""URL-to-filename mapping, metadata I/O, and per-page save-to-disk.""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6import json 

7import logging 

8import re 

9import tempfile 

10from dataclasses import dataclass 

11from pathlib import Path 

12from urllib.parse import urlparse 

13 

14from lilbee.core.config import cfg 

15from lilbee.core.security import validate_path_within 

16from lilbee.crawler.models import CrawlResult 

17 

18log = logging.getLogger(__name__) 

19 

20# Maximum filename length before truncation (most filesystems cap at 255 bytes) 

21_MAX_FILENAME_LEN = 200 

22 

23# Sentinel for index pages (trailing slash or empty path) 

24_INDEX_FILENAME = "index.md" 

25 

26# Length of the query-string hash folded into a filename to keep distinct 

27# queries on the same path from mapping to (and overwriting) the same file. 

28_QUERY_HASH_LEN = 8 

29 

30# How often the crawl metadata JSON is rewritten during a streaming crawl. 

31# Markdown files are durable per-page; metadata batches to keep write volume 

32# bounded. Worst-case loss on crash is N-1 entries, recoverable from the files. 

33METADATA_FLUSH_INTERVAL = 10 

34 

35 

36def url_to_filename(url: str) -> str: 

37 """Convert a URL to a safe filesystem path ending in .md. 

38 

39 Examples: 

40 https://docs.python.org/3/tutorial/ → docs.python.org/3/tutorial/index.md 

41 https://example.com/page?q=1 → example.com/page/index_q<hash>.md 

42 https://example.com/ → example.com/index.md 

43 

44 A distinct query string folds a short hash of the query into the filename so 

45 two URLs differing only in their query do not map to the same file and 

46 overwrite each other. The fragment is client-side and intentionally ignored. 

47 """ 

48 parsed = urlparse(url) 

49 host = parsed.hostname or "unknown" 

50 path = parsed.path.strip("/") 

51 

52 if not path: 

53 rel = _INDEX_FILENAME 

54 else: 

55 # Neutralize path traversal segments and unsafe filesystem characters. 

56 path = re.sub(r"\.\.+", "_", path) 

57 path = re.sub(r'[<>:"|?*]', "_", path) 

58 last_segment = path.rsplit("/", 1)[-1] 

59 if re.search(r"\.[^./]+$", last_segment): 

60 rel = re.sub(r"\.[^./]+$", ".md", path) # real extension: swap it for .md 

61 else: 

62 # No real extension (incl. a bare "." segment) -> treat as a directory. 

63 # This guarantees rel ends in .md so the query suffix below always applies. 

64 rel = f"{path}/{_INDEX_FILENAME}" 

65 

66 if parsed.query: 

67 query_hash = hashlib.sha256(parsed.query.encode()).hexdigest()[:_QUERY_HASH_LEN] 

68 rel = re.sub(r"\.md$", f"_q{query_hash}.md", rel) 

69 

70 full = f"{host}/{rel}" 

71 

72 # Truncate if too long, preserving .md extension 

73 if len(full) > _MAX_FILENAME_LEN: 

74 url_hash = hashlib.sha256(url.encode()).hexdigest()[:12] 

75 full = full[: _MAX_FILENAME_LEN - 16] + f"_{url_hash}.md" 

76 

77 return full 

78 

79 

80def _web_dir() -> Path: 

81 """Return the _web/ subdirectory under documents.""" 

82 return cfg.documents_dir / "_web" 

83 

84 

85def _crawl_meta_path() -> Path: 

86 """Path to the crawl metadata sidecar JSON.""" 

87 return cfg.data_dir / "crawl_meta.json" 

88 

89 

90@dataclass 

91class CrawlMeta: 

92 """Metadata for a single crawled URL.""" 

93 

94 file: str 

95 content_hash: str 

96 crawled_at: str 

97 

98 

99def load_crawl_metadata() -> dict[str, CrawlMeta]: 

100 """Load URL→metadata mapping from the JSON sidecar.""" 

101 path = _crawl_meta_path() 

102 if not path.exists(): 

103 return {} 

104 try: 

105 raw = json.loads(path.read_text(encoding="utf-8")) 

106 except (json.JSONDecodeError, UnicodeDecodeError, OSError) as exc: 

107 # A corrupt sidecar means every known URL looks new and gets re-crawled; 

108 # surface why rather than silently discarding the whole mapping. 

109 log.warning("Discarding unreadable crawl metadata sidecar %s: %s", path, exc) 

110 return {} 

111 result: dict[str, CrawlMeta] = {} 

112 for url, data in raw.items(): 

113 try: 

114 result[url] = CrawlMeta(**data) 

115 except (TypeError, KeyError): 

116 log.warning("Skipping malformed crawl metadata entry: %s", url) 

117 return result 

118 

119 

120def save_crawl_metadata(meta: dict[str, CrawlMeta]) -> None: 

121 """Persist URL→metadata mapping to the JSON sidecar (atomic write).""" 

122 path = _crawl_meta_path() 

123 path.parent.mkdir(parents=True, exist_ok=True) 

124 serializable = { 

125 url: {"file": m.file, "content_hash": m.content_hash, "crawled_at": m.crawled_at} 

126 for url, m in meta.items() 

127 } 

128 tmp_name: str | None = None 

129 try: 

130 with tempfile.NamedTemporaryFile(dir=path.parent, suffix=".tmp", delete=False) as tmp: 

131 tmp_name = tmp.name 

132 tmp.write(json.dumps(serializable, indent=2).encode("utf-8")) 

133 Path(tmp_name).replace(path) 

134 except BaseException: 

135 if tmp_name is not None: 

136 Path(tmp_name).unlink(missing_ok=True) 

137 raise 

138 

139 

140def content_hash(text: str) -> str: 

141 """SHA-256 hex digest of text content.""" 

142 return hashlib.sha256(text.encode()).hexdigest() 

143 

144 

145# Reference-style nested-bracket links, e.g. Wikipedia footnote markers like 

146# ``[[1]](https://en.wikipedia.org/wiki/Foo#cite_note-1)``. The inner brackets 

147# make this a normal Markdown link with the text ``[1]``, but readers that 

148# treat ``[[...]]`` as a wikilink (Obsidian) mis-parse it as a broken wikilink 

149# followed by the literal URL. 

150_REFERENCE_LINK_RE = re.compile(r"\[\[([^\]]*)\]\]\(([^)]*)\)") 

151 

152 

153def normalize_crawled_markdown(markdown: str) -> str: 

154 """Collapse reference-style ``[[N]](url)`` links to plain ``[N](url)``. 

155 

156 This fixes the double-bracket/wikilink collision without dropping the 

157 link text or URL. Ordinary single-bracket links are left untouched. 

158 """ 

159 return _REFERENCE_LINK_RE.sub(r"[\1](\2)", markdown) 

160 

161 

162@dataclass(frozen=True) 

163class SaveOutcome: 

164 """Return value of ``_save_single_result``: written path and the hash/filename used.""" 

165 

166 path: Path 

167 filename: str 

168 content_hash: str 

169 

170 

171def _save_single_result(result: CrawlResult, meta: dict[str, CrawlMeta]) -> SaveOutcome | None: 

172 """Write one crawl result to disk if it's new or changed. 

173 

174 Returns the outcome (written path plus reusable filename/hash), or 

175 None if skipped (failure, empty markdown, unchanged hash with file 

176 on disk, or blocked by path traversal). 

177 """ 

178 if not result.success or not result.markdown.strip(): 

179 return None 

180 markdown = normalize_crawled_markdown(result.markdown) 

181 filename = url_to_filename(result.url) 

182 web_dir = _web_dir() 

183 file_path = web_dir / filename 

184 resolved_web_dir = web_dir.resolve() 

185 try: 

186 validate_path_within(file_path, resolved_web_dir) 

187 except ValueError: 

188 log.warning("Path traversal blocked: %s -> %s", result.url, file_path) 

189 return None 

190 new_hash = content_hash(markdown) 

191 prev = meta.get(result.url) 

192 if prev is not None and prev.content_hash == new_hash and file_path.exists(): 

193 log.info("Content unchanged, skipping save: %s", result.url) 

194 return None 

195 file_path.parent.mkdir(parents=True, exist_ok=True) 

196 file_path.write_text(markdown, encoding="utf-8") 

197 return SaveOutcome(path=file_path, filename=filename, content_hash=new_hash) 

198 

199 

200def _update_single_metadata( 

201 meta: dict[str, CrawlMeta], 

202 url: str, 

203 outcome: SaveOutcome, 

204 now: str, 

205) -> None: 

206 """Update the metadata dict in place with a previously-computed outcome.""" 

207 meta[url] = CrawlMeta( 

208 file=outcome.filename, 

209 content_hash=outcome.content_hash, 

210 crawled_at=now, 

211 )