40-validation/validate-links.py
#!/usr/bin/env python3
"""Read-only link checker for source Markdown and published HTML."""
from __future__ import annotations
import argparse, json, os, re, time, urllib.error, urllib.request
from datetime import datetime, timezone
from html.parser import HTMLParser
from pathlib import Path
MARKDOWN_LINK = re.compile(r"(?<!!)\[([^\]]*)\]\(([^)]+)\)")
REPO_REF = re.compile(r"(?<![\w.-])([A-Za-z0-9][A-Za-z0-9._-]*):([^\s`<>]+)")
GENERATED = ("reports/", "test/reports/", "99-berichte/")
SKIPPED = ("mailto:", "data:", "javascript:")
def stable_timestamp(value=None):
if value: return value
epoch = int(os.environ.get("SOURCE_DATE_EPOCH", "0"))
return datetime.fromtimestamp(epoch, timezone.utc).isoformat().replace("+00:00", "Z")
def anchor(text):
value = re.sub(r"[^\w\s-]", "", text.strip().lower(), flags=re.UNICODE)
return re.sub(r"[\s-]+", "-", value).strip("-")
def anchors(path):
return {anchor(match.group(1)) for line in path.read_text(encoding="utf-8").splitlines()
if (match := re.match(r"^#{1,6}\s+(.+?)\s*#*\s*$", line))}
class HtmlLinks(HTMLParser):
def __init__(self): super().__init__(); self.links=[]
def handle_starttag(self, tag, attrs):
for key, value in attrs:
if key.lower() in ("href", "src") and value: self.links.append(value)
def _resolve_case(root, relative):
current = root
for part in relative.parts:
if part in ("", "."): continue
if part == "..": current = current.parent; continue
# The overwhelmingly common case in materialized reports is an exact
# path. Avoid scanning every directory for every navigation link.
# Directory enumeration is retained solely for the case-mismatch
# diagnostic promised by this validator.
direct = current / part
if direct.exists():
current = direct
continue
if not current.is_dir(): return None, "MISSING_PATH"
matches = [item for item in current.iterdir() if item.name.lower() == part.lower()]
if not matches: return None, "MISSING_PATH"
exact = [item for item in matches if item.name == part]
if not exact: return matches[0], "CASE_MISMATCH"
current = exact[0]
return current, None
def finding(code, location, target, hint, **extra):
return {"severity":"ERROR", "errorClass":code, "sourceArtifact":location,
"target":target, "correctionHint":hint, **extra}
def reference(repository, relative, line=None):
value = f"{repository}:{relative}"
return f"{value}#L{line}" if line else value
def enrich(item, *, source_reference, expected_target, link_type, preview, relationship="verweist auf"):
"""Give every finding enough context for a person to navigate its cause."""
item.update({"sourceReference":source_reference, "expectedTarget":expected_target,
"targetReference":expected_target, "linkType":link_type,
"preview":preview or "(ohne sichtbaren Linktext)", "relationship":relationship,
"technicalReference":"engineering-tools:40-validation/validate-links.py"})
return item
def grouped_findings(findings):
fields={"byExpectedTarget":"expectedTarget", "bySourceLocation":"sourceReference", "byPreview":"preview", "byRelationship":"relationship", "byTechnicalReference":"technicalReference"}
return {name:{key:[item["errorClass"] for item in findings if item.get(field)==key]
for key in sorted({item.get(field, "") for item in findings})} for name,field in fields.items()}
class _NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl): return None
def _external_finding(code, location, value, hint, **extra):
return {"severity":"WARNING" if code in ("ACCESS_RESTRICTED", "RETRY") else "ERROR",
"errorClass":code, "sourceArtifact":location, "target":value,
"correctionHint":hint, **extra}
def _probe_external(value, location, cache, host_last_probe, rate_limit):
cached = cache.get(value)
if cached: return None if cached.get("status") == "REACHABLE" else cached
current, chain, content_type = value, [], None
opener = urllib.request.build_opener(_NoRedirect())
for _ in range(6):
host = urllib.parse.urlparse(current).netloc.casefold()
wait = rate_limit - (time.monotonic() - host_last_probe.get(host, 0))
if wait > 0: time.sleep(wait)
host_last_probe[host] = time.monotonic()
def request(method):
headers={"User-Agent":"engineering-link-checker/1.0"}
if method == "GET": headers["Range"]="bytes=0-4096"
return opener.open(urllib.request.Request(current, method=method, headers=headers), timeout=10)
try:
try:
response=request("HEAD")
except urllib.error.HTTPError as error:
if error.code not in (405, 501): raise
response=request("GET")
with response:
status=response.status; content_type=response.headers.get_content_type()
except urllib.error.HTTPError as error:
status=error.code
if status in (301,302,303,307,308):
target=error.headers.get("Location")
if not target: return _external_finding("HTTP_STATUS", location, value, "Redirect ohne Ziel prüfen.", httpStatus=status, redirectChain=chain)
next_url=urllib.parse.urljoin(current, target); chain.append({"status":status,"url":current,"location":next_url}); current=next_url; continue
if status in (401,403): return _external_finding("ACCESS_RESTRICTED", location, value, "Quelle existiert, ist aber nicht öffentlich prüfbar.", httpStatus=status, finalUrl=current, redirectChain=chain)
if status in (429,500,502,503,504): return _external_finding("RETRY", location, value, "Temporären externen Fehler später erneut prüfen.", httpStatus=status, finalUrl=current, redirectChain=chain)
return _external_finding("BROKEN" if status in (404,410) else "HTTP_STATUS", location, value, "HTTP-Status des Zielsystems prüfen; keine automatische Umleitung übernehmen.", httpStatus=status, finalUrl=current, redirectChain=chain)
except (urllib.error.URLError, TimeoutError, ValueError) as error:
return _external_finding("RETRY", location, value, "Erreichbarkeit später erneut prüfen; kein lokaler Publish-Fehler.", detail=str(error.reason) if hasattr(error, "reason") else str(error), finalUrl=current, redirectChain=chain)
# A reachable target is not a finding. Redirect metadata is retained
# only when it needs human attention, so ordinary successful links do
# not turn the local quality report into noise.
if chain:
return _external_finding("REDIRECT_REVIEW", location, value, "Permanente oder hostübergreifende Umleitung fachlich prüfen.", httpStatus=status, finalUrl=current, contentType=content_type, redirectChain=chain, redirectsToOtherHost=urllib.parse.urlparse(value).netloc.casefold()!=urllib.parse.urlparse(current).netloc.casefold())
cache[value]={"status":"REACHABLE", "checkedAt":stable_timestamp(), "finalUrl":current, "httpStatus":status, "contentType":content_type, "redirectChain":chain}
return None
return _external_finding("RETRY", location, value, "Redirect-Kette später erneut prüfen.", finalUrl=current, redirectChain=chain)
def check_target(value, location, base, roots, anchor_cache, check_external, external_cache=None, host_last_probe=None, rate_limit=.2):
value = value.strip().split()[0].strip("<>")
if not value or value.startswith(SKIPPED): return None
if value.startswith(("http://", "https://")):
if not check_external: return {"severity":"INFO", "errorClass":"EXTERNAL_NOT_REQUESTED", "sourceArtifact":location, "target":value, "correctionHint":"Externen Zieltest im Batch mit --check-external ausführen."}
return _probe_external(value, location, external_cache if external_cache is not None else {}, host_last_probe if host_last_probe is not None else {}, rate_limit)
if value.startswith("#"):
resolved, fragment = base, value[1:]
else:
fragment=""
if "#" in value: value, fragment=value.split("#",1)
resolved, error = _resolve_case(base, Path(value))
if error: return finding(error, location, value, "Pfad und Groß-/Kleinschreibung an der Quelle bewusst korrigieren.")
if fragment:
if not resolved or resolved.suffix.lower() not in (".md", ".html", ".htm"):
return finding("ANCHOR_TARGET_UNSUPPORTED", location, value, "Ankerziel muss ein prüfbares Markdown- oder HTML-Dokument sein.")
known=anchor_cache.setdefault(resolved, anchors(resolved) if resolved.suffix.lower()==".md" else set())
if resolved.suffix.lower() == ".md" and fragment.lower() not in known:
return finding("MISSING_ANCHOR", location, value, "Ankername oder Zielüberschrift bewusst angleichen.")
return None
def validate(roots, published_html=(), check_external=False, timestamp=None, work_item="SE-0059E", check_sources=True, external_cache=None, rate_limit=.2):
findings=[]; checked=0; generated=0; anchor_cache={}; host_last_probe={}
if check_sources:
for repository, root in sorted(roots.items()):
for path in sorted(root.rglob("*.md")):
rel=path.relative_to(root).as_posix()
if any(rel.startswith(prefix) for prefix in GENERATED): generated += 1; continue
text=path.read_text(encoding="utf-8"); location=f"{repository}:{rel}"
refs=[]
for match in MARKDOWN_LINK.finditer(text):
label,value=match.groups(); line=text.count("\n",0,match.start())+1
value=value.strip(); qualified=REPO_REF.fullmatch(value)
if qualified and qualified.group(1) in roots:
target_repository,target=qualified.groups()
refs.append((target,roots[target_repository],reference(repository,rel,line),value,"repository-Referenz",label.strip()))
else:
refs.append((value,path if value.startswith("#") else path.parent,reference(repository,rel,line),value,"Markdown-Link",label.strip()))
# A qualified reference in Markdown was already handled above.
for match in REPO_REF.finditer(MARKDOWN_LINK.sub("",text)):
repo,target=match.groups()
if repo in roots:
line=text.count("\n",0,match.start())+1; value=target.rstrip(".,;")
refs.append((value,roots[repo],reference(repository,rel,line),f"{repo}:{value}","repository-Referenz",value))
for value,base,source_ref,expected_target,link_type,preview in refs:
checked += 1
result=check_target(value, location, base, roots, anchor_cache, check_external, external_cache, host_last_probe, rate_limit)
if result: findings.append(enrich(result,source_reference=source_ref,expected_target=expected_target,link_type=link_type,preview=preview))
seen_published=set()
for path in published_html:
parser=HtmlLinks(); parser.feed(path.read_text(encoding="utf-8")); location=f"published:{path.name}"
for index,value in enumerate(parser.links,start=1):
key=(path.parent.resolve(), value)
if key in seen_published: continue
seen_published.add(key)
checked += 1
result=check_target(value, location, path if value.startswith("#") else path.parent, roots, anchor_cache, check_external, external_cache, host_last_probe, rate_limit)
if result: findings.append(enrich(result,source_reference=f"published:{path.name}#link-{index}",expected_target=value,link_type="HTML-Link",preview=value))
errors=[item for item in findings if item["severity"] == "ERROR"]
return {"schemaVersion":"2.0", "workItem":work_item, "generatedAt":stable_timestamp(timestamp),
"status":"FAILED" if errors else "PASSED", "repositories":len(roots), "checkedReferences":checked,
"excludedGeneratedReports":generated, "automaticCorrectionsApplied":False,
"findings":findings, "groups":grouped_findings(findings)}
def markdown_report(result):
rows=["# Link-Befundbericht","",f"- Status: **{result['status']}**",f"- Zeitstempel: `{result['generatedAt']}`",f"- Geprüfte Referenzen: {result['checkedReferences']}","- Automatische Korrekturen: keine","","## Befunde nach erwartetem Ziel"]
items=[x for x in result["findings"] if x["severity"]=="ERROR"]
if not items: return "\n".join(rows+["","Keine fehlerhaften Links festgestellt."])+"\n"
for target in sorted({x["expectedTarget"] for x in items}):
rows += ["",f"### `{target}`","","| Quelle und Stelle | Vorschau | Linktyp / Beziehung | Schwere | Befund | Direkter Zielverweis |","|---|---|---|---|---|---|"]
for x in (item for item in items if item["expectedTarget"]==target):
rows.append(f"| [{x['sourceReference']}]({x['sourceReference']}) | {x['preview']} | {x['linkType']} / {x['relationship']} | {x['severity']} | {x['errorClass']}: {x['correctionHint']} | [{x['targetReference']}]({x['targetReference']}) |")
rows += ["","## Technischer Bezug","","- `engineering-tools:40-validation/validate-links.py`"]
return "\n".join(rows)+"\n"
def main():
p=argparse.ArgumentParser(); p.add_argument("--repository", action="append", default=[], help="name=path")
p.add_argument("--published-html", type=Path, action="append", default=[]); p.add_argument("--published-root", type=Path, action="append", default=[]); p.add_argument("--check-external", action="store_true")
p.add_argument("--allow-findings", action="store_true", help="Befund schreiben, aber Prozess nicht mit Exit 1 abbrechen")
p.add_argument("--external-cache", type=Path, help="JSON-Cache für explizite externe Audits")
p.add_argument("--external-rate-limit", type=float, default=.2, help="Mindestabstand je Host in Sekunden")
p.add_argument("--published-only", action="store_true", help="Nur materialisierte HTML-Links prüfen")
p.add_argument("--timestamp"); p.add_argument("--work-item", default="SE-0059E"); p.add_argument("--output", type=Path); p.add_argument("--markdown-output", type=Path); a=p.parse_args()
roots={}
if not a.repository and not a.published_only: p.error("mindestens ein --repository oder --published-only benötigt")
for value in a.repository:
name, sep, path=value.partition("=")
if not sep: p.error("--repository benötigt name=path")
root=Path(path).resolve()
if not root.is_dir(): p.error(f"Repository nicht lesbar: {root}")
roots[name]=root
published=[x.resolve() for x in a.published_html]
for root in a.published_root:
resolved=root.resolve()
if not resolved.is_dir(): p.error(f"Published HTML Root nicht lesbar: {resolved}")
published.extend(sorted(resolved.rglob("*.html")))
cache={}
if a.external_cache and a.external_cache.exists():
try: cache=json.loads(a.external_cache.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError): cache={}
result=validate(roots, published, a.check_external, a.timestamp, a.work_item, check_sources=not a.published_only, external_cache=cache, rate_limit=max(0, a.external_rate_limit))
if a.external_cache and a.check_external:
a.external_cache.parent.mkdir(parents=True, exist_ok=True); a.external_cache.write_text(json.dumps(cache, ensure_ascii=False, indent=2)+"\n", encoding="utf-8")
payload=json.dumps(result,ensure_ascii=False,indent=2)+"\n"
if a.output: a.output.parent.mkdir(parents=True,exist_ok=True); a.output.write_text(payload,encoding="utf-8")
else: print(payload,end="")
if a.markdown_output: a.markdown_output.parent.mkdir(parents=True,exist_ok=True); a.markdown_output.write_text(markdown_report(result),encoding="utf-8")
raise SystemExit(0 if result["status"]=="PASSED" or a.allow_findings else 1)
if __name__=="__main__": main()