pig.py/neopig/arborist_sink.py

# This is free software for the public good of a permacomputer hosted at
# permacomputer.com, an always-on computer by the people, for the people.
# One which is durable, easy to repair, & distributed like tap water
# for machine learning intelligence.
#
# The permacomputer is community-owned infrastructure optimized around
# four values:
#
#   TRUTH      First principles, math & science, open source code freely distributed
#   FREEDOM    Voluntary partnerships, freedom from tyranny & corporate control
#   HARMONY    Minimal waste, self-renewing systems with diverse thriving connections
#   LOVE       Be yourself without hurting others, cooperation through natural law
#
# This software contributes to that vision by archiving the web, preserving digital knowledge before it disappears.
# Code is seeds to sprout on any abandoned technology.

"""Optional arborist provenance sink.

When enabled, every crawled page neopig stores is ALSO committed to an
arborist content-addressed store: a Merkle ``document_root`` plus an
append-only audit chain, sitting beside neopig's md5/FileVault storage.
It adds verifiable dedup + an FTS5 index + tamper-evident provenance; it
replaces nothing.

Default OFF, and a hard no-op when the flag is unset OR arborist is not
installed — neopig behaves identically without it. arborist is AGPL and
neopig is public domain, so the dependency stays strictly opt-in: an
operator who enables it pulls AGPL code by choice.

Enable::

    pip install -e ~/git/arborist          # core only (httpx/zstandard/crypto)
    export NEOPIG_ARBORIST_ENABLED=1
    export NEOPIG_ARBORIST_DB=data/arborist.db   # optional, default shown

neopig never touches arborist's tables — every write goes through
``arborist.embed``, so neopig's no-raw-SQL rule is preserved (there is no
SQL to write).
"""

from __future__ import annotations

import asyncio
import hashlib
import logging
import os
from typing import List, Optional

logger = logging.getLogger(__name__)

try:
    from arborist.embed import Document, Edge, ingest_documents, open_store

    _ARBORIST_AVAILABLE = True
except ImportError:
    _ARBORIST_AVAILABLE = False


def _truthy(value: Optional[str]) -> bool:
    return (value or "").strip().lower() in {"1", "true", "yes", "on"}


class ArboristSink:
    """Opt-in mirror of crawled pages into an arborist store.

    Every public method is a no-op unless the sink is enabled — which
    requires both the ``NEOPIG_ARBORIST_ENABLED`` flag AND arborist being
    importable. arborist's sync SQLite writes run in a worker thread so
    they never block neopig's event loop, serialized by a lock so
    concurrent pages don't contend on the one writer.
    """

    def __init__(self, db_path: Optional[str] = None, enabled: Optional[bool] = None):
        flag = _truthy(os.environ.get("NEOPIG_ARBORIST_ENABLED")) if enabled is None else bool(enabled)
        self.available = _ARBORIST_AVAILABLE
        self.enabled = bool(flag and self.available)
        self.db_path = db_path or os.environ.get("NEOPIG_ARBORIST_DB", "data/arborist.db")
        self._lock = asyncio.Lock()
        if flag and not self.available:
            logger.warning(
                "NEOPIG_ARBORIST_ENABLED is set but arborist is not installed; "
                "provenance sink disabled. `pip install -e ~/git/arborist` to enable."
            )
        elif self.enabled:
            logger.info("arborist provenance sink enabled -> %s", self.db_path)

    def _page_to_document(
        self,
        uri: str,
        title: Optional[str],
        content: str,
        *,
        path: str = "",
        description: str = "",
        keywords: Optional[list] = None,
        crawl_job_id: Optional[int] = None,
        media_uris: Optional[List[str]] = None,
    ):
        """Map a crawled page onto an arborist Document.

        Provenance neopig already tracks (its md5 uri-hash, crawl job,
        meta) rides along in ``extra``; page->media links become edges.
        """
        edges = [Edge(edge_type="embeds_media", dst_uri=m) for m in (media_uris or []) if m]
        return Document(
            uri=uri,
            content=content or "",
            source_type="neopig_html",
            title=title or None,
            edges=edges,
            extra={
                "path": path,
                "description": description,
                "keywords": keywords or [],
                "crawl_job_id": crawl_job_id,
                "md5_uri_hash": hashlib.md5(uri.encode()).hexdigest(),
            },
        )

    async def ingest_page(self, uri: str, title: Optional[str], content: str, **kwargs) -> bool:
        """Commit one crawled page to arborist.

        No-op returning ``False`` when disabled. Runs the sync arborist
        write off the event loop. Failures are swallowed and logged — a
        provenance mirror must NEVER break a crawl.
        """
        if not self.enabled:
            return False
        document = self._page_to_document(uri, title, content, **kwargs)
        async with self._lock:
            try:
                await asyncio.to_thread(self._ingest_sync, [document])
                return True
            except Exception as exc:  # never let the mirror break a crawl
                logger.warning("arborist ingest failed for %s: %s", uri, exc)
                return False

    def _ingest_sync(self, documents) -> None:
        """Open a fresh connection in this worker thread, ingest, close.

        Per-page open/close keeps each SQLite connection bound to one
        thread (safe) and leans on arborist's idempotent ingest. Batching
        is a later optimization; correctness first.
        """
        conn = open_store(self.db_path)
        try:
            ingest_documents(conn, documents)
        finally:
            conn.close()