path: root/webmentions_ssg/tasks/scanner.py

from collections.abc import Iterator
from dataclasses import dataclass
from datetime import UTC, datetime
from fnmatch import fnmatchcase
from itertools import chain
from pathlib import Path
from typing import Any
from urllib.parse import quote, urldefrag, urljoin, urlsplit

import mf2py
import sqlalchemy as sa
import xxhash
from bs4 import BeautifulSoup, Tag
from flask import current_app
from sqlalchemy.orm import selectinload

from .. import DATABASE as db
from .. import HUEY as huey
from ..models import SentWebmention, Source
from ..url_security import is_http_url
from .sender import send_webmention

REACTION_PROPERTIES = ("in-reply-to", "like-of", "repost-of", "bookmark-of")


class SourceScanError(Exception):
    """A source document cannot safely be processed."""


@dataclass(frozen=True)
class ScannedSource:
    path: str
    url: str
    content_hash: str
    targets: frozenset[str]


def source_url_for_path(relative_path: Path, base_url: str) -> str:
    """
    Derive the public source URL from a relative filesystem path.

    :param relative_path: Path of the source document relative to the source root.
    :param base_url: Base URL under which source documents are published.
    :return: Public URL corresponding to the source document.
    """
    directory = relative_path.parent.as_posix()

    return urljoin(base_url, f"{quote(directory, safe='/')}/")


def canonical_url(
    document: BeautifulSoup, *, relative_path: Path, base_url: str | None
) -> str | None:
    """
    Return the canonical URL declared by a source document.

    Relative canonical URLs are resolved against the configured source base URL.

    :param document: Parsed HTML source document.
    :param relative_path: Path of the source document relative to the source root.
    :param base_url: Configured source base URL, if available.
    :return: Canonical URL, or ``None`` if none is declared.
    :raises SourceScanError: If the canonical URL is empty, cannot be resolved,
        or does not resolve to an HTTP or HTTPS URL.
    """
    if (link := document.select_one('link[rel~="canonical"][href]')) is not None:
        href = link.get("href")

        if not isinstance(href, str) or not (href := href.strip()):
            raise SourceScanError("Document contains an empty canonical URL")

        if is_http_url(href):
            return href

        if base_url is None:
            raise SourceScanError(
                "Document contains a relative canonical URL, but "
                "WEBMENTIONS_SSG_SOURCE_BASE_URL is not configured"
            )

        resolved = urljoin(source_url_for_path(relative_path, base_url), href)

        if not is_http_url(resolved):
            raise SourceScanError(
                f"Canonical URL is not a valid HTTP or HTTPS URL: {href!r}"
            )

        return resolved


def property_urls(entry: dict[str, Any], property_name: str) -> Iterator[str]:
    """
    Yield URL values from a microformats property.

    :param entry: Parsed microformats entry.
    :param property_name: Name of the property to inspect.
    :return: Iterator over string values of the property.
    """
    properties = entry.get("properties")

    if not isinstance(properties, dict):
        return

    values = properties.get(property_name)

    if not isinstance(values, list):
        return

    yield from (value for value in values if isinstance(value, str))


def parse_entry(element: Tag, base_url: str) -> dict[str, Any]:
    """
    Parse and validate an h-entry with mf2py.

    :param element: HTML element containing the h-entry.
    :param base_url: Base URL used when parsing the microformats data.
    :return: Parsed h-entry.
    :raises SourceScanError: If parsing does not produce exactly one valid h-entry.
    """
    parsed = mf2py.parse(doc=str(element), url=base_url)

    if not isinstance(parsed, dict):
        raise SourceScanError("Microformats parser did not return a document object")

    items = parsed.get("items")

    if not isinstance(items, list) or len(items) != 1:
        raise SourceScanError("Expected exactly one parsed h-entry")

    entry = items[0]

    if not isinstance(entry, dict):
        raise SourceScanError("Parsed h-entry is not an object")

    types = entry.get("type")

    if not isinstance(types, list) or "h-entry" not in types:
        raise SourceScanError("Parsed microformats item is not an h-entry")

    return entry


def primary_entry(
    document: BeautifulSoup, source_url: str
) -> tuple[Tag, dict[str, Any]]:
    """
    Return and validate the primary h-entry of a source document.

    The document must contain exactly one h-entry whose ``u-url`` matches the
    source URL.

    :param document: Parsed HTML source document.
    :param source_url: Public URL of the source document.
    :return: h-entry element and its parsed microformats representation.
    :raises SourceScanError: If the document does not contain exactly one h-entry
        or its ``u-url`` does not match the source URL.
    """
    entries = document.find_all(class_="h-entry")

    if len(entries) != 1:
        raise SourceScanError(f"Expected exactly one h-entry, found {len(entries)}")

    entry_element = entries[0]
    mf2_entry = parse_entry(entry_element, source_url)

    urls = tuple(property_urls(mf2_entry, "url"))

    if source_url not in urls:
        raise SourceScanError(
            f"The h-entry u-url does not match the source URL {source_url!r}: {urls!r}"
        )

    return entry_element, mf2_entry


def content_element(entry: Tag) -> Tag:
    """
    Return the e-content element of an h-entry.

    :param entry: HTML element containing the h-entry.
    :return: The h-entry's e-content element.
    :raises SourceScanError: If the h-entry does not contain exactly one
        e-content element.
    """
    contents = entry.find_all(class_="e-content")

    if len(contents) != 1:
        raise SourceScanError(f"Expected exactly one e-content, found {len(contents)}")

    return contents[0]


def normalize_target(value: str, *, base_url: str, source_url: str) -> str | None:
    """
    Resolve and validate a possible Webmention target URL.

    Empty values, non-HTTP URLs, and URLs referring to the source itself are
    discarded.

    :param value: URL reference to normalize.
    :param base_url: Base URL against which relative references are resolved.
    :param source_url: URL of the source document.
    :return: Normalized target URL, or ``None`` if the value is not a valid
        Webmention target.
    """
    value = value.strip()

    if not value:
        return None

    target = urljoin(base_url, value)

    if not is_http_url(target):
        return None

    if urldefrag(target)[0] == urldefrag(source_url)[0]:
        return None

    return target


def iter_targets(
    content: Tag,
    mf2_entry: dict[str, Any],
    *,
    base_url: str,
    source_url: str,
    ignored_hostnames: tuple[str, ...] = (),
) -> Iterator[str]:
    """
    Yield outgoing Webmention targets from an h-entry.

    Targets are collected from links in the entry content and supported
    microformats reaction properties. Invalid, self-referencing, and ignored
    targets are excluded.

    :param content: e-content element of the h-entry.
    :param mf2_entry: Parsed microformats representation of the h-entry.
    :param base_url: Base URL against which relative targets are resolved.
    :param source_url: URL of the source document.
    :param ignored_hostnames: Hostname patterns whose targets should be ignored.
    :return: Iterator over outgoing Webmention target URLs.
    """
    hrefs = (
        href
        for element in content.find_all("a", href=True)
        if isinstance(href := element.get("href"), str)
    )

    reactions = chain.from_iterable(
        property_urls(mf2_entry, property_name) for property_name in REACTION_PROPERTIES
    )

    for value in chain(hrefs, reactions):
        target = normalize_target(value, base_url=base_url, source_url=source_url)
        if target is not None:
            hostname = urlsplit(target).hostname or ""
            if not any(fnmatchcase(hostname, pattern) for pattern in ignored_hostnames):
                yield target


def scan_source_file(
    path: Path,
    *,
    root: Path,
    base_url: str | None,
    ignored_hostnames: tuple[str, ...] = (),
) -> ScannedSource:
    """
    Scan one generated source document for outgoing Webmentions.

    :param path: Path of the generated HTML document.
    :param root: Root directory containing generated source documents.
    :param base_url: Configured source base URL, if available.
    :param ignored_hostnames: Hostname patterns whose targets should be ignored.
    :return: Scanned source metadata and discovered targets.
    :raises SourceScanError: If the document cannot be safely interpreted as a
        Webmention source.
    :raises OSError: If the source document cannot be read.
    """
    document = BeautifulSoup(path.read_bytes(), "html5lib")

    relative_path = path.relative_to(root)

    source_url = canonical_url(document, relative_path=relative_path, base_url=base_url)

    if source_url is None:
        if base_url is None:
            raise SourceScanError(
                "Document has no canonical URL and "
                "WEBMENTIONS_SSG_SOURCE_BASE_URL is not configured"
            )

        source_url = source_url_for_path(relative_path, base_url)

    entry_element, mf2_entry = primary_entry(document, source_url)

    content = content_element(entry_element)

    return ScannedSource(
        path=relative_path.as_posix(),
        url=source_url,
        content_hash=xxhash.xxh3_128_hexdigest(str(entry_element).encode("utf-8")),
        targets=frozenset(
            iter_targets(
                content,
                mf2_entry,
                base_url=source_url,
                source_url=source_url,
                ignored_hostnames=ignored_hostnames,
            )
        ),
    )


def create_source(scanned: ScannedSource, scan_time: datetime) -> Source:
    """
    Create a source model from a newly discovered document.

    A sent Webmention is created for each discovered target at the initial source
    revision.

    :param scanned: Scanned source data.
    :param scan_time: Time at which the source was discovered.
    :return: Newly created source model.
    """
    source = Source(
        path=scanned.path,
        url=scanned.url,
        content_hash=scanned.content_hash,
        revision=1,
        last_seen_at=scan_time,
        revised_at=scan_time,
        deleted_at=None,
    )

    for target in scanned.targets:
        source.sent_webmentions.append(
            SentWebmention(target=target, active=True, desired_revision=1)
        )

    return source


def update_source(
    source: Source, scanned: ScannedSource, scan_time: datetime
) -> Source:
    """
    Update an existing source from newly scanned data.

    A changed or previously deleted source receives a new revision. Existing sent
    Webmentions are updated to reflect the current targets, and newly discovered
    targets are added.

    :param source: Existing source model to update.
    :param scanned: Newly scanned source data.
    :param scan_time: Time at which the source was scanned.
    :return: Updated source model.
    :raises SourceScanError: If the public URL of the source has changed.
    """
    if source.url != scanned.url:
        raise SourceScanError(
            f"Source path {source.path!r} changed public URL "
            f"from {source.url!r} to {scanned.url!r}"
        )

    source.last_seen_at = scan_time

    if source.deleted_at is None and source.content_hash == scanned.content_hash:
        return source

    source.revision += 1
    source.content_hash = scanned.content_hash
    source.revised_at = scan_time
    source.deleted_at = None

    existing = {webmention.target: webmention for webmention in source.sent_webmentions}

    for webmention in existing.values():
        webmention.active = webmention.target in scanned.targets
        webmention.desired_revision = source.revision

        if not webmention.active and webmention.sent_revision is None:
            webmention.processed_revision = source.revision

    for target in scanned.targets:
        if target not in existing:
            source.sent_webmentions.append(
                SentWebmention(
                    target=target, active=True, desired_revision=source.revision
                )
            )

    return source


@huey.lock_task("scan-webmention-sources")
def scan_sources() -> None:
    """
    Scan generated source documents and queue pending Webmentions.

    New and changed sources are persisted, missing sources are marked as deleted,
    and Webmentions requiring processing are queued for sending.

    :raises RuntimeError: If the configured source directory or base URL is
        invalid.
    """
    directory = current_app.config.get("WEBMENTIONS_SSG_SOURCE_DIRECTORY")

    if directory is None:
        return

    root = Path(directory).expanduser().resolve(strict=True)

    if not root.is_dir():
        raise RuntimeError(f"Source directory is not a directory: {root}")

    base_url = current_app.config.get("WEBMENTIONS_SSG_SOURCE_BASE_URL")

    if base_url is not None:
        if not is_http_url(base_url):
            raise RuntimeError(
                "WEBMENTIONS_SSG_SOURCE_BASE_URL must be an absolute HTTP or HTTPS URL"
            )
        base_url = f"{base_url.rstrip('/')}/"

    ignored_hostnames = tuple(
        pattern.lower().rstrip(".")
        for pattern in current_app.config.get("WEBMENTIONS_SSG_IGNORED_HOSTNAMES", ())
    )

    sources_by_path = {
        source.path: source
        for source in db.session.scalars(
            sa.select(Source).options(selectinload(Source.sent_webmentions))
        )
    }

    seen_paths: set[str] = set()
    scanned_count = 0
    scan_time = datetime.now(UTC)

    for index_path in root.rglob("index.html"):
        relative_path = index_path.relative_to(root)

        if relative_path.parent == Path("."):
            continue

        seen_paths.add(relative_path.as_posix())

        try:
            scanned = scan_source_file(
                index_path,
                root=root,
                base_url=base_url,
                ignored_hostnames=ignored_hostnames,
            )

            if (source := sources_by_path.get(scanned.path)) is None:
                source = create_source(scanned, scan_time)
                db.session.add(source)
            else:
                source = update_source(source, scanned, scan_time)

            sources_by_path[source.path] = source
            scanned_count += 1

        except (OSError, SourceScanError) as exc:
            current_app.logger.warning("Could not scan source %s: %s", index_path, exc)

    for path, source in sources_by_path.items():
        if source.deleted_at is not None or path in seen_paths:
            continue

        source.revision += 1
        source.revised_at = scan_time
        source.deleted_at = scan_time

        for webmention in source.sent_webmentions:
            webmention.active = False
            webmention.desired_revision = source.revision

            if webmention.sent_revision is None:
                webmention.processed_revision = source.revision

    # Persist the desired state before queueing any work.
    db.session.commit()

    pending_count = 0

    for identifier in db.session.scalars(
        sa.select(SentWebmention.uuid)
        .where(
            sa.or_(
                SentWebmention.processed_revision.is_(None),
                SentWebmention.processed_revision < SentWebmention.desired_revision,
            )
        )
        .order_by(SentWebmention.uuid)
    ):
        send_webmention(identifier)
        pending_count += 1

    current_app.logger.info(
        "Webmention source scan complete: %d sources scanned, %d pending sends",
        scanned_count,
        pending_count,
    )


@huey.task()
def manual_scan_sources() -> None:
    """
    Run a source scan as a Huey task.
    """
    return scan_sources()