status-django

  1"""
  2A wrapper around the lighthouse node CLI.
  3
  4Raises LighthouseError with a descriptive message on failure so callers can
  5log/persist the reason instead of silently dropping the result.
  6"""
  7
  8import json
  9import logging
 10import shutil
 11import subprocess
 12
 13from django.conf import settings
 14
 15logger = logging.getLogger(__name__)
 16
 17
 18# Lighthouse's chrome-launcher searches for a browser on its own; we only
 19# pin CHROME_PATH when we can resolve a known name, for determinism in
 20# production (Alpine ships `chromium`). If nothing is found, fall through
 21# and let chrome-launcher do its own lookup.
 22CHROMIUM_BINARY = (
 23    shutil.which("chromium")
 24    or shutil.which("chromium-browser")
 25    or shutil.which("google-chrome")
 26)
 27
 28CHROME_FLAGS = "--headless --no-sandbox --disable-dev-shm-usage --disable-gpu"
 29
 30# Lighthouse itself can take 60-90s on a slow site; the outer timeout is a
 31# backstop against Chromium hangs that would otherwise wedge the scheduler.
 32SUBPROCESS_TIMEOUT_SECONDS = 180
 33
 34
 35class LighthouseError(Exception):
 36    pass
 37
 38
 39def fetch_lighthouse_results(url):
 40    command = [
 41        f"{settings.BASE_DIR}/node_modules/.bin/lighthouse",
 42        url,
 43        f"--chrome-flags={CHROME_FLAGS}",
 44        "--output=json",
 45        "--output-path=stdout",
 46        "--quiet",
 47    ]
 48    env = {"PATH": "/usr/bin:/bin:/usr/local/bin"}
 49    if CHROMIUM_BINARY:
 50        env["CHROME_PATH"] = CHROMIUM_BINARY
 51
 52    try:
 53        process = subprocess.run(
 54            command,
 55            check=True,
 56            stdout=subprocess.PIPE,
 57            stderr=subprocess.PIPE,
 58            timeout=SUBPROCESS_TIMEOUT_SECONDS,
 59            env=env,
 60        )
 61    except subprocess.TimeoutExpired:
 62        raise LighthouseError(
 63            f"lighthouse timed out after {SUBPROCESS_TIMEOUT_SECONDS}s"
 64        )
 65    except subprocess.CalledProcessError as e:
 66        stderr = (e.stderr or b"").decode("utf-8", errors="replace").strip()
 67        raise LighthouseError(f"lighthouse exited {e.returncode}: {stderr[-500:]}")
 68    except FileNotFoundError as e:
 69        raise LighthouseError(f"lighthouse binary missing: {e}")
 70
 71    try:
 72        return json.loads(process.stdout)
 73    except json.JSONDecodeError as e:
 74        raise LighthouseError(f"could not parse lighthouse output: {e}")
 75
 76
 77def parse_lighthouse_results(results):
 78    try:
 79        scores = {
 80            "Performance": results["categories"]["performance"]["score"],
 81            "Accessibility": results["categories"]["accessibility"]["score"],
 82            "Best practices": results["categories"]["best-practices"]["score"],
 83            "SEO": results["categories"]["seo"]["score"],
 84        }
 85    except KeyError as e:
 86        raise LighthouseError(f"missing category in lighthouse output: {e}")
 87
 88    if any(v is None for v in scores.values()):
 89        missing = [k for k, v in scores.items() if v is None]
 90        raise LighthouseError(f"null score(s) returned by lighthouse: {missing}")
 91
 92    return {k: round(v * 100) for k, v in scores.items()}
 93
 94
 95def parse_performance_details(results):
 96    """
 97    Extract the weighted metrics and top opportunities behind the Performance
 98    score. Returns None if the category is missing — callers should treat that
 99    as "no breakdown available" rather than an error.
100    """
101    try:
102        category = results["categories"]["performance"]
103        audits = results["audits"]
104    except KeyError:
105        return None
106
107    metrics = []
108    opportunities = []
109
110    for ref in category.get("auditRefs", []):
111        audit = audits.get(ref.get("id"))
112        if not audit:
113            continue
114        group = ref.get("group")
115        score = audit.get("score")
116        weight = ref.get("weight", 0)
117
118        if group == "metrics" and weight > 0:
119            metrics.append(
120                {
121                    "id": audit.get("id"),
122                    "acronym": ref.get("acronym") or audit.get("id"),
123                    "title": audit.get("title"),
124                    "display_value": audit.get("displayValue"),
125                    "score": score,
126                    "weight": weight,
127                }
128            )
129            continue
130
131        # Opportunities/diagnostics: skip passing, manual, and not-applicable
132        # audits — we only want actionable findings.
133        mode = audit.get("scoreDisplayMode")
134        if mode in ("manual", "notApplicable", "informative"):
135            continue
136        if score is None or score >= 0.9:
137            continue
138
139        savings_ms = 0
140        details = audit.get("details") or {}
141        if isinstance(details, dict):
142            savings_ms = details.get("overallSavingsMs") or 0
143
144        opportunities.append(
145            {
146                "id": audit.get("id"),
147                "title": audit.get("title"),
148                "display_value": audit.get("displayValue"),
149                "score": score,
150                "savings_ms": savings_ms,
151                "weight": weight,
152            }
153        )
154
155    # Sort metrics by weight desc so the most impactful ones lead.
156    metrics.sort(key=lambda m: m["weight"], reverse=True)
157    # Sort opportunities by estimated savings, then by how badly they failed.
158    opportunities.sort(key=lambda o: (o["savings_ms"], -o["score"]), reverse=True)
159
160    return {
161        "metrics": metrics,
162        "opportunities": opportunities[:10],
163    }