mycelium Integration

# Curio export for personal search

Adapted by Zengobi from John's mycelium integration, shared with permission.
See the walkthrough: https://www.zengobi.com/curio/cli/examples/?item=mycelium
Video: https://vimeo.com/1224764392/803afa7ec8

## Run

Requires Python 3.10+ and, for a live export, macOS with the Curio CLI installed
from Curio 34 or later. No Python packages are required. Save the project first.

```bash
python3 curio_export.py --project "/path/to/Notes.curio" --name notes --output notes.jsonl
```

Alternatively, pass `--export Notes.json` instead of `--project` to consume the
JSON envelope from the CLI query shown in the walkthrough. Conversion of saved
JSON does not require macOS or the CLI. Use an existing output directory.

## Output

Each line is a JSON object. `note` records include a project-namespaced ID,
title, body, tags, section/organizer paths, dates, and a Curio deep link in `url`.
Figures are grouped by idea space, following parent IDs for nested collections.
Text stays in export order; collection text may overlap its children's text.
Dates retain Curio's source representation; the script does not assume UTC.

`asset_reference` records include the asset ID, exported file path and kind,
referencing figure IDs, and the first referencing figure's Curio link. They are
deduplicated by asset ID (falling back to file path), not by file contents.
The script does not open attachment files or follow aliases.

## Connect an indexer

Index note title, body and tags; retain `url` as the result's open-in-Curio target.
For attachments, add your own format-specific text extraction, OCR or
transcription. Validate resolved paths against your project mount before reading
files, skip external aliases, and map host paths when working in a container.
`asset_file` describes the exported environment, not necessarily the current one.

Choose a unique, stable `--name` for each project. Reuse it on subsequent exports
so an indexer can upsert by `external_id`. This script exports a full snapshot;
it implements neither incremental indexing nor index deletion. Before deleting
old index entries, validate the snapshot and check for unexpected count drops.
The script rejects failed/malformed responses and exports without idea spaces,
and atomically replaces the output only after conversion succeeds.

## Scope

This is a standalone adaptation, not John's original mycelium adapter. It omits
his service-specific imports, extractors, caches, database, scheduler, deployment
configuration, rsync, and staged-copy fallback. It calls the CLI directly with a
15-minute timeout. It does not reproduce his network-denying sandbox wrapper;
normal CLI behavior, including any update checks, still applies.

The CLI reads the saved project without modifying it. The only persistent output
created by this script is the specified JSONL file, which contains your exported
project data. Nothing in this script uploads that output.

#!/usr/bin/env python3
"""Export Curio notes and attachment references for a search index.
Adapted by Zengobi from John's mycelium integration, shared with permission.
Python 3.10+, standard library only. See README.md for scope and usage.
"""
from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile

FIELDS = ('id,kind,title,plain_text,note,tags,section_path,organizer_path,link,'
          'modified_date,added_date,asset_id,asset_file,asset_file_kind,url,parent_id')
SPACE = 'organizerideaspace'
ORGANIZERS = {SPACE, 'organizersection', 'organizerfolder', 'organizeralias', 'organizerdocument'}
TEXT_KINDS = {'text', 'indexcard', 'list', 'table', 'stack', 'mindmap', 'pinboard',
              'group', 'album', 'tabloid', 'matrix', 'ideaspacelink'}
LINK = re.compile(r'curio://[A-Za-z0-9._~%+-]+\?[A-Za-z0-9=&_.%~+-]+')


def clean(value):
    return value.strip() if isinstance(value, str) else ''


def tags(item):
    values = item.get('tags')
    return {v.strip() for v in values if isinstance(v, str) and v.strip()} if isinstance(values, list) else set()


def link(item):
    value = clean(item.get('link'))
    return value if LINK.fullmatch(value) else None


def load_items(stream):
    raw = stream.read(64_000_001)
    if len(raw) > 64_000_000:
        raise ValueError('Export exceeds 64 MB')
    data = json.loads(raw)
    if not isinstance(data, dict) or data.get('response_ok') is not True:
        raise ValueError('CLI export did not report response_ok: true')
    result = data.get('result')
    items = result.get('items') if isinstance(result, dict) else None
    if not isinstance(items, list) or len(items) > 200_000:
        raise ValueError('Missing or oversized result.items')
    if any(not isinstance(i, dict) or not isinstance(i.get('id'), str) for i in items):
        raise ValueError('Every item must have a string id')
    return items


def documents(items, project):
    by_id = {i['id']: i for i in items}
    spaces = {i['id']: i for i in items if i.get('kind') == SPACE}
    groups = {sid: [] for sid in spaces}
    assets = {}
    for item in items:
        if item.get('kind') in ORGANIZERS:
            continue
        # Follow ancestors for figures inside collections; guard malformed cycles.
        parent = item.get('parent_id')
        seen = set()
        while isinstance(parent, str) and parent not in spaces and parent in by_id and parent not in seen:
            seen.add(parent)
            parent = by_id[parent].get('parent_id')
        if isinstance(parent, str) and parent in groups:
            groups[parent].append(item)
        aid = clean(item.get('asset_id')) or clean(item.get('asset_file'))
        if aid and item.get('kind') not in TEXT_KINDS:
            asset = assets.setdefault(aid, {
                'type': 'asset_reference', 'external_id': f'{project}:asset:{aid}',
                'project': project, 'title': clean(item.get('title')),
                'asset_id': item.get('asset_id'), 'asset_file': item.get('asset_file'),
                'asset_file_kind': item.get('asset_file_kind'), 'url': link(item),
                'figure_ids': [],
            })
            asset['figure_ids'].append(item['id'])
    for sid, space in spaces.items():
        parts = [clean(space.get('note'))]
        all_tags = tags(space)
        dates = [clean(space.get('modified_date'))]
        for figure in groups[sid]:
            parts.extend([clean(figure.get('plain_text')) or clean(figure.get('title')),
                          clean(figure.get('note')), clean(figure.get('url'))])
            all_tags.update(tags(figure))
            dates.append(clean(figure.get('modified_date')))
        yield {
            'type': 'note', 'external_id': f'{project}:is:{sid}', 'project': project,
            'title': clean(space.get('title')) or 'Untitled',
            'body': '\n\n'.join(p for p in parts if p), 'tags': sorted(all_tags),
            'section_path': clean(space.get('section_path')),
            'organizer_path': clean(space.get('organizer_path')),
            'url': link(space), 'created_at': space.get('added_date'),
            'updated_at': max(dates) or None,
        }
    yield from assets.values()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument('--project', type=Path, help='Read a saved .curio project on this Mac')
    source.add_argument('--export', type=Path, help='Read a previously saved CLI JSON response')
    parser.add_argument('--name', required=True, help='Unique, stable project name for index IDs')
    parser.add_argument('--output', type=Path, required=True, help='Destination JSON Lines file')
    args = parser.parse_args()
    try:
        if args.project:
            project = args.project.expanduser().resolve(strict=True)
            if project.suffix.lower() != '.curio' or not project.is_dir():
                raise ValueError('--project must be a .curio project directory')
            # Capture to a temporary file so a large response does not fill memory.
            with tempfile.TemporaryFile() as response:
                subprocess.run(['curio', 'query', 'not kind=nothing', '--fields', FIELDS,
                                '--project', str(project)], stdout=response, check=True, timeout=900)
                response.seek(0)
                items = load_items(response)
        else:
            with args.export.expanduser().open('rb') as response:
                items = load_items(response)
        if not any(i.get('kind') == SPACE for i in items):
            raise ValueError('No idea spaces found; check the project and file access permissions')
        output = args.output.expanduser().resolve()
        if args.project and (output == project or project in output.parents):
            raise ValueError('Choose an output location outside the Curio project')
        if args.export and output == args.export.expanduser().resolve():
            raise ValueError('Output must differ from the input export')
        # Replace only after a complete successful conversion. The temp file is private.
        temporary = None
        try:
            with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=output.parent,
                                             delete=False) as stream:
                temporary = Path(stream.name)
                count = 0
                for document in documents(items, args.name):
                    stream.write(json.dumps(document, ensure_ascii=False) + '\n')
                    count += 1
            os.replace(temporary, output)
        finally:
            if temporary is not None:
                temporary.unlink(missing_ok=True)
        print(f'Wrote {count} records to {output}', file=sys.stderr)
    except (OSError, ValueError, subprocess.SubprocessError) as exc:
        parser.exit(1, f'Export failed: {exc}\n')


if __name__ == '__main__':
    main()