#!/usr/bin/env python3 """任务分配健康检查脚本""" import json from datetime import datetime, timedelta, timezone # 代码评审专家 ID CODE_REVIEWER_ID = "34d7c53d-bd70-45a8-bbbb-77dbb1da16b5" # 当前时间 (UTC) NOW = datetime.now(timezone.utc) # 读取任务数据 with open('/home/multica/.claude/projects/-home-multica-multica-workspaces-b5fdce19-2a82-455d-b644-5b83da2b3078-65deff22-workdir/6e361410-5bd5-42f7-87dd-3b6a095c5b91/tool-results/b6ko1exyf.txt', 'r') as f: data = json.load(f) issues = data['issues'] # 筛选未完成的任务 active_issues = [ issue for issue in issues if issue['status'] not in ['done', 'cancelled', 'backlog'] ] print(f"总任务数: {len(issues)}") print(f"未完成任务数: {len(active_issues)}") print(f"当前时间: {NOW.isoformat()}") print("\n" + "="*80 + "\n") # 分类问题 problems = { 'in_review_wrong_assignee': [], 'todo_stale': [], 'in_progress_stale': [], 'blocked_no_assignee': [], 'todo_no_assignee': [], 'in_progress_no_assignee': [] } actions = [] for issue in active_issues: issue_id = issue['id'] identifier = issue['identifier'] status = issue['status'] assignee_id = issue['assignee_id'] assignee_type = issue['assignee_type'] title = issue['title'] updated_at = datetime.fromisoformat(issue['updated_at'].replace('Z', '+00:00')) hours_since_update = (NOW - updated_at).total_seconds() / 3600 # 规则 1: in_review 状态必须分配给代码评审专家 if status == 'in_review': if assignee_type != 'agent' or assignee_id != CODE_REVIEWER_ID: problems['in_review_wrong_assignee'].append({ 'issue': identifier, 'title': title, 'current_assignee': assignee_id, 'assignee_type': assignee_type }) actions.append({ 'type': 'reassign', 'issue_id': issue_id, 'identifier': identifier, 'title': title, 'new_assignee_id': CODE_REVIEWER_ID, 'reason': 'in_review 状态必须分配给代码评审专家' }) # 规则 2: todo 状态且有 assignee，超过2小时未更新 elif status == 'todo' and assignee_id: if hours_since_update > 2: problems['todo_stale'].append({ 'issue': identifier, 'title': title, 'assignee_id': assignee_id, 'assignee_type': assignee_type, 'hours_since_update': round(hours_since_update, 1) }) # 根据 assignee_type 决定如何 mention if assignee_type == 'agent': actions.append({ 'type': 'mention', 'issue_id': issue_id, 'identifier': identifier, 'title': title, 'mention_type': 'agent', 'mention_id': assignee_id, 'message': f'此任务已分配超过 {round(hours_since_update, 1)} 小时，请开始处理。' }) elif assignee_type == 'squad': actions.append({ 'type': 'comment', 'issue_id': issue_id, 'identifier': identifier, 'title': title, 'message': f'⚠️ 此任务已分配给 squad 超过 {round(hours_since_update, 1)} 小时，状态仍为 todo，请检查。' }) # 规则 3: in_progress 状态，超过48小时未更新 elif status == 'in_progress': if hours_since_update > 48: problems['in_progress_stale'].append({ 'issue': identifier, 'title': title, 'assignee_id': assignee_id, 'assignee_type': assignee_type, 'hours_since_update': round(hours_since_update, 1) }) if assignee_id: if assignee_type == 'agent': actions.append({ 'type': 'mention', 'issue_id': issue_id, 'identifier': identifier, 'title': title, 'mention_type': 'agent', 'mention_id': assignee_id, 'message': f'此任务已 {round(hours_since_update, 1)} 小时未更新，请提供进度更新或遇到的问题。' }) elif assignee_type == 'squad': actions.append({ 'type': 'comment', 'issue_id': issue_id, 'identifier': identifier, 'title': title, 'message': f'⚠️ 此任务已 {round(hours_since_update, 1)} 小时未更新，请检查进度。' }) # 规则 4: 任何状态但 assignee 为 null if not assignee_id: if status == 'blocked': problems['blocked_no_assignee'].append({ 'issue': identifier, 'title': title, 'status': status }) actions.append({ 'type': 'comment', 'issue_id': issue_id, 'identifier': identifier, 'title': title, 'message': '⚠️ 此任务状态为 blocked，但没有 assignee。blocked 的任务应该有负责人追踪解除阻塞。' }) elif status == 'todo': problems['todo_no_assignee'].append({ 'issue': identifier, 'title': title, 'status': status }) actions.append({ 'type': 'comment', 'issue_id': issue_id, 'identifier': identifier, 'title': title, 'message': '⚠️ 此任务状态为 todo，但没有 assignee。todo 状态的任务应该分配给具体的 agent 或 squad。' }) elif status == 'in_progress': problems['in_progress_no_assignee'].append({ 'issue': identifier, 'title': title, 'status': status }) actions.append({ 'type': 'comment', 'issue_id': issue_id, 'identifier': identifier, 'title': title, 'message': '⚠️ 此任务状态为 in_progress，但没有 assignee。in_progress 的任务必须有明确的负责人。' }) # 输出检查报告 print("## 任务分配健康检查报告\n") print("### 1. in_review 状态分配错误") if problems['in_review_wrong_assignee']: print(f"发现 {len(problems['in_review_wrong_assignee'])} 个问题：") for p in problems['in_review_wrong_assignee']: print(f" - {p['issue']}: {p['title']}") print(f" 当前分配: {p['assignee_type']} / {p['current_assignee']}") print(f" ✅ 将自动重新分配给代码评审专家") else: print("✅ 无问题") print("\n### 2. todo 状态超过2小时未更新") if problems['todo_stale']: print(f"发现 {len(problems['todo_stale'])} 个问题：") for p in problems['todo_stale']: print(f" - {p['issue']}: {p['title']}") print(f" 分配给: {p['assignee_type']} / {p['assignee_id']}") print(f" 未更新时长: {p['hours_since_update']} 小时") print(f" ✅ 将添加 mention 提醒") else: print("✅ 无问题") print("\n### 3. in_progress 状态超过48小时未更新") if problems['in_progress_stale']: print(f"发现 {len(problems['in_progress_stale'])} 个问题：") for p in problems['in_progress_stale']: print(f" - {p['issue']}: {p['title']}") print(f" 分配给: {p['assignee_type']} / {p['assignee_id']}") print(f" 未更新时长: {p['hours_since_update']} 小时") print(f" ✅ 将添加 mention 提醒") else: print("✅ 无问题") print("\n### 4. 缺少 assignee 的任务") total_no_assignee = (len(problems['blocked_no_assignee']) + len(problems['todo_no_assignee']) + len(problems['in_progress_no_assignee'])) if total_no_assignee > 0: print(f"发现 {total_no_assignee} 个问题：") if problems['blocked_no_assignee']: print(f"\n blocked 状态 ({len(problems['blocked_no_assignee'])}):") for p in problems['blocked_no_assignee']: print(f" - {p['issue']}: {p['title']}") if problems['todo_no_assignee']: print(f"\n todo 状态 ({len(problems['todo_no_assignee'])}):") for p in problems['todo_no_assignee']: print(f" - {p['issue']}: {p['title']}") if problems['in_progress_no_assignee']: print(f"\n in_progress 状态 ({len(problems['in_progress_no_assignee'])}):") for p in problems['in_progress_no_assignee']: print(f" - {p['issue']}: {p['title']}") print(f" ✅ 将添加评论提醒") else: print("✅ 无问题") print("\n" + "="*80) print(f"\n总共发现 {len(actions)} 个需要处理的问题") print("\n需要执行的操作：") print(f" - 自动重新分配: {len([a for a in actions if a['type'] == 'reassign'])}") print(f" - 添加 mention: {len([a for a in actions if a['type'] == 'mention'])}") print(f" - 添加评论: {len([a for a in actions if a['type'] == 'comment'])}") # 保存 actions 到文件 with open('/tmp/actions.json', 'w') as f: json.dump(actions, f, indent=2, ensure_ascii=False) print("\n操作列表已保存到 /tmp/actions.json")