207 lines
6.8 KiB
Python
207 lines
6.8 KiB
Python
import json
|
|
import os
|
|
from datetime import date, datetime, timedelta
|
|
from typing import List, Optional, Literal
|
|
|
|
from modules.tasks.task import Task
|
|
|
|
|
|
class TaskHandler:
|
|
def __init__(self, base_dir: Optional[str] = None):
|
|
# Par défaut: Data/tasks (un fichier JSON par tâche)
|
|
self.base_dir = base_dir or os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
|
|
"Data",
|
|
"tasks",
|
|
)
|
|
os.makedirs(self.base_dir, exist_ok=True)
|
|
|
|
def _task_path(self, task_id: str) -> str:
|
|
return os.path.join(self.base_dir, f"{task_id}.json")
|
|
|
|
def add_task(self, task: Task) -> str:
|
|
path = self._task_path(task.id)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(task.to_dict(), f, ensure_ascii=False, indent=2)
|
|
return task.id
|
|
|
|
def get_task(self, task_id: str) -> Optional[Task]:
|
|
path = self._task_path(task_id)
|
|
if not os.path.exists(path):
|
|
return None
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return Task.from_dict(data)
|
|
|
|
def update_task(self, task: Task) -> bool:
|
|
path = self._task_path(task.id)
|
|
if not os.path.exists(path):
|
|
return False
|
|
# pas de gestion updated_at pour rester simple
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(task.to_dict(), f, ensure_ascii=False, indent=2)
|
|
return True
|
|
|
|
def delete_task(self, task_id: str) -> bool:
|
|
path = self._task_path(task_id)
|
|
if os.path.exists(path):
|
|
try:
|
|
os.remove(path)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
return False
|
|
|
|
def list_tasks(self, status: Optional[Literal["todo", "done", "canceled"]] = None) -> List[Task]:
|
|
tasks: List[Task] = []
|
|
for filename in os.listdir(self.base_dir):
|
|
if filename.endswith(".json"):
|
|
try:
|
|
with open(os.path.join(self.base_dir, filename), "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
t = Task.from_dict(data)
|
|
if status is None or t.status == status:
|
|
tasks.append(t)
|
|
except Exception:
|
|
continue
|
|
# Tri par (due_date, priorité, création)
|
|
def _prio_val(p: str) -> int:
|
|
return {"haute": 0, "normale": 1, "basse": 2}.get(p, 1)
|
|
|
|
tasks.sort(key=lambda t: (t.due_date or "", _prio_val(t.priority), t.created_at or ""))
|
|
return tasks
|
|
|
|
def list_tasks_for_date(
|
|
self,
|
|
day_iso: str,
|
|
status: Optional[Literal["todo", "done", "canceled"]] = None,
|
|
) -> List[Task]:
|
|
return [t for t in self.list_tasks(status=status) if (t.due_date or "") == day_iso]
|
|
|
|
def list_today_tasks(self, status: Optional[Literal["todo", "done", "canceled"]] = None) -> List[Task]:
|
|
return self.list_tasks_for_date(date.today().isoformat(), status=status)
|
|
|
|
def list_tomorrow_tasks(self, status: Optional[Literal["todo", "done", "canceled"]] = None) -> List[Task]:
|
|
tomorrow = (date.today() + timedelta(days=1)).isoformat()
|
|
return self.list_tasks_for_date(tomorrow, status=status)
|
|
|
|
def list_overdue_tasks(self) -> List[Task]:
|
|
"""Retourne les tâches en retard (due_date < aujourd'hui) encore à faire."""
|
|
today_iso = date.today().isoformat()
|
|
tasks = self.list_tasks(status="todo")
|
|
return [t for t in tasks if (t.due_date or "") < today_iso]
|
|
|
|
def list_tasks_by_entity(
|
|
self,
|
|
entity_type: Literal["client", "prospect", "project", "campaign"],
|
|
entity_id: str,
|
|
status: Optional[Literal["todo", "done", "canceled"]] = None,
|
|
) -> List[Task]:
|
|
return [
|
|
t
|
|
for t in self.list_tasks(status=status)
|
|
if t.entity_type == entity_type and t.entity_id == entity_id
|
|
]
|
|
|
|
def complete_task(self, task_id: str) -> bool:
|
|
t = self.get_task(task_id)
|
|
if not t:
|
|
return False
|
|
t.status = "done"
|
|
t.completed_at = datetime.utcnow().isoformat()
|
|
return self.update_task(t)
|
|
|
|
def cancel_task(self, task_id: str) -> bool:
|
|
t = self.get_task(task_id)
|
|
if not t:
|
|
return False
|
|
t.status = "canceled"
|
|
return self.update_task(t)
|
|
|
|
def postpone_task(self, task_id: str, new_due_date: str) -> bool:
|
|
t = self.get_task(task_id)
|
|
if not t:
|
|
return False
|
|
t.due_date = new_due_date
|
|
return self.update_task(t)
|
|
|
|
# Raccourcis pour créer des tâches liées
|
|
def add_task_for_client(
|
|
self,
|
|
client_id: str,
|
|
title: str,
|
|
due_date: str,
|
|
description: str = "",
|
|
priority: Literal["basse", "normale", "haute"] = "normale",
|
|
metadata: Optional[dict] = None,
|
|
) -> str:
|
|
task = Task(
|
|
title=title,
|
|
description=description,
|
|
due_date=due_date,
|
|
entity_type="client",
|
|
entity_id=client_id,
|
|
priority=priority,
|
|
metadata=metadata or {},
|
|
)
|
|
return self.add_task(task)
|
|
|
|
def add_task_for_prospect(
|
|
self,
|
|
prospect_id: str,
|
|
title: str,
|
|
due_date: str,
|
|
description: str = "",
|
|
priority: Literal["basse", "normale", "haute"] = "normale",
|
|
metadata: Optional[dict] = None,
|
|
) -> str:
|
|
task = Task(
|
|
title=title,
|
|
description=description,
|
|
due_date=due_date,
|
|
entity_type="prospect",
|
|
entity_id=prospect_id,
|
|
priority=priority,
|
|
metadata=metadata or {},
|
|
)
|
|
return self.add_task(task)
|
|
|
|
def add_task_for_project(
|
|
self,
|
|
project_id: str,
|
|
title: str,
|
|
due_date: str,
|
|
description: str = "",
|
|
priority: Literal["basse", "normale", "haute"] = "normale",
|
|
metadata: Optional[dict] = None,
|
|
) -> str:
|
|
task = Task(
|
|
title=title,
|
|
description=description,
|
|
due_date=due_date,
|
|
entity_type="project",
|
|
entity_id=project_id,
|
|
priority=priority,
|
|
metadata=metadata or {},
|
|
)
|
|
return self.add_task(task)
|
|
|
|
def add_task_for_campaign(
|
|
self,
|
|
campaign_id: str,
|
|
title: str,
|
|
due_date: str,
|
|
description: str = "",
|
|
priority: Literal["basse", "normale", "haute"] = "normale",
|
|
metadata: Optional[dict] = None,
|
|
) -> str:
|
|
task = Task(
|
|
title=title,
|
|
description=description,
|
|
due_date=due_date,
|
|
entity_type="campaign",
|
|
entity_id=campaign_id,
|
|
priority=priority,
|
|
metadata=metadata or {},
|
|
)
|
|
return self.add_task(task)
|