import datetime
from collections import defaultdict
from encord import EncordUserClient
from encord.analytics import TaskActionType
SSH_PATH = "<your_private_key>"
PROJECT_ID = "<project_id>"
# Authenticate
user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
ssh_private_key_path=SSH_PATH,
domain="https://api.encord.com",
)
# Get the project
project = user_client.get_project(PROJECT_ID)
# Define time range
start_time = datetime.datetime.now() - datetime.timedelta(days=29)
# Get task actions
actions = list(project.get_task_actions(
after=start_time,
action_type=[TaskActionType.SUBMIT, TaskActionType.APPROVE, TaskActionType.REJECT],
))
# Group by task, sorted by time
by_task = defaultdict(list)
for action in actions:
by_task[action.task_uuid].append(action)
for events in by_task.values():
events.sort(key=lambda a: a.timestamp)
# For each approve/reject, attribute it to the annotator who last submitted
annotator_stats = defaultdict(lambda: {"approved": 0, "rejected": 0})
for events in by_task.values():
current_annotator = None
for event in events:
if event.action_type == TaskActionType.SUBMIT:
current_annotator = event.actor_email
elif event.action_type == TaskActionType.APPROVE and current_annotator:
annotator_stats[current_annotator]["approved"] += 1
elif event.action_type == TaskActionType.REJECT and current_annotator:
annotator_stats[current_annotator]["rejected"] += 1
print("=== Approval & Rejection Rates per Annotator ===")
if not annotator_stats:
print("No data found.")
else:
print(f"{'Annotator':<35} {'Approved':>10} {'Rejected':>10} {'Total':>8} {'Approval Rate':>15}")
print("-" * 80)
for email, stats in sorted(annotator_stats.items()):
total = stats["approved"] + stats["rejected"]
rate = stats["approved"] / total * 100 if total else 0
print(f"{email:<35} {stats['approved']:>10} {stats['rejected']:>10} {total:>8} {rate:>14.1f}%")