python auth_token.py --host 'https://app.cvat.ai' --token '<your token>'

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Connect to CVAT with a Personal Access Token (PAT) — the recommended way.

Steps:
  1. Open an authenticated client.
  2. Print the server version.
  3. Print who you are authenticated as (a quick sanity check for scripts).

Usage (run ``python auth_token.py --help`` for the full list of options):
  python auth_token.py --host 'https://app.cvat.ai' --token '<your token>'

Create a token in the CVAT UI under Profile -> Security.
"""

import argparse

from cvat_sdk import make_client


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (create one in the CVAT UI: Profile -> Security)",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        print("Server version:", client.get_server_version())
        me = client.users.retrieve_current_user()
        print(f"Authenticated as {me.username} (id={me.id})")


if __name__ == "__main__":
    main()

cvat-cli --server-host 'https://app.cvat.ai' profile create --name app --set-default

python auth_profile.py --profile app
python auth_profile.py               # uses the default profile

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Authenticate without putting a token in your code: use a saved profile.

Create a profile once on the command line, then any script can use it:

  cvat-cli --server-host 'https://app.cvat.ai' profile create --name app --set-default

Steps:
  1. If --profile is passed, use that profile; otherwise use the default profile.
  2. Print who you are authenticated as.

Usage (run ``python auth_profile.py --help`` for the full list of options):
  python auth_profile.py --profile app
  python auth_profile.py               # uses the default profile
"""

import argparse
import sys

from cvat_sdk import make_client_from_profile
from cvat_sdk.core.auth import AuthStore


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument(
        "--profile", help="name of a saved profile; omit to use the default profile"
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    store = AuthStore()
    if args.profile:
        profile = store.get_profile(args.profile)
        if profile is None:
            sys.exit(f"Profile {args.profile!r} not found. Create it with cvat-cli.")
        print(f"Using profile {args.profile!r}")
    else:
        default = store.get_default_profile()
        if default is None:
            sys.exit(
                "No default profile configured. Create one with:\n"
                "    cvat-cli --server-host 'https://app.cvat.ai' profile create"
                " --name app --set-default"
            )
        name, profile = default
        print(f"Using default profile {name!r}")

    with make_client_from_profile(profile) as client:
        me = client.users.retrieve_current_user()
        print(f"Authenticated as {me.username} (id={me.id})")


if __name__ == "__main__":
    main()

python auth_cli.py --profile app
python auth_cli.py --server-host 'https://app.cvat.ai'          # uses CVAT_ACCESS_TOKEN env
python auth_cli.py --server-host 'https://app.cvat.ai' --auth me:secret

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Build a CLI-compatible script that reuses ``cvat-cli``'s auth argument set:
``--server-host`` / ``--server-port`` / ``--auth`` / ``--profile`` / ``--insecure`` / ``--organization``.

This is the go-to pattern when your script should feel like an extension of
``cvat-cli`` — it accepts the same flags, honors the ``CVAT_ACCESS_TOKEN`` and
``PASS`` env variables, and resolves profiles the same way (explicit
``--profile``, else the default profile if no host/auth is passed).

Steps:
  1. Register the shared auth flags with ``configure_client_auth_arguments()``.
  2. Add your own script-specific arguments on top.
  3. Hand the parsed namespace to ``make_client_from_cli()`` to create a server API client object.

Usage (run ``python auth_cli.py --help`` for the full list of options):
  python auth_cli.py --profile app

  # export CVAT_ACCESS_TOKEN='<token>'  # for macOS/Linux
  # $env:CVAT_ACCESS_TOKEN = "<token>"  # for PowerShell
  python auth_cli.py --server-host 'https://app.cvat.ai'

  python auth_cli.py --server-host 'https://app.cvat.ai' --auth me:secret
"""

import argparse

from cvat_sdk import make_client_from_cli
from cvat_sdk.core.auth import configure_client_auth_arguments


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    configure_client_auth_arguments(parser)
    # Add your script's own arguments here, e.g.
    # parser.add_argument("--task-id", type=int, required=True)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    with make_client_from_cli(args) as client:
        me = client.users.retrieve_current_user()
        print(f"Authenticated as {me.username} (id={me.id})")


if __name__ == "__main__":
    main()

python project_create_and_list.py --host 'https://app.cvat.ai' --token '<your token>' \
    --name 'My project' --labels car person

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Create a project with labels, then list, filter, retrieve, and rename it.

Steps:
  1. Create a project with a simple label schema.
  2. List all projects visible to you (pagination is handled by the SDK).
  3. Filter projects by a name substring.
  4. Retrieve one project by id and read its labels.
  5. Rename it.
  6. Optionally delete it (--cleanup).

Usage (run ``python project_create_and_list.py --help`` for the full list of options):
  python project_create_and_list.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --name 'My project' --labels car person
"""

import argparse

from cvat_sdk import make_client, models
from cvat_sdk.core.filters import F


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--name", default="Example project", help="project name (default: '%(default)s')"
    )
    parser.add_argument(
        "--labels",
        nargs="+",
        default=["car", "person"],
        help="label names (default: %(default)s)",
    )
    parser.add_argument(
        "--cleanup", action="store_true", help="delete the created project at the end"
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        # 1. Create a project with labels
        project = client.projects.create(
            models.ProjectWriteRequest(
                name=args.name,
                labels=[models.PatchedLabelRequest(name=name) for name in args.labels],
            )
        )
        print(f"Created project {project.id}: {args.host}/projects/{project.id}")

        # 2. List all projects
        projects = client.projects.list()
        print(f"Projects visible to you: {len(projects)}")

        # 3. Filter by name substring
        matches = client.projects.list(filter=F.name.contains(args.name))
        print(f"Projects with {args.name!r} in the name: {[p.id for p in matches]}")

        # 4. Retrieve by id
        fetched = client.projects.retrieve(project.id)
        print(f"Project {fetched.id} labels: {[label.name for label in fetched.get_labels()]}")

        # 5. Rename
        renamed = fetched.update(models.PatchedProjectWriteRequest(name=f"{args.name} (renamed)"))
        print(f"Renamed to: {renamed.name}")

        # 6. Opt-in cleanup
        if args.cleanup:
            renamed.remove()
            print(f"Deleted project {project.id}")
        else:
            print("Keeping the project; pass --cleanup to delete it")


if __name__ == "__main__":
    main()

python project_add_labels.py --host 'https://app.cvat.ai' --token '<your token>' \
    --project-id 7 --labels car person
python project_add_labels.py --host 'https://app.cvat.ai' --token '<your token>' \
    --project-id 7 --labels car --attr car color red green blue

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Add labels, optionally with selectable attributes, to an existing project.

Steps:
  1. Retrieve the project and read the labels it already has; the requested
     labels that already exist are skipped, so the script is safe to re-run.
  2. Attach the --attr definitions to their new labels.
  3. Send one project update with the new labels. Labels of the tasks inside
     the project come from the project itself, so they all pick up the change.

Usage (run ``python project_add_labels.py --help`` for the full list of options):
  python project_add_labels.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --project-id 7 --labels car person
  python project_add_labels.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --project-id 7 --labels car --attr car color red green blue
"""

import argparse

from cvat_sdk import make_client, models


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--project-id", type=int, required=True, help="id of an existing project, e.g. 7"
    )
    parser.add_argument(
        "--labels", nargs="+", metavar="NAME", required=True, help="label names to add"
    )
    parser.add_argument(
        "--attr",
        nargs="+",
        action="append",
        default=[],
        metavar=("LABEL NAME", "VALUE"),
        help="selectable attribute for one of the --labels: label name, attribute "
        "name, then its values (repeat --attr for more attributes)",
    )
    args = parser.parse_args()
    for attr in args.attr:
        if len(attr) < 3:
            parser.error("--attr needs a label, an attribute name, and at least one value")
        if attr[0] not in args.labels:
            parser.error(f"--attr refers to label {attr[0]!r}, which is not in --labels")
    return args


def main() -> None:
    args = parse_args()
    attributes_per_label: dict[str, list[models.AttributeRequest]] = {}
    for label_name, attribute_name, *values in args.attr:
        attributes_per_label.setdefault(label_name, []).append(
            models.AttributeRequest(
                name=attribute_name,
                input_type=models.InputTypeEnum("select"),
                values=values,
                default_value=values[0],
                mutable=True,
            )
        )

    with make_client(args.host, access_token=args.token) as client:
        project = client.projects.retrieve(args.project_id)
        existing = {label.name for label in project.get_labels()}

        new_labels = []
        for name in args.labels:
            if name in existing:
                print(f"Label {name!r} already exists, skipping")
                continue
            new_labels.append(
                models.PatchedLabelRequest(name=name, attributes=attributes_per_label.get(name, []))
            )

        if new_labels:
            project.update(models.PatchedProjectWriteRequest(labels=new_labels))
        print(
            f"Added {len(new_labels)} labels to project {project.id}: "
            f"{', '.join(label.name for label in new_labels) or '-'}"
        )
        print(f"Project {project.id} labels: {[label.name for label in project.get_labels()]}")


if __name__ == "__main__":
    main()

python project_backup.py --host 'https://app.cvat.ai' --token '<your token>' \
    --project-id 42

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Download a backup zip of an existing project.

A backup contains the project's tasks, jobs, annotations, and settings. Pair
this recipe with project_restore.py to migrate or clone a project.

Steps:
  1. Retrieve the project by id.
  2. Download its backup to --output (default: project_<id>_backup.zip).

Usage (run ``python project_backup.py --help`` for the full list of options):
  python project_backup.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --project-id 42
"""

import argparse
from pathlib import Path

from cvat_sdk import make_client


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--project-id", type=int, required=True, help="id of an existing project, e.g. 42"
    )
    parser.add_argument(
        "--output",
        type=Path,
        help="destination file path (default: project_<id>_backup.zip)",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        project = client.projects.retrieve(args.project_id)
        output = args.output or Path(f"project_{project.id}_backup.zip")
        project.download_backup(output)
        print(f"Backed up project {project.id} to {output.resolve()}")


if __name__ == "__main__":
    main()

python project_restore.py --host 'https://app.cvat.ai' --token '<your token>' \
    --backup './project_42_backup.zip'

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Restore a project from a backup zip as a new project.

Pair with project_backup.py to migrate or clone a project.

Steps:
  1. Restore --backup as a brand-new project.
  2. Optionally delete the restored copy (--cleanup) — useful when testing a
     backup file.

Usage (run ``python project_restore.py --help`` for the full list of options):
  python project_restore.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --backup './project_42_backup.zip'
"""

import argparse
import sys
from pathlib import Path

from cvat_sdk import make_client


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument("--backup", type=Path, required=True, help="path to a project backup zip")
    parser.add_argument(
        "--cleanup",
        action="store_true",
        help="delete the restored project at the end (never touches the source backup)",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    if not args.backup.is_file():
        sys.exit(f"--backup {args.backup} does not exist")

    with make_client(args.host, access_token=args.token) as client:
        restored = client.projects.create_from_backup(args.backup)
        print(f"Restored a copy as project {restored.id}: {args.host}/projects/{restored.id}")

        if args.cleanup:
            restored.remove()
            print(f"Deleted restored project {restored.id}")
        else:
            print("Keeping the restored project; pass --cleanup to delete it")


if __name__ == "__main__":
    main()

python project_export_dataset.py --host 'https://app.cvat.ai' --token '<your token>' \
    --project-id 42 --cloud-storage-id 7 --export-format 'COCO 1.0'

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Export a project's tasks individually, without images, to local zips AND to
a registered cloud storage.

By default every task in the project is exported; pass --task-id to export
only a specific subset. This is the SDK-only stand-in for what could become a
bulk per-task export command in cvat-cli.

Steps:
  1. Fetch the server's export format list and validate --export-format.
  2. Resolve which tasks to export: --task-id filters to a subset of the
     project's tasks; omit it to export every task in the project.
  3. For each task: export to task_<id>_dataset.zip in the current directory,
     then export the same dataset straight to the cloud storage (no local
     download).

Usage (run ``python project_export_dataset.py --help`` for the full list of options):
  # every task in the project
  python project_export_dataset.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --project-id 42 --cloud-storage-id 7 --export-format 'COCO 1.0'

  # only tasks 10 and 11
  python project_export_dataset.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --project-id 42 --cloud-storage-id 7 --task-id 10 11
"""

import argparse
import sys
from pathlib import Path

from cvat_sdk import make_client
from cvat_sdk.core.proxies.types import Location


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--project-id", type=int, required=True, help="id of an existing project, e.g. 42"
    )
    parser.add_argument(
        "--cloud-storage-id",
        type=int,
        required=True,
        help="a registered cloud storage id (see cloud_storage_register.py)",
    )
    parser.add_argument(
        "--export-format",
        default="COCO 1.0",
        help="exporter name, e.g. 'COCO 1.0' (default: '%(default)s')",
    )
    parser.add_argument(
        "--task-id",
        type=int,
        nargs="+",
        metavar="ID",
        help="export only these task ids (must belong to the project); "
        "omit to export every task in the project",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        # 1. Validate the format against the server's list.
        # Low-level API: there is no high-level proxy for the format list yet.
        formats, _ = client.api_client.server_api.retrieve_annotation_formats()
        names = [f.name for f in formats.exporters]
        if args.export_format not in names:
            sys.exit(
                f"Unknown export format {args.export_format!r}. Choose one of: {', '.join(names)}"
            )

        # 2. Resolve which tasks to export.
        project = client.projects.retrieve(args.project_id)
        tasks_by_id = {task.id: task for task in project.get_tasks()}
        if args.task_id:
            missing = [str(tid) for tid in args.task_id if tid not in tasks_by_id]
            if missing:
                sys.exit(f"Task id(s) {', '.join(missing)} not found in project {project.id}")
            tasks = [tasks_by_id[tid] for tid in args.task_id]
        else:
            tasks = list(tasks_by_id.values())
        if not tasks:
            sys.exit(f"Project {project.id} has no tasks to export")

        # 3. Export each task individually: a local zip AND straight to the cloud storage.
        for task in tasks:
            local_path = Path(f"task_{task.id}_dataset.zip")
            task.export_dataset(
                args.export_format, local_path, include_images=False, location=Location.LOCAL
            )
            print(f"Exported {local_path.resolve()}")

            remote_name = f"task_{task.id}_dataset.zip"
            task.export_dataset(
                args.export_format,
                remote_name,
                include_images=False,
                location=Location.CLOUD_STORAGE,
                cloud_storage_id=args.cloud_storage_id,
            )
            print(f"Exported {remote_name} to cloud storage {args.cloud_storage_id}")

        print(f"Exported {len(tasks)} task dataset(s) from project {project.id}")


if __name__ == "__main__":
    main()

python task_create_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \
    --cloud-storage-id 7 --cloud-keys 'images/0001.jpg' 'images/0002.jpg' \
    --labels car person

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Create an annotation task from images that already live in a registered
cloud storage.

Steps:
  1. Create a task whose data is a list of object keys in the bucket.
  2. Print the result.
  3. Optionally delete it (--cleanup).

Register a bucket first with cloud_storage_register.py to get the storage id.

Usage (run ``python task_create_from_cloud.py --help`` for the full list of options):
  python task_create_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --cloud-storage-id 7 --cloud-keys 'images/0001.jpg' 'images/0002.jpg' \\
      --labels car person
"""

import argparse

from cvat_sdk import make_client, models
from cvat_sdk.core.proxies.tasks import ResourceType


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--cloud-storage-id",
        type=int,
        required=True,
        help="a registered cloud storage id (see cloud_storage_register.py)",
    )
    parser.add_argument(
        "--cloud-keys",
        nargs="+",
        required=True,
        help="object keys in the bucket, e.g. 'images/0001.jpg' 'images/0002.jpg'",
    )
    parser.add_argument(
        "--name",
        default="Task from cloud storage",
        help="task name (default: '%(default)s')",
    )
    parser.add_argument(
        "--labels", nargs="+", default=["object"], help="label names (default: %(default)s)"
    )
    parser.add_argument("--cleanup", action="store_true", help="delete the created task at the end")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        # ResourceType.SHARE + cloud_storage_id = read images from the bucket
        task = client.tasks.create_from_data(
            spec=models.TaskWriteRequest(
                name=args.name,
                labels=[models.PatchedLabelRequest(name=name) for name in args.labels],
            ),
            resource_type=ResourceType.SHARE,
            resources=args.cloud_keys,
            data_params={"cloud_storage_id": args.cloud_storage_id},
        )
        print(f"Created task {task.id} with {task.size} frames: {args.host}/tasks/{task.id}")

        if args.cleanup:
            task.remove()
            print(f"Deleted task {task.id}")
        else:
            print("Keeping the task; pass --cleanup to delete it")


if __name__ == "__main__":
    main()

# three video tasks in project 42
python tasks_bulk_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \
    --cloud-storage-id 7 --project-id 42 \
    --task 'videos/clip_01.mp4' --task 'videos/clip_02.mp4' --task 'videos/clip_03.mp4'

# two image-batch tasks in project 42
python tasks_bulk_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \
    --cloud-storage-id 7 --project-id 42 \
    --task 'batch_a/img_1.jpg,batch_a/img_2.jpg' \
    --task 'batch_b/img_1.jpg,batch_b/img_2.jpg'

# the same two batches, without listing every key: one task per wildcard match
python tasks_bulk_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \
    --cloud-storage-id 7 --project-id 42 --manifest manifest.jsonl \
    --task-pattern 'batch_a/*.jpg' --task-pattern 'batch_b/*.jpg'

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Bulk-create tasks inside a project, each task's data read from a registered
cloud storage.

Two ways to spell a task's data, repeatable and mixable:
  --task KEY[,KEY,...]    explicit object keys, in order:
                             * a single key -> a video task (or single-image task);
                             * several keys -> an image task, in the given order.
  --task-pattern PATTERN  every bucket file matching a fnmatch wildcard (e.g.
                           'batch_a/*.jpg'), resolved from the bucket's
                           manifest instead of being listed one by one.

All tasks land in the same project, so they share its label schema.

Steps:
  1. For each --task, create a task in --project-id from its explicit keys.
  2. For each --task-pattern, create a task in --project-id from every bucket
     file the wildcard matches, resolved via the bucket's manifest.
  3. Print the created ids and a summary count.
  4. Optionally delete every created task (--cleanup).

Register a bucket first with cloud_storage_register.py to get the storage id.
A --task-pattern also needs a manifest file already generated for the bucket -
see "How to generate manifest file" in the CVAT docs on attaching cloud storage.

Usage (run ``python tasks_bulk_from_cloud.py --help`` for the full list of options):
  # three video tasks in project 42
  python tasks_bulk_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --cloud-storage-id 7 --project-id 42 \\
      --task 'videos/clip_01.mp4' --task 'videos/clip_02.mp4' --task 'videos/clip_03.mp4'

  # two image-batch tasks in project 42
  python tasks_bulk_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --cloud-storage-id 7 --project-id 42 \\
      --task 'batch_a/img_1.jpg,batch_a/img_2.jpg' \\
      --task 'batch_b/img_1.jpg,batch_b/img_2.jpg'

  # the same two batches, without listing every key: one task per wildcard match
  python tasks_bulk_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --cloud-storage-id 7 --project-id 42 --manifest manifest.jsonl \\
      --task-pattern 'batch_a/*.jpg' --task-pattern 'batch_b/*.jpg'
"""

import argparse

from cvat_sdk import make_client, models
from cvat_sdk.core.proxies.tasks import ResourceType


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--cloud-storage-id",
        type=int,
        required=True,
        help="a registered cloud storage id (see cloud_storage_register.py)",
    )
    parser.add_argument(
        "--project-id",
        type=int,
        required=True,
        help="tasks are created in this project and inherit its labels",
    )
    parser.add_argument(
        "--task",
        dest="tasks",
        action="append",
        default=[],
        metavar="KEY[,KEY,...]",
        help="comma-separated object keys for one task; repeat for more tasks",
    )
    parser.add_argument(
        "--task-pattern",
        dest="task_patterns",
        action="append",
        default=[],
        metavar="PATTERN",
        help="one task from every bucket file matching this fnmatch wildcard "
        "(e.g. 'batch_a/*.jpg'); repeat for more tasks. Needs --manifest. "
        "(default: '%(default)s')",
    )
    parser.add_argument(
        "--manifest",
        default="manifest.jsonl",
        help="manifest object key in the bucket, used to resolve --task-pattern "
        "(default: '%(default)s')",
    )
    parser.add_argument(
        "--name-prefix",
        default="Bulk task",
        help="task name prefix; each task is named '<prefix> N' (default: '%(default)s')",
    )
    parser.add_argument(
        "--cleanup", action="store_true", help="delete every created task at the end"
    )
    args = parser.parse_args()
    if not args.tasks and not args.task_patterns:
        parser.error("at least one --task or --task-pattern is required")
    return args


def main() -> None:
    args = parse_args()
    task_key_groups = [
        [key.strip() for key in spec.split(",") if key.strip()] for spec in args.tasks
    ]
    if any(not group for group in task_key_groups):
        raise SystemExit("each --task must contain at least one non-empty key")

    with make_client(args.host, access_token=args.token) as client:
        created = []
        for keys in task_key_groups:
            # Tasks in a project inherit the project's labels — do NOT pass labels.
            # ResourceType.SHARE + cloud_storage_id reads the objects from the bucket.
            task = client.tasks.create_from_data(
                spec=models.TaskWriteRequest(
                    name=f"{args.name_prefix} {len(created) + 1}", project_id=args.project_id
                ),
                resource_type=ResourceType.SHARE,
                resources=keys,
                data_params={"cloud_storage_id": args.cloud_storage_id},
            )
            created.append(task)
            print(f"Created task {task.id} ({task.size} frames): {args.host}/tasks/{task.id}")

        for pattern in args.task_patterns:
            # A wildcard task needs the bucket's manifest as its only resource;
            # the server expands filename_pattern against it (fnmatch syntax).
            # use_cache=True is required to serve data straight from the bucket.
            task = client.tasks.create_from_data(
                spec=models.TaskWriteRequest(
                    name=f"{args.name_prefix} {len(created) + 1}", project_id=args.project_id
                ),
                resource_type=ResourceType.SHARE,
                resources=[args.manifest],
                data_params={
                    "cloud_storage_id": args.cloud_storage_id,
                    "use_cache": True,
                    "filename_pattern": pattern,
                },
            )
            created.append(task)
            print(
                f"Created task {task.id} ({task.size} frames) from pattern {pattern!r}: "
                f"{args.host}/tasks/{task.id}"
            )

        print(f"Created {len(created)} tasks in project {args.project_id}")

        if args.cleanup:
            for task in created:
                task.remove()
            print(f"Deleted {len(created)} tasks")
        else:
            print("Keeping the tasks; pass --cleanup to delete them")


if __name__ == "__main__":
    main()

python task_inspect_and_export.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42 --export-format 'COCO 1.0'

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Inspect an existing task (labels, jobs, frames), export its dataset to a
local zip, and export its event log to report quick analytics.

Steps:
  1. Retrieve the task and print a summary: labels, jobs (stage/state), frames.
  2. Fetch the server's export format list and validate --export-format.
  3. Export the dataset to task_<id>_dataset.zip in the current directory.
  4. Export the task's event log to task_<id>_events.csv and report two
     analytics: how many people are currently assigned to a job, and how
     many jobs were rejected in review and sent back for rework - the second
     one needs the log, since a job's current state doesn't show its history.

Usage (run ``python task_inspect_and_export.py --help`` for the full list of options):
  python task_inspect_and_export.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42 --export-format 'COCO 1.0'
"""

import argparse
import csv
import sys
from pathlib import Path

from cvat_sdk import make_client
from cvat_sdk.core.downloading import Downloader
from cvat_sdk.core.proxies.types import Location


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--task-id", type=int, required=True, help="id of an existing task, e.g. 42"
    )
    parser.add_argument(
        "--export-format",
        default="COCO 1.0",
        help="exporter name, e.g. 'COCO 1.0' (default: '%(default)s')",
    )
    return parser.parse_args()


def count_reworks(events_path: Path) -> int:
    """Count how many times a job in the log was rejected in review, i.e. sent
    back to the annotator for rework. A job's current state only shows where
    it stands now, not how many times it got there, so this needs the log.
    """
    with events_path.open(newline="") as f:
        return sum(
            1
            for row in csv.DictReader(f)
            if row["scope"] == "update:job"
            and row["obj_name"] == "state"
            and row["obj_val"] == "rejected"
        )


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        # 1. Inspect
        task = client.tasks.retrieve(args.task_id)
        jobs = task.get_jobs()
        print(f"Task {task.id}: {task.name!r}, {task.size} frames")
        print(f"  labels: {[label.name for label in task.get_labels()]}")
        for job in jobs:
            print(f"  job {job.id}: stage={job.stage}, state={job.state}")

        # 2. Validate the export format against the server's list.
        # Low-level API: there is no high-level proxy for the format list yet.
        formats, _ = client.api_client.server_api.retrieve_annotation_formats()
        names = [f.name for f in formats.exporters]
        if args.export_format not in names:
            sys.exit(
                f"Unknown export format {args.export_format!r}. Choose one of: {', '.join(names)}"
            )

        # 3. Export the dataset to a local zip
        local_path = Path(f"task_{task.id}_dataset.zip")
        task.export_dataset(
            args.export_format, local_path, include_images=False, location=Location.LOCAL
        )
        print(f"Exported {local_path.resolve()}")

        # 4. Export the task's event log and report quick analytics.
        events_path = Path(f"task_{task.id}_events.csv")
        Downloader(client).prepare_and_download_file_from_endpoint(
            client.api_client.events_api.create_export_endpoint,
            events_path,
            query_params={"task_id": task.id},
        )
        print(f"Exported {events_path.resolve()}")

        assigned = {job.assignee.id for job in jobs if job.assignee}
        print(f"  {len(assigned)} people currently assigned, {count_reworks(events_path)} reworks")


if __name__ == "__main__":
    main()

python tasks_create_per_label_group.py --host 'https://app.cvat.ai' --token '<your token>' \
    --image-dir ./images \
    --task 'boxes:rectangle:car,person' \
    --task 'roads:polygon:road,lane' \
    --task 'weather:tag:rain,snow'

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Create one task per shape type and label group over the same set of images,
so boxes, polygons, and tags are annotated in parallel by different people
instead of all at once in one crowded task.

Each --task spec is 'NAME:TYPE:label1,label2': the task's name, the shape type
its labels are drawn with, and the labels themselves.

The tasks are standalone, not tasks of one project: tasks in a project share the
project's label set, so a per-task label set cannot exist inside a single
project.

Steps:
  1. Parse and validate every --task spec before creating anything.
  2. Collect the files from --image-dir.
  3. Create one task per spec, with that spec's labels typed to its shape type.
  4. Print a summary of what each task got.

Usage (run ``python tasks_create_per_label_group.py --help`` for the full list of options):
  python tasks_create_per_label_group.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --image-dir ./images \\
      --task 'boxes:rectangle:car,person' \\
      --task 'roads:polygon:road,lane' \\
      --task 'weather:tag:rain,snow'
"""

import argparse
import sys
from pathlib import Path

from cvat_sdk import make_client, models
from cvat_sdk.core.proxies.tasks import ResourceType

LABEL_TYPES = {
    "any",
    "cuboid",
    "ellipse",
    "mask",
    "points",
    "polygon",
    "polyline",
    "rectangle",
    "tag",
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--image-dir",
        type=Path,
        required=True,
        help="directory with the images to annotate; every file in it is uploaded, "
        "and the server decides which media it accepts",
    )
    parser.add_argument(
        "--task",
        action="append",
        required=True,
        metavar="NAME:TYPE:LABELS",
        help="one task, e.g. 'boxes:rectangle:car,person' (repeat for more)",
    )
    parser.add_argument("--segment-size", type=int, help="frames per job, in every task")
    parser.add_argument(
        "--cleanup", action="store_true", help="delete the created tasks at the end"
    )
    return parser.parse_args()


def parse_spec(spec: str) -> tuple[str, str, list[str]]:
    """'boxes:rectangle:car,person' -> ('boxes', 'rectangle', ['car', 'person'])."""
    name, _, rest = spec.partition(":")
    type_, _, labels = rest.partition(":")
    names = [label.strip() for label in labels.split(",") if label.strip()]
    if not (name and type_ and names):
        sys.exit(f"Bad --task {spec!r}: expected 'NAME:TYPE:label1,label2'")
    if type_ not in LABEL_TYPES:
        sys.exit(
            f"Unknown label type {type_!r} in --task {spec!r}. "
            f"Choose one of: {', '.join(sorted(LABEL_TYPES))}"
        )
    return name, type_, names


def main() -> None:
    args = parse_args()
    # 1. Validate every spec first, so a typo in the last one costs nothing.
    specs = [parse_spec(spec) for spec in args.task]

    # 2. The same files go into every task. They are passed as they are found:
    # the server is the authority on which media formats it supports, so
    # filtering by extension here would only reject files CVAT can read.
    images = sorted(p for p in args.image_dir.iterdir() if p.is_file())
    if not images:
        sys.exit(f"No files found in {args.image_dir}")

    created = []
    with make_client(args.host, access_token=args.token) as client:
        try:
            for name, type_, label_names in specs:
                task = client.tasks.create_from_data(
                    spec=models.TaskWriteRequest(
                        name=name,
                        labels=[
                            models.PatchedLabelRequest(name=label, type=type_)
                            for label in label_names
                        ],
                        **({"segment_size": args.segment_size} if args.segment_size else {}),
                    ),
                    resource_type=ResourceType.LOCAL,
                    resources=images,
                )
                created.append(task)
                print(
                    f"Created task {task.id} {name!r} "
                    f"({type_}: {', '.join(label_names)}, {len(task.get_jobs())} job(s)): "
                    f"{args.host}/tasks/{task.id}"
                )
            print(f"Created {len(created)} task(s) from {len(images)} file(s)")
        finally:
            # Clean up whatever was created, including after a mid-run failure.
            if args.cleanup:
                for task in created:
                    task.remove()
                    print(f"Deleted task {task.id}")
            elif created:
                print("Keeping the tasks; pass --cleanup to delete them")


if __name__ == "__main__":
    main()

python task_create_job_mapping.py --host 'https://app.cvat.ai' --token '<your token>' \
    --image-dir ./images \
    --job 'cam1_001.png' 'cam1_002.png' --job 'cam2_001.png' 'cam2_002.png'

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Create a task whose jobs are defined by you, file by file, instead of by a
frame count: one job per camera, per scene, per delivery batch.

Two ways to group:
  --job FILE [FILE ...]   one job per occurrence, in the given file order
  --files-per-job N       a count, not a file list: chunk the sorted directory
                          listing into jobs of N files each, the explicit
                          equivalent of the task's segment_size

Steps:
  1. Build the file groups and check them against --image-dir: every file must
     exist, appear once, and belong to a job.
  2. Create the task with the job_file_mapping data parameter.
  3. Read the jobs back from the server and print/write the resulting mapping,
     so what you see is what the server built, not what was requested.

job_file_mapping implies predefined file ordering and takes one file per frame,
so it applies to image tasks only: a video is a single file the server cuts into
frames itself, and there is nothing to map. A directory of images is therefore
what this recipe takes, and CVAT rejects the request if the data turns out to be
a video.

Usage (run ``python task_create_job_mapping.py --help`` for the full list of options):
  python task_create_job_mapping.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --image-dir ./images \\
      --job 'cam1_001.png' 'cam1_002.png' --job 'cam2_001.png' 'cam2_002.png'
  python task_create_job_mapping.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --image-dir ./images --files-per-job 50
"""

import argparse
import csv
import sys
from pathlib import Path

from cvat_sdk import make_client, models
from cvat_sdk.core.proxies.tasks import ResourceType


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--image-dir",
        type=Path,
        required=True,
        help="directory with the task's images, one file per frame (a video cannot be "
        "mapped to jobs file by file)",
    )
    grouping = parser.add_mutually_exclusive_group(required=True)
    grouping.add_argument(
        "--job",
        action="append",
        nargs="+",
        metavar="FILE",
        help="the files of one job (repeat for more jobs)",
    )
    grouping.add_argument(
        "--files-per-job",
        type=int,
        metavar="N",
        help="number of files per job: chunk the sorted directory listing into jobs of N "
        "files each (the explicit equivalent of the task's segment_size)",
    )
    parser.add_argument("--name", default="Task with a job file mapping", help="task name")
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("job_file_mapping.csv"),
        help="path to write the resulting mapping to (default: %(default)s)",
    )
    parser.add_argument("--cleanup", action="store_true", help="delete the created task at the end")
    return parser.parse_args()


def build_groups(args: argparse.Namespace, available: list[str]) -> list[list[str]]:
    """The file groups to send as job_file_mapping, validated against the directory."""
    if args.files_per_job:
        if args.files_per_job < 1:
            sys.exit("--files-per-job must be at least 1")
        return [
            available[start : start + args.files_per_job]
            for start in range(0, len(available), args.files_per_job)
        ]

    groups = [list(group) for group in args.job]
    known = set(available)
    assigned = []
    for group in groups:
        assigned.extend(group)

    unknown = [name for name in assigned if name not in known]
    if unknown:
        sys.exit(f"File(s) {', '.join(unknown)} not found in {args.image_dir}")

    duplicates = sorted({name for name in assigned if assigned.count(name) > 1})
    if duplicates:
        sys.exit(f"File(s) {', '.join(duplicates)} appear in more than one job")

    unassigned = sorted(known - set(assigned))
    if unassigned:
        sys.exit(
            f"File(s) {', '.join(unassigned)} are not assigned to any job. "
            "Every file in --image-dir must belong to exactly one --job."
        )
    return groups


def main() -> None:
    args = parse_args()
    # The files are taken as they are found: the server is the authority on
    # which media formats it supports, so filtering by extension here would
    # only reject files CVAT can read.
    available = sorted(p.name for p in args.image_dir.iterdir() if p.is_file())
    if not available:
        sys.exit(f"No files found in {args.image_dir}")

    groups = build_groups(args, available)
    resources = [args.image_dir / name for group in groups for name in group]
    print(f"Requesting {len(groups)} job(s) over {len(resources)} file(s)")

    with make_client(args.host, access_token=args.token) as client:
        task = client.tasks.create_from_data(
            spec=models.TaskWriteRequest(
                name=args.name,
                labels=[models.PatchedLabelRequest(name="object")],
            ),
            resource_type=ResourceType.LOCAL,
            resources=resources,
            data_params={"job_file_mapping": groups},
        )
        print(f"Created task {task.id} with {task.size} frames: {args.host}/tasks/{task.id}")

        # 3. The mapping is built using one row per file, carrying
        # the frame that file ended up on. A job's frames come back in order,
        # so the frame is the job's start plus the file's position in it.
        rows = []
        for job in sorted(task.get_jobs(), key=lambda job: job.start_frame):
            names = [frame.name for frame in job.get_frames_info()]
            print(f"  job {job.id} frames {job.start_frame}-{job.stop_frame}: {', '.join(names)}")
            rows.extend(
                {"job_id": job.id, "frame": job.start_frame + offset, "file_name": name}
                for offset, name in enumerate(names)
            )

        with args.output.open("w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=["job_id", "frame", "file_name"])
            writer.writeheader()
            writer.writerows(rows)
        print(f"Wrote {args.output.resolve()}")

        if args.cleanup:
            task.remove()
            print(f"Deleted task {task.id}")
        else:
            print("Keeping the task; pass --cleanup to delete it")


if __name__ == "__main__":
    main()