給 AI 助理的 Colab 說明書

# 安裝 Google 官方的 Colab 指令列工具（Linux 與 macOS）
uv tool install google-colab-cli
colab version

# 第一次登入：用 oauth2，不要用 ADC（重做 ADC 會蓋掉使用者既有的 gcloud 憑證）
# 這一步要請人類在瀏覽器完成授權、把驗證碼貼回終端機
colab --auth=oauth2 usage

# 例：在 A100 上跑 job.py，先上傳 input.jpg，跑完下載 /content/out.mp4
python3 colab_detached.py job.py --gpu A100 --high-mem --upload input.jpg --download /content/out.mp4 --max-minutes 60

"""Run one long job on a Google Colab GPU without babysitting a fragile connection.

Why this exists: a single long `colab exec` keeps one websocket open for the whole job.
If your home network blips, the CLI can lose the connection and hang forever while the
VM keeps billing. This script instead starts the job *detached* on the VM, checks its
log with short commands, downloads the results, and always stops the VM.

Usage:
  python3 colab_detached.py job.py --gpu A100 --upload input.jpg --download /content/out.mp4
  python3 colab_detached.py job.py --gpu T4 --max-minutes 30

Requirements: the official Colab CLI (`uv tool install google-colab-cli`), already logged in
with `colab --auth=oauth2 usage`. Tested with Colab CLI 0.7.4 on macOS, 2026-09-28.
"""
import argparse
import subprocess
import sys
import time
from pathlib import Path

AUTH = "--auth=oauth2"


def colab(args, stdin=None, timeout=120):
    """Run one colab CLI command with a hard timeout, so a dropped connection can't hang us."""
    try:
        r = subprocess.run(["colab", AUTH, *args], input=stdin, capture_output=True, text=True, timeout=timeout)
        return r.returncode, (r.stdout + r.stderr).strip()
    except subprocess.TimeoutExpired:
        return -9, f"timed out after {timeout}s"


def remote(session, code, timeout=90):
    return colab(["exec", "-s", session], stdin=code, timeout=timeout)


def usage_line():
    _, out = colab(["usage"], timeout=60)
    return " | ".join(out.splitlines()[-3:])


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("job", help="local Python script to run on the VM (runs from /content)")
    ap.add_argument("--gpu", default="T4", help="CPU, T4, L4, G4, A100 or H100 (A100 High-RAM measured 6.77 units/hour)")
    ap.add_argument("--high-mem", action="store_true", help="request the high-RAM machine shape")
    ap.add_argument("--session", default="detached-job")
    ap.add_argument("--upload", action="append", default=[], help="local file to copy into /content (repeatable)")
    ap.add_argument("--download", action="append", default=[], help="remote file to fetch when the job ends (repeatable)")
    ap.add_argument("--max-minutes", type=float, default=60, help="hard wall-clock limit; the VM is stopped after this")
    a = ap.parse_args()

    # Only one GPU VM at a time is allowed on consumer plans; never reuse a live session by accident.
    _, sessions = colab(["sessions"], timeout=60)
    if any(line.startswith(f"[{a.session}]") for line in sessions.splitlines()):
        sys.exit(f"session {a.session} is already running; stop it first with: colab {AUTH} stop -s {a.session}")

    print("before:", usage_line(), flush=True)
    new_args = ["new", "-s", a.session] + ([] if a.gpu.upper() == "CPU" else ["--gpu", a.gpu]) + (["--high-mem"] if a.high_mem else [])
    rc, out = colab(new_args, timeout=600)
    print(out.splitlines()[-1] if out else out, flush=True)
    if rc != 0:
        sys.exit("could not allocate a VM (no charge). Try another --gpu or wait a few minutes.")

    t0 = time.monotonic()
    try:
        # Warm up the kernel BEFORE uploading: on 2026-09-28 the first exec after an upload
        # lost its connection twice in a row on A100 VMs; a warm-up exec first never did.
        for _ in range(4):
            rc, out = remote(a.session, "print('warm', 1 + 1)")
            if rc == 0 and "warm 2" in out:
                break
            time.sleep(20)
        else:
            sys.exit("kernel never answered")

        for f in [a.job, *a.upload]:
            rc, out = colab(["upload", "-s", a.session, f, f"/content/{Path(f).name}"], timeout=600)
            print(out.splitlines()[-1] if out else out, flush=True)
            if rc != 0:
                sys.exit(f"upload failed: {f}")

        job = f"/content/{Path(a.job).name}"
        start = (
            "import os, subprocess, sys\n"
            "if not os.path.exists('/content/job.pid'):\n"
            "    log = open('/content/job.log', 'a')\n"
            f"    p = subprocess.Popen([sys.executable, '-u', {job!r}], cwd='/content', stdout=log,\n"
            "                         stderr=subprocess.STDOUT, start_new_session=True)\n"
            "    open('/content/job.pid', 'w').write(str(p.pid))\n"
            "print('started', open('/content/job.pid').read())\n"
        )
        for _ in range(4):
            rc, out = remote(a.session, start)
            if rc == 0 and "started" in out:
                break
            time.sleep(20)
        else:
            sys.exit("could not start the job")
        print("job started; checking every 45 s", flush=True)

        poll = (
            "import os\n"
            "t = open('/content/job.log').read() if os.path.exists('/content/job.log') else ''\n"
            "pid = int(open('/content/job.pid').read())\n"
            "alive = os.path.exists(f'/proc/{pid}') and open(f'/proc/{pid}/stat').read().split()[2] != 'Z'\n"
            "tail = [l for l in t.splitlines() if l.strip()][-1:] or ['']\n"
            "print('ALIVE' if alive else 'EXITED', '|', tail[0][-160:])\n"
        )
        last = ""
        while True:
            if time.monotonic() - t0 > a.max_minutes * 60:
                print("wall-clock limit reached; stopping the VM", flush=True)
                return 1
            time.sleep(45)
            rc, out = remote(a.session, poll)
            status = out.splitlines()[-1] if out else ""
            if rc != 0:
                continue  # a dropped check is harmless; the job keeps running on the VM
            if status != last:
                print(f"[{(time.monotonic() - t0) / 60:.1f} min] {status}", flush=True)
                last = status
            if status.startswith("EXITED"):
                break

        _, tail = remote(a.session, "print(open('/content/job.log').read()[-3000:])")
        Path("job.log").write_text(tail)
        print("remote log saved to job.log", flush=True)
        for f in a.download:
            rc, out = colab(["download", "-s", a.session, f, Path(f).name], timeout=900)
            print(out.splitlines()[-1] if out else out, flush=True)
        return 0
    finally:
        rc, out = colab(["stop", "-s", a.session], timeout=300)
        print(out.splitlines()[-1] if out else out, flush=True)
        print(f"total {(time.monotonic() - t0) / 60:.1f} min;", "after:", usage_line(), flush=True)


if __name__ == "__main__":
    sys.exit(main())

subject_definitions:
<Picture 1> is the first frame of [Shot 1], showing <一句話描述你畫好的第一格>.
<Subject 1> is <主角> in <Picture 1>, with <三到五個外觀特徵>.

summary:
[keyframe completion + reference generation] The target video starts exactly on <Picture 1>. <一句話講整支影片發生什麼>.

retention_analysis:
<Picture 1> (appears in [Shot 1]): fully_preserved - used unchanged as the first frame, keeping <畫風>.
<Subject 1> (appears in [Shot 1], [Shot 2]): fully_preserved - <要保留的外觀特徵> are retained.

detailed_description:
The target video is <畫風一句話，每一支都逐字重貼同一句>.
[Shot 1] The video opens exactly on <Picture 1>. <這一鏡唯一的動作>.
[Shot 2] At 00:03.500, the shot cuts to <景別> of <Subject 1> (S1). It says in <聲音描述> with a Taiwanese Mandarin accent, <d>[Chinese] <台詞>。</d> <說完之後的一個動作>.

overall_soundscape:
<環境音與動作音>

non_diegetic_music:
<配樂一句話，沒有就寫 N/A>