b3ef3a90d4
Neue 4xx/5xx-Seiten (nicht in Baseline) werden bewertet (+20 Pkt) und lösen ab Schwelle einen Alert aus. Bekannte kaputte Links (in Baseline vorhanden) erscheinen im Report als [bekannt] ohne Score-Beitrag. Motivation: erster Crawl fand bredelar.info/www.stadtmarketing-marsberg.de (404) — CMS-Fehler durch Schema-losen href. Soll dauerhaft sichtbar sein. Änderungen: checker.find_broken_pages(), differ.compare_snapshots() mit new_broken_urls/known_broken_urls, score_diff(), Report-Abschnitt, config-Default new_broken_link:20, 4 neue Tests. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
364 lines
13 KiB
Python
364 lines
13 KiB
Python
"""
|
|
Diff logic: text, link, metadata, and URL-set comparison.
|
|
Scoring maps diff results to a risk level (green/yellow/red).
|
|
"""
|
|
import difflib
|
|
import hashlib
|
|
import re
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Normalisation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def normalize_text(text: str, cfg: dict) -> str:
|
|
"""Apply noise-reduction before diffing."""
|
|
norm_cfg = cfg.get("normalization", {})
|
|
if norm_cfg.get("collapse_whitespace", True):
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
for pattern in (norm_cfg.get("ignore_patterns") or []):
|
|
text = re.sub(pattern, "[IGNORED]", text)
|
|
return text
|
|
|
|
|
|
def text_hash(text: str) -> str:
|
|
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Atomic diffs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def diff_text(old: str, new: str, context: int = 2) -> list[str]:
|
|
"""Unified diff of two normalised texts."""
|
|
return list(difflib.unified_diff(
|
|
old.splitlines(),
|
|
new.splitlines(),
|
|
fromfile="baseline",
|
|
tofile="current",
|
|
lineterm="",
|
|
n=context,
|
|
))
|
|
|
|
|
|
def diff_link_set(old_list: list[dict], new_list: list[dict]) -> dict:
|
|
"""Compare two link lists by URL."""
|
|
old_urls = {l["url"] for l in old_list if isinstance(l, dict) and "url" in l}
|
|
new_urls = {l["url"] for l in new_list if isinstance(l, dict) and "url" in l}
|
|
return {
|
|
"added": sorted(new_urls - old_urls),
|
|
"removed": sorted(old_urls - new_urls),
|
|
}
|
|
|
|
|
|
def diff_metadata(old: dict, new: dict) -> dict:
|
|
"""Compare metadata fields that matter for integrity monitoring."""
|
|
changes: dict = {}
|
|
for key in ("title", "description", "canonical", "robots"):
|
|
if old.get(key) != new.get(key):
|
|
changes[key] = {"old": old.get(key), "new": new.get(key)}
|
|
|
|
for heading in ("h1", "h2"):
|
|
old_set = set(old.get(heading, []))
|
|
new_set = set(new.get(heading, []))
|
|
if old_set != new_set:
|
|
changes[heading] = {
|
|
"added": sorted(new_set - old_set),
|
|
"removed": sorted(old_set - new_set),
|
|
}
|
|
return changes
|
|
|
|
|
|
def _all_external_domains(pages: dict) -> set[str]:
|
|
"""Collect every external domain referenced across all pages."""
|
|
domains: set[str] = set()
|
|
for page in pages.values():
|
|
for link_type, links in page.get("links", {}).items():
|
|
if not isinstance(links, list):
|
|
continue
|
|
for l in links:
|
|
if isinstance(l, dict) and l.get("class") == "external":
|
|
netloc = urlparse(l.get("url", "")).netloc
|
|
if netloc:
|
|
domains.add(netloc)
|
|
return domains
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Full snapshot comparison
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def compare_snapshots(baseline_pages: dict, current_pages: dict, cfg: dict) -> dict:
|
|
"""
|
|
Compare two page sets (url → page_dict) and produce a structured diff.
|
|
|
|
Returns a dict with:
|
|
new_internal_urls, missing_internal_urls,
|
|
new_external_domains, removed_external_domains,
|
|
page_diffs (list of per-page change records),
|
|
new_pages_analysis (freshly seen pages checked for immediate issues)
|
|
"""
|
|
b_urls = set(baseline_pages)
|
|
c_urls = set(current_pages)
|
|
|
|
new_urls = sorted(c_urls - b_urls)
|
|
missing_urls = sorted(b_urls - c_urls)
|
|
common_urls = sorted(b_urls & c_urls)
|
|
|
|
page_diffs: list[dict] = []
|
|
|
|
for url in common_urls:
|
|
old = baseline_pages[url]
|
|
new = current_pages[url]
|
|
|
|
old_text = normalize_text(old.get("text", ""), cfg)
|
|
new_text = normalize_text(new.get("text", ""), cfg)
|
|
text_changed = text_hash(old_text) != text_hash(new_text)
|
|
|
|
# Quick-skip if nothing changed (text + links + meta all identical)
|
|
if (
|
|
not text_changed
|
|
and old.get("links") == new.get("links")
|
|
and old.get("metadata") == new.get("metadata")
|
|
and old.get("hidden_content") == new.get("hidden_content")
|
|
and old.get("inline_scripts") == new.get("inline_scripts")
|
|
and old.get("jsonld") == new.get("jsonld")
|
|
):
|
|
continue
|
|
|
|
text_diff = diff_text(old_text, new_text) if text_changed else []
|
|
|
|
# Per-type link diff
|
|
link_diff: dict = {}
|
|
for ltype in ("a", "script", "iframe", "link_rel", "form"):
|
|
old_links = old.get("links", {}).get(ltype, [])
|
|
new_links = new.get("links", {}).get(ltype, [])
|
|
ld = diff_link_set(old_links, new_links)
|
|
if ld["added"] or ld["removed"]:
|
|
link_diff[ltype] = ld
|
|
|
|
meta_diff = diff_metadata(old.get("metadata", {}), new.get("metadata", {}))
|
|
|
|
# New hidden content (compare by fingerprint to avoid dupe noise)
|
|
old_hc_fps = {_fingerprint(h) for h in old.get("hidden_content", [])}
|
|
new_hidden = [
|
|
h for h in new.get("hidden_content", [])
|
|
if _fingerprint(h) not in old_hc_fps
|
|
]
|
|
|
|
# New suspicious inline scripts
|
|
old_scripts_fps = {_fingerprint(s) for s in old.get("inline_scripts", [])}
|
|
new_scripts = [
|
|
s for s in new.get("inline_scripts", [])
|
|
if _fingerprint(s) not in old_scripts_fps
|
|
]
|
|
|
|
# New JSON-LD
|
|
old_jld_fps = {_fingerprint(j) for j in old.get("jsonld", [])}
|
|
new_jsonld = [
|
|
j for j in new.get("jsonld", [])
|
|
if _fingerprint(j) not in old_jld_fps
|
|
]
|
|
|
|
# New meta-refresh targets
|
|
old_mr = set(old.get("links", {}).get("meta_refresh", []))
|
|
new_mr = [u for u in new.get("links", {}).get("meta_refresh", []) if u not in old_mr]
|
|
|
|
# New canonical
|
|
old_canonical = old.get("links", {}).get("canonical") or old.get("metadata", {}).get("canonical")
|
|
new_canonical = new.get("links", {}).get("canonical") or new.get("metadata", {}).get("canonical")
|
|
canonical_changed = old_canonical != new_canonical
|
|
|
|
# New comment links
|
|
old_cl = set(old.get("comment_links", []))
|
|
new_cl = [u for u in new.get("comment_links", []) if u not in old_cl]
|
|
|
|
if any([
|
|
text_diff, link_diff, meta_diff, new_hidden, new_scripts,
|
|
new_jsonld, new_mr, canonical_changed, new_cl,
|
|
]):
|
|
added_chars = sum(
|
|
len(line[1:]) for line in text_diff
|
|
if line.startswith("+") and not line.startswith("+++")
|
|
)
|
|
page_diffs.append({
|
|
"url": url,
|
|
"text_diff": text_diff,
|
|
"added_chars": added_chars,
|
|
"link_diff": link_diff,
|
|
"meta_diff": meta_diff,
|
|
"new_hidden_content": new_hidden,
|
|
"new_inline_scripts": new_scripts,
|
|
"new_jsonld": new_jsonld,
|
|
"new_meta_refresh": new_mr,
|
|
"canonical_changed": canonical_changed,
|
|
"old_canonical": old_canonical,
|
|
"new_canonical": new_canonical,
|
|
"new_comment_links": new_cl,
|
|
})
|
|
|
|
# Site-wide external domain diff
|
|
old_ext = _all_external_domains(baseline_pages)
|
|
new_ext = _all_external_domains(current_pages)
|
|
|
|
# Broken link diff (4xx/5xx pages)
|
|
baseline_broken = {
|
|
url for url, p in baseline_pages.items() if p.get("status", 0) >= 400
|
|
}
|
|
current_broken_map = {
|
|
url: p.get("status", 0)
|
|
for url, p in current_pages.items()
|
|
if p.get("status", 0) >= 400
|
|
}
|
|
current_broken = set(current_broken_map)
|
|
new_broken = [
|
|
{"url": url, "status": current_broken_map[url]}
|
|
for url in sorted(current_broken - baseline_broken)
|
|
]
|
|
known_broken = [
|
|
{"url": url, "status": current_broken_map[url]}
|
|
for url in sorted(current_broken & baseline_broken)
|
|
]
|
|
|
|
# Analysis of brand-new pages (no baseline to compare against)
|
|
new_pages_analysis = [
|
|
_quick_check_new_page(current_pages[url])
|
|
for url in new_urls
|
|
if url in current_pages
|
|
]
|
|
|
|
return {
|
|
"new_internal_urls": new_urls,
|
|
"missing_internal_urls": missing_urls,
|
|
"new_external_domains": sorted(new_ext - old_ext),
|
|
"removed_external_domains": sorted(old_ext - new_ext),
|
|
"page_diffs": page_diffs,
|
|
"new_pages_analysis": new_pages_analysis,
|
|
"new_broken_urls": new_broken,
|
|
"known_broken_urls": known_broken,
|
|
"total_pages_checked": len(common_urls),
|
|
"changed_pages": len(page_diffs),
|
|
}
|
|
|
|
|
|
def _quick_check_new_page(page: dict) -> dict:
|
|
"""Summarise a new (not in baseline) page for immediate risk indicators."""
|
|
return {
|
|
"url": page["url"],
|
|
"hidden_content": page.get("hidden_content", []),
|
|
"inline_scripts": page.get("inline_scripts", []),
|
|
"unexpected_jsonld": [j for j in page.get("jsonld", []) if j.get("unexpected")],
|
|
"meta_refresh": page.get("links", {}).get("meta_refresh", []),
|
|
"comment_links": page.get("comment_links", []),
|
|
"external_link_count": sum(
|
|
1 for links in page.get("links", {}).values()
|
|
if isinstance(links, list)
|
|
for l in links
|
|
if isinstance(l, dict) and l.get("class") == "external"
|
|
),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scoring
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def score_diff(diff: dict, cfg: dict) -> dict:
|
|
"""
|
|
Map a diff result to a risk score and level.
|
|
|
|
Returns {score, level, reasons, exit_code}.
|
|
"""
|
|
sc = cfg.get("scoring", {})
|
|
thr = cfg.get("thresholds", {"yellow": 20, "red": 60})
|
|
large_text = thr.get("large_text_block_chars", 200)
|
|
|
|
score = 0
|
|
reasons: list[str] = []
|
|
|
|
def add(pts: int, msg: str) -> None:
|
|
nonlocal score
|
|
score += pts
|
|
reasons.append(f"{msg} (+{pts})")
|
|
|
|
for domain in diff.get("new_external_domains", []):
|
|
add(sc.get("new_external_domain", 50), f"Neue externe Domain: {domain}")
|
|
|
|
for url in diff.get("new_internal_urls", []):
|
|
add(sc.get("new_internal_url", 30), f"Neue interne URL: {url}")
|
|
|
|
for url in diff.get("missing_internal_urls", []):
|
|
add(sc.get("missing_internal_url", 20), f"Fehlende URL: {url}")
|
|
|
|
for pd in diff.get("page_diffs", []):
|
|
u = pd["url"]
|
|
|
|
for h in pd.get("new_hidden_content", []):
|
|
add(sc.get("hidden_content", 40),
|
|
f"{u}: Hidden Content [{', '.join(h.get('indicators', []))}]")
|
|
|
|
for mr in pd.get("new_meta_refresh", []):
|
|
add(sc.get("meta_refresh", 40), f"{u}: Meta-Refresh → {mr}")
|
|
|
|
for s in pd.get("new_inline_scripts", []):
|
|
add(sc.get("new_inline_script_suspicious", 30),
|
|
f"{u}: Verdächtiges Inline-Script [{', '.join(s.get('patterns', []))}]")
|
|
|
|
for j in pd.get("new_jsonld", []):
|
|
if j.get("unexpected"):
|
|
add(sc.get("unexpected_jsonld", 35),
|
|
f"{u}: Unerwartetes JSON-LD @type={j.get('type')}")
|
|
|
|
if pd.get("canonical_changed"):
|
|
add(sc.get("unexpected_canonical", 35),
|
|
f"{u}: Canonical geändert → {pd.get('new_canonical')}")
|
|
|
|
for cl in pd.get("new_comment_links", []):
|
|
add(sc.get("comment_links", 25), f"{u}: Link in HTML-Kommentar: {cl}")
|
|
|
|
if pd.get("added_chars", 0) > large_text:
|
|
add(sc.get("large_text_addition", 20),
|
|
f"{u}: +{pd['added_chars']} Zeichen neuer Text")
|
|
|
|
for entry in diff.get("new_broken_urls", []):
|
|
add(sc.get("new_broken_link", 20),
|
|
f"Neuer kaputte Link: {entry['url']} (HTTP {entry['status']})")
|
|
# known_broken_urls are reported but not scored — they were already in the baseline
|
|
|
|
# New pages: immediate suspicious content counts too
|
|
for npa in diff.get("new_pages_analysis", []):
|
|
u = npa["url"]
|
|
if npa.get("hidden_content"):
|
|
add(sc.get("hidden_content", 40), f"{u} (neu): Hidden Content")
|
|
if npa.get("inline_scripts"):
|
|
add(sc.get("new_inline_script_suspicious", 30), f"{u} (neu): Verdächtiges Script")
|
|
if npa.get("meta_refresh"):
|
|
add(sc.get("meta_refresh", 40), f"{u} (neu): Meta-Refresh")
|
|
if npa.get("comment_links"):
|
|
add(sc.get("comment_links", 25), f"{u} (neu): Kommentar-Link")
|
|
if npa.get("unexpected_jsonld"):
|
|
add(sc.get("unexpected_jsonld", 35), f"{u} (neu): Unerwartetes JSON-LD")
|
|
|
|
yellow = thr.get("yellow", 20)
|
|
red = thr.get("red", 60)
|
|
|
|
if score >= red:
|
|
level = "red"
|
|
exit_code = 2
|
|
elif score >= yellow:
|
|
level = "yellow"
|
|
exit_code = 1
|
|
else:
|
|
level = "green"
|
|
exit_code = 0
|
|
|
|
return {"score": score, "level": level, "reasons": reasons, "exit_code": exit_code}
|
|
|
|
|
|
def _fingerprint(obj: dict | list | str) -> str:
|
|
"""Stable hash for deduplicating diff entries."""
|
|
import json as _json
|
|
return hashlib.md5( # noqa: S324 — not crypto, just dedup
|
|
_json.dumps(obj, sort_keys=True, ensure_ascii=False).encode()
|
|
).hexdigest()
|