Drive Video Reader

1"""drive-video-reader — read a video file from Drive, output it as a Video file.23Migrated from the legacy monolithic template to the new multi-file format.4Privileged Drive IO is delegated to the gais.drive SDK; this node imports no5HTTP client and uses no os/shutil (both lint-blocked). The downloaded file is6stream-copied into /data/output with plain open() so the executor uploads it7as the Video output.8"""910from __future__ import annotations1112import json13import re14import sys15import traceback16from typing import Any1718from gais import Gais1920OUTPUT_DIR = "/data/output"2122_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)262728def _normalize_drive_item_id(value: Any) -> str:29    """Accept a bare UUID or a mention://drive_{file,folder}/<uuid> URI."""30    if not isinstance(value, str):31        raise ValueError(f"Drive item id must be a string, got {type(value).__name__}")32    s = value.strip()33    for prefix in ("mention://drive_file/", "mention://drive_folder/"):34        if s.startswith(prefix):35            s = s[len(prefix):]36            break37    if not _UUID_RE.match(s):38        raise ValueError(f"Drive item id is not a valid UUID: {value!r}")39    return s404142def _safe_output_name(name: Any, default: str) -> str:43    """Reduce a Drive-supplied name to a bare filename so it cannot escape44    OUTPUT_DIR (path traversal). The name comes from user-controlled DriveItem45    metadata; os/pathlib are lint-blocked, so strip directory components by46    hand and reject empty / dot names."""47    base = str(name or "").replace("\\", "/").rsplit("/", 1)[-1].strip()48    if not base or base in (".", ".."):49        return default50    return base515253def _copy_to_output(src_path: str, name: str) -> None:54    """Stream-copy the downloaded file into /data/output (os/shutil are55    lint-blocked, so use plain open() in 1 MiB chunks). `name` must already be56    a sanitized basename (see _safe_output_name)."""57    with open(src_path, "rb") as src, open(f"{OUTPUT_DIR}/{name}", "wb") as dst:58        while True:59            chunk = src.read(1024 * 1024)60            if not chunk:61                break62            dst.write(chunk)636465def process(inputs: dict[str, Any]) -> dict[str, Any]:66    """Validate the input, download via gais.drive, stage the Video output."""67    raw = inputs.get("driveItemId")68    if not raw:69        raise ValueError("Required input 'driveItemId' not provided")7071    # Single-video reader: reject multi-item arrays.72    if isinstance(raw, list):73        if len(raw) == 1:74            raw = raw[0]75        else:76            raise ValueError(77                f"Drive Video Reader accepts a single video file, got {len(raw)} items. "78                "Use one Drive Video Reader per video file."79            )80    elif not isinstance(raw, str):81        raise ValueError(f"driveItemId must be a string, got {type(raw).__name__}")8283    item_id = _normalize_drive_item_id(raw)8485    result = Gais.drive.download(item_id=item_id)86    file_name = _safe_output_name(result.metadata.get("name", "video.mp4"), "video.mp4")87    mime_type = result.metadata.get("mimeType", "") or ""88    if mime_type and not mime_type.startswith("video/"):89        raise ValueError(90            f"Selected Drive item is not a video file (mimeType: {mime_type}). "91            "Please select a video file."92        )9394    _copy_to_output(str(result.file("file")), file_name)95    return {"video": file_name}969798def main() -> None:99    try:100        envelope = json.loads(sys.stdin.read() or "{}")101        if not isinstance(envelope, dict):102            envelope = {}103        inputs = envelope.get("inputs", {}) or {}104        json.dump(process(inputs), sys.stdout)105    except Exception as e:106        print(107            json.dumps({108                "error": str(e),109                "errorType": type(e).__name__,110                "traceback": traceback.format_exc(),111            }),112            file=sys.stderr,113        )114        sys.exit(1)115116117if __name__ == "__main__":118    main()