from datetime import datetime, timedelta
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
import json
class SprintStatus(Enum):
PLANNING = "planning"
ACTIVE = "active"
COMPLETED = "completed"
CANCELLED = "cancelled"
class StoryStatus(Enum):
BACKLOG = "backlog"
IN_PROGRESS = "in_progress"
IN_REVIEW = "in_review"
DONE = "done"
BLOCKED = "blocked"
class TaskPriority(Enum):
LOW = 1
MEDIUM = 2
HIGH = 3
CRITICAL = 4
@dataclass
class UserStory:
"""用户故事"""
id: str
title: str
description: str
acceptance_criteria: List[str]
story_points: Optional[int] = None
priority: TaskPriority = TaskPriority.MEDIUM
status: StoryStatus = StoryStatus.BACKLOG
assignee: Optional[str] = None
created_date: datetime = None
updated_date: datetime = None
def __post_init__(self):
if self.created_date is None:
self.created_date = datetime.now()
if self.updated_date is None:
self.updated_date = datetime.now()
@dataclass
class Sprint:
"""Sprint"""
id: str
name: str
start_date: datetime
end_date: datetime
stories: List[str] # story IDs
status: SprintStatus = SprintStatus.PLANNING
capacity: Optional[int] = None # 团队容量（故事点）
velocity: Optional[float] = None # 实际完成速度
@property
def duration_days(self) -> int:
"""Sprint持续天数"""
return (self.end_date - self.start_date).days + 1
@dataclass
class TeamMember:
"""团队成员"""
id: str
name: str
role: str # Product Owner, Scrum Master, Developer
skills: List[str]
availability: float = 1.0 # 可用性系数
current_stories: List[str] = None
def __post_init__(self):
if self.current_stories is None:
self.current_stories = []
class AgileProjectManager:
"""敏捷项目管理器"""
def __init__(self, project_name: str):
self.project_name = project_name
self.stories: Dict[str, UserStory] = {}
self.sprints: Dict[str, Sprint] = {}
self.team_members: Dict[str, TeamMember] = {}
self.product_backlog: List[str] = [] # story IDs
self.sprint_history: List[Dict] = []
def add_user_story(self, story: UserStory):
"""添加用户故事"""
self.stories[story.id] = story
self.product_backlog.append(story.id)
self._sort_backlog()
def _sort_backlog(self):
"""按优先级排序待办列表"""
self.product_backlog.sort(
key=lambda story_id: self.stories[story_id].priority.value,
reverse=True
)
def create_sprint(self, sprint_id: str, name: str, start_date: datetime,
duration_days: int = 14, capacity: int = None) -> Sprint:
"""创建Sprint"""
end_date = start_date + timedelta(days=duration_days - 1)
sprint = Sprint(
id=sprint_id,
name=name,
start_date=start_date,
end_date=end_date,
stories=[],
capacity=capacity
)
self.sprints[sprint_id] = sprint
return sprint
def plan_sprint(self, sprint_id: str, story_ids: List[str]) -> bool:
"""Sprint规划"""
if sprint_id not in self.sprints:
return False
sprint = self.sprints[sprint_id]
# 计算总故事点
total_points = 0
for story_id in story_ids:
if story_id not in self.stories:
return False
story = self.stories[story_id]
if story.story_points is None:
return False
total_points += story.story_points
# 检查容量限制
if sprint.capacity and total_points > sprint.capacity:
print(f"警告: Sprint容量不足 ({total_points} > {sprint.capacity})")
return False
# 分配故事到Sprint
sprint.stories = story_ids
sprint.status = SprintStatus.ACTIVE
# 更新故事状态
for story_id in story_ids:
self.stories[story_id].status = StoryStatus.IN_PROGRESS
# 从待办列表移除
if story_id in self.product_backlog:
self.product_backlog.remove(story_id)
return True
def assign_story(self, story_id: str, assignee_id: str) -> bool:
"""分配用户故事"""
if story_id not in self.stories or assignee_id not in self.team_members:
return False
story = self.stories[story_id]
assignee = self.team_members[assignee_id]
story.assignee = assignee_id
if story_id not in assignee.current_stories:
assignee.current_stories.append(story_id)
return True
def update_story_status(self, story_id: str, status: StoryStatus) -> bool:
"""更新故事状态"""
if story_id not in self.stories:
return False
story = self.stories[story_id]
old_status = story.status
story.status = status
story.updated_date = datetime.now()
# 如果故事完成，从成员当前任务中移除
if status == StoryStatus.DONE and story.assignee:
assignee = self.team_members.get(story.assignee)
if assignee and story_id in assignee.current_stories:
assignee.current_stories.remove(story_id)
return True
def get_sprint_progress(self, sprint_id: str) -> Dict:
"""获取Sprint进度"""
if sprint_id not in self.sprints:
return {}
sprint = self.sprints[sprint_id]
total_stories = len(sprint.stories)
completed_stories = sum(
1 for story_id in sprint.stories
if self.stories[story_id].status == StoryStatus.DONE
)
total_points = sum(
self.stories[story_id].story_points or 0
for story_id in sprint.stories
)
completed_points = sum(
self.stories[story_id].story_points or 0
for story_id in sprint.stories
if self.stories[story_id].status == StoryStatus.DONE
)
return {
'sprint_id': sprint_id,
'total_stories': total_stories,
'completed_stories': completed_stories,
'story_completion_rate': completed_stories / total_stories if total_stories > 0 else 0,
'total_points': total_points,
'completed_points': completed_points,
'point_completion_rate': completed_points / total_points if total_points > 0 else 0,
'status': sprint.status.value
}
def calculate_team_velocity(self, num_sprints: int = 3) -> float:
"""计算团队速度（平均每Sprint完成的故事点）"""
completed_sprints = [
sprint for sprint in self.sprints.values()
if sprint.status == SprintStatus.COMPLETED
]
if not completed_sprints:
return 0.0
# 取最近N个完成的Sprint
recent_sprints = sorted(
completed_sprints,
key=lambda s: s.end_date,
reverse=True
)[:num_sprints]
total_points = 0
for sprint in recent_sprints:
sprint_points = sum(
self.stories[story_id].story_points or 0
for story_id in sprint.stories
if self.stories[story_id].status == StoryStatus.DONE
)
total_points += sprint_points
return total_points / len(recent_sprints) if recent_sprints else 0.0
def generate_burndown_chart(self, sprint_id: str) -> Dict:
"""生成燃尽图数据"""
if sprint_id not in self.sprints:
return {}
sprint = self.sprints[sprint_id]
# 理想燃尽线
total_points = sum(
self.stories[story_id].story_points or 0
for story_id in sprint.stories
)
days = sprint.duration_days
ideal_burndown = [
{'day': i, 'remaining': total_points - (total_points * i / days)}
for i in range(days + 1)
]
# 实际燃尽线（简化版本，实际应该基于每日数据）
actual_burndown = []
# 这里需要实际的每日完成数据，简化处理
return {
'sprint_id': sprint_id,
'total_points': total_points,
'duration_days': days,
'ideal_burndown': ideal_burndown,
'actual_burndown': actual_burndown
}
def conduct_sprint_review(self, sprint_id: str) -> Dict:
"""进行Sprint评审"""
if sprint_id not in self.sprints:
return {}
sprint = self.sprints[sprint_id]
progress = self.get_sprint_progress(sprint_id)
review_results = {
'sprint_id': sprint_id,
'sprint_name': sprint.name,
'review_date': datetime.now(),
'completion_rate': progress['point_completion_rate'],
'total_stories': progress['total_stories'],
'completed_stories': progress['completed_stories'],
'velocity': progress['completed_points']
}
# 记录Sprint历史
self.sprint_history.append(review_results)
# 更新Sprint状态
sprint.status = SprintStatus.COMPLETED
sprint.velocity = progress['completed_points']
return review_results
def suggest_next_sprint_capacity(self) -> int:
"""建议下一个Sprint的容量"""
avg_velocity = self.calculate_team_velocity()
# 基于历史速度的80%-120%范围建议
min_capacity = int(avg_velocity * 0.8)
max_capacity = int(avg_velocity * 1.2)
return max_capacity if avg_velocity > 0 else 20 # 默认容量
def export_project_data(self, file_path: str):
"""导出项目数据"""
data = {
'project_name': self.project_name,
'export_date': datetime.now().isoformat(),
'stories': [
{
'id': story.id,
'title': story.title,
'description': story.description,
'story_points': story.story_points,
'priority': story.priority.value,
'status': story.status