chorus/echo-memory.plugin.src/skills/echo-memory/scripts/echo_links.py

#!/usr/bin/env python3
"""echo_links.py — cross-link primitives.

Parse and add `## Related` wikilinks, and create **bidirectional** links between
notes (read-modify-write, idempotent). Links use path-style targets without the
`.md` extension (e.g. `[[resources/people/bob-smith]]`), which are unambiguous
and resolve in Obsidian. Network goes through echo.py.
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import echo  # noqa: E402

WIKILINK = re.compile(r"\[\[([^\]|#]+)(?:[#|][^\]]*)?\]\]")


def first_h1(text: str) -> str:
    for line in text.splitlines():
        if line.startswith("# "):
            return line[2:].strip()
    return ""


def all_wikilinks(text: str) -> list[str]:
    return [m.group(1).strip() for m in WIKILINK.finditer(text)]


def related_targets(text: str) -> list[str]:
    """Wikilink targets under the `## Related` heading only."""
    out, cap = [], False
    for line in text.splitlines():
        if line.strip().lower() == "## related":
            cap = True
            continue
        if cap and line.startswith("## "):
            break
        if cap:
            out += [m.group(1).strip() for m in WIKILINK.finditer(line)]
    return out


def source_notes(text: str) -> list[str]:
    """Plain relative paths listed in the frontmatter `source_notes:` field."""
    if not text.startswith("---"):
        return []
    end = text.find("\n---", 3)
    fm = text[:end] if end != -1 else text
    m = re.search(r"(?m)^source_notes:\s*\[(.*?)\]", fm)
    if not m:
        return []
    return [p.strip().strip('"').strip("'") for p in m.group(1).split(",") if p.strip()]


def link_token(path: str) -> str:
    """Path-style wikilink target without the .md extension."""
    return path[:-3] if path.endswith(".md") else path


def add_related(text: str, target_path: str):
    """Return (new_text, changed): ensure `- [[<target>]]` exists under ## Related."""
    token = link_token(target_path)
    if token in related_targets(text):
        return text, False
    bullet = f"- [[{token}]]"
    trailing_nl = text.endswith("\n")
    lines = text.splitlines()
    idx = next((i for i, l in enumerate(lines) if l.strip().lower() == "## related"), None)
    if idx is None:
        sep = "" if trailing_nl else "\n"
        return text + f"{sep}\n## Related\n{bullet}\n", True
    # insert after the section's existing content (before the next heading / EOF)
    j = idx + 1
    while j < len(lines) and not lines[j].startswith("## "):
        j += 1
    k = j
    while k > idx + 1 and not lines[k - 1].strip():
        k -= 1
    lines.insert(k, bullet)
    return "\n".join(lines) + ("\n" if trailing_nl else ""), True


def get_text(path: str) -> str | None:
    status, body = echo.request("GET", echo.vault_url(path))
    if status == 404:
        return None
    echo.check(status, body, f"get {path}")
    return body.decode(errors="replace")


def put_text(path: str, text: str) -> None:
    status, body = echo.request("PUT", echo.vault_url(path), data=text.encode("utf-8"),
                                headers={"Content-Type": "text/markdown"})
    echo.check(status, body, f"put {path}")


def add_one(path: str, target_path: str) -> bool:
    """Add [[target]] under path's ## Related. Returns True if the file changed."""
    text = get_text(path)
    if text is None:
        return False
    new, changed = add_related(text, target_path)
    if changed:
        put_text(path, new)
    return changed


def link_bidirectional(a_path: str, b_path: str):
    """Add A<->B reciprocal links. Returns (a_changed, b_changed)."""
    if a_path == b_path:
        return (False, False)
    return (add_one(a_path, b_path), add_one(b_path, a_path))