ambrosiana-document-intake/ambrosiana_intake.py

#!/usr/bin/env python3
# Created by John A. Hoeven with the ethical assistance of Claude AI
# ---------------------------------------------------------------------------
# ambrosiana_intake.py
# ~/projects/ambrosiana-document-intake/ambrosiana_intake.py
# Version: v0.1.0 | Status: BETA
# ---------------------------------------------------------------------------
# Purpose:  Textual TUI that collects document-intake metadata and writes
#           a structured markdown ticket for Claude Code to act on.
# Target:   Any Ambrosiana device (per-device deployment, no cross-device
#           routing)
# Entry:    python3 ambrosiana_intake.py
# Depends:  textual (see requirements.txt)
# Config:   ~/.config/ambrosiana-intake/config.toml (tickets_root), created
#           with a default on first run if not present
# ---------------------------------------------------------------------------
"""
Ambrosiana Intake — CE OS document intake ticket generator

Collects the finalized intake variable set for an incoming document and
writes a structured markdown ticket (YAML front matter) that is handed
to Claude Code as the instruction to run the actual scrape/vet/commit/
index pipeline. This app never touches the pipeline, any agent, or any
model — it is a pure human-input-to-structured-document step.

Built standing on the shoulders of billions of dwarves.
Created by John A. Hoeven with the ethical assistance of Claude AI.
License: The Unlicense — https://unlicense.org
"""

from __future__ import annotations

import re
import socket
import tomllib
from datetime import datetime
from pathlib import Path

from textual.app import App, ComposeResult
from textual.containers import Horizontal, VerticalScroll
from textual.widgets import (
    Button,
    Footer,
    Header,
    Input,
    Label,
    Select,
    Static,
    TextArea,
)

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

# XDG for now by explicit choice — planned move to a ~/.local/etc
# subdirectory later. CONFIG_DIR is the only line that needs to change
# when that happens.
CONFIG_DIR = Path.home() / ".config" / "ambrosiana-intake"
CONFIG_FILE = CONFIG_DIR / "config.toml"
DEFAULT_TICKETS_ROOT = Path.home() / "documents" / "rag-administration" / "ticket-queue"


def load_tickets_root() -> Path:
    """Read tickets_root from config.toml, writing a default config on first run."""
    if not CONFIG_FILE.exists():
        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        CONFIG_FILE.write_text(
            "# Ambrosiana Intake configuration\n"
            f'tickets_root = "{DEFAULT_TICKETS_ROOT}"\n',
            encoding="utf-8",
        )
        return DEFAULT_TICKETS_ROOT

    with CONFIG_FILE.open("rb") as config_file:
        config = tomllib.load(config_file)
    raw_path = config.get("tickets_root", str(DEFAULT_TICKETS_ROOT))
    return Path(raw_path).expanduser()


TICKETS_ROOT = load_tickets_root()
TICKETS_ROOT.mkdir(parents=True, exist_ok=True)

SOURCE_TYPE_OPTIONS = [
    ("upstream-doc", "upstream-doc"),
    ("ce-authored", "ce-authored"),
    ("ce-experience", "ce-experience"),
]

SOURCE_TIER_OPTIONS = [
    ("primary", "primary"),
    ("secondary", "secondary"),
    ("forum/unmoderated", "forum-unmoderated"),
]

PRIORITY_OPTIONS = [
    ("Immediate need", "immediate"),
    ("Add to queue", "queue"),
]

REQUIRED_FIELDS = ("source_location", "source_type", "source_tier", "licence", "priority")


def slugify(text: str, fallback: str) -> str:
    text = text.strip().lower()
    if not text:
        text = fallback
    text = re.sub(r"[^\w\s-]", "", text)
    text = re.sub(r"[\s_-]+", "-", text).strip("-")
    return text[:60] or fallback


# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------


class AmbrosianaIntakeApp(App):
    CSS = """
    Screen {
        background: $surface;
        align: center top;
    }

    #form-scroll {
        width: 90;
        max-width: 100%;
        padding: 1 2;
    }

    #device-banner {
        width: 100%;
        content-align: center middle;
        color: $text-muted;
        padding-bottom: 1;
    }

    .field-label {
        padding-top: 1;
        color: $text;
        text-style: bold;
    }

    .required-marker {
        color: $error;
    }

    .optional-marker {
        color: $text-muted;
        text-style: italic;
    }

    Input, Select, TextArea {
        width: 100%;
    }

    #notes-area {
        height: 5;
        border: round $primary-darken-1;
    }

    #button-row {
        width: 100%;
        height: auto;
        padding-top: 2;
        align-horizontal: center;
    }

    #button-row Button {
        margin: 0 1;
    }

    #status-line {
        width: 100%;
        padding-top: 1;
        text-align: center;
        color: $success;
    }

    #status-line.-error {
        color: $error;
    }
    """

    BINDINGS = [
        ("ctrl+s", "submit", "Save ticket"),
        ("ctrl+n", "reset_form", "Clear form"),
        ("ctrl+q", "quit", "Quit"),
    ]

    def __init__(self) -> None:
        super().__init__()
        self.hostname = socket.gethostname()

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        with VerticalScroll(id="form-scroll"):
            yield Static(
                f"Ambrosiana Intake — running on [b]{self.hostname}[/b]",
                id="device-banner",
            )

            yield Label("Source location  [required]", classes="field-label")
            yield Input(placeholder="filepath or URL", id="source_location")

            yield Label("Source type  [required]", classes="field-label")
            yield Select(SOURCE_TYPE_OPTIONS, id="source_type", prompt="Choose source type")

            yield Label("Source tier  [required]", classes="field-label")
            yield Select(SOURCE_TIER_OPTIONS, id="source_tier", prompt="Choose source tier")

            yield Label("Licence  [required]", classes="field-label")
            yield Input(placeholder="e.g. The Unlicense, CC-BY-4.0, proprietary...", id="licence")

            yield Label("Priority  [required]", classes="field-label")
            yield Select(PRIORITY_OPTIONS, id="priority", prompt="Choose priority")

            yield Label("Title / description  (optional)", classes="field-label optional-marker")
            yield Input(placeholder="short human-readable title", id="title")

            yield Label("Notes / context  (optional)", classes="field-label optional-marker")
            yield TextArea(id="notes-area")

            with Horizontal(id="button-row"):
                yield Button("Save ticket", id="submit", variant="success")
                yield Button("Clear", id="reset", variant="warning")
                yield Button("Quit", id="quit", variant="error")

            yield Static("", id="status-line")
        yield Footer()

    # -- helpers ------------------------------------------------------

    def _collect(self) -> dict[str, str]:
        return {
            "source_location": self.query_one("#source_location", Input).value.strip(),
            "source_type": self.query_one("#source_type", Select).value,
            "source_tier": self.query_one("#source_tier", Select).value,
            "licence": self.query_one("#licence", Input).value.strip(),
            "priority": self.query_one("#priority", Select).value,
            "title": self.query_one("#title", Input).value.strip(),
            "notes": self.query_one("#notes-area", TextArea).text.strip(),
        }

    def _set_status(self, message: str, error: bool = False) -> None:
        status = self.query_one("#status-line", Static)
        status.update(message)
        status.set_class(error, "-error")
        if message:
            self.notify(message, severity="error" if error else "information")

    def _clear_form(self) -> None:
        self.query_one("#source_location", Input).value = ""
        self.query_one("#source_type", Select).clear()
        self.query_one("#source_tier", Select).clear()
        self.query_one("#licence", Input).value = ""
        self.query_one("#priority", Select).clear()
        self.query_one("#title", Input).value = ""
        self.query_one("#notes-area", TextArea).text = ""
        self.query_one("#source_location", Input).focus()

    def _write_ticket(self, data: dict[str, str]) -> Path | None:
        missing = [f for f in REQUIRED_FIELDS if not data.get(f) or data.get(f) is Select.BLANK]
        if missing:
            self._set_status(f"Missing required field(s): {', '.join(missing)}", error=True)
            return None

        now = datetime.now()

        fallback_slug = Path(data["source_location"]).stem or "untitled"
        slug = slugify(data["title"] or fallback_slug, fallback_slug)
        priority_tag = "immediate_" if data["priority"] == "immediate" else ""
        filename = f"{priority_tag}{now.strftime('%Y%m%d-%H%M%S')}_{slug}.md"
        ticket_path = TICKETS_ROOT / filename

        front_matter_lines = [
            "---",
            "ticket_type: ambrosiana-intake",
            f"created: {now.isoformat(timespec='seconds')}",
            f"device: {self.hostname}",
            f"source_location: \"{data['source_location']}\"",
            f"source_type: {data['source_type']}",
            f"source_tier: {data['source_tier']}",
            f"licence: \"{data['licence']}\"",
            f"priority: {data['priority']}",
            f"title: \"{data['title']}\"" if data["title"] else "title: \"\"",
            "---",
            "",
        ]

        body_lines = []
        if data["title"]:
            body_lines.append(f"# {data['title']}")
            body_lines.append("")
        body_lines.append(f"**Source:** {data['source_location']}")
        body_lines.append(f"**Source-type:** `{data['source_type']}`")
        body_lines.append(f"**Source-tier:** `{data['source_tier']}`")
        body_lines.append(f"**Licence:** {data['licence']}")
        body_lines.append(f"**Priority:** {data['priority']}")
        body_lines.append("")
        if data["notes"]:
            body_lines.append("## Notes / context")
            body_lines.append("")
            body_lines.append(data["notes"])
            body_lines.append("")
        body_lines.append("---")
        body_lines.append("*Ambrosiana intake ticket — awaiting pipeline run via Claude Code*")

        ticket_path.write_text("\n".join(front_matter_lines + body_lines), encoding="utf-8")
        return ticket_path

    # -- actions --------------------------------------------------------

    def action_submit(self) -> None:
        self._do_submit()

    def action_reset_form(self) -> None:
        self._clear_form()
        self._set_status("")

    def _do_submit(self) -> None:
        data = self._collect()
        path = self._write_ticket(data)
        if path is not None:
            self._set_status(f"Saved: {path}")
            self._clear_form()

    def on_button_pressed(self, event: Button.Pressed) -> None:
        if event.button.id == "submit":
            self._do_submit()
        elif event.button.id == "reset":
            self._clear_form()
            self._set_status("")
        elif event.button.id == "quit":
            self.exit()


if __name__ == "__main__":
    AmbrosianaIntakeApp().run()