Self-hostable uptime monitor and status page on Django: HTTP checks, Lighthouse audits, SEO crawls, and email and Discord alerts.
djangodockerhandcodedpythonself-hostedsqlitestatus-pageuptime-monitoringvite
1import logging
2import re
3import time
4import uuid
5
6import requests
7from django.conf import settings
8from django.contrib.auth import get_user_model
9from django.core.mail import EmailMessage
10from django.db import models, transaction
11from django.template.loader import render_to_string
12from django.utils import timezone
13from django.utils.functional import cached_property
14
15from crawler.runner import run_seo_spider
16from status.lighthouse import (
17 LighthouseError,
18 fetch_lighthouse_results,
19 parse_lighthouse_results,
20 parse_performance_details,
21)
22
23logger = logging.getLogger(__name__)
24
25
26User = get_user_model()
27
28
29class SecurityMixin:
30
31 @property
32 def invalid_cert(self):
33 return self.statuses.latest("created_at").status_code == 526
34
35 @property
36 def is_https(self):
37 return self.url.startswith("https://")
38
39 @property
40 def has_mime_type(self):
41 return self.latest_headers.get("content-type", None) is not None
42
43 @property
44 def has_content_sniffing_protection(self):
45 return self.latest_headers.get("x-content-type-options", None) == "nosniff"
46
47 @property
48 def has_xss_protection(self):
49 return self.latest_headers.get("x-xss-protection", None) == "1; mode=block"
50
51 @property
52 def has_clickjack_protection(self):
53 return self.latest_headers.get("x-frame-options", None) in [
54 "deny",
55 "sameorigin",
56 "allow-from",
57 ]
58
59 @property
60 def hides_server_version(self):
61 if (
62 self.latest_headers.get("server", None) is None
63 and self.latest_headers.get("x-server", None) is None
64 and self.latest_headers.get("powered-by", None) is None
65 and self.latest_headers.get("x-powered-by", None) is None
66 ):
67 return True
68 return False
69
70 @property
71 def has_hsts(self):
72 # hsts has at least one year set
73 hsts = self.latest_headers.get("strict-transport-security", None)
74 if hsts is None:
75 return False
76 # use re to get the max-age value
77 max_age = re.search(r"max-age=(\d+)", hsts)
78 if max_age is None:
79 return False
80 # convert to int and compare
81 max_age = int(max_age.group(1))
82 return max_age >= 31536000
83
84 @property
85 def has_hsts_preload(self):
86 hsts = self.latest_headers.get("strict-transport-security", None)
87 if hsts is None:
88 return False
89 return "preload" in hsts.lower()
90
91 @property
92 def has_security_issue(self):
93 if not self.is_https:
94 return True
95 if not self.has_mime_type:
96 return True
97 if not self.has_content_sniffing_protection:
98 return True
99 if not self.has_xss_protection:
100 return True
101 if not self.has_clickjack_protection:
102 return True
103 if not self.hides_server_version:
104 return True
105 if not self.has_hsts:
106 return True
107 if not self.has_hsts_preload:
108 return True
109 return False
110
111
112class AlertsMixin:
113
114 def send_down_email(self):
115 subject = f"Status: {self.name} is down!"
116 context = {"property": self, "BASE_URL": settings.BASE_URL}
117 message = render_to_string("emails/property_down.html", context)
118 from_email = "[email protected]"
119 to_emails = [self.user.email]
120 email = EmailMessage(subject, message, from_email, to_emails)
121 email.content_subtype = "html"
122 try:
123 email.send()
124 except Exception:
125 logger.exception("Failed to send down email for %s", self.url)
126
127 def send_recovery_email(self):
128 subject = f"Status: {self.name} is back up!"
129 context = {"property": self, "BASE_URL": settings.BASE_URL}
130 message = render_to_string("emails/property_recovery.html", context)
131 from_email = "[email protected]"
132 to_emails = [self.user.email]
133 email = EmailMessage(subject, message, from_email, to_emails)
134 email.content_subtype = "html"
135 try:
136 email.send()
137 except Exception:
138 logger.exception("Failed to send recovery email for %s", self.url)
139
140 def send_down_discord_message(self):
141 if self.user.discord_webhook_url:
142 payload = {
143 "username": "Status",
144 "embeds": [
145 {
146 "title": "Status Alert",
147 "description": f"{self.url} is down!",
148 "color": 16711680, # Red
149 "timestamp": timezone.now().isoformat(),
150 }
151 ],
152 }
153 try:
154 requests.post(self.user.discord_webhook_url, json=payload, timeout=5)
155 except requests.RequestException:
156 logger.exception("Discord down webhook failed for %s", self.url)
157
158 def send_recovery_discord_message(self):
159 if self.user.discord_webhook_url:
160 payload = {
161 "username": "Status",
162 "embeds": [
163 {
164 "title": "Status Recovery",
165 "description": f"{self.url} is back up!",
166 "color": 65280, # Green
167 "timestamp": timezone.now().isoformat(),
168 }
169 ],
170 }
171 try:
172 requests.post(self.user.discord_webhook_url, json=payload, timeout=5)
173 except requests.RequestException:
174 logger.exception("Discord recovery webhook failed for %s", self.url)
175
176 def send_alerts(self, current_status_code):
177 """
178 Send alerts based on state transitions:
179 - Send 'down' alert when site goes from UP to DOWN
180 - Send 'recovery' alert when site goes from DOWN to UP
181 - No alerts for consecutive failures or consecutive successes
182 """
183 is_currently_up = current_status_code == 200
184
185 # Commit the state transition inside the atomic block BEFORE firing
186 # notifications. If the save raises (e.g. SQLite "database is locked"
187 # from a concurrent writer), the transaction rolls back and nothing
188 # is emailed — the next check will retry from the same state. Sending
189 # first would mean a failed save leaves us firing the same alert on
190 # every subsequent check.
191 transition = None
192 with transaction.atomic():
193 locked = Property.objects.select_for_update().get(pk=self.pk)
194
195 if is_currently_up and locked.alert_state == "down":
196 transition = "recovery"
197 elif not is_currently_up and locked.alert_state == "up":
198 checks = self.statuses.order_by("-created_at")[:2]
199 if (
200 len(checks) >= 2
201 and checks[0].status_code != 200
202 and checks[1].status_code != 200
203 ):
204 transition = "down"
205
206 if transition is not None:
207 locked.alert_state = "up" if transition == "recovery" else "down"
208 locked.last_alert_sent = timezone.now()
209 locked.save(update_fields=["alert_state", "last_alert_sent"])
210 self.alert_state = locked.alert_state
211 self.last_alert_sent = locked.last_alert_sent
212
213 if transition == "recovery":
214 self.send_recovery_email()
215 self.send_recovery_discord_message()
216 elif transition == "down":
217 self.send_down_email()
218 self.send_down_discord_message()
219
220
221class CrawlerMixin:
222 def get_next_run_at_crawl(self):
223 """Weekly crawl by default; users can trigger a recrawl anytime."""
224 return timezone.now() + timezone.timedelta(days=7)
225
226 def should_check_crawl(self):
227 now = timezone.now()
228 if self.last_run_at_crawler is None:
229 return True
230 if self.next_run_at_crawler is None:
231 return True
232 return self.next_run_at_crawler <= now
233
234 def _report_crawl_progress(self, pages_count):
235 Property.objects.filter(pk=self.pk).update(last_crawl_pages_count=pages_count)
236
237 def crawl_site(self):
238 Property.objects.filter(pk=self.pk).update(
239 crawl_state="running",
240 crawl_started_at=timezone.now(),
241 last_crawl_pages_count=0,
242 )
243 start = time.monotonic()
244 try:
245 insights = run_seo_spider(self.url, progress_cb=self._report_crawl_progress)
246 except Exception as e:
247 logger.exception("Crawl failed for %s", self.url)
248 Property.objects.filter(pk=self.pk).update(
249 crawl_state="idle",
250 last_crawl_error=f"{type(e).__name__}: {e}",
251 last_crawl_duration_ms=int((time.monotonic() - start) * 1000),
252 )
253 return
254 duration_ms = int((time.monotonic() - start) * 1000)
255 Property.objects.filter(pk=self.pk).update(
256 crawler_insights=insights,
257 crawl_state="idle",
258 last_crawl_success_at=timezone.now(),
259 last_crawl_error=None,
260 last_crawl_duration_ms=duration_ms,
261 )
262
263
264class Property(CrawlerMixin, AlertsMixin, SecurityMixin, models.Model):
265 id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
266 user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="properties")
267
268 url = models.CharField(max_length=255)
269
270 is_public = models.BooleanField(default=False)
271
272 last_run_at = models.DateTimeField(blank=True, null=True)
273 next_run_at = models.DateTimeField(blank=True, null=True)
274
275 last_run_at_crawler = models.DateTimeField(blank=True, null=True)
276 next_run_at_crawler = models.DateTimeField(blank=True, null=True)
277 crawler_insights = models.JSONField(blank=True, null=True)
278 crawl_state = models.CharField(
279 max_length=10,
280 choices=[("idle", "Idle"), ("queued", "Queued"), ("running", "Running")],
281 default="idle",
282 )
283 crawl_started_at = models.DateTimeField(blank=True, null=True)
284 last_crawl_success_at = models.DateTimeField(blank=True, null=True)
285 last_crawl_error = models.TextField(blank=True, null=True)
286 last_crawl_duration_ms = models.IntegerField(blank=True, null=True)
287 last_crawl_pages_count = models.IntegerField(blank=True, null=True)
288
289 lighthouse_scores = models.JSONField(blank=True, null=True)
290 lighthouse_details = models.JSONField(blank=True, null=True)
291 last_lighthouse_run_at = models.DateTimeField(blank=True, null=True)
292 last_lighthouse_success_at = models.DateTimeField(blank=True, null=True)
293 last_lighthouse_error = models.TextField(blank=True, null=True)
294 last_lighthouse_duration_ms = models.IntegerField(blank=True, null=True)
295 next_lighthouse_run_at = models.DateTimeField(blank=True, null=True)
296 lighthouse_state = models.CharField(
297 max_length=10,
298 choices=[("idle", "Idle"), ("queued", "Queued"), ("running", "Running")],
299 default="idle",
300 )
301 lighthouse_started_at = models.DateTimeField(blank=True, null=True)
302
303 # Alert state tracking
304 last_alert_sent = models.DateTimeField(blank=True, null=True)
305 alert_state = models.CharField(
306 max_length=10, choices=[("up", "Up"), ("down", "Down")], default="up"
307 )
308
309 created_at = models.DateTimeField(auto_now_add=True)
310 updated_at = models.DateTimeField(auto_now=True)
311
312 class Meta:
313 verbose_name = "Property"
314 verbose_name_plural = "Properties"
315
316 indexes = [
317 models.Index(fields=["url"]),
318 models.Index(fields=["user"]),
319 ]
320
321 def __str__(self):
322 return self.url
323
324 @property
325 def name(self):
326 return self.url.split("/")[2].replace("www.", "")
327
328 def get_next_run_at(self):
329 now = timezone.now()
330 return now.replace(
331 minute=(now.minute // 3) * 3, second=0, microsecond=0
332 ) + timezone.timedelta(minutes=3)
333
334 def should_check(self):
335 now = timezone.now()
336 if self.last_run_at is None:
337 return True
338 if self.next_run_at is None:
339 return True
340 return self.next_run_at <= now
341
342 def run_check(self):
343 try:
344 headers = {
345 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.5005.115 Safari/537.36 Status/1.0.0"
346 }
347 response = requests.get(self.url, timeout=(3, 10), headers=headers)
348 response_time = response.elapsed.total_seconds() * 1000
349 status_code = response.status_code
350 headers = response.headers
351 except requests.exceptions.SSLError:
352 response_time = 10000
353 status_code = 526
354 headers = {}
355 except (requests.exceptions.RequestException, requests.exceptions.Timeout):
356 response_time = 10000
357 status_code = 408
358 headers = {}
359 return Check.objects.create(
360 property=self,
361 status_code=status_code,
362 response_time=response_time,
363 headers=dict(headers),
364 )
365
366 def process_check(self):
367 check = self.run_check()
368 # Always check for state changes, regardless of current status
369 self.send_alerts(check.status_code)
370
371 def get_next_run_at_lighthouse(self):
372 """
373 Should check daily.
374 """
375 return timezone.now() + timezone.timedelta(days=1)
376
377 def should_check_lighthouse(self):
378 now = timezone.now()
379 if self.last_lighthouse_run_at is None:
380 return True
381 if self.next_lighthouse_run_at is None:
382 return True
383 return self.next_lighthouse_run_at <= now
384
385 def process_check_lighthouse(self):
386 self.run_check_lighthouse()
387
388 def run_check_lighthouse(self):
389 Property.objects.filter(pk=self.pk).update(
390 lighthouse_state="running",
391 lighthouse_started_at=timezone.now(),
392 )
393 start = time.monotonic()
394 try:
395 results = fetch_lighthouse_results(self.url)
396 scores = parse_lighthouse_results(results)
397 details = parse_performance_details(results)
398 except LighthouseError as e:
399 logger.warning("Lighthouse failed for %s: %s", self.url, e)
400 Property.objects.filter(pk=self.pk).update(
401 lighthouse_state="idle",
402 last_lighthouse_error=str(e),
403 last_lighthouse_duration_ms=int((time.monotonic() - start) * 1000),
404 )
405 return
406 except Exception as e:
407 logger.exception("Unexpected lighthouse error for %s", self.url)
408 Property.objects.filter(pk=self.pk).update(
409 lighthouse_state="idle",
410 last_lighthouse_error=f"{type(e).__name__}: {e}",
411 last_lighthouse_duration_ms=int((time.monotonic() - start) * 1000),
412 )
413 return
414
415 Property.objects.filter(pk=self.pk).update(
416 lighthouse_scores=scores,
417 lighthouse_details=details,
418 last_lighthouse_success_at=timezone.now(),
419 last_lighthouse_error=None,
420 last_lighthouse_duration_ms=int((time.monotonic() - start) * 1000),
421 lighthouse_state="idle",
422 )
423
424 @property
425 def total_checks(self):
426 return self.statuses.count()
427
428 @cached_property
429 def current_status(self):
430 try:
431 return self.statuses.latest("created_at").status_code
432 except Check.DoesNotExist:
433 return 200
434
435 @property
436 def avg_response_time(self):
437 try:
438 return int(
439 self.statuses.all()[:31].aggregate(models.Avg("response_time"))[
440 "response_time__avg"
441 ]
442 )
443 except TypeError:
444 return 0
445
446 @cached_property
447 def latest_headers(self):
448 try:
449 # return all headers lowercase to make them easier to use
450 return {
451 k.lower(): v.lower() for k, v in self.statuses.latest().headers.items()
452 }
453 except Check.DoesNotExist:
454 return {}
455
456 @cached_property
457 def avg_lighthouse_score(self):
458 if self.lighthouse_scores:
459 scores = [score for score in self.lighthouse_scores.values()]
460 return round(sum(scores) / len(scores))
461
462 def recent_tick_stream(self, limit=30):
463 """Most-recent-first list of 'up' / 'down' strings for the uptime strip."""
464 codes = list(
465 self.statuses.order_by("-created_at").values_list("status_code", flat=True)[:limit]
466 )
467 return ["up" if c == 200 else "down" for c in reversed(codes)]
468
469 @cached_property
470 def recent_uptime_pct(self):
471 """Uptime percentage over the most recent 100 checks, rounded to one decimal."""
472 codes = list(
473 self.statuses.order_by("-created_at").values_list("status_code", flat=True)[:100]
474 )
475 if not codes:
476 return None
477 up = sum(1 for c in codes if c == 200)
478 return round((up / len(codes)) * 100, 1)
479
480
481class Check(models.Model):
482 property = models.ForeignKey(
483 Property, on_delete=models.CASCADE, related_name="statuses", editable=False
484 )
485
486 status_code = models.IntegerField()
487 response_time = models.IntegerField(default=0)
488 headers = models.JSONField(default=dict)
489
490 created_at = models.DateTimeField(auto_now_add=True, editable=False)
491
492 class Meta:
493 verbose_name = "Check"
494 verbose_name_plural = "Checks"
495 indexes = [
496 models.Index(fields=["created_at"]),
497 models.Index(fields=["property", "-created_at"]),
498 ]
499 get_latest_by = "created_at"
500
501 def __str__(self):
502 return f"{self.property.url} - {self.created_at} - {self.status_code}"