xiaoxia-saas/packages/application/project_management_use_cases.py

"""项目管理 Use Cases"""

from packages.domain import Milestone, Task, TaskIssue, TaskPriority, TaskStatus
from packages.ports import MilestoneRepository, TaskIssueRepository, TaskRepository


class CreateTaskUseCase:
    """创建任务用例"""

    def __init__(self, task_repo: TaskRepository):
        self.task_repo = task_repo

    def execute(
        self,
        project_id: str,
        name: str,
        description: str = "",
        priority: TaskPriority = TaskPriority.MEDIUM,
        parent_task_id: str = "",
        assignee_user_id: str = "",
    ) -> Task:
        task = Task.create(
            project_id=project_id,
            name=name,
            description=description,
            priority=priority,
            parent_task_id=parent_task_id,
            assignee_user_id=assignee_user_id,
        )
        return self.task_repo.create(task)


class ListProjectTasksUseCase:
    """获取项目任务列表用例"""

    def __init__(self, task_repo: TaskRepository):
        self.task_repo = task_repo

    def execute(self, project_id: str) -> list[Task]:
        return self.task_repo.list_by_project(project_id)


class UpdateTaskStatusUseCase:
    """更新任务状态用例"""

    def __init__(self, task_repo: TaskRepository):
        self.task_repo = task_repo

    def execute(self, task_id: str, new_status: TaskStatus) -> Task:
        task = self.task_repo.get_by_id(task_id)
        if not task:
            raise ValueError(f"Task {task_id} not found")
        task.update_status(new_status)
        return self.task_repo.update(task)


class UpdateTaskProgressUseCase:
    """更新任务进度用例"""

    def __init__(self, task_repo: TaskRepository):
        self.task_repo = task_repo

    def execute(self, task_id: str, progress: float) -> Task:
        task = self.task_repo.get_by_id(task_id)
        if not task:
            raise ValueError(f"Task {task_id} not found")
        task.update_progress(progress)
        return self.task_repo.update(task)


class CreateMilestoneUseCase:
    """创建里程碑用例"""

    def __init__(self, milestone_repo: MilestoneRepository):
        self.milestone_repo = milestone_repo

    def execute(
        self,
        project_id: str,
        name: str,
        description: str = "",
    ) -> Milestone:
        milestone = Milestone.create(
            project_id=project_id,
            name=name,
            description=description,
        )
        return self.milestone_repo.create(milestone)


class ListProjectMilestonesUseCase:
    """获取项目里程碑列表用例"""

    def __init__(self, milestone_repo: MilestoneRepository):
        self.milestone_repo = milestone_repo

    def execute(self, project_id: str) -> list[Milestone]:
        return self.milestone_repo.list_by_project(project_id)


class CreateTaskIssueUseCase:
    """创建任务问题用例"""

    def __init__(self, issue_repo: TaskIssueRepository):
        self.issue_repo = issue_repo

    def execute(
        self,
        task_id: str,
        project_id: str,
        title: str,
        description: str = "",
        created_by_user_id: str = "",
    ) -> TaskIssue:
        issue = TaskIssue.create(
            task_id=task_id,
            project_id=project_id,
            title=title,
            description=description,
            created_by_user_id=created_by_user_id,
        )
        return self.issue_repo.create(issue)


class ListTaskIssuesUseCase:
    """获取任务问题列表用例"""

    def __init__(self, issue_repo: TaskIssueRepository):
        self.issue_repo = issue_repo

    def execute(self, task_id: str) -> list[TaskIssue]:
        return self.issue_repo.list_by_task(task_id)


class ResolveTaskIssueUseCase:
    """解决任务问题用例"""

    def __init__(self, issue_repo: TaskIssueRepository):
        self.issue_repo = issue_repo

    def execute(self, issue_id: str) -> TaskIssue:
        issue = self.issue_repo.get_by_id(issue_id)
        if not issue:
            raise ValueError(f"TaskIssue {issue_id} not found")
        issue.mark_resolved()
        return self.issue_repo.update(issue)