portfolio-optimiser-claude/src/portfolio_optimiser_claude/ingest.py

"""Deterministic ingest layer: manifest → file/CSV connector → OKF bundle (ingest-spec).

An addition IN FRONT of the loop (ingest-spec §1): a deterministic step that couples the
framework to a real data source and materializes the extract as an OKF knowledge bundle,
which the existing 8-step loop then consumes UNCHANGED. Zero model calls; no network.

D7 scope (I3–I5, mirror of MAF I2/I4): the ``file`` source type (a local CSV catalogue) and
the ``sql`` source type (a local SQL database). The ``http`` source type is a later, OPTIONAL
extension point (§1) — a manifest naming it is rejected fail-fast at validation, never
silently accepted.

Contract discipline mirrors ``contracts.py``: the manifest is schema-validated fail-fast
BEFORE any source call (§4), and queries are declarative configuration, never evaluated as
code. The provenance frontmatter layer (§7) is ITS OWN contract — separate from the method
spec's §9 proposal provenance, never mixed.
"""

from __future__ import annotations

import csv
import hashlib
import json
import logging
import os
import re
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Annotated, Literal

from pydantic import BaseModel, Field, field_validator

_logger = logging.getLogger(__name__)

_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
_VERDICT_TYPE = "verdict"
_INGEST_PREFIX = "ingest-"
_STAMP_HASH_LEN = 16
# An index cross-link owned by ingest: ``- [label](ingest-{id}.md)`` (§6). Only these
# are managed on re-materialization; curated and promoted links are preserved verbatim.
_INGEST_LINK_RE = re.compile(r"\]\((ingest-[a-z0-9][a-z0-9-]*\.md)\)")
_CROSSLINK_RE = re.compile(r"\]\(([^)]+\.md)\)")

# Frontmatter keys, in the exact §5 order.
_FRONTMATTER_ORDER = (
    "type",
    "title",
    "source_system",
    "source_query",
    "ingested_at",
    "ingest_manifest",
    "generated",
)


class FileSource(BaseModel):
    """A local file catalogue (``source.type == "file"``, ingest-spec §4)."""

    type: Literal["file"]
    id: str
    root: str

    @field_validator("id")
    @classmethod
    def _id_grammar(cls, value: str) -> str:
        if not _ID_RE.match(value):
            raise ValueError(f"source.id {value!r} must match [a-z0-9][a-z0-9-]*")
        return value


class SqlSource(BaseModel):
    """A local SQL database (``source.type == "sql"``, ingest-spec §4).

    ``connection_ref`` is the NAME of a runtime-resolved environment variable whose value is
    the database path (or connection string) — the secret/location never lives in the manifest
    (§4, §8), so the manifest stays versionable and shareable without credentials.
    """

    type: Literal["sql"]
    id: str
    connection_ref: str

    @field_validator("id")
    @classmethod
    def _id_grammar(cls, value: str) -> str:
        if not _ID_RE.match(value):
            raise ValueError(f"source.id {value!r} must match [a-z0-9][a-z0-9-]*")
        return value


class Extraction(BaseModel):
    """One extraction description (ingest-spec §4). ``max_rows`` is a required cap (§8)."""

    id: str
    title: str
    query: str
    okf_type: str
    max_rows: int = Field(gt=0)

    @field_validator("id")
    @classmethod
    def _id_grammar(cls, value: str) -> str:
        if not _ID_RE.match(value):
            raise ValueError(f"extraction.id {value!r} must match [a-z0-9][a-z0-9-]*")
        return value

    @field_validator("title")
    @classmethod
    def _title_single_line(cls, value: str) -> str:
        if "\n" in value or "\r" in value:
            raise ValueError("extraction.title must be single-line")
        return value

    @field_validator("okf_type")
    @classmethod
    def _okf_type_not_verdict(cls, value: str) -> str:
        # §3 verdict-layer reservation: the promotion gate is the ONLY path into the
        # verdict layer — an ingest mapping to type: verdict would inject machine-made
        # "approved" verdicts around the gate. Case-insensitive, fail-fast.
        if value.lower() == _VERDICT_TYPE:
            raise ValueError("okf_type MUST NOT be 'verdict' (verdict layer is reserved, §3)")
        return value


class ManifestContract(BaseModel):
    """The ingest manifest (§4) — schema-validated fail-fast before any source call."""

    manifest_version: Literal[1]
    source: Annotated[FileSource | SqlSource, Field(discriminator="type")]
    bundle_summary: str
    extractions: list[Extraction] = Field(min_length=1)

    @field_validator("extractions")
    @classmethod
    def _unique_extraction_ids(cls, value: list[Extraction]) -> list[Extraction]:
        ids = [e.id for e in value]
        if len(ids) != len(set(ids)):
            raise ValueError("extraction ids must be unique within the manifest")
        return value


@dataclass(frozen=True)
class LoadedManifest:
    """A validated manifest plus its provenance stamp and on-disk location."""

    contract: ManifestContract
    stamp: str  # ``{manifest stem}@{sha256(raw bytes)[:16]}`` (§5)
    path: Path


def load_manifest(manifest_path: Path) -> LoadedManifest:
    """Validate the manifest fail-fast (§4) and compute its provenance stamp (§5).

    Raises ``pydantic.ValidationError`` on a malformed manifest — never starts a run on a
    bad contract. The stamp is over the manifest's RAW bytes, so any edit re-stamps.
    """
    raw = manifest_path.read_bytes()
    contract = ManifestContract(**json.loads(raw))
    digest = hashlib.sha256(raw).hexdigest()[:_STAMP_HASH_LEN]
    stamp = f"{manifest_path.stem}@{digest}"
    return LoadedManifest(contract=contract, stamp=stamp, path=manifest_path)


def _escape_cell(value: str) -> str:
    """Render a cell value: text verbatim with §5 escaping.

    Backslash FIRST (so the pipe escape it introduces is not re-escaped), then pipe,
    then any newline collapsed to a single space. For the ``file`` source every cell is
    a ``str`` — the integer/float/NULL typing rules (§5) belong to the ``sql`` source.
    """
    escaped = value.replace("\\", "\\\\").replace("|", "\\|")
    escaped = escaped.replace("\r\n", " ").replace("\r", " ").replace("\n", " ")
    return escaped


def _render_table(rows: list[list[str]]) -> str:
    """Header row, separator, data rows — GitHub-flavoured markdown table (§5)."""
    header, *data = rows
    lines = [
        "| " + " | ".join(_escape_cell(c) for c in header) + " |",
        "| " + " | ".join("---" for _ in header) + " |",
    ]
    for row in data:
        lines.append("| " + " | ".join(_escape_cell(c) for c in row) + " |")
    return "\n".join(lines)


def _read_csv(csv_path: Path, max_rows: int) -> list[list[str]]:
    """Read a CSV (first row = header). Enforce ``max_rows`` fail-fast on DATA rows (§8)."""
    with csv_path.open(encoding="utf-8", newline="") as handle:
        rows = list(csv.reader(handle))
    data_rows = len(rows) - 1 if rows else 0
    if data_rows > max_rows:
        raise ValueError(
            f"extraction exceeds max_rows: {data_rows} > {max_rows} for {csv_path.name}"
        )
    return rows


def _render_sql_cell(value: object) -> str:
    """Type a SQL cell to its §5 string form (BEFORE §5 escaping via ``_escape_cell``).

    ``None`` (SQL NULL) → the empty string; ``int`` → plain decimal; ``float`` → its shortest
    round-trip decimal form (Python ``repr``); ``str`` → verbatim. Any other value type (a
    BLOB, say) MUST fail — never a silent coercion (§5).
    """
    if value is None:
        return ""
    if isinstance(value, int):
        return str(value)
    if isinstance(value, float):
        return repr(value)
    if isinstance(value, str):
        return value
    raise ValueError(f"SQL cell of type {type(value).__name__} is not a supported value type (§5)")


def _resolve_connection_ref(connection_ref: str) -> str:
    """Resolve the ``connection_ref`` env var to the database location at run time (§4, §8).

    Locations/credentials never live in the manifest — the reference is resolved from the
    environment. An unset reference fails fast (never a silent empty connection).
    """
    dsn = os.environ.get(connection_ref)
    if not dsn:
        raise ValueError(f"connection_ref {connection_ref!r} is not set in the environment")
    return dsn


def _read_sql(dsn: str, query: str, max_rows: int) -> list[list[str]]:
    """Run one read-only SELECT against a local sqlite database (ingest-spec §4, §5, §8).

    The connection is opened read-only (§4: the connector SHOULD enforce read-only access), and
    sqlite executes exactly ONE statement per ``execute`` call — so the single-statement rule is
    enforced by the driver. ``max_rows`` is enforced fail-fast on the returned rows (§8).
    Returns the column-name header row followed by typed-then-stringified data rows, in source
    order (§5); a stable order is the manifest query's responsibility via ORDER BY (§4).
    """
    uri = Path(dsn).resolve().as_uri() + "?mode=ro"
    connection = sqlite3.connect(uri, uri=True)
    try:
        cursor = connection.execute(query)
        header = [description[0] for description in cursor.description]
        data = cursor.fetchall()
    finally:
        connection.close()
    if len(data) > max_rows:
        raise ValueError(f"extraction exceeds max_rows: {len(data)} > {max_rows}")
    rows: list[list[str]] = [header]
    for record in data:
        rows.append([_render_sql_cell(value) for value in record])
    return rows


def _read_extraction(
    source: FileSource | SqlSource, manifest_dir: Path, extraction: Extraction
) -> list[list[str]]:
    """Dispatch to the connector for ``source.type`` — header + data rows in source order (§5).

    ``file``: resolve the query path within ``root`` (boundary-checked fail-closed, §4) and read
    the CSV. ``sql``: resolve ``connection_ref`` from the environment (§4, §8) and run the
    read-only SELECT. Both enforce ``max_rows`` fail-fast (§8).
    """
    if source.type == "file":
        csv_path = _resolve_within(manifest_dir / source.root, extraction.query)
        return _read_csv(csv_path, extraction.max_rows)
    dsn = _resolve_connection_ref(source.connection_ref)
    return _read_sql(dsn, extraction.query, extraction.max_rows)


def _resolve_within(root: Path, relative: str) -> Path:
    """Resolve ``relative`` under ``root``, boundary-checked fail-closed (the OKF path rule)."""
    resolved = (root / relative).resolve()
    if not resolved.is_relative_to(root.resolve()):
        raise ValueError(f"extraction path {relative!r} escapes the source root")
    return resolved


def _frontmatter_block(values: dict[str, str]) -> str:
    return "---\n" + "".join(f"{k}: {values[k]}\n" for k in _FRONTMATTER_ORDER) + "---"


def _concept_file(
    extraction: Extraction, source_id: str, ingested_at: str, stamp: str, body: str
) -> str:
    frontmatter = _frontmatter_block(
        {
            "type": extraction.okf_type,
            "title": extraction.title,
            "source_system": source_id,
            "source_query": _escape_ws(extraction.query),
            "ingested_at": ingested_at,
            "ingest_manifest": stamp,
            "generated": "true",
        }
    )
    return f"{frontmatter}\n\n{body}\n"


def _escape_ws(value: str) -> str:
    """Collapse whitespace runs (incl. newlines) to single spaces (§5) — single-line values."""
    return " ".join(value.split())


def _is_ingest_stamped(md_path: Path) -> bool:
    """True iff the file carries the ingest stamp: ``generated: true`` + an ``ingest_manifest``."""
    frontmatter: dict[str, str] = {}
    lines = md_path.read_text(encoding="utf-8").splitlines()
    if not lines or lines[0].strip() != "---":
        return False
    for line in lines[1:]:
        if line.strip() == "---":
            break
        key, sep, val = line.partition(":")
        if sep:
            frontmatter[key.strip()] = val.strip()
    return frontmatter.get("generated") == "true" and "ingest_manifest" in frontmatter


def _update_index(bundle_dir: Path, bundle_summary: str, extractions: list[Extraction]) -> None:
    """Create or update ``index.md`` (§6).

    A fresh index carries ``bundle_summary`` as its body; an existing index keeps every
    line it does not manage byte for byte. Ingest cross-links (targets ``ingest-*.md``)
    are the ONLY managed lines: stale ones are dropped and the current set re-appended in
    extraction order, idempotently by target — curated and promoted links are preserved
    (so a promoted verdict's index link survives re-ingest, load-bearing §11).
    """
    index_path = bundle_dir / "index.md"
    if index_path.is_file():
        kept = [
            line
            for line in index_path.read_text(encoding="utf-8").splitlines()
            if not _INGEST_LINK_RE.search(line)
        ]
    else:
        kept = [bundle_summary]
    present = {t for line in kept for t in _CROSSLINK_RE.findall(line)}
    lines = list(kept)
    for extraction in extractions:
        target = f"{_INGEST_PREFIX}{extraction.id}.md"
        if target not in present:
            lines.append(f"- [{extraction.title}]({target})")
            present.add(target)
    index_path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def materialize(manifest_path: Path, bundle_dir: Path, ingested_at: str) -> list[Path]:
    """Materialize the manifest's extractions into ``bundle_dir`` (deterministic, §5).

    ``ingested_at`` is an EXPLICIT required argument (no wall-clock default, §5) — the ISO
    timestamp stamped verbatim into every generated file, which is what makes extractions
    bit-deterministic. Returns the generated concept-file paths in extraction order.

    Replacement semantics (§3, §5): first every ingest-stamped file is removed, then the
    new set is written (a name colliding with a NON-stamped file fails — curated content is
    never overwritten), then the index is updated. Curated and promoted files always survive.
    """
    loaded = load_manifest(manifest_path)
    source = loaded.contract.source

    bundle_dir.mkdir(parents=True, exist_ok=True)

    # 1. Remove exactly the files carrying the ingest stamp (§5) — never curated/promoted.
    for md_path in sorted(bundle_dir.glob("*.md")):
        if md_path.name != "index.md" and _is_ingest_stamped(md_path):
            md_path.unlink()

    # 2. Write the new set.
    generated: list[Path] = []
    for extraction in loaded.contract.extractions:
        rows = _read_extraction(source, manifest_path.parent, extraction)
        _logger.info(
            "ingest source=%s extraction=%s rows=%d",
            source.id,
            extraction.id,
            max(len(rows) - 1, 0),
        )
        target = bundle_dir / f"{_INGEST_PREFIX}{extraction.id}.md"
        if target.exists():
            # Post-removal, an existing target is a NON-stamped collision (§3): never overwrite.
            raise ValueError(f"generated name {target.name} collides with a non-ingest file")
        target.write_text(
            _concept_file(extraction, source.id, ingested_at, loaded.stamp, _render_table(rows)),
            encoding="utf-8",
        )
        generated.append(target)

    # 3. Update the index (§6).
    _update_index(bundle_dir, loaded.contract.bundle_summary, loaded.contract.extractions)
    return generated