CVAT SDK 作业（Job）自动化实战：列表查询、轮询分配与批量阶段流转三则脚本详解
【免费下载链接】cvatComputer Vision Annotation Tool (CVAT) is a leading platform for building high-quality visual datasets for vision AI. It offers open-source, cloud, and enterprise products, as well as labeling services, for image, video, and 3D annotation with AI-assisted labeling, quality assurance, team collaboration, analytics, and developer APIs.项目地址: https://gitcode.com/GitHub_Trending/cvat/cvat
本篇技术指南以 CVAT 官方 SDK 示例文档 jobs.md 为主体，围绕三则开箱即用的 Python 脚本展开：
job_list.py（列出任务/项目的作业并输出 CSV 报告）、job_assign.py（将未分配作业按轮询方式分发给用户池）、job_workflow.py（批量把已完成作业推进到下一阶段）。读完本篇，你将掌握 CVAT 作业（Job）的stage/state状态模型、基于 JSON Logic 的服务端过滤写法，以及如何用个人访问令牌（PAT）通过 cvat-sdk 驱动真实标注工作流中的分配与流转自动化。
一、背景：CVAT 中的 Job 是什么
在 CVAT 中，一个任务（Task）在创建时会按segment_size（分段大小）自动切分为一个或多个作业（Job），作业是标注工作的最小单元，通常对应一段连续帧区间。每个作业都携带两个核心状态字段：
stage（阶段）：annotation（标注）→validation（校验）→acceptance（验收），描述作业所处的流程环节；state（状态）：new（新建）、in progress（进行中）、rejected（被打回）、completed（已完成），描述当前阶段内的完成程度。
需要特别注意的是：作业由任务自动创建，无法单独创建（只能更新和分配），因此针对作业的自动化必须建立在"枚举已有作业 → 逐个更新"的脚本化模式上。这也正是本篇三则脚本存在的意义——CVAT 本身没有内置的自动分配机制，job_assign.py 就是官方提供的脚本化分配范式。
二、运行准备
三则脚本位于 cvat-sdk/examples/ 目录，依赖cvat_sdk包。它们都通过make_client(host, access_token=token)建立客户端连接（见 job_list.py），因此需要两项必填参数：
| 参数 | 必填 | 含义 |
|---|---|---|
--host | 是 | CVAT 服务器地址，如https://app.cvat.ai或自部署实例地址 |
--token | 是 | 个人访问令牌（Personal Access Token），在 CVAT UI 的 Profile → Security 中生成 |
所有脚本均支持python <脚本名>.py --help查看完整选项列表。
三、脚本一：job_list.py——列出任务或项目的作业
job_list.py查询某个任务或某个项目的全部作业，支持可选的--stage/--state服务端过滤，结果按最近更新时间排序；传入--csv时会在当前目录额外写出report.csv报告。
3.1 参数说明
| Flag | 必填 | 含义 |
|---|---|---|
--host | 是 | 服务器地址 |
--token | 是 | 个人访问令牌 |
--task-id | --task-id/--project-id二选一 | 要列出作业的任务 id |
--project-id | --task-id/--project-id二选一 | 要列出作业的项目 id |
--stage | 否 | 只列出处于该阶段的作业，如annotation |
--state | 否 | 只列出处于该状态的作业，如new |
--csv | 否 | 同时在当前目录写出report.csv |
--task-id与--project-id在脚本中通过add_mutually_exclusive_group(required=True)强制互斥且必选其一（见 job_list.py）。
3.2 使用示例
# 列出任务 42 的全部作业 python job_list.py --host 'https://app.cvat.ai' --token '<your token>' \ --task-id 42 # 只列出任务 42 中处于 annotation 阶段、new 状态的作业 python job_list.py --host 'https://app.cvat.ai' --token '<your token>' \ --task-id 42 --stage annotation --state new # 列出项目 7 的全部作业并写出 report.csv python job_list.py --host 'https://app.cvat.ai' --token '<your token>' \ --project-id 7 --csvreport.csv的列结构为：project_id, project_name, task_id, task_name, job_id, stage, state, assignee, frames，其中frames即作业的frame_count。当作业没有 assignee 时，表格与 CSV 中分别以-和空字符串表示（见 job_list.py）。
3.3 源码级原理
脚本核心查询语句为（job_list.py）：
conditions = [F.task_id == args.task_id] # 或 F.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")几个值得深挖的实现细节：
F过滤 DSL：cvat_sdk.core.filters.F是一个字段访问器，F.task_id == 42会生成{"==": [{"var": "task_id"}, 42]}这样的JSON Logic表达式，all_()将其合并为 AND 节点，最终序列化为filter查询参数由服务端执行（见 filters.py）。这意味着--stage/--state是服务端过滤而非客户端过滤——对于作业数量巨大的任务/项目，分页与传输成本都能保持很低。文档注释中特别提到，同一接口还接受自由文本search，例如search='alice'。- 排序：
sort="-updated_date"表示按最近更新时间倒序，先展示最新改动的作业。 - 分页透明化：
client.jobs.list(...)内部通过get_paginated_collection自动拉取全部分页结果，使用者拿到的就是一个完整列表（见 proxies/jobs.py 中JobsRepo对ModelListMixin的继承）。
四、脚本二：job_assign.py——轮询分配任务的未分配作业
job_assign.py把一个任务的全部未分配作业均摊给一个解析好的用户池，并写出assignments.csv（列结构：job_id, previous_assignee, new_assignee, new_assignee_id）。用户池有三种解析方式：用--assignees精确按用户名查找、用--search在组织成员中搜索（命中者全部成为分配对象）、两者都不传则自我分配给当前认证用户。
4.1 参数说明
| Flag | 必填 | 含义 |
|---|---|---|
--host | 是 | 服务器地址 |
--token | 是 | 个人访问令牌 |
--task-id | 是 | 要分配作业的任务 id |
--org SLUG | 否 | 组织 slug，用于限定用户与作业查询范围 |
--org-id ID | 否 | 组织 id，作为--org的替代 |
--assignees USERNAME [...] | 否 | 参与轮询的用户名列表（精确匹配） |
--search QUERY | 否 | 搜索组织成员，所有命中者都成为分配对象 |
约束关系（脚本通过add_mutually_exclusive_group与启动时校验强制，见 job_assign.py）：
--assignees与--search互斥；--org与--org-id互斥；--search必须搭配--org或--org-id使用（否则parser.error直接退出），因为搜索匹配的是username、first_name、last_name三个字段，只有限定在团队范围内才有意义；- 两者都不传时，自动取当前认证用户作为唯一分配对象。
4.2 使用示例
# 把所有未分配作业分配给自己 python job_assign.py --host 'https://app.cvat.ai' --token '<your token>' \ --task-id 42 # 在显式用户池 alice、bob 之间轮询分配 python job_assign.py --host 'https://app.cvat.ai' --token '<your token>' \ --task-id 42 --assignees alice bob # 用户池 = 组织 'annotators' 中所有匹配 'annotator-team' 的成员 python job_assign.py --host 'https://app.cvat.ai' --token '<your token>' \ --task-id 42 --org 'annotators' --search 'annotator-team'4.3 源码级原理
用户池解析（resolve_pool，见 job_assign.py）三条分支：
--search：调用client.users.list(search=args.search, **org_filters)做服务端搜索，无命中时打印错误并以非零码退出；命中则先打印id + username再全部纳入池。--assignees：对每个用户名执行client.users.list(filter=F.username == username, **org_filters)精确查找，找不到则退出；找到后取found[0]加入池。- 都不传：
client.users.retrieve_current_user()取当前用户，即自我分配。
未分配作业筛选（job_assign.py）：
unassigned = client.jobs.list( filter=all_(F.task_id == args.task_id, not_(F.assignee.is_set())), **organization_filters(args), )F.assignee.is_set()生成{"var": "assignee"}，not_()为其取反（{"!": ...}），组合起来表达"assignee 未设置"（见 filters.py 与 filters.py）。organization_filters会把--org/--org-id作为org/org_id参数透传给查询，将作业与用户查询都限定在指定组织内（job_assign.py）。
轮询与更新（job_assign.py）：
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])分配的核心是一次局部更新：job.update(models.PatchedJobWriteRequest(assignee=user.id))。Job代理类混入了ModelUpdateMixin，其局部更新参数为patched_job_write_request（见 proxies/jobs.py），因此可以只提交assignee字段而不触碰作业的其他属性。每完成一次分配，assignments.csv都会记录job_id, previous_assignee, new_assignee, new_assignee_id，便于事后审计与回滚。
五、脚本三：job_workflow.py——批量推进已完成作业
job_workflow.py找出所有处于--from-stage阶段且state == "completed"的作业，将它们逐个推进到下一阶段（annotation → validation → acceptance），并打印被修改的作业清单。可选--task-id将扫描范围限制在单个任务内。
5.1 参数说明
| Flag | 必填 | 含义 |
|---|---|---|
--host | 是 | 服务器地址 |
--token | 是 | 个人访问令牌 |
--from-stage | 是 | 推进处于该阶段的已完成作业（annotation或validation） |
--task-id | 否 | 把扫描范围限制在单个任务（默认扫描你能看到的所有任务） |
5.2 使用示例
# 把标注人员完成的所有作业送入校验（review） python job_workflow.py --host 'https://app.cvat.ai' --token '<your token>' \ --from-stage annotation # 把通过校验的作业全部验收，限定在任务 42 内 python job_workflow.py --host 'https://app.cvat.ai' --token '<your token>' \ --from-stage validation --task-id 425.3 源码级原理
阶段映射表定义在模块顶层（job_workflow.py）：
NEXT_STAGE = {"annotation": "validation", "validation": "acceptance"}--from-stage的合法值正是sorted(NEXT_STAGE)（即annotation、validation），argparse的choices会在参数解析阶段就拦截非法输入（job_workflow.py）。主流程（job_workflow.py）：
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)) for job in jobs: job.update(models.PatchedJobWriteRequest(stage=to_stage)) print(f" job {job.id}: {args.from_stage} -> {to_stage}")该脚本与job_assign.py共享同一套过滤与更新机制，区别仅在于筛选条件（stage+state == "completed"）与更新字段（stage）。由于annotation/validation之外不定义更远的阶段，脚本在acceptance阶段会自然停止，形成一条完整的"标注完成 → 送审 → 验收"流水线。
六、进阶：Job 代理对象的更多 API 能力
除了上述三则脚本演示的列表、分配、推进之外，Job代理对象（定义于 cvat-sdk/cvat_sdk/core/proxies/jobs.py）还封装了完整的作业级操作，官方文档以速查表形式给出：
| SDK 方法 / 参数 | 作用说明 |
|---|---|
Job.update(models.PatchedJobWriteRequest(stage=...)) | 修改作业的stage（先取回作业再更新），取值必须是annotation、validation、acceptance之一 |
Job.update(models.PatchedJobWriteRequest(state=...)) | 修改作业的state，取值必须是new、in progress、rejected、completed之一 |
Job.import_annotations(..., import_mode="replace" \| "append") | 导入标注："replace"覆盖作业已有标注（默认）；"append"将导入的标注合并进去 |
Job.import_annotations(..., conv_mask_to_poly=True \| False) | 是否把导入的 mask 标注转换为多边形（布尔值，服务端默认True） |
Job.import_annotations(..., pbar=ProgressReporter()) | 上报上传进度（传入cvat_sdk.core.progress.ProgressReporter） |
Job.get_issues() | 获取某个作业上提出的评审问题（review issues）列表 |
Job.export_dataset(format_name, path) | 导出单个作业的数据集——是import_annotations的导出对应物 |
Job.get_frame(frame_id: int, *, quality="original" \| "compressed") | 以文件类对象（io.RawIOBase）返回单帧图像字节；quality为可选关键字参数（"original"或"compressed"），省略时使用服务端默认值 |
Job.download_frames(frame_ids: Sequence[int], outdir=".", quality="original", image_extension=None, filename_pattern="frame_{frame_id:06d}{frame_ext}") | 把指定帧保存到outdir下的磁盘文件；image_extension（如"png"）可覆盖自动检测的扩展名；quality取"original"或"compressed" |
Job.get_meta()/Job.get_labels() | 读取作业的帧元数据与标签（label）模式 |
从源码看这些能力的底层实现：
- 导入标注：
import_annotations委托AnnotationUploader.upload_file_and_wait调用create_annotations_endpoint完成上传与轮询等待（proxies/jobs.py）。上传参数conv_mask_to_poly、import_mode会在请求层被序列化为"true"/"false"与对应枚举字符串（见 cvat-sdk/cvat_sdk/core/uploading.py）。 - 取帧与下载：
get_frame调用retrieve_data（type="frame"）并把响应字节包装成io.BytesIO返回；download_frames在其上叠加 PIL 解码与扩展名推断（.jpe/.jpeg统一归一为.jpg），按filename_pattern模板落盘（proxies/jobs.py）。 - 评审问题：
get_issues通过issues_api.list_endpoint分页拉取该作业的全部Issue对象并包装为 SDK 代理（proxies/jobs.py）。
七、注意事项与最佳实践
回顾原文档的 Notes 与本仓库实现，以下几点在实际使用中最容易踩坑：
- 枚举值固定：
stage只能是annotation、validation、acceptance；state只能是new、in progress、rejected、completed。传错值会直接导致服务端过滤无结果或更新被拒。 - 作业不可单独创建：作业随任务自动生成（由任务创建时的
segment_size控制切分粒度），脚本只能"更新与分配"，不能"新建作业"。 - CVAT 没有内置自动分配：
job_assign.py正是官方认可的脚本化分配范式；而job_list.py则适合作为报表与监控的基石（--csv报告可直接导入表格工具）。 - 过滤在服务端：三个脚本都通过
filter=all_(...)将条件编码为 JSON Logic 交给服务端执行，数据量大时性能优于客户端过滤，这也是job_list.py特别标注--stage/--state过滤"保持大任务廉价"的原因。 - 安全与作用域：涉及组织成员搜索时务必搭配
--org/--org-id，否则--search会被脚本直接拒绝；令牌建议使用最小权限的 PAT。
三则脚本的完整可运行源码位于：
- cvat-sdk/examples/job_list.py
- cvat-sdk/examples/job_assign.py
- cvat-sdk/examples/job_workflow.py
它们与本文对应的官方文档页 site/content/en/docs/api_sdk/sdk/examples/jobs.md 一起，构成了从"查询 → 分配 → 流转"完整闭环的作业自动化参考实现；结合 cvat-sdk 文档 中Job代理与F过滤 DSL 的说明，你可以在此基础上继续扩展出属于自己团队的标注流水线工具。
【免费下载链接】cvatComputer Vision Annotation Tool (CVAT) is a leading platform for building high-quality visual datasets for vision AI. It offers open-source, cloud, and enterprise products, as well as labeling services, for image, video, and 3D annotation with AI-assisted labeling, quality assurance, team collaboration, analytics, and developer APIs.项目地址: https://gitcode.com/GitHub_Trending/cvat/cvat
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
