beam: URL canonicalization utility with tests

wizard-work/beam/canonical_url.py

@@ -0,0 +1,86 @@
+"""Canonicalize URLs for deduplication.
+
+Handles: scheme/host lowercasing, default-port removal, trailing-slash
+normalization, fragment stripping, query-key sorting, percent-decoding of
+unreserved chars, IDNA host encoding, userinfo removal, dot-segment removal.
+"""
+from __future__ import annotations
+import re
+from urllib.parse import (
+    urlparse, urlunparse, parse_qsl, urlencode, unquote, quote,
+)
+
+_UNRESERVED = set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~")
+_DEFAULT_PORTS = {"http": 80, "https": 443, "ftp": 21}
+
+def _norm_host(host: str) -> str:
+    host = host.strip().lower().rstrip(".")
+    # IDNA encode if non-ascii
+    try:
+        host = host.encode("idna").decode("ascii")
+    except (UnicodeError, UnicodeDecodeError):
+        pass
+    return host
+
+def _norm_path(path: str) -> str:
+    # remove dot segments
+    segs = path.split("/")
+    out = []
+    for s in segs:
+        if s == "." or s == "":
+            # collapse empty only if not leading? keep single leading
+            if s == "":
+                if not out or out[-1] != "":
+                    out.append("")
+                continue
+            continue
+        if s == "..":
+            if len(out) > 1:
+                out.pop()
+            continue
+        out.append(s)
+    # percent-decode unreserved, then re-encode
+    p = "/".join(out)
+    if p == "":
+        p = "/"
+    # normalize percent encoding of unreserved
+    decoded = unquote(p)
+    reenc = "".join(c if c in _UNRESERVED or c == "/" else quote(c, safe="") for c in decoded)
+    # collapse multiple slashes
+    reenc = re.sub(r"/{2,}", "/", reenc)
+    if reenc != "/" and reenc.endswith("/"):
+        # keep trailing slash only for root already; remove for non-root? We'll keep root "/"
+        reenc = reenc.rstrip("/")
+    return reenc if reenc else "/"
+
+def canonicalize(url: str) -> str:
+    if not isinstance(url, str) or not url.strip():
+        raise ValueError("url must be non-empty string")
+    url = url.strip()
+    # add scheme if missing
+    if "://" not in url and not url.startswith("//"):
+        url = "http://" + url
+    if url.startswith("//"):
+        url = "http:" + url
+    u = urlparse(url)
+    scheme = (u.scheme or "http").lower()
+    if scheme not in ("http", "https", "ftp"):
+        scheme = u.scheme.lower()
+    host = _norm_host(u.hostname or "")
+    port = u.port
+    if port is not None and port == _DEFAULT_PORTS.get(scheme):
+        port = None
+    netloc = host
+    if port is not None:
+        netloc = f"{host}:{port}"
+    # userinfo dropped
+    path = _norm_path(u.path or "/")
+    # query sort
+    q = parse_qsl(u.query, keep_blank_values=True)
+    q.sort(key=lambda kv: (kv[0], kv[1]))
+    query = urlencode(q, doseq=True)
+    # fragment dropped
+    return urlunparse((scheme, netloc, path, "", query, ""))
+
+def equivalent(a: str, b: str) -> bool:
+    return canonicalize(a) == canonicalize(b)

wizard-work/beam/test_canonical_url.py

@@ -0,0 +1,30 @@
+import sys
+sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")
+from canonical_url import canonicalize, equivalent
+
+cases = [
+    ("HTTP://Example.COM/Foo/", "http://example.com/Foo"),
+    ("http://example.com:80/path?b=2&a=1#frag", "http://example.com/path?a=1&b=2"),
+    ("http://example.com/a/../b/./c", "http://example.com/b/c"),
+    ("https://example.com//a//b", "https://example.com/a/b"),
+    ("http://user:pass@example.com/x", "http://example.com/x"),
+    ("http://example.com/%7Efoo", "http://example.com/~foo"),
+    ("example.com/path", "http://example.com/path"),
+    ("//example.com/path", "http://example.com/path"),
+]
+for a,b in cases:
+    ca, cb = canonicalize(a), canonicalize(b)
+    assert ca == cb, (a, ca, b, cb)
+    assert equivalent(a,b)
+print("all cases pass")
+# distinct
+assert canonicalize("http://a.com/?x=1") != canonicalize("http://a.com/?x=2")
+assert canonicalize("http://a.com") != canonicalize("https://a.com")
+print("distinct ok")
+# invalid
+try:
+    canonicalize("")
+    raise SystemExit("expected ValueError")
+except ValueError:
+    pass
+print("invalid ok")