chemenu/tools/chemenu/commands/_util.py

"""Shared helpers for wikitool subcommands."""
from __future__ import annotations

import re
from datetime import date
from pathlib import Path
from typing import Any, Dict, Optional

import typer
from rich.console import Console

console = Console()

# A third outcome alongside success (0) and validation error (1): the command
# is refusing until a *human* has seen its output and cleared it. It exists as
# its own code so the caller - an agent, a harness hook, a CI job, a trajectory
# scorer - can tell "stop and ask the user" apart from "your input was wrong,
# fix it and retry". Nothing about *why* clearance is needed lives in the agent
# instructions: the command's own output carries the reason, the evidence, and
# the exact re-run line.
EXIT_NEEDS_CLEARANCE = 42


def success(msg: str) -> None:
    console.print(f"[green]OK[/green] {msg}")


# Set by `fail()`, read once per process by the CLI entry point. Exit 1 raised
# through `fail()` means the command declined and did the thing it was asked
# for: the argument was rejected, or a read-only check reported findings.
# Neither is an iteration step on the wiki, so the Iteration Budget Gate gives
# the slot back (see run_budget.refund). A command that has already done its
# work and then reports a non-zero result - `lint --fail-on-error` writes its
# report first - raises `typer.Exit(1)` directly and stays counted.
_declined = False


def declined() -> bool:
    """Whether this process left through `fail()`."""
    return _declined


def fail(msg: str) -> None:
    global _declined
    _declined = True
    console.print(f"[bold red]ERROR[/bold red] {msg}")
    raise typer.Exit(code=1)


def needs_clearance(msg: str) -> None:
    """Refuse with EXIT_NEEDS_CLEARANCE. The message is written to be shown to
    a human verbatim - it is the whole user-facing artifact of this gate."""
    console.print(f"[bold yellow]NEEDS USER CLEARANCE[/bold yellow] {msg}")
    raise typer.Exit(code=EXIT_NEEDS_CLEARANCE)


# A comma preceded by a backslash is a literal comma, not a separator.
_UNESCAPED_COMMA = re.compile(r"(?<!\\),")


def parse_list(value: str | None) -> list[str]:
    """Split a comma-separated CLI value into list elements.

    `\\,` is an escaped literal comma: it survives the split and lands inside
    the element. Without it a list format simply cannot express an element
    that contains a comma - and shell quoting is no help, because the quotes
    are gone long before this sees the string. Paths and page titles carry
    commas often enough for that to matter: it once cost a `raw/` file its
    original name, which `raw/CONTRACT.md` forbids.
    """
    if not value:
        return []
    parts = (part.replace("\\,", ",").strip() for part in _UNESCAPED_COMMA.split(value))
    return [part for part in parts if part]


def coerce_set_value(raw_value: str, field_schema: Optional[Dict[str, Any]]) -> Any:
    """Coerce a `--set field=value` string to the type its schema declares.

    Arrays are comma-split (see `parse_list` for the escape), numbers are
    parsed as float/int, booleans as true/false; everything else stays a
    string. Unknown fields (no schema entry) pass through as strings and are
    then caught by schema validation's `additionalProperties: false`.
    """
    declared = (field_schema or {}).get("type")
    if declared == "array":
        return parse_list(raw_value)
    if declared == "number":
        try:
            return float(raw_value)
        except ValueError:
            return raw_value
    if declared == "integer":
        try:
            return int(raw_value)
        except ValueError:
            return raw_value
    if declared == "boolean":
        if raw_value.lower() in ("true", "false"):
            return raw_value.lower() == "true"
    return raw_value


def parse_set_fields(
    set_fields: Optional[list[str]], schema: Optional[Dict[str, Any]], flag: str = "--set"
) -> Dict[str, Any]:
    """Parse repeated `<flag> field=value` pairs into a frontmatter dict,
    coercing each value by the field's declared schema type.

    Repeating the flag for an *array* field appends rather than replaces, so
    `--set raw_files=a --set raw_files=b` yields both. That is the form that
    needs no separator at all, and therefore the one to reach for when an
    element contains a comma; `\\,` inside a single value does the same job
    for a one-liner. Repeating a scalar field still means "last one wins" -
    there is nothing to append to.

    Note that this is per *invocation*. What a parsed value then means for a
    page already on disk is the caller's decision: `new` writes it as the
    page's initial value, while `touch` replaces, extends or subtracts
    depending on which flag it came from.
    """
    explicit: Dict[str, Any] = {}
    properties = (schema or {}).get("properties", {})
    for pair in set_fields or []:
        if "=" not in pair:
            fail(f"{flag} expects field=value, got: {pair}")
        field_name, raw_value = pair.split("=", 1)
        field_name = field_name.strip()
        if not field_name:
            fail(f"{flag} expects field=value, got: {pair}")
        value = coerce_set_value(raw_value, properties.get(field_name))
        previous = explicit.get(field_name)
        if isinstance(value, list) and isinstance(previous, list):
            previous.extend(value)
        else:
            explicit[field_name] = value
    return explicit


def check_raw_files_exist(raw_files: Any) -> None:
    """Verify every `raw_files:` entry is an existing file.

    This is the one validation that genuinely cannot live in the schema:
    it is filesystem I/O, not a data-shape constraint. Cardinality
    (`minItems: 1`) is already enforced by the schema itself, so only
    existence and file-vs-directory are checked here.

    Shared by `new` and `touch` - both write the field, and a page pointing at
    a raw file that is not there is the same defect whichever wrote it.
    """
    from chemenu import config

    for raw_path in raw_files or []:
        full_path = config.ROOT / raw_path
        if not full_path.exists():
            fail(
                f"raw_files path does not exist: {raw_path}\n"
                "  This is one element after splitting the value on commas. If the real "
                "filename contains a comma, escape it as `\\,` or pass one `--set "
                "raw_files=<path>` per file - never rename the raw file to fit the flag."
            )
        if full_path.is_dir():
            fail(f"raw_files must be a file, not a directory: {raw_path}")


def today_iso() -> str:
    return date.today().isoformat()


def rel_path(path: Path) -> str:
    """Format a path relative to the repo root for display, falling back to
    the raw path if it lies outside the root (e.g. in tests)."""
    from chemenu import config

    try:
        return str(Path(path).relative_to(config.ROOT))
    except ValueError:
        return str(path)


def check_collision(name: str) -> None:
    """Fail if any page under kb/ already has `name` as its filename stem.

    The stem *is* the page title and wikilinks resolve by title alone, so two
    files sharing a stem in different directories are indistinguishable to
    every link in the wiki. Shared by `new` and `rename`.
    """
    from chemenu import config

    for path in config.KB_DIR.rglob("*.md"):
        if path.stem == name:
            fail(f"A page titled '{name}' already exists at {rel_path(path)}")