repos
/ status-django master

status-django

mirror archived upstream

Self-hostable uptime monitor and status page on Django: HTTP checks, Lighthouse audits, SEO crawls, and email and Discord alerts.

djangodockerhandcodedpythonself-hostedsqlitestatus-pageuptime-monitoringvite

10.2 KB · 297 lines · Python Raw History
  1import uuid
  2
  3from django.conf import settings
  4from django.contrib import messages
  5from django.core.files.storage import default_storage
  6from django.core.paginator import Paginator
  7from django.db import models
  8from django.http import HttpResponse, JsonResponse
  9from django.shortcuts import redirect, render
 10from django.template.loader import render_to_string
 11from django.utils import timezone
 12
 13from status.chromium import generate_pdf_from_html
 14
 15from .forms import PropertyForm
 16from .models import Property
 17
 18
 19def properties(request):
 20    if not request.user.is_authenticated:
 21        return redirect("/")
 22
 23    if request.method == "POST":
 24        form = PropertyForm(request.POST)
 25        if form.is_valid():
 26            new_property = form.save(commit=False)
 27            new_property.user = request.user
 28            new_property.save()
 29            messages.success(request, "Property added successfully.")
 30            return redirect("properties")
 31    else:
 32        form = PropertyForm()
 33
 34    properties = request.user.properties.order_by("url")
 35
 36    q = request.GET.get("q")
 37    if q:
 38        properties = properties.filter(url__icontains=q)
 39
 40    page = request.GET.get("page")
 41    properties = Paginator(properties, 25)
 42    properties = properties.get_page(page)
 43
 44    return render(
 45        request,
 46        "properties/properties.html",
 47        {
 48            "form": form,
 49            "title": "Properties",
 50            "description": "Manage your properties.",
 51            "q": q,
 52            "properties": properties,
 53        },
 54    )
 55
 56
 57def property_delete(request, property_id):
 58    if not request.user.is_authenticated:
 59        return redirect("/")
 60
 61    try:
 62        property_obj = request.user.properties.get(pk=property_id)
 63    except Property.DoesNotExist:
 64        return redirect("properties")
 65
 66    property_obj.delete()
 67    messages.success(request, "Property deleted successfully.")
 68    return redirect("properties")
 69
 70
 71def adjust_is_public_property(request, property_id):
 72    """
 73    Sets the property to public or private
 74    """
 75    if not request.user.is_authenticated:
 76        return redirect("/")
 77
 78    try:
 79        property_obj = request.user.properties.get(pk=property_id)
 80    except Property.DoesNotExist:
 81        return redirect("properties")
 82
 83    if request.method == "POST":
 84        property_obj.is_public = property_obj.is_public is False
 85        property_obj.save()
 86        return JsonResponse({"success": True})
 87
 88    return JsonResponse({"success": False})
 89
 90
 91def property(request, property_id):
 92    context = {}
 93
 94    try:
 95        property_obj = Property.objects.get(pk=property_id)
 96        context["property"] = property_obj
 97    except Property.DoesNotExist:
 98        return redirect("properties")
 99
100    if not property_obj.is_public and property_obj.user != request.user:
101        return redirect("properties")
102
103    # Set some basic page context variables
104    context["title"] = property_obj.name
105    context["description"] = "Status for " + property_obj.name
106    context["BASE_URL"] = settings.BASE_URL
107
108    status_response_times = []
109    for status in reversed(property_obj.statuses.order_by("-created_at")[:31]):
110        status_response_times.append(
111            {"label": status.created_at.isoformat(), "count": status.response_time}
112        )
113    context["status_response_times_graph"] = status_response_times
114
115    status_codes = property_obj.statuses.values("status_code").annotate(
116        count=models.Count("status_code")
117    )
118    context["status_codes_graph"] = [
119        {"label": x["status_code"], "count": x["count"]} for x in status_codes
120    ]
121
122    uptime = property_obj.statuses.filter(status_code=200).count()
123    downtime = property_obj.statuses.exclude(status_code=200).count()
124    total = uptime + downtime
125    try:
126        uptime_pct = round(uptime / total * 100, 2)
127    except ZeroDivisionError:
128        uptime_pct = 0
129    try:
130        downtime_pct = round(downtime / total * 100, 2)
131    except ZeroDivisionError:
132        downtime_pct = 0
133    context["uptime_graph"] = [
134        {"label": "Uptime", "count": uptime_pct},
135        {"label": "Downtime", "count": downtime_pct},
136    ]
137
138    # Report formats. `?report` (or `?report=pdf`) returns a printed PDF
139    # rendered from the light-themed report template; `?report=md` returns a
140    # plain-text Markdown report suited for piping into an LLM.
141    if "report" in request.GET:
142        fmt = request.GET.get("report") or "pdf"
143        if fmt == "md":
144            md = render_to_string("properties/property_report.md", context)
145            response = HttpResponse(md, content_type="text/markdown; charset=utf-8")
146            response["Content-Disposition"] = f'inline; filename="{property_obj.name}.md"'
147            return response
148        if fmt == "pdf":
149            html = render_to_string("properties/property_report.html", context)
150            filename = f"reports/{uuid.uuid4()}.pdf"
151            generate_pdf_from_html(html, filename)
152            with open(default_storage.path(filename), "rb") as pdf:
153                response = HttpResponse(pdf.read(), content_type="application/pdf")
154                response["Content-Disposition"] = "inline; filename=report.pdf"
155                return response
156
157    return render(request, "properties/property.html", context)
158
159
160def _crawl_progress(property_obj):
161    """Return the fraction (0-1) of the discovered work that's complete."""
162    from crawler.fetcher import PAGE_CAP
163
164    pages = property_obj.last_crawl_pages_count or 0
165    if pages <= 0:
166        return 0.05  # show *some* movement once we start
167    # We don't know the total ahead of time, so use a log-ish ratio capped at
168    # ~90% — the last 10% is reserved for post-crawl check processing.
169    return min(pages / PAGE_CAP, 0.9)
170
171
172def _serialize_status(property_obj):
173    now = timezone.now()
174
175    crawl_next = property_obj.next_run_at_crawler
176    lh_next = property_obj.next_lighthouse_run_at
177
178    insights = property_obj.crawler_insights or []
179    severity_counts = {"error": 0, "warning": 0, "info": 0}
180    for insight in insights:
181        sev = insight.get("severity", "info")
182        if sev in severity_counts:
183            severity_counts[sev] += 1
184
185    return {
186        "crawler": {
187            "state": property_obj.crawl_state,
188            "started_at": property_obj.crawl_started_at.isoformat()
189            if property_obj.crawl_started_at
190            else None,
191            "last_attempt_at": property_obj.last_run_at_crawler.isoformat()
192            if property_obj.last_run_at_crawler
193            else None,
194            "last_success_at": property_obj.last_crawl_success_at.isoformat()
195            if property_obj.last_crawl_success_at
196            else None,
197            "last_error": property_obj.last_crawl_error,
198            "last_duration_ms": property_obj.last_crawl_duration_ms,
199            "pages_count": property_obj.last_crawl_pages_count,
200            "next_run_at": crawl_next.isoformat() if crawl_next else None,
201            "is_overdue": bool(crawl_next and crawl_next <= now),
202            "insights_total": len(insights),
203            "insights_by_severity": severity_counts,
204            "progress": _crawl_progress(property_obj)
205            if property_obj.crawl_state == "running"
206            else None,
207        },
208        "lighthouse": {
209            "state": property_obj.lighthouse_state,
210            "started_at": property_obj.lighthouse_started_at.isoformat()
211            if property_obj.lighthouse_started_at
212            else None,
213            "last_attempt_at": property_obj.last_lighthouse_run_at.isoformat()
214            if property_obj.last_lighthouse_run_at
215            else None,
216            "last_success_at": property_obj.last_lighthouse_success_at.isoformat()
217            if property_obj.last_lighthouse_success_at
218            else None,
219            "last_error": property_obj.last_lighthouse_error,
220            "last_duration_ms": property_obj.last_lighthouse_duration_ms,
221            "next_run_at": lh_next.isoformat() if lh_next else None,
222            "is_overdue": bool(lh_next and lh_next <= now),
223            "scores": property_obj.lighthouse_scores,
224        },
225        "server_time": now.isoformat(),
226    }
227
228
229def property_status(request, property_id):
230    try:
231        property_obj = Property.objects.get(pk=property_id)
232    except Property.DoesNotExist:
233        return JsonResponse({"error": "not_found"}, status=404)
234
235    if not property_obj.is_public and property_obj.user != request.user:
236        return JsonResponse({"error": "forbidden"}, status=403)
237
238    return JsonResponse(_serialize_status(property_obj))
239
240
241def property_recrawl(request, property_id):
242    if not request.user.is_authenticated:
243        return JsonResponse({"error": "forbidden"}, status=403)
244
245    if request.method != "POST":
246        return JsonResponse({"error": "method_not_allowed"}, status=405)
247
248    try:
249        property_obj = request.user.properties.get(pk=property_id)
250    except Property.DoesNotExist:
251        return JsonResponse({"error": "not_found"}, status=404)
252
253    if property_obj.crawl_state in ("queued", "running"):
254        return JsonResponse(
255            {
256                "ok": False,
257                "reason": "already_running",
258                **_serialize_status(property_obj),
259            }
260        )
261
262    property_obj.next_run_at_crawler = timezone.now()
263    property_obj.last_crawl_error = None
264    property_obj.save(update_fields=["next_run_at_crawler", "last_crawl_error"])
265    return JsonResponse({"ok": True, **_serialize_status(property_obj)})
266
267
268def property_rerun_lighthouse(request, property_id):
269    if not request.user.is_authenticated:
270        return JsonResponse({"error": "forbidden"}, status=403)
271
272    if request.method != "POST":
273        return JsonResponse({"error": "method_not_allowed"}, status=405)
274
275    try:
276        property_obj = request.user.properties.get(pk=property_id)
277    except Property.DoesNotExist:
278        return JsonResponse({"error": "not_found"}, status=404)
279
280    if property_obj.lighthouse_state in ("queued", "running"):
281        return JsonResponse(
282            {
283                "ok": False,
284                "reason": "already_running",
285                **_serialize_status(property_obj),
286            }
287        )
288
289    property_obj.next_lighthouse_run_at = timezone.now()
290    property_obj.last_lighthouse_error = None
291    property_obj.save(
292        update_fields=["next_lighthouse_run_at", "last_lighthouse_error"]
293    )
294    return JsonResponse({"ok": True, **_serialize_status(property_obj)})
295
296