logseq-cli: query Logseq via local HTTP API and render markdown-friendly output

logseq-cli

#!/usr/bin/env python3
"""logseq-cli — talk to the local Logseq HTTP API."""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import textwrap
import urllib.error
import urllib.request

DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 12315
DEFAULT_DEPTH = 10
TIMEOUT_SECONDS = 15

UUID_RE = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")
# Logseq block-property line, e.g. "id:: 6a04...", "logseq.order-list-type:: number"
PROP_RE = re.compile(r"^\s*[A-Za-z][A-Za-z0-9_.-]*::\s")

DESCRIPTION = "Talk to the local Logseq HTTP API. Retrieves blocks and renders Datalog query results."

EPILOG = textwrap.dedent("""\
    Setup:
      1. Logseq desktop must be running on this machine.
      2. Enable Logseq → Settings → Features → HTTP APIs (default port 12315).
      3. Export your Logseq API token before running:
           export LOGSEQ_API_TOKEN=<token>
         Optional overrides: LOGSEQ_API_HOST (default 127.0.0.1),
                             LOGSEQ_API_PORT (default 12315).

    Block examples:
      logseq-cli <block-uuid>
      logseq-cli '((<block-uuid>))'
      logseq-cli --header <block-uuid>
      logseq-cli --depth 2 <block-uuid>
      logseq-cli --json   <block-uuid>

    Datalog query examples:
      logseq-cli query '[:find (pull ?p [*]) :where [?p :block/name ?n] [(clojure.string/starts-with? ?n "blog/")]]'
      logseq-cli query --file ~/queries/blogs.edn
      logseq-cli query - < ~/queries/blogs.edn

    Notes:
      • Block mode accepts a raw UUID, the ((uuid)) form, or any string containing a UUID.
      • Query mode uses logseq.DB.customQuery and expects Datalog that returns page or block entities.
      • Query snippets may be plain Datalog vectors or Logseq advanced query maps containing :query.
      • Block-property lines (id::, collapsed::, logseq.order-list-type::, …) are stripped from markdown output.
      • If a subtree extends beyond --depth, a "… (children truncated)" marker is printed in its place.
""")

QUERY_EPILOG = textwrap.dedent("""\
    Examples:
      logseq-cli query '[:find (pull ?p [*]) :where [?p :block/name ?n] [(clojure.string/starts-with? ?n "blog/")]]'
      logseq-cli query --file ~/queries/blogs.edn
      logseq-cli query --stdin < ~/queries/blogs.edn
      logseq-cli query - < ~/queries/blogs.edn

    Query result shape:
      Return page entities when you want whole pages:
        [:find (pull ?p [*])
         :where
         [?p :block/name ?n]
         [(clojure.string/starts-with? ?n "blog/")]]

      Return block entities when you want matching block subtrees:
        [:find (pull ?b [*])
         :where
         [?b :block/content ?c]
         [(clojure.string/includes? ?c "#blog")]]
""")


def call_api(method: str, args: list, host: str, port: int, token: str):
    payload = json.dumps({"method": method, "args": args}).encode("utf-8")
    req = urllib.request.Request(
        f"http://{host}:{port}/api",
        data=payload,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {token}",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as resp:
            body = resp.read().decode("utf-8")
    except urllib.error.HTTPError as e:
        sys.exit(f"HTTP {e.code} from Logseq API: {e.read().decode('utf-8', 'replace')}")
    except urllib.error.URLError as e:
        sys.exit(f"Could not reach Logseq API at {host}:{port}: {e.reason}")
    if not body:
        return None
    return json.loads(body)


def extract_uuid(text: str) -> str:
    """Accept raw UUID, ((uuid)) form, or any string containing one."""
    m = UUID_RE.search(text.lower())
    if not m:
        sys.exit(f"No valid block UUID found in: {text!r}")
    return m.group(0)


def clean_content(content: str | None) -> list[str]:
    """Drop block-property lines; return the remaining lines (≥1)."""
    if not content:
        return [""]
    lines = [ln for ln in content.split("\n") if not PROP_RE.match(ln)]
    while lines and not lines[-1].strip():
        lines.pop()
    return lines or [""]


def fetch_block_tree(identifier, host: str, port: int, token: str,
                     max_depth: int, depth: int = 0) -> dict | None:
    """Fetch a block, then recurse into each child (API expands only 1 level)."""
    block = call_api(
        "logseq.Editor.getBlock",
        [identifier, {"includeChildren": True}],
        host, port, token,
    )
    if not block:
        return None

    raw_children = block.get("children") or []
    if depth >= max_depth:
        block["_truncated"] = bool(raw_children)
        block["children"] = []
        return block

    expanded = []
    for child in raw_children:
        child_uuid = None
        if isinstance(child, dict):
            child_uuid = child.get("uuid")
        elif isinstance(child, list) and len(child) == 2 and child[0] == "uuid":
            child_uuid = child[1]
        if not child_uuid:
            continue
        sub = fetch_block_tree(child_uuid, host, port, token, max_depth, depth + 1)
        if sub is not None:
            expanded.append(sub)
    block["children"] = expanded
    return block


def limit_tree_depth(block: dict, max_depth: int, depth: int = 0) -> None:
    children = block.get("children") or []
    if depth >= max_depth:
        block["_truncated"] = bool(children)
        block["children"] = []
        return
    for child in children:
        if isinstance(child, dict):
            limit_tree_depth(child, max_depth, depth + 1)


def fetch_page_blocks_tree(identifier, host: str, port: int, token: str, max_depth: int) -> list[dict]:
    blocks = call_api(
        "logseq.Editor.getPageBlocksTree",
        [identifier],
        host, port, token,
    )
    if not blocks:
        return []
    for block in blocks:
        if isinstance(block, dict):
            limit_tree_depth(block, max_depth)
    return blocks


def render_block(block: dict, depth: int = 0, indent: str = "  ") -> list[str]:
    lines = clean_content(block.get("content"))
    pad = indent * depth
    out = [f"{pad}- {lines[0]}"]
    cont_pad = pad + "  "
    out.extend(f"{cont_pad}{ln}" for ln in lines[1:])
    if block.get("_truncated"):
        out.append(f"{cont_pad}… (children truncated, increase --depth)")
    for child in block.get("children") or []:
        out.extend(render_block(child, depth + 1, indent))
    return out


def page_display_name(page: dict) -> str:
    return (
        page.get("originalName")
        or page.get("original-name")
        or page.get("name")
        or str(page.get("uuid") or page.get("id") or "?")
    )


def page_lookup_key(page: dict):
    return (
        page.get("uuid")
        or page.get("originalName")
        or page.get("original-name")
        or page.get("name")
        or page.get("id")
    )


def block_lookup_key(block: dict):
    return block.get("uuid") or block.get("id")


def render_page(page: dict, blocks: list[dict], include_header: bool = True) -> str:
    out = []
    if include_header:
        out.extend([f"# {page_display_name(page)}", ""])
    for block in blocks:
        out.extend(render_block(block))
    while out and out[-1] == "":
        out.pop()
    return "\n".join(out)


def strip_markdown_wrapper(text: str) -> str:
    lines = text.strip().splitlines()
    if lines and lines[0].lstrip().startswith("```"):
        lines = lines[1:]
        if lines and lines[-1].lstrip().startswith("```"):
            lines = lines[:-1]
    lines = [
        line for line in lines
        if line.strip() not in {"#+BEGIN_QUERY", "#+END_QUERY"}
    ]
    return "\n".join(lines).strip()


def extract_query_from_map(text: str) -> str | None:
    idx = text.find(":query")
    if idx == -1:
        return None
    i = idx + len(":query")
    while i < len(text) and text[i].isspace():
        i += 1
    if i >= len(text):
        return None

    pairs = {"[": "]", "(": ")", "{": "}"}
    opener = text[i]
    closer = pairs.get(opener)
    if not closer:
        return None

    depth = 0
    in_string = False
    escaped = False
    for j in range(i, len(text)):
        ch = text[j]
        if in_string:
            if escaped:
                escaped = False
            elif ch == "\\":
                escaped = True
            elif ch == '"':
                in_string = False
            continue

        if ch == '"':
            in_string = True
        elif ch == opener:
            depth += 1
        elif ch == closer:
            depth -= 1
            if depth == 0:
                return text[i:j + 1].strip()
    return None


def normalize_datalog_query(text: str) -> str:
    query = strip_markdown_wrapper(text)
    if query.startswith("{") and ":query" in query:
        return extract_query_from_map(query) or query
    return query


def is_block_entity(value) -> bool:
    return isinstance(value, dict) and "content" in value and ("uuid" in value or "id" in value)


def is_page_entity(value) -> bool:
    return (
        isinstance(value, dict)
        and "content" not in value
        and any(key in value for key in ("name", "originalName", "original-name"))
    )


def entity_key(entity: dict) -> tuple:
    return (
        entity.get("id"),
        entity.get("uuid"),
        entity.get("originalName") or entity.get("original-name") or entity.get("name"),
    )


def collect_result_entities(result) -> tuple[list[dict], list[dict]]:
    pages: list[dict] = []
    blocks: list[dict] = []
    seen_pages: set[tuple] = set()
    seen_blocks: set[tuple] = set()

    def add_page(page: dict) -> None:
        key = entity_key(page)
        if key not in seen_pages:
            seen_pages.add(key)
            pages.append(page)

    def add_block(block: dict) -> None:
        key = entity_key(block)
        if key not in seen_blocks:
            seen_blocks.add(key)
            blocks.append(block)

    def walk(value) -> None:
        if is_block_entity(value):
            add_block(value)
            return
        if is_page_entity(value):
            add_page(value)
            return
        if isinstance(value, list):
            for item in value:
                walk(item)
        elif isinstance(value, dict):
            for child in value.values():
                walk(child)

    walk(result)
    return pages, blocks


def read_query(args: argparse.Namespace, parser: argparse.ArgumentParser) -> str:
    query_text = " ".join(args.query_parts).strip() if args.query_parts else ""
    wants_stdin = args.stdin or query_text == "-" or args.file == "-"
    sources = sum(bool(source) for source in (query_text and query_text != "-", args.file and args.file != "-", wants_stdin))
    if sources > 1:
        parser.error("provide the Datalog query using only one source: argument, --file, or stdin")

    if args.file and args.file != "-":
        with open(os.path.expanduser(args.file), "r", encoding="utf-8") as f:
            query_text = f.read()
    elif wants_stdin:
        query_text = sys.stdin.read()
    elif not query_text and not sys.stdin.isatty():
        query_text = sys.stdin.read()

    query = normalize_datalog_query(query_text)
    if not query:
        parser.error("provide a Datalog query as an argument, with --file, or on stdin")
    return query


def get_token() -> str:
    token = os.environ.get("LOGSEQ_API_TOKEN")
    if not token:
        sys.exit("LOGSEQ_API_TOKEN is not set. Export your Logseq API token first.")
    return token


def run_block_command(argv: list[str]) -> None:
    parser = argparse.ArgumentParser(
        prog="logseq-cli",
        description=DESCRIPTION,
        epilog=EPILOG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("uuid", help="Block UUID — raw, ((uuid)) form, or any string containing one.")
    parser.add_argument("--host", default=os.environ.get("LOGSEQ_API_HOST", DEFAULT_HOST),
                        help=f"API host (default: {DEFAULT_HOST}, env: LOGSEQ_API_HOST).")
    parser.add_argument("--port", type=int,
                        default=int(os.environ.get("LOGSEQ_API_PORT", DEFAULT_PORT)),
                        help=f"API port (default: {DEFAULT_PORT}, env: LOGSEQ_API_PORT).")
    parser.add_argument("--depth", type=int, default=DEFAULT_DEPTH,
                        help=f"Max nesting depth to recurse (default: {DEFAULT_DEPTH}).")
    parser.add_argument("--json", action="store_true",
                        help="Print the fetched tree as JSON instead of markdown.")
    parser.add_argument("--header", action="store_true",
                        help="Prepend a header line with the source page name and uuid.")
    args = parser.parse_args(argv)

    token = get_token()
    uuid = extract_uuid(args.uuid)
    tree = fetch_block_tree(uuid, args.host, args.port, token, args.depth)
    if not tree:
        sys.exit(f"Block not found: {uuid}")

    if args.json:
        print(json.dumps(tree, indent=2, ensure_ascii=False))
        return

    if args.header:
        page = tree.get("page") or {}
        page_name = page_display_name(page)
        print(f"# {page_name}  ·  {uuid}\n")
    print("\n".join(render_block(tree)))


def run_query_command(argv: list[str]) -> None:
    parser = argparse.ArgumentParser(
        prog="logseq-cli query",
        description="Run a Logseq Datalog query and render returned page or block entities as markdown.",
        epilog=QUERY_EPILOG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("query_parts", nargs="*", help="Datalog query text, or '-' to read from stdin.")
    parser.add_argument("--file", "-f", help="Read the Datalog query from this file. Use '-' for stdin.")
    parser.add_argument("--stdin", action="store_true", help="Read the Datalog query from stdin.")
    parser.add_argument("--host", default=os.environ.get("LOGSEQ_API_HOST", DEFAULT_HOST),
                        help=f"API host (default: {DEFAULT_HOST}, env: LOGSEQ_API_HOST).")
    parser.add_argument("--port", type=int,
                        default=int(os.environ.get("LOGSEQ_API_PORT", DEFAULT_PORT)),
                        help=f"API port (default: {DEFAULT_PORT}, env: LOGSEQ_API_PORT).")
    parser.add_argument("--depth", type=int, default=DEFAULT_DEPTH,
                        help=f"Max nesting depth to render (default: {DEFAULT_DEPTH}).")
    parser.add_argument("--json", action="store_true",
                        help="Print fetched page/block trees as JSON instead of markdown.")
    parser.add_argument("--no-header", action="store_true",
                        help="Do not print page headers before rendered page results.")
    args = parser.parse_args(argv)

    token = get_token()
    query = read_query(args, parser)
    result = call_api("logseq.DB.customQuery", [query], args.host, args.port, token)
    if isinstance(result, dict) and result.get("error"):
        sys.exit(f"Logseq query error: {result['error']}")

    pages, blocks = collect_result_entities(result)
    if not pages and not blocks:
        sys.exit("Query returned no page or block entities. Use (pull ?p [*]) or (pull ?b [*]) in :find.")

    rendered_json = []
    rendered_markdown = []
    page_ids = {page.get("id") for page in pages if page.get("id") is not None}

    for page in pages:
        identifier = page_lookup_key(page)
        if identifier is None:
            continue
        page_blocks = fetch_page_blocks_tree(identifier, args.host, args.port, token, args.depth)
        rendered_json.append({"type": "page", "page": page, "blocks": page_blocks})
        rendered_markdown.append(render_page(page, page_blocks, include_header=not args.no_header))

    for block in blocks:
        page = block.get("page") or {}
        if page.get("id") in page_ids:
            continue
        identifier = block_lookup_key(block)
        if identifier is None:
            continue
        tree = fetch_block_tree(identifier, args.host, args.port, token, args.depth)
        if not tree:
            continue
        rendered_json.append({"type": "block", "block": tree})
        if not args.no_header:
            page_name = page_display_name(tree.get("page") or page)
            block_uuid = tree.get("uuid") or identifier
            rendered_markdown.append(f"# {page_name}  ·  {block_uuid}\n\n" + "\n".join(render_block(tree)))
        else:
            rendered_markdown.append("\n".join(render_block(tree)))

    if args.json:
        print(json.dumps(rendered_json, indent=2, ensure_ascii=False))
        return

    print("\n\n".join(markdown for markdown in rendered_markdown if markdown))


def main() -> None:
    argv = sys.argv[1:]
    if argv and argv[0] in {"query", "datalog"}:
        run_query_command(argv[1:])
    elif argv and argv[0] == "block":
        run_block_command(argv[1:])
    else:
        run_block_command(argv)


if __name__ == "__main__":
    main()