diff --git a/ambrosiana_intake.py b/ambrosiana_intake.py index 6ea1ffb..87c92e1 100644 --- a/ambrosiana_intake.py +++ b/ambrosiana_intake.py @@ -36,10 +36,13 @@ 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, @@ -49,6 +52,8 @@ from textual.widgets import ( TextArea, ) +from link_discovery import DiscoveredPage, discover_links + # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- @@ -93,12 +98,9 @@ SOURCE_TIER_OPTIONS = [ ("forum/unmoderated", "forum-unmoderated"), ] -PRIORITY_OPTIONS = [ - ("Immediate need", "immediate"), - ("Add to queue", "queue"), -] +PERMISSION_QUESTION = "Do you have permission to access this document and add it to the RAG?" -REQUIRED_FIELDS = ("source_location", "source_type", "source_tier", "licence", "priority") +REQUIRED_FIELDS = ("source_location", "source_type", "source_tier", "permission") def slugify(text: str, fallback: str) -> str: @@ -110,6 +112,81 @@ def slugify(text: str, fallback: str) -> str: 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 # --------------------------------------------------------------------------- @@ -123,8 +200,7 @@ class AmbrosianaIntakeApp(App): } #form-scroll { - width: 90; - max-width: 100%; + width: 100%; padding: 1 2; } @@ -136,6 +212,7 @@ class AmbrosianaIntakeApp(App): } .field-label { + width: 100%; padding-top: 1; color: $text; text-style: bold; @@ -203,17 +280,24 @@ class AmbrosianaIntakeApp(App): 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("Licence [required]", classes="field-label") - yield Input(placeholder="e.g. The Unlicense, CC-BY-4.0, proprietary...", id="licence") + yield Label( + f"{PERMISSION_QUESTION} [required — type yes/no]", classes="field-label" + ) + yield Input(placeholder="yes / no", id="permission") - yield Label("Priority [required]", classes="field-label") - yield Select(PRIORITY_OPTIONS, id="priority", prompt="Choose priority") + 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") @@ -223,6 +307,7 @@ class AmbrosianaIntakeApp(App): 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") @@ -236,8 +321,8 @@ class AmbrosianaIntakeApp(App): "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, + "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(), } @@ -253,8 +338,8 @@ class AmbrosianaIntakeApp(App): 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("#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() @@ -264,14 +349,25 @@ class AmbrosianaIntakeApp(App): 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 "" - filename = f"{priority_tag}{now.strftime('%Y%m%d-%H%M%S')}_{slug}.md" - ticket_path = TICKETS_ROOT / filename + 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 = [ "---", @@ -281,7 +377,7 @@ class AmbrosianaIntakeApp(App): f"source_location: \"{data['source_location']}\"", f"source_type: {data['source_type']}", f"source_tier: {data['source_tier']}", - f"licence: \"{data['licence']}\"", + f"permission_confirmed: \"{data['permission']}\"", f"priority: {data['priority']}", f"title: \"{data['title']}\"" if data["title"] else "title: \"\"", "---", @@ -295,9 +391,15 @@ class AmbrosianaIntakeApp(App): 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"**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("") @@ -328,12 +430,72 @@ class AmbrosianaIntakeApp(App): 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() diff --git a/link_discovery.py b/link_discovery.py new file mode 100644 index 0000000..1c1d703 --- /dev/null +++ b/link_discovery.py @@ -0,0 +1,243 @@ +#!/usr/bin/env python3 +# Created by John A. Hoeven with the ethical assistance of Claude AI +# --------------------------------------------------------------------------- +# link_discovery.py +# ~/projects/ambrosiana-document-intake/link_discovery.py +# Version: v0.2.0 | Status: BETA +# --------------------------------------------------------------------------- +# Purpose: URL discovery from an index page, for the "add index page, +# review discovered sub-pages" intake flow. Fetches pages to find +# URLs only — never writes a ticket itself; that stays a separate, +# human-approved step in the App. +# Target: Any Ambrosiana device. +# Entry: import as a module. +# Depends: requests, beautifulsoup4 (see requirements.txt) +# --------------------------------------------------------------------------- +""" +Security note (deliberate, not an oversight): this module never extracts +free-text content from a fetched page (titles, headings, link text) into +anything that ends up in a ticket. A ticket is later read by Claude Code +as a pipeline instruction, so pulling untrusted third-party prose into a +ticket field would be a real prompt-injection vector. Only URLs — which +this tool itself discovered under a scope the human chose — are ever +returned. Display titles shown in the review UI are derived from the URL +path alone, never from fetched page content. + +Two discovery strategies, tried in this order: + +1. Sitemap (preferred) — sitemap.xml is pure structured data (, + ), nothing to inject into. Checked at the domain root and via + robots.txt's Sitemap: directive. +2. HTML link crawl (fallback) — same-domain breadth-first crawl for sites + without a sitemap. Extracts hrefs only, never page titles or body text. +""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET +from dataclasses import dataclass +from urllib.parse import urljoin, urlparse, urlunparse + +import requests +from bs4 import BeautifulSoup + +USER_AGENT = "ambrosiana-intake-discovery/0.2 (personal RAG intake tool)" +REQUEST_TIMEOUT = 10 +DEFAULT_MAX_PAGES = 300 +SKIP_EXTENSIONS = ( + ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".ico", + ".woff", ".woff2", ".ttf", ".xml", ".json", ".pdf", ".zip", ".md", +) +SITEMAP_XML_NS = "{http://www.sitemaps.org/schemas/sitemap/0.9}" + + +@dataclass +class DiscoveredPage: + url: str + title: str # derived from the URL path only — never from fetched content + depth: int + sitemap_confirmed: bool | None = None # None = no sitemap available to check against + + +def _display_title(url: str) -> str: + """Human-readable label derived purely from the URL — no page content.""" + parsed = urlparse(url) + path = parsed.path.strip("/") + if not path: + return parsed.netloc or url # site root — use the domain, not the raw URL + last_segment = path.rsplit("/", 1)[-1] + return last_segment.replace("-", " ").replace("_", " ").strip() or url + + +def _normalize(url: str) -> str: + """Strip fragment/query for de-dup purposes; keep the path as-is.""" + parsed = urlparse(url) + return urlunparse((parsed.scheme, parsed.netloc, parsed.path.rstrip("/") or "/", "", "", "")) + + +def _is_same_domain(url: str, domain: str) -> bool: + return urlparse(url).netloc == domain + + +def _is_content_link(url: str) -> bool: + parsed = urlparse(url) + if parsed.scheme not in ("http", "https"): + return False + return not parsed.path.lower().endswith(SKIP_EXTENSIONS) + + +def _fetch(url: str) -> requests.Response | None: + try: + response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=REQUEST_TIMEOUT) + response.raise_for_status() + return response + except requests.RequestException: + return None + + +# ── Strategy 1: sitemap ────────────────────────────────────────────────── + +def find_sitemap_url(start_url: str) -> str | None: + """Check robots.txt's Sitemap: directive first, then the conventional path.""" + parsed = urlparse(start_url) + root = f"{parsed.scheme}://{parsed.netloc}" + + robots = _fetch(f"{root}/robots.txt") + if robots is not None: + for line in robots.text.splitlines(): + if line.strip().lower().startswith("sitemap:"): + return line.split(":", 1)[1].strip() + + candidate = f"{root}/sitemap.xml" + response = _fetch(candidate) + if response is not None and "xml" in response.headers.get("content-type", ""): + return candidate + return None + + +def fetch_sitemap_urls(sitemap_url: str, _depth: int = 0) -> set[str]: + """ + Parse a sitemap.xml (or a sitemap index listing nested sitemaps, handled + up to 2 levels deep) and return the full set of normalized URLs — + used only as a cross-reference trust signal, never as the primary + discovery source (see discover_links). ElementTree's default parser + does not resolve external entities/DTDs, so this is safe against + classic XXE without needing an extra dependency. + """ + if _depth > 2: + return set() + + response = _fetch(sitemap_url) + if response is None: + return set() + + try: + root = ET.fromstring(response.content) + except ET.ParseError: + return set() + + locs = [el.text.strip() for el in root.iter(f"{SITEMAP_XML_NS}loc") if el.text] + + if root.tag == f"{SITEMAP_XML_NS}sitemapindex": + urls: set[str] = set() + for nested_url in locs[:20]: # bounded — a sitemap index fan-out cap + urls |= fetch_sitemap_urls(nested_url, _depth=_depth + 1) + return urls + + return {_normalize(loc) for loc in locs} + + +# ── Strategy 2: HTML crawl fallback ────────────────────────────────────── + +def discover_from_crawl( + start_url: str, + max_depth: int, + max_pages: int = DEFAULT_MAX_PAGES, + sitemap_urls: set[str] | None = None, +) -> list[DiscoveredPage]: + """ + Breadth-first same-domain crawl from start_url, following the actual + links on each page — this is what "N links deep from the index" + literally means, including for hub/TOC-style index pages whose real + sub-content lives at sibling paths rather than nested under the index's + own URL. Extracts hrefs only — never page titles or body text (see + module docstring). The start page itself is not included in the + returned list, only pages found from it. + + If sitemap_urls is given, each discovered page is annotated with + whether it's also present in the sitemap — a trust signal shown in the + review UI, not a filter (a sitemap can be incomplete or stale). + """ + domain = urlparse(start_url).netloc + visited = {_normalize(start_url)} + queue: list[tuple[str, int]] = [(start_url, 0)] + discovered: list[DiscoveredPage] = [] + fetched_count = 0 + + while queue and fetched_count < max_pages: + url, depth = queue.pop(0) + if depth >= max_depth: + continue + + response = _fetch(url) + fetched_count += 1 + if response is None: + continue + + soup = BeautifulSoup(response.text, "html.parser") + for anchor in soup.find_all("a", href=True): + link = urljoin(url, anchor["href"]) + if not _is_content_link(link) or not _is_same_domain(link, domain): + continue + link_norm = _normalize(link) + if link_norm in visited: + continue + visited.add(link_norm) + + confirmed = (link_norm in sitemap_urls) if sitemap_urls is not None else None + discovered.append( + DiscoveredPage(url=link, title=_display_title(link), depth=depth + 1, sitemap_confirmed=confirmed) + ) + if len(discovered) + fetched_count >= max_pages: + break + queue.append((link, depth + 1)) + + return [p for p in discovered if p.sitemap_confirmed or _is_reachable(p.url)] + + +def _is_reachable(url: str) -> bool: + """ + Cheap liveness check for links not already confirmed via the sitemap. + Relative-link resolution against an already-nested page can produce + plausible-looking but nonexistent URLs (e.g. a same-name breadcrumb + href resolving into a doubled path segment) — verified live 2026-08-07 + against textual.textualize.io, where this caught real 404s a sitemap- + confirmed link would never need checking for. + """ + try: + response = requests.head( + url, headers={"User-Agent": USER_AGENT}, timeout=REQUEST_TIMEOUT, allow_redirects=True + ) + return response.status_code < 400 + except requests.RequestException: + return False + + +# ── Entry point ─────────────────────────────────────────────────────────── + +def discover_links( + start_url: str, + max_depth: int, + max_pages: int = DEFAULT_MAX_PAGES, +) -> tuple[list[DiscoveredPage], str]: + """ + Returns (discovered_pages, method) — method is "sitemap-verified" or + "crawl" so the UI can tell the human whether a sitemap was available to + cross-check against. + """ + sitemap_url = find_sitemap_url(start_url) + sitemap_urls = fetch_sitemap_urls(sitemap_url) if sitemap_url else None + + pages = discover_from_crawl(start_url, max_depth=max_depth, max_pages=max_pages, sitemap_urls=sitemap_urls) + method = "sitemap-verified" if sitemap_urls else "crawl" + return pages, method diff --git a/requirements.txt b/requirements.txt index ac6218b..e7f04cd 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,3 @@ textual>=8.2.0 +requests>=2.34.0 +beautifulsoup4>=4.15.0