chore: 整理项目结构,清理归档文件,更新部署配置

This commit is contained in:
Simon
2026-07-04 21:01:39 +08:00
parent 8bd4ab0366
commit 64ff1bf7d5
508 changed files with 43575 additions and 14129 deletions
@@ -0,0 +1,87 @@
# 看板变更监听系统 (TaskBoard Monitor)
## 功能概述
通用型任务看板变更监听系统,可监控 Markdown 格式任务看板的状态变化,自动触发通知和激活逻辑。
## 适用场景
- 多小组并行开发项目
- 任务看板状态变更需要即时通知
- 依赖触发:当某任务完成时自动激活下游任务
- 阻塞解除:当阻塞问题解决时自动通知相关小组
## 核心能力
1. **状态解析**:从 Markdown 看板中提取任务状态
2. **变更检测**:对比上一次状态,检测新增变化
3. **触发动作**:根据配置执行相应动作(通知、记录日志等)
4. **可配置**:支持自定义看板路径、触发规则、通知方式
5. **跨项目复用**:只需指定看板路径即可复用
## 使用方式
### 基础监控
```
TaskBoard Monitor: 检查 docs/小组任务书/任务执行状态看板.md
```
### 带触发条件的监控
```
TaskBoard Monitor:
看板路径: docs/任务看板.md
触发条件: 任何任务状态变为"✅已完成"
动作: 输出变更报告
```
### 完整配置
```
TaskBoard Monitor:
看板路径: docs/任务看板.md
状态字段: 编号|任务|状态|Owner
触发规则:
- 当状态变为"✅已完成" → 记录完成时间,输出完成报告
- 当状态变为"🟢可立即启动" → 检查依赖是否满足,输出激活建议
- 当状态变为"🔴阻塞" → 记录阻塞原因
输出: 变更报告 + 动作建议
```
## 输出格式
系统会输出:
1. **变更摘要**:本次检测到的所有变化
2. **触发动作**:每个变化对应的建议动作
3. **统计信息**:各状态任务数量
## 技术实现
- 读取 Markdown 看板文件
- 使用正则表达式解析任务表格
- 维护状态缓存(.taskboard-cache.json
- 支持自定义触发规则
- 状态图标:✅已完成、🟢可立即启动、🔵进行中、⏳等待中、🔴阻塞、⚪未启动、🟡延期
## 复用方法
### 1. 复制 Skill 到其他项目
```bash
# 复制整个目录
cp -r .workbuddy/skills/taskboard-monitor /目标项目/.workbuddy/skills/
```
### 2. 在新项目中使用
```bash
python .workbuddy/skills/taskboard-monitor/taskboard_monitor.py "docs/你的任务看板.md"
```
### 3. 自定义看板格式
看板需满足以下格式:
- 包含表头:`| 编号 | 任务 | 状态 | Owner | ...`
- 任务编号格式:`X-Tn`(如 A-T1, B-T2, C-T3
- 状态列包含图标:✅🟢🔵⏳🔴⚪🟡
@@ -0,0 +1,537 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
看板变更监听系统 (TaskBoard Monitor)
通用型任务看板变更检测工具
功能:
1. 读取 Markdown 格式的任务看板
2. 解析任务表格,提取任务状态
3. 与上一次状态对比,检测变更
4. 输出变更报告和建议动作
"""
import json
import re
import os
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field, asdict
from enum import Enum
class TaskStatus(Enum):
"""任务状态枚举"""
NOT_STARTED = "" # 未启动
READY = "🟢" # 可立即启动
IN_PROGRESS = "🔵" # 进行中
COMPLETED = "" # 已完成
BLOCKED = "🔴" # 阻塞
DELAYED = "🟡" # 延期
WAITING = "" # 等待中
UNKNOWN = "" # 未知
@dataclass
class Task:
"""单个任务"""
id: str # 任务编号,如 "A-T1"
name: str # 任务名称
status: str # 状态图标
owner: str # 负责人
start_date: str # 开始日期
end_date: str # 预计完成
actual_end: str # 实际完成
notes: str # 备注
@property
def status_enum(self) -> TaskStatus:
"""获取状态枚举"""
for s in TaskStatus:
if s.value in self.status:
return s
return TaskStatus.UNKNOWN
@dataclass
class TaskChange:
"""任务变更记录"""
task_id: str
task_name: str
old_status: str
new_status: str
change_type: str # "completed", "ready", "blocked", "waiting", "in_progress"
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
@dataclass
class MonitorResult:
"""监控结果"""
board_path: str
total_tasks: int
changes: List[TaskChange]
stats: Dict[str, int]
actions: List[str]
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
class TaskBoardMonitor:
"""任务看板监听器"""
def __init__(self, board_path: str, cache_dir: str = ".taskboard-cache"):
"""
初始化监听器
Args:
board_path: 看板文件路径(Markdown格式)
cache_dir: 缓存目录,用于存储上一次的状态
"""
self.board_path = Path(board_path)
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
self.cache_file = self.cache_dir / f"{self.board_path.stem}_cache.json"
def read_board(self) -> str:
"""读取看板文件内容"""
if not self.board_path.exists():
raise FileNotFoundError(f"看板文件不存在: {self.board_path}")
return self.board_path.read_text(encoding="utf-8")
def parse_tasks(self, content: str) -> List[Task]:
"""解析看板内容,提取所有任务列表"""
tasks = []
# 按行扫描,直接处理
lines = content.split('\n')
for i, line in enumerate(lines):
# 检测任务表格区域的表头
if '| 编号 | 任务 | 状态 |' in line:
# 继续扫描后续行,直到下一个非任务表区域
for j in range(i + 1, min(i + 30, len(lines))):
row = lines[j]
# 跳过分隔行(如 |------|------|...|
if re.match(r'^\|[\s\-]+\|', row):
continue
if '|' not in row:
continue
parts = [p.strip() for p in row.split('|') if p.strip()]
# 遇到完整的下一个表头或里程碑,停止这个任务表
if len(parts) >= 2:
# 里程碑行
if '里程碑' in parts[0]:
break
# 完整的表头行(包含"任务"和"状态"
if '任务' in row and '状态' in row:
break
# 匹配任务编号(如 A-T1, B-T1, C-T1
if len(parts) >= 2 and re.match(r'^[A-Z]-T\d+$', parts[0]):
task = Task(
id=parts[0],
name=parts[1] if len(parts) > 1 else "",
status=parts[2] if len(parts) > 2 else "",
owner=parts[3] if len(parts) > 3 else "",
start_date=parts[4] if len(parts) > 4 else "",
end_date=parts[5] if len(parts) > 5 else "",
actual_end=parts[6] if len(parts) > 6 else "",
notes=parts[7] if len(parts) > 7 else ""
)
tasks.append(task)
return tasks
return tasks
def load_cache(self) -> Dict[str, Any]:
"""加载上一次的缓存状态"""
if not self.cache_file.exists():
return {}
try:
return json.loads(self.cache_file.read_text(encoding="utf-8"))
except Exception:
return {}
def save_cache(self, tasks: List[Task], stats: Dict[str, int]):
"""保存当前状态到缓存"""
cache_data = {
"timestamp": datetime.now().isoformat(),
"tasks": {
t.id: {
"status": t.status,
"owner": t.owner,
"actual_end": t.actual_end
}
for t in tasks
},
"stats": stats
}
self.cache_file.write_text(json.dumps(cache_data, ensure_ascii=False, indent=2), encoding="utf-8")
def detect_changes(self, tasks: List[Task], old_cache: Dict[str, Any]) -> List[TaskChange]:
"""检测任务变更"""
changes = []
old_tasks = old_cache.get("tasks", {})
for task in tasks:
old_task = old_tasks.get(task.id, {})
if not old_task:
# 新任务
continue
old_status = old_task.get("status", "")
new_status = task.status
if old_status != new_status:
# 状态发生变化
change_type = self._get_change_type(old_status, new_status)
change = TaskChange(
task_id=task.id,
task_name=task.name,
old_status=old_status,
new_status=new_status,
change_type=change_type
)
changes.append(change)
return changes
def _get_change_type(self, old_status: str, new_status: str) -> str:
"""判断变更类型"""
if TaskStatus.COMPLETED.value in new_status:
return "completed"
elif TaskStatus.READY.value in new_status:
return "ready"
elif TaskStatus.BLOCKED.value in new_status:
return "blocked"
elif TaskStatus.WAITING.value in new_status:
return "waiting"
elif TaskStatus.IN_PROGRESS.value in new_status:
return "in_progress"
elif TaskStatus.DELAYED.value in new_status:
return "delayed"
return "changed"
def parse_dependencies(self, tasks: List[Task]) -> Dict[str, List[str]]:
"""
解析任务依赖关系
从任务的备注/阻塞字段中提取依赖任务编号
支持格式:
- "依赖A-T8"
- "依赖A-T8, B-T1"
- "A-T1~T6完成后"
- "A-T8通过后"
"""
deps = {}
for task in tasks:
task_deps = []
# 查找备注字段中的依赖
notes = task.notes
# 模式1: 依赖X-Tn (如 "依赖A-T8")
import re
dep_pattern1 = r'依赖([A-Z]-T\d+)'
task_deps.extend(re.findall(dep_pattern1, notes))
# 模式2: X-Tn~X-Tn完成后 (如 "A-T1~T6完成后")
dep_pattern2 = r'([A-Z]-T\d+)~\1'
matches = re.findall(dep_pattern2, notes)
for match in matches:
task_deps.append(match)
# 模式3: X-Tn通过后 / X-Tn完成后 (如 "A-T8通过后")
dep_pattern3 = r'([A-Z]-T\d+)通过后|([A-Z]-T\d+)完成后'
for match in re.finditer(dep_pattern3, notes):
if match.group(1):
task_deps.append(match.group(1))
elif match.group(2):
task_deps.append(match.group(2))
if task_deps:
deps[task.id] = list(set(task_deps))
return deps
def check_and_activate_tasks(self, tasks: List[Task]) -> List[str]:
"""
检查依赖是否满足,自动激活等待中的任务
Returns:
激活的任务列表
"""
# 解析依赖关系
deps = self.parse_dependencies(tasks)
# 构建已完成任务集合
completed = {t.id for t in tasks if TaskStatus.COMPLETED.value in t.status}
# 检查每个等待中的任务
activated = []
for task in tasks:
if TaskStatus.WAITING.value not in task.status:
continue
task_deps = deps.get(task.id, [])
# 如果没有依赖,或者所有依赖都已完成
if not task_deps or all(dep in completed for dep in task_deps):
activated.append(task.id)
return activated
def auto_start_ready_tasks(self, tasks: List[Task]) -> List[str]:
"""
自动开始可立即启动的任务(🟢 → 🔵)
实现全自动流转:可立即启动的任务自动开始执行
Returns:
自动开始的任务列表
"""
started = []
for task in tasks:
# 只处理"可立即启动"状态的任务
if TaskStatus.READY.value in task.status:
started.append(task.id)
if started:
try:
content = self.read_board()
lines = content.split('\n')
new_lines = []
for line in lines:
for task_id in started:
if f"| {task_id} |" in line and "🟢可立即启动" in line:
line = line.replace("🟢可立即启动", "🔵进行中")
print(f" 🚀 自动开始执行: {task_id}")
new_lines.append(line)
content = '\n'.join(new_lines)
self.board_path.write_text(content, encoding='utf-8')
except Exception as e:
print(f" ❌ 自动开始任务失败: {e}")
return started
def auto_update_board(self, activated_tasks: List[str]) -> bool:
"""
自动更新看板,将激活的任务状态从等待中改为可立即启动
Args:
activated_tasks: 要激活的任务ID列表
Returns:
是否成功更新
"""
if not activated_tasks:
return False
try:
content = self.read_board()
for task_id in activated_tasks:
# 替换等待中状态为可立即启动
# 格式: | task_id | ... | ⏳等待中 | ... → | task_id | ... | 🟢可立即启动 | ...
old_pattern = f"| {task_id} |"
# 需要找到包含 task_id 和 ⏳等待中 的行
lines = content.split('\n')
new_lines = []
for line in lines:
if f"| {task_id} |" in line and "⏳等待中" in line:
line = line.replace("⏳等待中", "🟢可立即启动")
print(f" 🔄 自动激活: {task_id}")
new_lines.append(line)
content = '\n'.join(new_lines)
# 写回文件
self.board_path.write_text(content, encoding='utf-8')
return True
except Exception as e:
print(f" ❌ 自动更新失败: {e}")
return False
def generate_actions(self, changes: List[TaskChange]) -> List[str]:
"""根据变更生成建议动作"""
actions = []
for change in changes:
if change.change_type == "completed":
actions.append(f"{change.task_id} 已完成:{change.task_name}")
actions.append(f" → 检查是否有依赖此任务的其他任务,准备激活")
elif change.change_type == "ready":
actions.append(f"🟢 {change.task_id} 可立即启动:{change.task_name}")
actions.append(f" → 通知负责人开始执行")
elif change.change_type == "blocked":
actions.append(f"🔴 {change.task_id} 阻塞:{change.task_name}")
actions.append(f" → 记录阻塞原因,通知项目经理协调")
elif change.change_type == "waiting":
actions.append(f"{change.task_id} 变为等待中:{change.task_name}")
actions.append(f" → 等待依赖任务完成后激活")
return actions
def calculate_stats(self, tasks: List[Task]) -> Dict[str, int]:
"""统计各状态任务数量"""
stats = {
"total": len(tasks),
"not_started": 0,
"ready": 0,
"in_progress": 0,
"completed": 0,
"blocked": 0,
"delayed": 0,
"waiting": 0
}
for task in tasks:
status = task.status_enum
if status == TaskStatus.NOT_STARTED:
stats["not_started"] += 1
elif status == TaskStatus.READY:
stats["ready"] += 1
elif status == TaskStatus.IN_PROGRESS:
stats["in_progress"] += 1
elif status == TaskStatus.COMPLETED:
stats["completed"] += 1
elif status == TaskStatus.BLOCKED:
stats["blocked"] += 1
elif status == TaskStatus.DELAYED:
stats["delayed"] += 1
elif status == TaskStatus.WAITING:
stats["waiting"] += 1
return stats
def monitor(self, verbose: bool = True, auto_activate: bool = True) -> MonitorResult:
"""
执行一次监控检查
Args:
verbose: 是否输出详细信息
auto_activate: 是否自动激活依赖满足的任务
Returns:
监控结果
"""
# 读取看板
content = self.read_board()
tasks = self.parse_tasks(content)
# 加载缓存
old_cache = self.load_cache()
# 检测变更
changes = self.detect_changes(tasks, old_cache)
# 检查并自动激活任务
activated_tasks = []
started_tasks = []
if auto_activate:
activated_tasks = self.check_and_activate_tasks(tasks)
if activated_tasks:
# 自动更新看板
self.auto_update_board(activated_tasks)
# 重新读取看板获取最新状态
content = self.read_board()
tasks = self.parse_tasks(content)
# 自动开始可立即执行的任务(🟢 → 🔵)
started_tasks = self.auto_start_ready_tasks(tasks)
if started_tasks:
# 重新读取看板
content = self.read_board()
tasks = self.parse_tasks(content)
# 统计
stats = self.calculate_stats(tasks)
# 生成动作建议
actions = self.generate_actions(changes)
# 添加激活信息
if activated_tasks:
actions.append(f"\n🎉 自动激活任务: {', '.join(activated_tasks)}")
if started_tasks:
actions.append(f"\n🚀 自动开始执行: {', '.join(started_tasks)}")
# 保存缓存
self.save_cache(tasks, stats)
# 输出结果
result = MonitorResult(
board_path=str(self.board_path),
total_tasks=len(tasks),
changes=changes,
stats=stats,
actions=actions
)
if verbose:
self._print_result(result, old_cache)
return result
def _print_result(self, result: MonitorResult, old_cache: Dict[str, Any]):
"""打印监控结果"""
print(f"\n{'='*60}")
print(f"📋 看板变更监听报告")
print(f"{'='*60}")
print(f"📂 看板: {result.board_path}")
print(f"⏰ 时间: {result.timestamp}")
print(f"📊 总任务数: {result.total_tasks}")
# 统计
stats = result.stats
print(f"\n📈 状态统计:")
print(f" ✅ 已完成: {stats['completed']}")
print(f" 🟢 可立即启动: {stats['ready']}")
print(f" 🔵 进行中: {stats['in_progress']}")
print(f" ⏳ 等待中: {stats['waiting']}")
print(f" 🔴 阻塞: {stats['blocked']}")
print(f" ⚪ 未启动: {stats['not_started']}")
# 变更
if result.changes:
print(f"\n🔄 检测到 {len(result.changes)} 项变更:")
for change in result.changes:
print(f"{change.task_id}: {change.old_status}{change.new_status}")
print(f" 任务: {change.task_name}")
print(f" 类型: {change.change_type}")
else:
print(f"\n✅ 无变更检测到")
# 动作建议
if result.actions:
print(f"\n🎯 建议动作:")
for action in result.actions:
print(f" {action}")
print(f"{'='*60}\n")
def main():
"""主函数 - 演示用法"""
import sys
# 默认看板路径
board_path = sys.argv[1] if len(sys.argv) > 1 else "docs/小组任务书/任务执行状态看板.md"
# 创建监控器并执行
monitor = TaskBoardMonitor(board_path)
result = monitor.monitor()
# 返回码表示是否有变更
return 0 if len(result.changes) > 0 else 1
if __name__ == "__main__":
exit(main())