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 job_list.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42
python job_list.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42 --stage annotation --state new
python job_list.py --host 'https://app.cvat.ai' --token '<your token>' \
    --project-id 7 --csv

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

"""List the jobs of an existing task or project with their stage, state, and
assignee, optionally as a CSV report.

Steps:
  1. Query jobs scoped to --task-id or --project-id, most recently updated
     first. --stage / --state filter server-side, so large tasks/projects
     stay cheap. The same endpoint also accepts free-text search, e.g.
     search='alice'.
  2. Print one row per job.
  3. If --csv is passed, also write report.csv into the current directory
     (project_id, project_name, task_id, task_name, job_id, stage, state,
     assignee, frames).

Usage (run ``python job_list.py --help`` for the full list of options):
  python job_list.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42
  python job_list.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42 --stage annotation --state new
  python job_list.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --project-id 7 --csv
"""

import argparse
import csv
from collections.abc import Iterable
from pathlib import Path

from cvat_sdk import make_client
from cvat_sdk.core.filters import F, all_
from cvat_sdk.core.proxies.jobs import Job


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)",
    )
    scope = parser.add_mutually_exclusive_group(required=True)
    scope.add_argument("--task-id", type=int, help="id of an existing task, e.g. 42")
    scope.add_argument("--project-id", type=int, help="id of an existing project, e.g. 7")
    parser.add_argument("--stage", help="only jobs at this stage, e.g. 'annotation'")
    parser.add_argument("--state", help="only jobs in this state, e.g. 'new'")
    parser.add_argument(
        "--csv", action="store_true", help="also write report.csv into the current directory"
    )
    return parser.parse_args()


def write_report(jobs: Iterable[Job], path: Path) -> None:
    with path.open("w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(
            [
                "project_id",
                "project_name",
                "task_id",
                "task_name",
                "job_id",
                "stage",
                "state",
                "assignee",
                "frames",
            ]
        )
        for job in jobs:
            assignee = job.assignee.username if job.assignee else ""
            writer.writerow(
                [
                    job.project_id or "",
                    job.project_name or "",
                    job.task_id,
                    job.task_name,
                    job.id,
                    job.stage,
                    job.state,
                    assignee,
                    job.frame_count,
                ]
            )


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        if args.task_id is not None:
            conditions = [F.task_id == args.task_id]
            scope_label = f"Task {args.task_id}"
        else:
            conditions = [F.project_id == args.project_id]
            scope_label = f"Project {args.project_id}"
        if args.stage:
            conditions.append(F.stage == args.stage)
        if args.state:
            conditions.append(F.state == args.state)

        jobs = client.jobs.list(filter=all_(*conditions), sort="-updated_date")
        print(f"{scope_label}: {len(jobs)} matching jobs")
        for job in jobs:
            assignee = job.assignee.username if job.assignee else "-"
            print(f"  job {job.id}: stage={job.stage}, state={job.state}, assignee={assignee}")

        if args.csv:
            report_path = Path("report.csv")
            write_report(jobs, report_path)
            print(f"Wrote {report_path.resolve()}")


if __name__ == "__main__":
    main()

# self-assign every unassigned job
python job_assign.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42
# round-robin across an explicit pool
python job_assign.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42 --assignees alice bob
# pool = every organization member matching the search
python job_assign.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42 --org 'annotators' --search 'annotator-team'

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

"""Round-robin the unassigned jobs of a task across a set of annotators and
write a CSV report of the assignments (job_id, previous_assignee, new_assignee).

The user API supports server-side search within an organization, so you rarely
need to know user ids — pass usernames, or an organization and search query,
and let the recipe resolve them.

Steps:
  1. Resolve the assignee pool:
       --assignees USERNAME [USERNAME ...] : look up each username exactly.
       --search QUERY --org SLUG           : search organization members,
                                             print the matches, use them all.
       --search QUERY --org-id ID          : same, using the organization id.
       neither                             : assign to me (the authenticated user).
  2. Filter the task's unassigned jobs.
  3. Round-robin the jobs across the resolved users.
  4. Write assignments.csv into the current directory.

Usage (run ``python job_assign.py --help`` for the full list of options):
  python job_assign.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42                              # self-assign
  python job_assign.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42 --assignees alice bob
  python job_assign.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42 --org 'annotators' --search 'annotator-team'
                                                  # pool = matches in the organization
"""

import argparse
import csv
import sys
from pathlib import Path

from cvat_sdk import make_client, models
from cvat_sdk.core.filters import F, all_, not_
from cvat_sdk.core.proxies.users import User


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"
    )
    organization = parser.add_mutually_exclusive_group()
    organization.add_argument(
        "--org", metavar="SLUG", help="organization slug to scope user and job queries"
    )
    organization.add_argument(
        "--org-id", type=int, metavar="ID", help="organization id to scope user and job queries"
    )
    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "--assignees",
        nargs="+",
        metavar="USERNAME",
        help="usernames to round-robin across (looked up exactly on the server)",
    )
    group.add_argument(
        "--search",
        metavar="QUERY",
        help="server-side search within --org/--org-id; every matching member becomes an assignee",
    )
    args = parser.parse_args()
    if args.search and args.org is None and args.org_id is None:
        parser.error("--search requires --org or --org-id")
    return args


def organization_filters(args: argparse.Namespace) -> dict[str, str | int]:
    if args.org is not None:
        return {"org": args.org}
    if args.org_id is not None:
        return {"org_id": args.org_id}
    return {}


def resolve_pool(client, args: argparse.Namespace) -> list[User]:
    """Resolve --assignees / --search / nothing to a list of User objects."""
    org_filters = organization_filters(args)
    if args.search:
        matches = client.users.list(search=args.search, **org_filters)
        if not matches:
            sys.exit(f"No users matched search {args.search!r}")
        print(f"Users matching {args.search!r}:")
        for user in matches:
            print(f"  {user.id}\t{user.username}")
        return matches

    if args.assignees:
        pool: list[User] = []
        for username in args.assignees:
            found = client.users.list(filter=F.username == username, **org_filters)
            if not found:
                sys.exit(f"User {username!r} not found")
            pool.append(found[0])
        return pool

    me = client.users.retrieve_current_user()
    print(f"No --assignees / --search; self-assigning as {me.username} (id={me.id})")
    return [me]


def main() -> None:
    args = parse_args()
    report_path = Path("assignments.csv")
    with make_client(args.host, access_token=args.token) as client:
        pool = resolve_pool(client, args)

        unassigned = client.jobs.list(
            filter=all_(F.task_id == args.task_id, not_(F.assignee.is_set())),
            **organization_filters(args),
        )
        print(f"Task {args.task_id}: {len(unassigned)} unassigned jobs to distribute")

        with report_path.open("w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["job_id", "previous_assignee", "new_assignee", "new_assignee_id"])
            for i, job in enumerate(unassigned):
                user = pool[i % len(pool)]
                previous = job.assignee.username if job.assignee else ""
                job.update(models.PatchedJobWriteRequest(assignee=user.id))
                writer.writerow([job.id, previous, user.username, user.id])
                print(f"Assigned job {job.id} -> {user.username} (id={user.id})")

        print(f"Wrote {report_path.resolve()}")


if __name__ == "__main__":
    main()