"""Knowtation vault link indexer (Phase 3.1). Parses three kinds of inter-note edges and assembles a directed graph keyed by vault-relative POSIX paths: 1. **Wikilinks** — Obsidian-style ``[[Note Title]]`` / ``[[note|alias]]`` / ``[[note#section]]``. Regex ported from ``knowtation/lib/wikilink.mjs`` with semantic parity. 2. **Markdown links to .md files** — standard CommonMark ``[text](relative/path.md)`` (and ``./path.md`` / ``../path.md``) references. This parser does **not** exist on the Knowtation side today and is new in Muse. 3. **Entity references** — every value in a note's ``entity`` frontmatter field that matches another note's ``entity`` field is treated as a co-occurrence edge. Useful for ``muse code entangle`` (Phase 3.3). 4. **Mist attachment references** — Phase 3.5 (extension) treats ``frontmatter.attachments[*]`` as a third edge type. Stub returned in the ``attachments`` field; full join logic lives in `link_index.py` Phase 3.5. Public API ---------- - :class:`ResolvedLink` — a single resolved edge from one note to another. - :class:`LinkIndex` — full vault index with forward / backward / entity edges. - :func:`extract_wikilinks` — extract raw wikilink targets from a note body. - :func:`extract_md_links` — extract raw markdown-to-md links from a note body. - :func:`build_link_index` — walk a vault directory and assemble a full index. Resolver semantics (OBA-reviewed) --------------------------------- ``case sensitivity`` The resolver is **case-insensitive on basenames** to match Obsidian and macOS HFS+/APFS default behaviour. Two notes with the same basename in different cases are treated as duplicates and a warning is logged. ``fragment handling`` ``[[note#section]]`` resolves the *note* portion; the section fragment is stored on the :class:`ResolvedLink` for downstream consumers but is not used to disambiguate the target note. ``relative path resolution`` Markdown links of the form ``[text](path.md)`` are resolved relative to the *current note's directory*. ``./`` and ``../`` are normalised via :class:`pathlib.PurePosixPath`. Absolute paths (leading ``/``) are treated as vault-root-relative. Paths escaping the vault root are rejected (broken-link semantics, see below). ``broken-link semantics`` Wikilinks or markdown links whose target cannot be resolved to a vault note become a :class:`ResolvedLink` with ``target=None`` and ``broken=True``. Callers (``muse code deps`` / ``impact``) can filter on ``broken`` to choose between strict and lenient modes. ``URL fragments and external links`` Markdown links starting with ``http://``, ``https://``, ``mailto:``, or any other scheme are ignored — only relative ``.md`` references count. """ from __future__ import annotations import logging import os import pathlib import re from dataclasses import dataclass, field from typing import Iterator from muse.plugins.knowtation._query import is_note from muse.plugins.knowtation.parser import parse_frontmatter logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Regex patterns # --------------------------------------------------------------------------- #: Obsidian wikilink — ``[[target|alias]]`` or ``[[target#section]]`` or #: ``[[target]]``. Capture groups: (1) target, (2) section, (3) alias. #: #: Ported from knowtation/lib/wikilink.mjs. The character class excludes ``]``, #: ``|``, ``#`` from the target to prevent runaway matches. _WIKILINK_RE: re.Pattern[str] = re.compile( r"\[\[([^\]|#]+)(?:#([^\]|]+))?(?:\|([^\]]+))?\]\]" ) #: Markdown link to a ``.md`` file (or fragment). Matches: #: [text](path.md) #: [text](./path.md) #: [text](../sub/note.md) #: [text](note.md#heading) #: #: Excludes URL schemes (``http:``, ``mailto:``, etc.) via the negative #: lookahead ``(?![a-z][a-z0-9+.\-]*:)``. _MD_LINK_RE: re.Pattern[str] = re.compile( r"\[([^\]]+)\]" # [text] r"\(" # ( r"(?![a-z][a-z0-9+.\-]*:)" # not http://, mailto:, etc. r"([^)\s]+?\.md)" # path ending in .md r"(?:#([^)\s]+))?" # optional #fragment r"\)" # ) ) #: Schemes that mark a markdown link as external — never treated as note edges. _EXTERNAL_SCHEMES: frozenset[str] = frozenset( {"http", "https", "ftp", "mailto", "tel", "file", "data"} ) # --------------------------------------------------------------------------- # Data classes # --------------------------------------------------------------------------- @dataclass(frozen=True) class ResolvedLink: """A single inter-note edge with both raw and resolved forms. Attributes: source: Vault-relative POSIX path of the note containing the link. target: Vault-relative POSIX path of the resolved target note, or ``None`` when the link is broken. raw: The original link text as written in the source note (e.g. ``"Some Note"`` or ``"../folder/note.md"``). kind: Edge kind: ``"wikilink"``, ``"md_link"``, ``"entity"``, or ``"attachment"``. fragment: Optional section/heading fragment (e.g. ``"§Summary"``) when present in the original link. Empty string when no fragment. broken: ``True`` when the link's target could not be resolved to an existing vault note. alias: Display alias for wikilinks (the text after ``|``); empty string when none. """ source: str target: str | None raw: str kind: str fragment: str = "" broken: bool = False alias: str = "" @dataclass class LinkIndex: """Forward and reverse edge index for an entire Knowtation vault. Attributes: forward: ``source → [ResolvedLink, …]`` — every link in every note. backward: ``target → [ResolvedLink, …]`` — only resolved (non-broken) links. broken_links: List of :class:`ResolvedLink` with ``broken=True``. entity_index: ``entity_label → set[note_path]`` for entity co-occurrence. attachment_index: ``mist_id → set[note_path]`` for mist-attachment co-occurrence (Phase 3.5). notes_indexed: Total notes walked during construction. duplicate_basenames: Basenames that appeared in more than one folder (logged as warnings; the *first* path in walk order wins). """ forward: dict[str, list[ResolvedLink]] = field(default_factory=dict) backward: dict[str, list[ResolvedLink]] = field(default_factory=dict) broken_links: list[ResolvedLink] = field(default_factory=list) entity_index: dict[str, set[str]] = field(default_factory=dict) attachment_index: dict[str, set[str]] = field(default_factory=dict) notes_indexed: int = 0 duplicate_basenames: dict[str, list[str]] = field(default_factory=dict) # ------------------------------------------------------------------ # Lookup helpers # ------------------------------------------------------------------ def outgoing(self, source: str) -> list[ResolvedLink]: """Return all links originating at *source*. Empty list if none.""" return list(self.forward.get(source, [])) def incoming(self, target: str) -> list[ResolvedLink]: """Return all resolved links pointing at *target*. Empty list if none.""" return list(self.backward.get(target, [])) def co_entity(self, note_path: str) -> set[str]: """Return notes that share at least one entity label with *note_path*. Args: note_path: Vault-relative POSIX path of the seed note. Returns: Set of other note paths sharing at least one entity label. Excludes *note_path* itself. """ out: set[str] = set() for entity_label, members in self.entity_index.items(): if note_path in members: out.update(members) out.discard(note_path) return out def co_attachment(self, note_path: str) -> set[str]: """Return notes that share at least one mist attachment with *note_path*. Two notes are "co-attached" when both reference the same mist artefact ID in their ``frontmatter.attachments[*]``. Useful for ``muse code entangle`` and ``muse code impact --include-attachments``. Args: note_path: Vault-relative POSIX path of the seed note. Returns: Set of other note paths sharing at least one mist attachment ID. Excludes *note_path* itself. """ out: set[str] = set() for _mist_id, members in self.attachment_index.items(): if note_path in members: out.update(members) out.discard(note_path) return out # --------------------------------------------------------------------------- # Wikilink helpers (semantic parity with lib/wikilink.mjs) # --------------------------------------------------------------------------- def _wikilink_target_key(raw: str) -> str: """Normalise wikilink inner text to a comparable basename stem. Mirrors ``wikilinkTargetKey`` in ``lib/wikilink.mjs``: - Trim whitespace. - Take the last path segment (split on ``/`` or ``\\``). - Strip a trailing ``.md`` (case-insensitive). - Lowercase. Args: raw: The raw wikilink target text (between ``[[`` and the next special character). Returns: Lowercase comparable key. """ s = str(raw).strip() if not s: return "" last = re.split(r"[/\\]", s)[-1] if last.lower().endswith(".md"): last = last[:-3] return last.lower() def _vault_basename_key(vault_path: str) -> str: """Return the comparable basename key for *vault_path*. Mirrors ``vaultBasenameTargetKey`` in ``lib/wikilink.mjs``. Args: vault_path: Vault-relative POSIX path. Returns: Lowercase basename stem (no ``.md`` extension). """ target = vault_path.replace("\\", "/") last = target.split("/")[-1] or target return _wikilink_target_key(last) # --------------------------------------------------------------------------- # Public extractors # --------------------------------------------------------------------------- def extract_wikilinks(body: str) -> list[tuple[str, str, str]]: """Extract all wikilinks from *body* in source order. Args: body: Markdown body text (NOT bytes, NOT including frontmatter). Returns: List of ``(target, fragment, alias)`` tuples. ``fragment`` and ``alias`` are empty strings when absent. """ out: list[tuple[str, str, str]] = [] for m in _WIKILINK_RE.finditer(body): target = m.group(1).strip() fragment = (m.group(2) or "").strip() alias = (m.group(3) or "").strip() out.append((target, fragment, alias)) return out def extract_md_links(body: str) -> list[tuple[str, str, str]]: """Extract all ``[text](path.md)`` markdown links from *body*. Args: body: Markdown body text. Returns: List of ``(path, fragment, text)`` tuples. Only relative ``.md`` paths are returned; URLs are excluded. """ out: list[tuple[str, str, str]] = [] for m in _MD_LINK_RE.finditer(body): text = m.group(1).strip() path = m.group(2).strip() fragment = (m.group(3) or "").strip() out.append((path, fragment, text)) return out # --------------------------------------------------------------------------- # Path resolution # --------------------------------------------------------------------------- def _normalise_rel_path(source_dir: str, rel_path: str) -> str | None: """Resolve *rel_path* relative to *source_dir* without leaving the vault. Args: source_dir: Vault-relative POSIX directory of the source note (e.g. ``"projects/born-free"``; ``""`` for vault root). rel_path: Relative path from the source note (e.g. ``"../foo.md"``). Returns: The resolved vault-relative POSIX path, or ``None`` if the resolved path escapes the vault root (security check). """ if rel_path.startswith("/"): candidate = pathlib.PurePosixPath(rel_path.lstrip("/")) else: candidate = pathlib.PurePosixPath(source_dir) / rel_path parts: list[str] = [] for part in candidate.parts: if part == "." or part == "": continue if part == "..": if not parts: return None parts.pop() continue parts.append(part) return "/".join(parts) def _resolve_wikilink( raw: str, basename_to_path: dict[str, str], ) -> str | None: """Resolve a wikilink target to a vault-relative path. Uses case-insensitive basename matching (Obsidian / macOS semantics). Args: raw: The raw wikilink target text. basename_to_path: Pre-built map of lowercase basename stem → vault path. Returns: The resolved vault-relative POSIX path, or ``None`` if no match. """ key = _wikilink_target_key(raw) if not key: return None return basename_to_path.get(key) def _resolve_md_link( source_path: str, raw_path: str, existing_paths: set[str], ) -> str | None: """Resolve a markdown link relative path against the vault. Args: source_path: Vault-relative POSIX path of the source note. raw_path: Raw link path (e.g. ``"../foo.md"``). existing_paths: Set of all known vault note paths. Returns: Resolved vault-relative path, or ``None`` if it doesn't exist or escapes the vault. """ source_dir = "/".join(source_path.split("/")[:-1]) resolved = _normalise_rel_path(source_dir, raw_path) if resolved is None: return None if resolved in existing_paths: return resolved # Case-insensitive fallback for macOS / Windows behaviour parity. lower = resolved.lower() for p in existing_paths: if p.lower() == lower: return p return None # --------------------------------------------------------------------------- # Vault traversal # --------------------------------------------------------------------------- #: Directories pruned from vault walks (template scaffolding, meta, hidden). _IGNORE_DIRS: frozenset[str] = frozenset( {"templates", "meta", "node_modules", "__pycache__", ".muse"} ) #: Per-note size cap for body reads — defends against oversized files. _MAX_NOTE_BYTES: int = 16 * 1024 * 1024 # 16 MiB def _iter_vault_notes(root: pathlib.Path) -> Iterator[pathlib.Path]: """Yield :class:`pathlib.Path` for every Markdown note in *root*. Args: root: Vault root directory. Yields: Absolute paths to ``.md`` / ``.markdown`` / ``.mdx`` files. """ root_str = str(root) for dirpath, dirnames, filenames in os.walk(root_str, followlinks=False): dirnames[:] = sorted( d for d in dirnames if d not in _IGNORE_DIRS and not d.startswith(".") ) for fname in sorted(filenames): if is_note(fname): yield pathlib.Path(dirpath) / fname def _read_note_safely(path: pathlib.Path) -> bytes | None: """Read *path* with a size cap; return ``None`` on failure or oversize.""" try: size = path.stat().st_size except OSError: return None if size > _MAX_NOTE_BYTES: logger.warning("link_index: skipping oversized note %s (%d bytes)", path, size) return None try: return path.read_bytes() except OSError: return None # --------------------------------------------------------------------------- # Main builder # --------------------------------------------------------------------------- def build_link_index(root: pathlib.Path) -> LinkIndex: """Build a :class:`LinkIndex` for the vault rooted at *root*. Walks the vault once, parsing wikilinks, markdown-md links, and entity frontmatter values for every note. Resolves edges using case-insensitive basename matching (Obsidian / macOS semantics) and relative-path resolution from each note's directory. Args: root: Vault root directory. Returns: Fully populated :class:`LinkIndex`. Raises: ValueError: When *root* does not exist or is not a directory. Performance: < 10s on a 5k-note vault (verified by test_knowtation_link_index.py). """ if not root.exists() or not root.is_dir(): raise ValueError(f"link_index: root {root!r} is not a directory") root = root.resolve() # ── Pass 1: enumerate all notes and build the basename lookup table. ── note_paths: list[str] = [] basename_to_path: dict[str, str] = {} duplicates: dict[str, list[str]] = {} for abs_path in _iter_vault_notes(root): rel_path = abs_path.relative_to(root).as_posix() note_paths.append(rel_path) key = _vault_basename_key(rel_path) if key in basename_to_path: existing = basename_to_path[key] duplicates.setdefault(key, [existing]).append(rel_path) logger.debug( "link_index: duplicate basename %r (keeping %r, ignoring %r)", key, existing, rel_path, ) else: basename_to_path[key] = rel_path existing_paths: set[str] = set(note_paths) # ── Pass 2: extract links and build edges. ─────────────────────────── index = LinkIndex( notes_indexed=len(note_paths), duplicate_basenames=duplicates, ) for rel_path in note_paths: abs_path = root / rel_path content = _read_note_safely(abs_path) if content is None: continue try: body_text = content.decode("utf-8", errors="replace") except Exception: continue # Strip frontmatter so links inside YAML aren't matched as body text. body = _strip_frontmatter(body_text) links: list[ResolvedLink] = [] # Wikilinks ──────────────────────────────────────────────────── for raw, fragment, alias in extract_wikilinks(body): resolved = _resolve_wikilink(raw, basename_to_path) link = ResolvedLink( source=rel_path, target=resolved, raw=raw, kind="wikilink", fragment=fragment, broken=resolved is None, alias=alias, ) links.append(link) # Markdown .md links ─────────────────────────────────────────── for raw_path, fragment, text in extract_md_links(body): resolved = _resolve_md_link(rel_path, raw_path, existing_paths) link = ResolvedLink( source=rel_path, target=resolved, raw=raw_path, kind="md_link", fragment=fragment, broken=resolved is None, alias=text, ) links.append(link) # Entity frontmatter values — build co-occurrence index ─────── # Mist attachment IDs — build third co-occurrence index (Phase 3.5). fm = parse_frontmatter(content) if fm is not None: for entity_label in fm.entity: lbl = entity_label.strip() if not lbl: continue index.entity_index.setdefault(lbl, set()).add(rel_path) from muse.plugins.knowtation._query import is_valid_mist_id # noqa: PLC0415 for mist_id in fm.attachments: mid = mist_id.strip() if not mid or not is_valid_mist_id(mid): continue index.attachment_index.setdefault(mid, set()).add(rel_path) index.forward[rel_path] = links for link in links: if link.target and not link.broken: index.backward.setdefault(link.target, []).append(link) else: index.broken_links.append(link) return index # --------------------------------------------------------------------------- # Frontmatter stripping # --------------------------------------------------------------------------- _FM_START_RE: re.Pattern[str] = re.compile(r"\A---\s*\n") def _strip_frontmatter(text: str) -> str: """Return *text* with the leading YAML frontmatter block removed. Args: text: Full note text including any frontmatter. Returns: Body text (everything after the closing ``---`` line), or the original text if no frontmatter is present. """ if not _FM_START_RE.match(text): return text after_start = text.split("\n", 1) if len(after_start) < 2: return text rest = after_start[1] for delim in ("\n---", "\n..."): idx = rest.find(delim) if idx != -1: return rest[idx + len(delim):] return text