Function prepare_cloud_import

openkb/indexer.py:290–336

Fetch a PageIndex Cloud doc and resolve its wiki name WITHOUT writing. Cloud fetch + collision-resistant name resolution only — no KB mutation — so the caller knows ``doc_name`` before writing and can snapshot just this doc's paths instead of copying the whole summaries/sources trees. N

(doc_id: str, kb_dir: Path, path_key: str)

def prepare_cloud_import(doc_id: str, kb_dir: Path, path_key: str) -> CloudImportData:
    """Fetch a PageIndex Cloud doc and resolve its wiki name WITHOUT writing.

    Cloud fetch + collision-resistant name resolution only — no KB mutation —
    so the caller knows ``doc_name`` before writing and can snapshot just this
    doc's paths instead of copying the whole summaries/sources trees. Name
    resolution reads the registry but does not mutate it.
    """
    from openkb.converter import resolve_doc_name_from_key
    from openkb.state import HashRegistry

    pageindex_api_key = os.environ.get("PAGEINDEX_API_KEY", "")
    if not pageindex_api_key:
        raise RuntimeError(
            "Importing from PageIndex Cloud requires the PAGEINDEX_API_KEY environment variable."
        )

    client = PageIndexClient(api_key=pageindex_api_key)
    col = client.collection()

    doc = col.get_document(doc_id, include_text=True)
    cloud_name: str = doc.get("doc_name") or doc_id
    description: str = doc.get("doc_description", "")
    structure: list = doc.get("structure", [])

    registry = HashRegistry(kb_dir / ".openkb" / "hashes.json")
    stem = _cloud_display_stem(cloud_name, doc_id)
    doc_name = resolve_doc_name_from_key(stem, path_key, registry)

    tree = {
        "doc_name": cloud_name,
        "doc_description": description,
        "structure": structure,
    }

    all_pages = _fetch_cloud_pages(col, doc_id)
    if not all_pages:
        raise RuntimeError(f"No page content returned from PageIndex Cloud for doc_id={doc_id}")

    return CloudImportData(
        doc_id=doc_id,
        doc_name=doc_name,
        cloud_name=cloud_name,
        description=description,
        tree=tree,
        all_pages=all_pages,
    )


def import_cloud_document(doc_id: str, kb_dir: Path, path_key: str) -> CloudImportResult: