Job recipes

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