Self-hostable uptime monitor and status page on Django: HTTP checks, Lighthouse audits, SEO crawls, and email and Discord alerts.
djangodockerhandcodedpythonself-hostedsqlitestatus-pageuptime-monitoringvite
1"""Entry point for the SEO crawler.
2
3Crawls a site, runs checks, writes per-page debug output, and returns
4a list of insights. Designed to run in-process; no subprocess required.
5"""
6import json
7import logging
8import os
9import time
10from collections import deque
11from concurrent.futures import ThreadPoolExecutor, as_completed
12from urllib.parse import urlparse
13
14from django.conf import settings
15
16from . import checks
17from .fetcher import (
18 CONCURRENCY,
19 CRAWL_DEADLINE_SECONDS,
20 PAGE_CAP,
21 fetch,
22 head_status,
23 load_robots,
24 load_sitemap,
25 make_session,
26 same_site,
27)
28from .parser import parse_html
29
30
31logger = logging.getLogger(__name__)
32
33
34def _output_path(host):
35 base = "crawler_output" if settings.DEBUG else "/data/crawler_output"
36 return os.path.join(base, f"{host}.json")
37
38
39def _normalize_url(url):
40 # Drop fragments; keep query strings since they often distinguish pages.
41 p = urlparse(url)
42 cleaned = p._replace(fragment="").geturl()
43 return cleaned.rstrip("/") or cleaned
44
45
46def crawl(start_url, progress_cb=None):
47 """Fetch up to PAGE_CAP pages from the same host and collect metadata.
48
49 `progress_cb(pages_count)` is invoked after each batch so callers can
50 surface live progress without blocking the crawl.
51 """
52 session = make_session()
53
54 parsed = urlparse(start_url)
55 host = parsed.netloc
56 base_origin = f"{parsed.scheme}://{parsed.netloc}"
57
58 rp, robots_url, robots_text = load_robots(session, base_origin)
59 sitemap_urls = load_sitemap(session, base_origin, robots_text)
60
61 seen = set()
62 queue = deque()
63 pages = []
64 fetched = set()
65 deadline = time.time() + CRAWL_DEADLINE_SECONDS
66
67 def enqueue(url):
68 n = _normalize_url(url)
69 if n in seen:
70 return
71 seen.add(n)
72 queue.append(url)
73
74 enqueue(start_url)
75 # Seed with sitemap URLs so sitemap-only pages get crawled too.
76 for url in sitemap_urls[:PAGE_CAP]:
77 if same_site(url, host):
78 enqueue(url)
79
80 with ThreadPoolExecutor(max_workers=CONCURRENCY) as ex:
81 while queue and len(pages) < PAGE_CAP and time.time() < deadline:
82 batch = []
83 while queue and len(batch) < CONCURRENCY and len(pages) + len(batch) < PAGE_CAP:
84 url = queue.popleft()
85 if not rp.can_fetch("*", url):
86 continue
87 batch.append(url)
88 if not batch:
89 break
90
91 futures = [ex.submit(fetch, session, u) for u in batch]
92 for f in as_completed(futures):
93 r = f.result()
94 # Different requested URLs can collapse to the same final URL
95 # after redirects. Drop duplicates so checks don't double-flag.
96 final_key = _normalize_url(r.url)
97 if final_key in fetched:
98 seen.add(final_key)
99 continue
100 fetched.add(final_key)
101 is_html = r.status == 200 and "text/html" in r.content_type
102 page = {
103 "url": r.url,
104 "requested_url": r.requested_url,
105 "status": r.status,
106 "content_type": r.content_type,
107 "elapsed_ms": r.elapsed_ms,
108 "bytes": len(r.body),
109 "headers": r.headers,
110 "redirect_chain": r.redirect_chain,
111 "error": r.error,
112 "is_html": is_html,
113 }
114 if is_html:
115 try:
116 page.update(parse_html(r.body, r.url))
117 except Exception:
118 logger.exception("[crawler] parse failed for %s", r.url)
119 page["is_html"] = False
120 else:
121 for link in page.get("links", []):
122 lu = link["url"]
123 if same_site(lu, host):
124 enqueue(lu)
125 # Ensure the final redirected URL is considered "seen" too, so
126 # we don't refetch it on another pass.
127 seen.add(_normalize_url(r.url))
128 pages.append(page)
129
130 if progress_cb is not None:
131 try:
132 progress_cb(len(pages))
133 except Exception:
134 logger.exception("[crawler] progress callback failed")
135
136 if time.time() >= deadline:
137 logger.warning(
138 "[crawler] hit deadline for %s after %d pages",
139 start_url,
140 len(pages),
141 )
142
143 # External link HEAD check. Only checks same unique URL once.
144 external_links = set()
145 for p in pages:
146 if not p.get("is_html"):
147 continue
148 for link in p.get("links", []):
149 if not same_site(link["url"], host):
150 external_links.add(link["url"])
151
152 external_link_status = {}
153 if external_links and time.time() < deadline:
154 with ThreadPoolExecutor(max_workers=CONCURRENCY) as ex:
155 futures = {ex.submit(head_status, session, u): u for u in external_links}
156 for f in as_completed(futures):
157 url = futures[f]
158 try:
159 external_link_status[url] = f.result()
160 except Exception:
161 external_link_status[url] = 0
162
163 return {
164 "start_url": start_url,
165 "host": host,
166 "pages": pages,
167 "external_link_status": external_link_status,
168 "sitemap_urls": sitemap_urls,
169 "robots": {
170 "url": robots_url,
171 "exists": robots_text is not None,
172 "raw": robots_text,
173 "references_sitemap": bool(
174 robots_text
175 and any(
176 line.lower().startswith("sitemap:")
177 for line in robots_text.splitlines()
178 )
179 ),
180 },
181 }
182
183
184def run_checks(crawl_result):
185 """Build a ctx dict and run every check. Returns the flat insight list."""
186 ctx = {
187 "start_url": crawl_result["start_url"],
188 "host": crawl_result["host"],
189 "pages": crawl_result["pages"],
190 "html_pages": [p for p in crawl_result["pages"] if p.get("is_html")],
191 "status_map": {p["url"]: p["status"] for p in crawl_result["pages"]},
192 "external_link_status": crawl_result["external_link_status"],
193 "sitemap_urls": crawl_result["sitemap_urls"],
194 "robots": crawl_result["robots"],
195 }
196
197 insights = []
198 for fn in checks.ALL_CHECKS:
199 try:
200 insights.extend(fn(ctx))
201 except Exception:
202 logger.exception("[crawler] check %s failed", fn.__name__)
203 return insights
204
205
206def _write_debug_output(crawl_result):
207 host = crawl_result["host"]
208 path = _output_path(host)
209 try:
210 os.makedirs(os.path.dirname(path), exist_ok=True)
211 with open(path, "w") as f:
212 for p in crawl_result["pages"]:
213 # Strip bulky fields from the debug file.
214 snapshot = {k: v for k, v in p.items() if k != "headers"}
215 f.write(json.dumps(snapshot, default=str) + "\n")
216 except OSError:
217 logger.exception("[crawler] failed writing debug output to %s", path)
218
219
220def run_seo_spider(url, progress_cb=None):
221 """Crawl `url`, write debug output, return list of insight dicts."""
222 start = time.time()
223 logger.info("[crawler] starting %s", url)
224 result = crawl(url, progress_cb=progress_cb)
225 insights = run_checks(result)
226 _write_debug_output(result)
227 logger.info(
228 "[crawler] done %s - %d pages, %d insights, %.1fs",
229 url,
230 len(result["pages"]),
231 len(insights),
232 time.time() - start,
233 )
234 return insights