Files
scud_ai/services/tasks/service.py

72 lines
3.3 KiB
Python

"""
===============================================================================
FILE: services/tasks/service.py
PROJECT: SCUD Orion AI (Unified Architecture)
MODULE: services / tasks
ROLE: Единый доменный сервис задач (бизнес-логика и диспетчер операций).
AI-CONTEXT-ANCHORS:
- ANCHOR[TASK_SERVICE_DISPATCHER]: Маршрутизация действий ADD/UPDATE/DELETE/EXPORT.
===============================================================================
"""
from typing import Dict, Any, Optional, List
from .repository import repo_get_tasks, repo_add_task, repo_update_task, repo_delete_task
from .exporter import export_tasks_to_markdown
def get_tasks(user_id: int, status: Optional[str] = None) -> List[Dict[str, Any]]:
"""Получить список задач."""
return repo_get_tasks(user_id, status)
def add_task(user_id: int, module: str, title: str, priority: str = "MEDIUM", due_date: Optional[str] = None, status: str = "BACKLOG") -> Dict[str, Any]:
"""Создать задачу."""
res = repo_add_task(user_id, module, title, priority, due_date, status)
return {"status": "success", "task_id": res["task_id"], "message": f"Задача #{res['id']} создана и добавлена в планы"}
def update_task_details(user_id: int, task_id: str, title: Optional[str] = None, priority: Optional[str] = None, status: Optional[str] = None, due_date: Optional[str] = None) -> Dict[str, Any]:
"""Обновить задачу."""
return repo_update_task(user_id, task_id, title, priority, status, due_date)
def delete_task(user_id: int, task_id: str) -> Dict[str, Any]:
"""Удалить задачу."""
return repo_delete_task(user_id, task_id)
# ANCHOR[TASK_SERVICE_DISPATCHER]
def execute_task_action(
user_id: int,
action: str,
task_id: Optional[str] = None,
title: Optional[str] = None,
priority: Optional[str] = "MEDIUM",
status: Optional[str] = None,
module: Optional[str] = "general",
due_date: Optional[str] = None,
filename: Optional[str] = "ROADMAP.md"
) -> Dict[str, Any]:
"""Консолидированный диспетчер операций над задачами."""
act = (action or "").strip().upper()
if act == "ADD":
if not title:
return {"status": "error", "message": "Для создания задачи требуется указать title"}
return add_task(user_id, module or "general", title, priority or "MEDIUM", due_date, status or "BACKLOG")
elif act == "UPDATE":
if not task_id:
return {"status": "error", "message": "Для обновления требуется указать task_id"}
return update_task_details(user_id, str(task_id), title, priority, status, due_date)
elif act == "DELETE":
if not task_id:
return {"status": "error", "message": "Для удаления требуется указать task_id"}
return delete_task(user_id, str(task_id))
elif act == "EXPORT":
return export_tasks_to_markdown(user_id, filename=filename or "ROADMAP.md", status_filter=status)
return {"status": "error", "message": f"Неизвестное действие action='{action}'"}