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_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()

# send everything annotators finished into review
python job_workflow.py --host 'https://app.cvat.ai' --token '<your token>' \
    --from-stage annotation
# accept everything that passed review, scoped to one task
python job_workflow.py --host 'https://app.cvat.ai' --token '<your token>' \
    --from-stage validation --task-id 42

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

"""Batch-advance completed jobs to the next workflow stage

Find every job whose state is 'completed' at --from-stage, move each one to
the next stage, and print the list of modified jobs. Optionally restrict the
sweep to a single task with --task-id.

Steps:
  1. Query jobs matching (stage == --from-stage, state == 'completed').
  2. Update each job's stage to the next one in the workflow.
  3. Print the modified job ids.

Usage (run ``python job_workflow.py --help`` for the full list of options):
  # Send everything annotators finished into review:
  python job_workflow.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --from-stage annotation
  # Accept everything that passed review, scoped to one task:
  python job_workflow.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --from-stage validation --task-id 42
"""

import argparse

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

NEXT_STAGE = {"annotation": "validation", "validation": "acceptance"}


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(
        "--from-stage",
        required=True,
        choices=sorted(NEXT_STAGE),
        help="advance completed jobs currently at this stage",
    )
    parser.add_argument(
        "--task-id",
        type=int,
        help="restrict the sweep to a single task (default: every task you can see)",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    to_stage = NEXT_STAGE[args.from_stage]

    with make_client(args.host, access_token=args.token) as client:
        conditions = [F.stage == args.from_stage, F.state == "completed"]
        if args.task_id is not None:
            conditions.append(F.task_id == args.task_id)

        jobs = client.jobs.list(filter=all_(*conditions))
        print(f"Found {len(jobs)} completed jobs at stage {args.from_stage!r}")

        for job in jobs:
            job.update(models.PatchedJobWriteRequest(stage=to_stage))
            print(f"  job {job.id}: {args.from_stage} -> {to_stage}")

        print(f"Moved {len(jobs)} jobs to stage {to_stage!r}")


if __name__ == "__main__":
    main()