Drive Folder Reader

1"""drive-folder-reader — list document DriveItem ids in a Drive folder.23Migrated from the legacy monolithic template to the new multi-file format.4Privileged Drive IO is delegated to the gais.drive SDK (Gais.drive.list);5this node imports no HTTP client. Returns DriveItem ids (NOT Document refIds)6so downstream nodes can validate permissions via the DriveItem path.78`files` adds each file's sub-folder path, so a loop can save its result into9the same sub-folder elsewhere (src/a/b/c/d.pdf -> dest/a/b/c/d.it.pdf).10"""1112from __future__ import annotations1314import json15import re16import sys17import traceback18from typing import Any1920from gais import Gais2122_UUID_RE = re.compile(23    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",24    re.IGNORECASE,25)2627# Deepest folder chain walked for one file. Past it, or on a cycle, the28# file's path is unknown.29_MAX_DEPTH = 100303132def _normalize_drive_item_id(value: Any) -> str:33    """Accept a bare UUID or a mention://drive_{file,folder}/<uuid> URI."""34    if not isinstance(value, str):35        raise ValueError(f"Drive item id must be a string, got {type(value).__name__}")36    s = value.strip()37    for prefix in ("mention://drive_file/", "mention://drive_folder/"):38        if s.startswith(prefix):39            s = s[len(prefix):]40            break41    if not _UUID_RE.match(s):42        raise ValueError(f"Drive item id is not a valid UUID: {value!r}")43    return s444546def _id_key(value: Any) -> str | None:47    """An id for comparison: UUIDs compare case-insensitively."""48    return value.lower() if isinstance(value, str) and value else None495051def _segment(name: Any) -> str | None:52    """One folder name as one path level, or None when it has no usable name.5354    A "/" inside the name becomes "-", so the path splits back into the right55    levels. So does a "\\": the Drive ensure-path route refuses both.56    """57    if not isinstance(name, str) or not name.strip():58        return None59    return name.replace("/", "-").replace("\\", "-")606162def _folder_path(63    parent_id: Any,64    root_key: str,65    folders: dict[str, dict[str, Any]],66    cache: dict[str, str | None],67) -> str | None:68    """Folder names from below the picked folder down to `parent_id`.6970    Walks parentId up through `folders` until it reaches the picked folder,71    whose own name is not included. Returns "a/b/c", "" for a file directly in72    the picked folder, or None when the chain cannot be resolved: a parent73    missing from `folders` (e.g. not readable), a folder without a name, a74    cycle, or a chain deeper than _MAX_DEPTH.75    """76    start = _id_key(parent_id)77    if start is not None and start in cache:78        return cache[start]7980    parts: list[str] = []81    seen: set[str] = set()82    current = start83    path: str | None = None84    while True:85        if current == root_key:86            path = "/".join(reversed(parts))87            break88        if current is None or current in seen or len(parts) >= _MAX_DEPTH:89            break90        seen.add(current)91        folder = folders.get(current)92        segment = _segment(folder.get("name")) if folder else None93        if segment is None:94            break95        parts.append(segment)96        current = _id_key(folder.get("parentId"))9798    if start is not None:99        cache[start] = path100    return path101102103def _sort_key(entry: dict[str, Any]) -> tuple:104    """Path level by level, then name, ignoring case; id breaks ties."""105    path = entry["path"]106    levels = tuple(part.casefold() for part in path.split("/")) if path else ()107    return (levels, entry["name"].casefold(), entry["name"], entry["driveItemId"])108109110def _build_files(111    file_items: list[dict[str, Any]],112    folder_items: list[dict[str, Any]],113    folder_id: str,114) -> list[dict[str, Any]]:115    """Every file with its sub-folder path, sorted by path then name."""116    root_key = folder_id.lower()117    # `path descendantOf` may return the picked folder itself; skip it.118    folders: dict[str, dict[str, Any]] = {}119    for folder in folder_items:120        key = _id_key(folder.get("id"))121        if key is not None and key != root_key:122            folders[key] = folder123124    cache: dict[str, str | None] = {}125    files: list[dict[str, Any]] = []126    for item in file_items:127        if not item.get("id"):128            continue129        name = item.get("name")130        entry: dict[str, Any] = {131            "driveItemId": item["id"],132            "name": name if isinstance(name, str) else "",133        }134        path = _folder_path(item.get("parentId"), root_key, folders, cache)135        if path is None:136            # Never drop the file: it goes at the top, flagged.137            entry["path"] = ""138            entry["pathUnknown"] = True139        else:140            entry["path"] = path141        files.append(entry)142143    files.sort(key=_sort_key)144    return files145146147def process(inputs: dict[str, Any]) -> dict[str, Any]:148    """List every descendant file/document of the folder via gais.drive.list.149150    `driveItemIds` keeps the files query's own order. `files` adds each151    file's sub-folder path, from a second query for the descendant folders.152    """153    folder_id = inputs.get("driveItemId")154    if not folder_id:155        raise ValueError("driveItemId input is required")156157    folder_id = _normalize_drive_item_id(folder_id)158159    # Recursive query: every descendant that is a file/document (refId != null),160    # NOT subfolders. The Drive API applies the user's permissions automatically161    # (gais.drive carries the user token).162    result = Gais.drive.list(163        query=f"path descendantOf '{folder_id}' and refId!=null",164        page_size=1000,165    )166167    # Return DriveItem ids (NOT Document refIds) so downstream nodes can168    # validate permissions via the DriveItem path.169    file_items = result.data.get("items", [])170    drive_item_ids = [item["id"] for item in file_items if item.get("id")]171172    # The descendant folders (refId = null), to turn each file's parentId173    # into a folder path. A failure propagates, like the files query.174    folder_result = Gais.drive.list(175        query=f"path descendantOf '{folder_id}' and refId=null",176        page_size=1000,177    )178    folder_items = folder_result.data.get("items", [])179180    return {181        "driveItemIds": drive_item_ids,182        "files": _build_files(file_items, folder_items, folder_id),183    }184185186def main() -> None:187    try:188        envelope = json.loads(sys.stdin.read() or "{}")189        if not isinstance(envelope, dict):190            envelope = {}191        inputs = envelope.get("inputs", {}) or {}192        json.dump(process(inputs), sys.stdout)193    except Exception as e:194        print(195            json.dumps({196                "error": str(e),197                "errorType": type(e).__name__,198                "traceback": traceback.format_exc(),199            }),200            file=sys.stderr,201        )202        sys.exit(1)203204205if __name__ == "__main__":206    main()207