Cloud Imports (GCS / S3 / Cloudflare R2 / Hugging Face Buckets)

setup = client.cloud_integrations.get_gcs_storage_transfer_setup()
print(setup["google_service_account_email"])

RECORDING_BUCKET="zendar-batch-bucket-us"
NOMADIC_TRANSFER_ACCOUNT="<google_service_account_email returned above>"
for role in roles/storage.objectViewer roles/storage.legacyBucketReader; do
  gcloud storage buckets add-iam-policy-binding "gs://$RECORDING_BUCKET" \
    --member="serviceAccount:$NOMADIC_TRANSFER_ACCOUNT" \
    --role="$role" --condition=None
done

job = client.upload(
    "gs://zendar-batch-bucket-us/segments/segment-0042/",
    source_format="zendar_parquet",
    folder="2026-09-21",
)
print(job["import_job_id"], job["video_id"])

job = client.upload(
    [
        "gs://zendar-batch-bucket-us/segments/segment-0042/",
        "gs://zendar-batch-bucket-us/segments/segment-0043/",
    ],
    source_format="zendar_parquet",
    folder="2026-09-21",
)
for video in job["videos"]:
    print(video["source_uri"], video["video_id"])

jobs = []
for i in range(0, len(dirs), 1000):
    job = client.upload(
        dirs[i : i + 1000],
        source_format="zendar_parquet",
        folder="2026-09-21",
    )
    jobs.append(job["import_job_id"])

job = client.upload(
    [
        {"source_uri": "gs://zendar-batch-bucket-us/segments/segment-0042/", "name": "Morning drive"},
        {"source_uri": "gs://zendar-batch-bucket-us/segments/segment-0043/", "folder": "2026-09-22"},
    ],
    source_format="zendar_parquet",
    folder="2026-09-21",
)

status = client.get_import_job(job["import_job_id"])
print(status["status"], status["summary"])

page = client.get_import_job_videos(job["import_job_id"], limit=1000)
for video in page["videos"]:
    print(video["name"], video["import_status"], video["progress"], video["error"])

# Optional: block until every input finishes (default timeout: 24 hours).
result = client.wait_for_import_job(job["import_job_id"])

failed_ids = [v["video_id"] for v in page["videos"] if (v["error"] or {}).get("retryable")]
if failed_ids:
    client.retry_import_job(job["import_job_id"], video_ids=failed_ids)

# Single cloud multi-view set
multi = client.upload(
  {
    "front": "s3://drive-monitor/uploads/trip-042/front.mp4",
    "left": "s3://drive-monitor/uploads/trip-042/left.mp4",
    "right": "s3://drive-monitor/uploads/trip-042/right.mp4",
  },
  folder="fleet_uploads",
  scope="org",
)
# Returns: {"import_job_id": "ij_xxx", "status": "importing"}

upload = client.upload(
    [
        "s3://drive-monitor-archive/2024-09-01/front.mp4",
        "s3://drive-monitor-archive/2024-09-01/rear.mp4",
    ],
    wait_for_uploaded=False,
)

job = client.get_import_job(upload["import_job_id"])
print(job["job_id"], job["total"])

integration = client.cloud_integrations.add_hf_bucket(
    name="HF footage",
    bucket="JohnnyMnenonic/test",
    token="hf_xxx",
    prefix="incoming/",
)

result = client.upload(
    "hf://buckets/JohnnyMnenonic/test/incoming/front.mp4",
    integration_id=integration["id"],
    wait_for_uploaded=False,
)

upload = client.upload(
    [
        "gs://drive-monitor/uploads/trip-042/front.mp4",
        "gs://drive-monitor/uploads/trip-042/rear.mp4",
    ],
    wait_for_uploaded=False,
)

cursor = None  # First page: no cursor
while True:
    result = client.get_import_job_videos(
        upload["import_job_id"],
        limit=500,
        cursor=cursor,  # Pass cursor from previous response
    )

    print(result["import_job_id"])
    print(result["video_count"], result["has_more"], result["next_cursor"])  # page_size, more pages?, next anchor
    print(result["videos"][:3])  # [{video_id, status, import_source_uri}, ...]

    if not result["has_more"]:
        break
    cursor = result["next_cursor"]  # Use this value for the next page

import time

# 1. Start the cloud import — returns an import_job_id, NOT a video_id
result = client.upload([
    "s3://my-bucket/videos/clip_001.mp4",
    "s3://my-bucket/videos/clip_002.mp4",
    "s3://my-bucket/videos/clip_003.mp4",
])
print(result)
# {"import_job_id": "ij_a1b2c3d4e5f6", "status": "importing"}

job_id = result["import_job_id"]

# 2. Poll import-job videos until returned IDs reach the requested total
#    and each returned row is terminal.
TERMINAL = {"UPLOADED", "UPLOADING_FAILED"}
seen = {}
while True:
    job = client.get_import_job(job_id)
    target_total = int(job.get("total") or 0)
    cursor = None
    while True:
        page = client.get_import_job_videos(job_id, limit=500, cursor=cursor)
        for row in page["videos"]:
            seen[row["video_id"]] = row["status"]
        if not page["has_more"]:
            break
        cursor = page["next_cursor"]

    if target_total > 0:
        all_ids_materialized = len(seen) >= target_total
        all_terminal = all(status in TERMINAL for status in seen.values())
        print(f"Import progress: ids={len(seen)}/{target_total}")
        if all_ids_materialized and all_terminal:
            break
    time.sleep(10)

# 3. Use uploaded IDs for analysis
video_ids = [vid for vid, status in seen.items() if status == "UPLOADED"]

print(f"Imported {len(video_ids)} videos: {video_ids[:5]}...")

# 4. Now use the video IDs for analysis
client.analyze(
    video_ids,
    prompt="detect lane departure events",
)