# retoor <retoor@molodetz.nl>
from __future__ import annotations
import asyncio
from urllib.parse import urljoin, urlparse
from xml.etree import ElementTree
import httpx
from devplacepy.net_guard import BlockedAddressError, guard_public_url
from .checks.base import PageContext, SiteContext
USER_AGENT = (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/131.0.0.0 Safari/537.36 DevPlaceSEOBot/1.0"
)
NAV_TIMEOUT_MS = 30000
RAW_FETCH_TIMEOUT = 15.0
MAX_RAW_BYTES = 3_000_000
MOBILE_VIEWPORT = {"width": 390, "height": 844}
DESKTOP_VIEWPORT = {"width": 1366, "height": 900}
INIT_SCRIPT = """
window.__seo = { lcp: 0, cls: 0, consoleErrors: 0, errorSamples: [] };
try {
new PerformanceObserver((list) => {
for (const entry of list.getEntries()) {
window.__seo.lcp = Math.max(window.__seo.lcp, entry.startTime || entry.renderTime || 0);
}
}).observe({ type: 'largest-contentful-paint', buffered: true });
} catch (e) {}
try {
new PerformanceObserver((list) => {
for (const entry of list.getEntries()) {
if (!entry.hadRecentInput) window.__seo.cls += entry.value;
}
}).observe({ type: 'layout-shift', buffered: true });
} catch (e) {}
"""
EXTRACT_SCRIPT = r"""
() => {
const abs = (href) => { try { return new URL(href, document.baseURI).href; } catch (e) { return href || ''; } };
const metaByName = (name) => {
const el = document.querySelector(`meta[name="${name}"]`);
return el ? (el.getAttribute('content') || '') : '';
};
const og = {};
document.querySelectorAll('meta[property^="og:"]').forEach((m) => {
og[m.getAttribute('property')] = m.getAttribute('content') || '';
});
const twitter = {};
document.querySelectorAll('meta[name^="twitter:"]').forEach((m) => {
twitter[m.getAttribute('name')] = m.getAttribute('content') || '';
});
const headings = [];
document.querySelectorAll('h1,h2,h3,h4,h5,h6').forEach((h) => {
headings.push({ level: parseInt(h.tagName.substring(1), 10), text: (h.textContent || '').trim().slice(0, 160) });
});
const h1 = Array.from(document.querySelectorAll('h1')).map((h) => (h.textContent || '').trim().slice(0, 160));
const images = Array.from(document.querySelectorAll('img')).slice(0, 200).map((img) => ({
src: abs(img.getAttribute('src') || ''),
alt: img.getAttribute('alt'),
width: img.getAttribute('width') || (img.naturalWidth ? String(img.naturalWidth) : ''),
height: img.getAttribute('height') || (img.naturalHeight ? String(img.naturalHeight) : ''),
loading: img.getAttribute('loading') || ''
}));
const links = Array.from(document.querySelectorAll('a[href]')).slice(0, 500).map((a) => ({
href: abs(a.getAttribute('href') || ''),
rel: a.getAttribute('rel') || '',
text: (a.textContent || '').trim().slice(0, 120)
}));
const jsonld = Array.from(document.querySelectorAll('script[type="application/ld+json"]')).map((s) => s.textContent || '');
const hreflang = Array.from(document.querySelectorAll('link[rel="alternate"][hreflang]')).map((l) => ({
hreflang: l.getAttribute('hreflang'), href: abs(l.getAttribute('href') || '')
}));
const canonicalEls = document.querySelectorAll('link[rel="canonical"]');
const isHttps = location.protocol === 'https:';
const mixed = [];
if (isHttps) {
document.querySelectorAll('[src],[href]').forEach((el) => {
const v = el.getAttribute('src') || el.getAttribute('href') || '';
if (v.startsWith('http://')) mixed.push(v);
});
}
const viewportEl = document.querySelector('meta[name="viewport"]');
const charsetEl = document.querySelector('meta[charset]');
let formFieldCount = 0;
let formsMissingLabels = 0;
document.querySelectorAll('input,select,textarea').forEach((field) => {
const type = (field.getAttribute('type') || '').toLowerCase();
if (type === 'hidden' || type === 'submit' || type === 'button') return;
formFieldCount += 1;
const id = field.getAttribute('id');
const labelled = (id && document.querySelector(`label[for="${id}"]`)) ||
field.getAttribute('aria-label') || field.getAttribute('aria-labelledby') ||
field.closest('label');
if (!labelled) formsMissingLabels += 1;
});
const nav = performance.getEntriesByType('navigation')[0] || {};
const resources = performance.getEntriesByType('resource') || [];
let transfer = nav.transferSize || 0;
resources.forEach((r) => { transfer += (r.transferSize || 0); });
const bodyText = (document.body ? document.body.innerText || '' : '');
const wordCount = bodyText.split(/\s+/).filter(Boolean).length;
return {
title: (document.title || '').trim(),
titleCount: document.querySelectorAll('title').length,
metaDescription: metaByName('description'),
metaRobots: metaByName('robots'),
canonical: canonicalEls.length ? abs(canonicalEls[0].getAttribute('href') || '') : '',
canonicalCount: canonicalEls.length,
htmlLang: document.documentElement.getAttribute('lang') || '',
charset: charsetEl ? (charsetEl.getAttribute('charset') || '') : (document.characterSet || ''),
hasViewport: !!viewportEl,
viewportContent: viewportEl ? (viewportEl.getAttribute('content') || '') : '',
favicon: !!document.querySelector('link[rel~="icon"]'),
h1: h1,
headings: headings,
images: images,
links: links,
jsonld: jsonld,
hreflang: hreflang,
microdata: !!document.querySelector('[itemscope]'),
rdfa: !!document.querySelector('[vocab],[typeof],[property]'),
og: og,
twitter: twitter,
iframeCount: document.querySelectorAll('iframe').length,
semantic: {
main: document.querySelectorAll('main').length,
article: document.querySelectorAll('article').length,
nav: document.querySelectorAll('nav').length,
header: document.querySelectorAll('header').length,
footer: document.querySelectorAll('footer').length
},
mixedContent: mixed,
formFieldCount: formFieldCount,
formsMissingLabels: formsMissingLabels,
wordCount: wordCount,
metrics: {
ttfb: Math.max(0, (nav.responseStart || 0) - (nav.requestStart || 0)),
fcp: (performance.getEntriesByName('first-contentful-paint')[0] || {}).startTime || 0,
domContentLoaded: nav.domContentLoadedEventEnd || 0,
load: nav.loadEventEnd || 0,
transferSize: transfer,
resourceCount: resources.length,
domNodes: document.getElementsByTagName('*').length,
protocol: nav.nextHopProtocol || '',
lcp: window.__seo ? window.__seo.lcp : 0,
cls: window.__seo ? window.__seo.cls : 0
}
};
}
"""
MOBILE_SCRIPT = r"""
() => {
let small = 0;
document.querySelectorAll('a,button,[role="button"],input[type="submit"]').forEach((el) => {
const r = el.getBoundingClientRect();
if (r.width > 0 && r.height > 0 && (r.width < 24 || r.height < 24)) small += 1;
});
return {
scrollWidth: document.documentElement.scrollWidth,
clientWidth: document.documentElement.clientWidth,
hasHorizontalOverflow: document.documentElement.scrollWidth > document.documentElement.clientWidth + 4,
smallTapTargets: small
};
}
"""
def _normalize_url(url: str) -> str:
url = (url or "").strip()
if not url:
return ""
if "://" not in url:
url = "https://" + url
return url
async def _fetch_text(client: httpx.AsyncClient, url: str) -> dict:
try:
response = await client.get(url)
body = response.text[:MAX_RAW_BYTES]
return {"status": response.status_code, "text": body, "url": str(response.url)}
except httpx.HTTPError as exc:
return {"status": 0, "text": "", "url": url, "error": str(exc)[:200]}
def _parse_robots(text: str, target_path: str) -> dict:
disallows = []
sitemap_urls = []
active = False
for raw in text.splitlines():
line = raw.split("#", 1)[0].strip()
if not line:
continue
key, _, value = line.partition(":")
key = key.strip().lower()
value = value.strip()
if key == "user-agent":
active = value == "*"
elif key == "sitemap":
sitemap_urls.append(value)
elif key == "disallow" and active and value:
disallows.append(value)
blocks_target = any(
target_path.startswith(rule) for rule in disallows if rule and rule != "/"
) or "/" in [r for r in disallows]
return {
"disallows": disallows,
"sitemap_urls": sitemap_urls,
"blocks_target": blocks_target,
}
def _parse_sitemap(text: str) -> dict:
result = {"valid_xml": False, "url_count": 0, "lastmod_count": 0, "urls": []}
try:
root = ElementTree.fromstring(text)
except ElementTree.ParseError:
return result
result["valid_xml"] = True
urls = []
lastmods = 0
for url_el in root.iter():
tag = url_el.tag.rsplit("}", 1)[-1]
if tag == "loc" and url_el.text:
urls.append(url_el.text.strip())
elif tag == "lastmod" and url_el.text:
lastmods += 1
result["url_count"] = len(urls)
result["lastmod_count"] = lastmods
result["urls"] = urls
return result
async def _site_resources(
client: httpx.AsyncClient, base_url: str, sitemap_url: str, target_path: str
) -> tuple[dict, dict, dict]:
parsed = urlparse(base_url)
root = f"{parsed.scheme}://{parsed.netloc}"
robots_raw = await _fetch_text(client, urljoin(root + "/", "robots.txt"))
robots = dict(robots_raw)
if robots_raw.get("status") == 200:
robots.update(_parse_robots(robots_raw.get("text", ""), target_path))
sm_target = sitemap_url or urljoin(root + "/", "sitemap.xml")
sitemap_raw = await _fetch_text(client, sm_target)
sitemap = dict(sitemap_raw)
if sitemap_raw.get("status") == 200:
sitemap.update(_parse_sitemap(sitemap_raw.get("text", "")))
llms_raw = await _fetch_text(client, urljoin(root + "/", "llms.txt"))
llms = {"found": llms_raw.get("status") == 200, "status": llms_raw.get("status")}
return robots, sitemap, llms
async def _build_page(context, client, url: str, output_dir, index: int) -> PageContext:
page_ctx = PageContext(requested_url=url, url=url)
page = await context.new_page()
console_errors = []
def _on_console(message):
if message.type == "error":
console_errors.append(message.text[:200])
page.on("console", _on_console)
page.on("pageerror", lambda exc: console_errors.append(str(exc)[:200]))
await page.add_init_script(INIT_SCRIPT)
try:
response = await page.goto(url, wait_until="load", timeout=NAV_TIMEOUT_MS)
try:
await page.wait_for_load_state("networkidle", timeout=8000)
except Exception:
pass
page_ctx.status = response.status if response else 0
page_ctx.url = page.url
page_ctx.ok = bool(response and response.ok)
if response is not None:
page_ctx.headers = {k.lower(): v for k, v in response.headers.items()}
chain = []
req = response.request.redirected_from
while req is not None:
resp = await req.response()
chain.append({"url": req.url, "status": resp.status if resp else 0})
req = req.redirected_from
page_ctx.redirect_chain = list(reversed(chain))
dom = await page.evaluate(EXTRACT_SCRIPT)
page_ctx.metrics = dom.pop("metrics", {})
page_ctx.metrics["consoleErrors"] = len(console_errors)
page_ctx.metrics["consoleErrorSamples"] = console_errors[:5]
page_ctx.dom = dom
page_ctx.rendered_html = (await page.content())[:MAX_RAW_BYTES]
try:
await page.set_viewport_size(MOBILE_VIEWPORT)
page_ctx.mobile = await page.evaluate(MOBILE_SCRIPT)
await page.set_viewport_size(DESKTOP_VIEWPORT)
except Exception:
page_ctx.mobile = {}
if output_dir is not None:
shot = output_dir / f"page-{index}.png"
try:
await page.screenshot(path=str(shot), full_page=False)
page_ctx.screenshot = shot.name
except Exception:
page_ctx.screenshot = ""
raw = await _fetch_text(client, page_ctx.url)
page_ctx.raw_html = raw.get("text", "")
except Exception as exc: # noqa: BLE001 - record the failure as page state
page_ctx.error = str(exc)[:300]
page_ctx.ok = False
finally:
await page.close()
return page_ctx
async def crawl_target(payload: dict, emit, output_dir) -> SiteContext:
target = _normalize_url(payload.get("url", ""))
mode = payload.get("mode", "url")
allow_private = bool(payload.get("allow_private"))
max_pages = max(1, min(int(payload.get("max_pages", 1) or 1), 50))
await guard_public_url(target, allow_private=allow_private)
parsed = urlparse(target)
site = SiteContext(
target_url=target,
mode=mode,
base_host=parsed.netloc,
base_scheme=parsed.scheme,
)
from playwright.async_api import async_playwright
async with httpx.AsyncClient(
follow_redirects=True,
timeout=RAW_FETCH_TIMEOUT,
headers={"User-Agent": USER_AGENT},
) as client:
emit({"type": "stage", "stage": "resources", "message": "Reading robots.txt and sitemap"})
sitemap_hint = target if mode == "sitemap" else ""
robots, sitemap, llms = await _site_resources(
client, target, sitemap_hint, parsed.path or "/"
)
site.robots = robots
site.sitemap = sitemap
site.llms_txt = llms
if mode == "sitemap":
candidates = sitemap.get("urls", [])[:max_pages]
else:
candidates = [target]
safe_urls = []
for url in candidates:
host = urlparse(url).netloc
if host and host == site.base_host:
# Same host as the audited target, already validated by the
# top-level guard above; skip the redundant per-URL DNS lookup.
safe_urls.append(url)
continue
try:
await guard_public_url(url, allow_private=allow_private)
safe_urls.append(url)
except BlockedAddressError:
continue
if not safe_urls:
if mode == "sitemap":
raise ValueError(
"No crawlable page URLs found in the sitemap "
f"({sitemap.get('url_count', 0)} entries; none reachable or all blocked)."
)
safe_urls = [target]
emit(
{
"type": "target",
"url": target,
"mode": mode,
"urls": safe_urls,
"sitemap_total": sitemap.get("url_count", 0) if mode == "sitemap" else 0,
}
)
async with async_playwright() as pw:
browser = await pw.chromium.launch(
headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"]
)
context = await browser.new_context(
viewport=DESKTOP_VIEWPORT,
user_agent=USER_AGENT,
ignore_https_errors=False,
)
try:
total = len(safe_urls)
for index, url in enumerate(safe_urls):
emit(
{
"type": "progress",
"done": index,
"total": total,
"url": url,
"message": f"Auditing {url}",
}
)
page_ctx = await _build_page(
context, client, url, output_dir, index
)
site.pages.append(page_ctx)
yield_frame = {
"type": "page_loaded",
"url": page_ctx.url,
"status": page_ctx.status,
"done": index + 1,
"total": total,
}
emit(yield_frame)
finally:
await context.close()
await browser.close()
return site
View Git Blame Copy Permalink