ceos/ambrosiana-document-intake
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
ambrosiana-document-intake/ambrosiana_intake.py
John A. Hoeven 16c592c971
 Add index-page link discovery and simplify licence/priority fields 
Adds link_discovery.py: sitemap-first, same-domain-crawl-fallback
  discovery of sub-pages from a single index URL, so a source with many
  sub-pages no longer needs each page entered by hand. Never pulls page
  titles or body text into a ticket field, only URLs, to avoid a
  prompt-injection path into tickets Claude Code later reads. Adds a
  review-and-approve screen so nothing is queued without a human
  checking the actual discovered list first.

  Replaces the free-text licence field with a required typed permission
  question, since licence itself is better researched from the source
  during actual processing rather than typed in at ticket time. Replaces
  the priority dropdown with a single urgent checkbox.

  Fixes a few real bugs found along the way: colliding ticket filenames
  when writing many tickets in one batch, a couple of layout issues
  where long labels or URLs were getting clipped instead of wrapping or
  scrolling, and dead links slipping through discovery from malformed
  relative link resolution
2026-08-07 22:23:28 +02:00

501 lines
18 KiB
Python
Raw Blame History

#!/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 import work
from textual.app import App, ComposeResult
from textual.containers import Horizontal, VerticalScroll
from textual.screen import Screen
from textual.widgets import (
Button,
Checkbox,
Footer,
Header,
Input,
Label,
Select,
Static,
TextArea,
)
from link_discovery import DiscoveredPage, discover_links
# ---------------------------------------------------------------------------
# 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"),
]
PERMISSION_QUESTION = "Do you have permission to access this document and add it to the RAG?"
REQUIRED_FIELDS = ("source_location", "source_type", "source_tier", "permission")
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
# ---------------------------------------------------------------------------
# Discovery review screen
# ---------------------------------------------------------------------------
class LinkReviewScreen(Screen[list[DiscoveredPage]]):
"""
Shows discovered sub-pages for approval before any ticket is written.
Titles shown here are derived from the URL only (see link_discovery.py)
— never from fetched page content, deliberately, to avoid a prompt-
injection vector into the ticket queue Claude Code later reads.
Dismisses with the list of approved DiscoveredPage entries (empty list
if cancelled).
"""
CSS = """
#review-scroll { width: 100%; padding: 1 2; overflow-x: auto; }
#review-header { width: 100%; padding-bottom: 1; color: $text-muted; }
.review-row { width: auto; min-width: 100%; }
.not-in-sitemap { color: $warning; }
#review-button-row { width: 100%; height: auto; padding-top: 1; align-horizontal: center; }
#review-button-row Button { margin: 0 1; }
"""
BINDINGS = [("escape", "cancel", "Cancel")]
def __init__(self, pages: list[DiscoveredPage], method: str) -> None:
super().__init__()
self.pages = pages
self.method = method
def compose(self) -> ComposeResult:
yield Header()
with VerticalScroll(id="review-scroll"):
confirmed_note = (
"sitemap-verified" if self.method == "sitemap-verified" else "HTML crawl (no sitemap found)"
)
yield Static(
f"Found {len(self.pages)} page(s) via {confirmed_note}. "
f"Review and deselect anything you don't want queued.",
id="review-header",
)
for index, page in enumerate(self.pages):
warn = page.sitemap_confirmed is False
label = f"{page.title} [dim]{page.url}[/dim]"
if warn:
label += " [warning](not in sitemap)[/warning]"
yield Checkbox(label, value=True, id=f"pick-{index}", classes="review-row")
with Horizontal(id="review-button-row"):
yield Button("Select all", id="select-all")
yield Button("Select none", id="select-none")
yield Button("Approve selected", id="approve", variant="success")
yield Button("Cancel", id="cancel", variant="error")
yield Footer()
def _checkboxes(self) -> list[Checkbox]:
return list(self.query(Checkbox))
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "select-all":
for box in self._checkboxes():
box.value = True
elif event.button.id == "select-none":
for box in self._checkboxes():
box.value = False
elif event.button.id == "approve":
approved = [page for page, box in zip(self.pages, self._checkboxes()) if box.value]
self.dismiss(approved)
elif event.button.id == "cancel":
self.dismiss([])
def action_cancel(self) -> None:
self.dismiss([])
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
class AmbrosianaIntakeApp(App):
CSS = """
Screen {
background: $surface;
align: center top;
}
#form-scroll {
width: 100%;
padding: 1 2;
}
#device-banner {
width: 100%;
content-align: center middle;
color: $text-muted;
padding-bottom: 1;
}
.field-label {
width: 100%;
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(
"Discovery depth (only used by \"Discover sub-pages\", below)",
classes="field-label optional-marker",
)
yield Input(placeholder="e.g. 4", id="discovery_depth", value="1")
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(
f"{PERMISSION_QUESTION} [required — type yes/no]", classes="field-label"
)
yield Input(placeholder="yes / no", id="permission")
yield Checkbox("Urgent (immediate need — otherwise added to queue)", value=False, id="urgent")
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("Discover sub-pages", id="discover", variant="primary")
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,
"permission": self.query_one("#permission", Input).value.strip(),
"priority": "immediate" if self.query_one("#urgent", Checkbox).value else "queue",
"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("#permission", Input).value = ""
self.query_one("#urgent", Checkbox).value = False
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
if data["permission"].strip().lower() not in ("yes", "y"):
self._set_status(
"Permission must be confirmed by typing 'yes' — ticket not saved.", 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 ""
base_filename = f"{priority_tag}{now.strftime('%Y%m%d-%H%M%S')}_{slug}"
ticket_path = TICKETS_ROOT / f"{base_filename}.md"
# Second-precision timestamps collide when writing many tickets in a
# tight loop (batch discovery) — disambiguate rather than overwrite.
suffix = 2
while ticket_path.exists():
ticket_path = TICKETS_ROOT / f"{base_filename}-{suffix}.md"
suffix += 1
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"permission_confirmed: \"{data['permission']}\"",
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"**Permission confirmed:** {data['permission']}")
body_lines.append(f"**Priority:** {data['priority']}")
body_lines.append("")
body_lines.append(
"*Licence is not captured here — research it from the source itself "
"during drafting; default to \"restricted, in-house use only\" if it "
"can't be determined.*"
)
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 == "discover":
self._do_discover()
elif event.button.id == "reset":
self._clear_form()
self._set_status("")
elif event.button.id == "quit":
self.exit()
# -- discovery flow ---------------------------------------------------
def _do_discover(self) -> None:
data = self._collect()
shared_missing = [f for f in REQUIRED_FIELDS if f != "source_location"
and (not data.get(f) or data.get(f) is Select.BLANK)]
if shared_missing:
self._set_status(
f"Fill in {', '.join(shared_missing)} first — every discovered "
f"page's ticket will reuse these.", error=True,
)
return
if data["permission"].strip().lower() not in ("yes", "y"):
self._set_status(
"Permission must be confirmed by typing 'yes' before discovering.", error=True
)
return
if not re.match(r"^https?://", data["source_location"]):
self._set_status("Discovery needs an http(s) URL in Source location.", error=True)
return
try:
depth = int(self.query_one("#discovery_depth", Input).value.strip() or "1")
except ValueError:
self._set_status("Discovery depth must be a whole number.", error=True)
return
self._set_status(f"Discovering (depth {depth})... this may take a moment.")
self._run_discovery(data["source_location"], depth, data)
@work(thread=True)
def _run_discovery(self, start_url: str, depth: int, shared_data: dict[str, str]) -> None:
pages, method = discover_links(start_url, max_depth=depth)
self.call_from_thread(self._on_discovery_done, pages, method, shared_data)
def _on_discovery_done(
self, pages: list[DiscoveredPage], method: str, shared_data: dict[str, str]
) -> None:
if not pages:
self._set_status(f"No pages discovered via {method} — nothing to review.", error=True)
return
def handle_approved(approved: list[DiscoveredPage]) -> None:
if not approved:
self._set_status("Discovery cancelled — no tickets written.")
return
written = 0
for page in approved:
page_data = dict(shared_data)
page_data["source_location"] = page.url
page_data["title"] = page.title
if self._write_ticket(page_data) is not None:
written += 1
self._set_status(f"Wrote {written} ticket(s) from discovery.")
self._clear_form()
self.push_screen(LinkReviewScreen(pages, method), handle_approved)
if __name__ == "__main__":
AmbrosianaIntakeApp().run()
Reference in a new issue View git blame Copy permalink