save-file

1"""save-file — persist a workspace file or text blob into the user's Drive.23Takes EITHER an upstream File-port input OR an inline Text input, saves it as a4Drive item under a folder chosen via the ``drive-item-picker`` widget, then5passes the source file/text straight through so downstream nodes can keep using6it. The upload + drive registration is delegated to the ``gais.drive`` SDK7(which holds the privileged ``USER_TOKEN`` and talks to the www-emblema Drive8API); this node stays sandbox-clean and imports no HTTP client.910Idempotent save: the DriveItem id is set to this workspace node's id11(``context["node_id"]``), so re-running the node UPSERTS the same item instead12of creating a duplicate every run.1314Contract15--------16- Reads one JSON envelope from stdin: {"inputs": {...}, "config": {...},17  "context": {...}, "inputSchema": {...}}.18- All declared schema fields arrive under ``envelope["inputs"]``.19- Exactly one of ``file`` / ``text`` must be provided.20- ``destinationFolder`` is a Drive folder UUID (drive-item-picker widget).21- ``name`` is the file name to store; defaults to the source filename for a22  file input and ``note.txt`` for text.23- Writes {"driveItemId", "name", "file", "text"} to stdout on success: the24  saved item id + stored name, plus the source file/text passed through25  unchanged (the unused one is empty).2627HARD-FAIL on any save error (exit 1). A sink node must never report a false28"saved": validation problems, permission denials, quota, and >50 MB payloads29all surface as a non-zero exit with the underlying message on stderr.30"""3132from __future__ import annotations3334import json35import sys36import traceback37from typing import Any3839from gais import Gais404142INPUT_DIR = "/data/input"43OUTPUT_DIR = "/data/output"444546def _extension_of(filename: str | None) -> str:47    """Return the extension (without the dot) of a filename, or "".4849    os/pathlib are lint-blocked, so split by hand. Strips directory components50    first, then treats the text after the last dot as the extension — but only51    when the dot isn't the leading char (so a hidden file like ".env" or a52    dotless name like "潮人图片 2" reports no extension).53    """54    if not filename:55        return ""56    base = filename.replace("\\", "/").rsplit("/", 1)[-1]57    if "." not in base[1:]:58        return ""59    return base.rsplit(".", 1)[1]606162def _resolve_name(raw_name: Any, source_filename: str | None, is_text: bool) -> str:63    """Pick the stored file name and guarantee it has an extension.6465    explicit name > source filename > default. A widget-supplied name often66    omits the extension (e.g. "潮人图片 2"), which would store the file67    extensionless and break the editor/preview's type detection — so when the68    chosen name has no extension we borrow one: the source file's extension for69    a file input, or ".md" for text.70    """71    if isinstance(raw_name, str) and raw_name.strip():72        name = raw_name.strip()73    elif source_filename:74        name = source_filename75    else:76        name = "note" if is_text else "file"7778    if _extension_of(name):79        return name80    if is_text:81        return f"{name}.md"82    src_ext = _extension_of(source_filename)83    return f"{name}.{src_ext}" if src_ext else name848586def _copy_to_output(src_path: str, name: str) -> None:87    """Stream-copy a staged input file into /data/output so it passes through as88    the File output. os/shutil are lint-blocked, so copy with plain open() in89    1 MiB chunks (never read the whole file into the 512 MB container)."""90    with open(src_path, "rb") as src, open(f"{OUTPUT_DIR}/{name}", "wb") as dst:91        while True:92            chunk = src.read(1024 * 1024)93            if not chunk:94                break95            dst.write(chunk)969798def process(inputs: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:99    """Validate inputs, save to Drive via gais.drive, pass the source through.100101    Raises ``ValueError`` on bad input (programmer/wiring error); the privileged102    network work is delegated to ``gais.drive.save`` and any SDK exception103    propagates (hard-fail).104    """105    file_name = inputs.get("file")106    text = inputs.get("text")107108    has_file = isinstance(file_name, str) and file_name.strip() != ""109    has_text = isinstance(text, str) and text != ""110111    if has_file and has_text:112        raise ValueError("provide exactly one of 'file' or 'text', not both")113    if not has_file and not has_text:114        raise ValueError("no input: connect a 'file' or provide 'text'")115116    folder_id = inputs.get("destinationFolder")117    if not isinstance(folder_id, str) or not folder_id.strip():118        raise ValueError("'destinationFolder' is required — pick a Drive folder")119    folder_id = folder_id.strip()120121    name = _resolve_name(122        inputs.get("name"),123        source_filename=(file_name if has_file else None),124        is_text=has_text,125    )126127    # Use the workspace node id as the DriveItem id so re-runs upsert the same128    # item instead of creating duplicates. Absent (e.g. a standalone test129    # harness) → None → the backend creates a fresh item.130    item_id = context.get("node_id") or None131132    if has_file:133        # The executor stages File-port inputs at /data/input/<filename>.134        result = Gais.drive.save(135            folder_id=folder_id,136            name=name,137            file=f"{INPUT_DIR}/{file_name}",138            item_id=item_id,139        )140    else:141        result = Gais.drive.save(142            folder_id=folder_id,143            name=name,144            text=text,145            item_id=item_id,146        )147148    md = result.metadata149    output: dict[str, Any] = {150        "driveItemId": md.get("itemId", ""),151        "name": md.get("name", name),152        "file": "",153        "text": "",154    }155    # Pass the source data through unchanged so downstream nodes can keep using156    # it: a File input is re-staged to /data/output; text is echoed.157    if has_file:158        src_name = str(file_name)159        _copy_to_output(f"{INPUT_DIR}/{src_name}", src_name)160        output["file"] = src_name161    else:162        output["text"] = text163    return output164165166def main() -> None:167    try:168        envelope = json.loads(sys.stdin.read() or "{}")169        if not isinstance(envelope, dict):170            envelope = {}171        inputs: dict[str, Any] = envelope.get("inputs", {}) or {}172        context: dict[str, Any] = envelope.get("context", {}) or {}173        output = process(inputs, context)174        json.dump(output, sys.stdout)175    except Exception as e:  # hard-fail: a save that didn't happen must be loud176        error_payload = {177            "error": str(e),178            "errorType": type(e).__name__,179            "traceback": traceback.format_exc(),180        }181        print(json.dumps(error_payload), file=sys.stderr)182        sys.exit(1)183184185if __name__ == "__main__":186    main()187