Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/tests.yml
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ jobs:

- name: Test with pytest
run: |
python -m pytest --cov=./ --cov-report=xml
python -m pytest --cov=courlan --cov-report=xml

- name: Test docs
# version matches .readthedocs.yml
Expand Down
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
dist/
build/
*.egg-info/
uv.lock

# tests
.cache/
Expand Down
128 changes: 83 additions & 45 deletions courlan/clean.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@

from .filters import is_valid_url
from .settings import ALLOWED_PARAMS, LANG_PARAMS, TARGET_LANGS
from .urlutils import _parse
from .urlutils import _parse, _strip_trailing_dot

LOGGER = logging.getLogger(__name__)

Expand All @@ -18,6 +18,9 @@

MIDDLE_URL = re.compile(r"https?://.+?(https?://.+?)(?:https?://|$)")

# netloc
DEFAULT_PORTS = {"http": 80, "https": 443}

# path
PATH1 = re.compile(r"/+")
PATH2 = re.compile(r"^(?:/\.\.(?![^/]))+")
Expand Down Expand Up @@ -74,29 +77,22 @@ def scrub_url(url: str) -> str:
match = SELECTION.match(url)
if match and is_valid_url(match[1]):
url = match[1]
LOGGER.debug("taking url: %s", url)
else:
match = MIDDLE_URL.match(url)
if match and is_valid_url(match[1]):
url = match[1]
LOGGER.debug("taking url: %s", url)

# too long and garbled URLs e.g. due to quotes URLs
match = TRAILING_PARTS.match(url)
if match:
url = match[1]
if len(url) > 500: # arbitrary choice
LOGGER.debug("invalid-looking link %s of length %d", url[:50] + "…", len(url))
# trailing slashes in URLs without path or in embedded URLs
if url.count("/") == 3 or url.count("://") > 1:
url = url.rstrip("/")

return url


def clean_query(
querystring: str, strict: bool = False, language: str | None = None
) -> str:
def clean_query(querystring: str, strict: bool, language: str | None = None) -> str:
"Strip unwanted query elements"
if not querystring:
return ""
Expand All @@ -119,7 +115,6 @@ def clean_query(
and teststr in LANG_PARAMS
and str(qdict[qelem][0]) not in TARGET_LANGS[language]
):
LOGGER.debug("bad lang: %s %s", language, qelem)
raise ValueError
# insert
newqdict[qelem] = qdict[qelem]
Expand Down Expand Up @@ -161,19 +156,25 @@ def normalize_fragment(fragment: str, language: str | None = None) -> str:
return normalize_part(fragment)


def normalize_authority(parsed_url: SplitResult) -> str:
"Lower-case the authority, decode punycode and strip the scheme default port."
netloc = decode_punycode(parsed_url.netloc.lower())
# port: strip only the scheme's default port (80 for http, 443 for https)
try:
port = parsed_url.port
except ValueError:
port = None # port could not be cast to integer value
def normalize_netloc_parts(parsed_url: SplitResult) -> tuple[str, str, str]:
"Return the normalized (scheme, netloc, hostport), hostport without userinfo."
userinfo, _, hostport = parsed_url.netloc.rpartition("@")
hostport = hostport.lower()
# split port before punycode decoding (e.g. "xn--n3h:8080")
host, sep, port = hostport.rpartition(":")
# a colon left in an unbracketed host means a malformed netloc (e.g. "x:80:80")
if not (sep and port.isascii() and port.isdecimal()) or (
":" in host and not host.endswith("]")
):
host, port = hostport, ""
host = decode_punycode(_strip_trailing_dot(host))
scheme = parsed_url.scheme.lower()
if (scheme == "http" and port == 80) or (scheme == "https" and port == 443):
# strip the trailing default port (IPv6-safe)
netloc = netloc.rsplit(":", 1)[0]
return netloc
# strip only the scheme's default port
if port and DEFAULT_PORTS.get(scheme) == int(port):
port = ""
hostport = f"{host}:{port}" if port else host
netloc = f"{userinfo}@{hostport}" if userinfo else hostport
return scheme, netloc, hostport


def normalize_path(path: str) -> str:
Expand All @@ -183,33 +184,70 @@ def normalize_path(path: str) -> str:
return normalize_part(PATH2.sub("", PATH1.sub("/", path)))


def rebuild_url(
scheme: str,
netloc: str,
path: str,
query: str,
fragment: str,
strict: bool,
language: str | None,
trailing_slash: bool,
) -> str:
"Assemble the final URL string from already normalized parts."
newfragment = "" if strict else normalize_fragment(fragment, language)
if query and not path:
path = "/"
elif path == "/" and not query and not newfragment:
# canonical root form has no trailing slash (matches scrub_url)
path = ""
elif not trailing_slash and not query and path.endswith("/"):
path = path.rstrip("/")
return urlunsplit((scheme, netloc, path, query, newfragment))


def normalize_and_split(
parsed_url: SplitResult,
strict: bool,
language: str | None,
trailing_slash: bool,
) -> tuple[str, str, str]:
"Return the normalized (url, base, path) triple from a single set of parts."
scheme, netloc, _ = normalize_netloc_parts(parsed_url)
url = rebuild_url(
scheme,
netloc,
normalize_path(parsed_url.path),
clean_query(parsed_url.query, strict, language),
parsed_url.fragment,
strict,
language,
trailing_slash,
)
# without a netloc urlunsplit drops the "//" and the slice below would cut
# into the path instead of past the host
if not netloc:
raise ValueError(f"no host to split off: {url}")
base = f"{scheme}://{netloc}"
return url, base, url[len(base) :] or "/"


def normalize_url(
parsed_url: SplitResult | str,
strict: bool = False,
language: str | None = None,
trailing_slash: bool = True,
netloc: str | None = None,
path: str | None = None,
query: str | None = None,
) -> str:
"""Takes a URL string or a parsed URL and returns a normalized URL string.
`netloc`, `path` and `query` skip the corresponding step when the caller has
already normalized that part (see check_url)."""
"Takes a URL string or a parsed URL and returns a normalized URL string."
parsed_url = _parse(parsed_url)
# lowercase + remove fragments + normalize punycode
scheme = parsed_url.scheme.lower()
if netloc is None:
netloc = normalize_authority(parsed_url)
if path is None:
path = normalize_path(parsed_url.path)
# strip unwanted query elements
if query is None:
query = clean_query(parsed_url.query, strict, language)
if query and not path:
path = "/"
elif not trailing_slash and not query and path.endswith("/"):
path = path.rstrip("/")
# fragment
newfragment = "" if strict else normalize_fragment(parsed_url.fragment, language)
# rebuild
return urlunsplit((scheme, netloc, path, query, newfragment))
scheme, netloc, _ = normalize_netloc_parts(parsed_url)
return rebuild_url(
scheme,
netloc,
normalize_path(parsed_url.path),
clean_query(parsed_url.query, strict, language),
parsed_url.fragment,
strict,
language,
trailing_slash,
)
88 changes: 44 additions & 44 deletions courlan/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,16 +2,15 @@
Core functions needed to make the module work.
"""

# import locale
import logging
import re
from urllib.robotparser import RobotFileParser

from .clean import (
clean_query,
normalize_authority,
normalize_netloc_parts,
normalize_path,
normalize_url,
rebuild_url,
scrub_url,
)
from .filters import (
Expand Down Expand Up @@ -58,7 +57,8 @@ def check_url(
with_redirects: set to True for redirection test (per HTTP HEAD request)
language: set target language (ISO 639-1 codes)
with_nav: set to True to include navigation pages instead of discarding them
trailing_slash: set to False to trim trailing slashes
trailing_slash: keep trailing slashes on non-root paths (default True);
the root slash is always stripped

Returns:
A tuple consisting of canonical URL and extracted domain
Expand All @@ -67,12 +67,11 @@ def check_url(
Nothing: invalid URLs are caught internally and None is returned.
"""

# first sanity check
# use standard parsing library, validate and strip fragments, then normalize
# scrub, parse and normalize, then filter on the normalized parts
# and the final form so the output is a fixed point
try:
# length test
if basic_filter(url) is False:
LOGGER.debug("rejected, basic filter: %s", url)
raise ValueError

# clean
Expand All @@ -82,69 +81,69 @@ def check_url(
if with_redirects:
url = redirection_test(url)

# spam & structural elements
if type_filter(url, strict=strict, with_nav=with_nav) is False:
LOGGER.debug("rejected, type filter: %s", url)
raise ValueError

# internationalization and language heuristics in URL
if (
language is not None
and lang_filter(url, language, strict, trailing_slash) is False
):
LOGGER.debug("rejected, lang filter: %s", url)
raise ValueError

# split and validate
validation_test, parsed_url = validate_url(url)
if validation_test is False or parsed_url is None:
LOGGER.debug("rejected, validation test: %s", url)
raise ValueError

# the filters below and normalize_url() need the same normalized parts, so
# compute each one once and pass them on
# normalized parts, shared with the rebuild; the query comes last as
# it is the expensive one and the filters below may reject first
path = normalize_path(parsed_url.path)

# content filter based on extensions
if extension_filter(path) is False:
LOGGER.debug("rejected, extension filter: %s", url)
raise ValueError

# unsuitable domain/host name (strip userinfo; domain_filter expects host[/port])
netloc = normalize_authority(parsed_url)
host = netloc.rsplit("@", 1)[-1]
scheme, netloc, host = normalize_netloc_parts(parsed_url)

# unsuitable domain/host name (without userinfo; domain_filter expects host[:port])
if not host or domain_filter(host) is False:
LOGGER.debug("rejected, domain name: %s", url)
raise ValueError

# spam & structural elements, also hidden in the query or the fragment
pre_normalized = url
if type_filter(url, strict=strict, with_nav=with_nav) is False:
raise ValueError

query = clean_query(parsed_url.query, strict, language)

# strict content filtering: the query has to be the one that survives
# normalization, else a stripped param keeps an index page that
# check_url would reject on a second pass
query = None
if strict:
query = clean_query(parsed_url.query, strict, language)
if path_filter(path, query) is False:
LOGGER.debug("rejected, path filter: %s", url)
raise ValueError

# normalize
url = normalize_url(
parsed_url,
if strict and path_filter(path, query) is False:
raise ValueError

# rebuild
url = rebuild_url(
scheme,
netloc,
path,
query,
parsed_url.fragment,
strict,
language,
trailing_slash,
netloc=netloc,
path=path,
query=query,
)

# again if normalization changed the URL, it can create patterns of its own
if (
url != pre_normalized
and type_filter(url, strict=strict, with_nav=with_nav) is False
):
raise ValueError

# internationalization and language heuristics in URL
if (
language is not None
and lang_filter(url, language, strict, trailing_slash) is False
):
raise ValueError

# domain info: use blacklist in strict mode only
domain = extract_domain(url, blacklist=BLACKLIST if strict else None)
if domain is None:
LOGGER.debug("rejected, domain name: %s", url)
return None

# handle exceptions
except (AttributeError, ValueError):
LOGGER.debug("discarded URL: %s", url)
return None
Expand Down Expand Up @@ -175,7 +174,8 @@ def extract_links(
no_filter: override settings and bypass checks to return all possible URLs
language: set target language (ISO 639-1 codes)
strict: set to True for stricter filtering
trailing_slash: set to False to trim trailing slashes
trailing_slash: keep trailing slashes on non-root paths (default True);
the root slash is always stripped
with_nav: set to True to include navigation pages instead of discarding them
redirects: set to True for redirection test (per HTTP HEAD request)
reference: provide a host reference for external/internal evaluation
Expand Down
Loading
Loading