strukto-ai--mirage/python/mirage/core/dify/path.py

from dataclasses import dataclass

from mirage.cache.index import IndexCacheStore, IndexEntry
from mirage.core.dify.tree import ensure_tree
from mirage.types import PathSpec
from mirage.utils.errors import enoent
from mirage.utils.key_prefix import mount_prefix_of


@dataclass(frozen=True)
class ResolvedDifyPath:
    virtual_key: str
    mount_prefix: str
    is_dir: bool
    entry: IndexEntry | None = None


async def resolve_path(accessor, path: PathSpec,
                       index: IndexCacheStore) -> ResolvedDifyPath:
    if isinstance(path, str):
        path = PathSpec(virtual=path,
                        directory=path,
                        resource_path=path.strip("/"))
    mount_prefix = mount_prefix_of(path.virtual, path.resource_path) or ""
    await ensure_tree(accessor, index, mount_prefix)
    virtual_key = virtual_key_for(path)
    result = await index.get(virtual_key)
    if result.entry is not None:
        return ResolvedDifyPath(
            virtual_key=virtual_key,
            mount_prefix=mount_prefix,
            is_dir=result.entry.resource_type == "folder",
            entry=result.entry,
        )
    listing = await index.list_dir(virtual_key)
    if listing.entries is not None:
        return ResolvedDifyPath(virtual_key=virtual_key,
                                mount_prefix=mount_prefix,
                                is_dir=True)
    raise enoent(path)


def virtual_key_for(path: PathSpec) -> str:
    raw = path.directory if path.pattern else path.virtual
    prefix = mount_prefix_of(path.virtual, path.resource_path) or ""
    if prefix:
        root = prefix.rstrip("/") or "/"
        if raw == root or raw.startswith(root + "/"):
            return raw.rstrip("/") or root
        rest = raw.strip("/")
        if not rest:
            return root
        return root + "/" + rest
    stripped = raw.strip("/")
    return "/" + stripped if stripped else "/"