281 lines
8.3 KiB
Python
281 lines
8.3 KiB
Python
# api/projects.py
|
|
# Projects Tab Backend — Rose's Projects & Tasks Dashboard
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
HERMES_HOME = Path.home() / ".hermes"
|
|
PROJECTS_DIR = HERMES_HOME / "projects"
|
|
DATA_FILE = HERMES_HOME / "data" / "projects.json"
|
|
|
|
DATA_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def _load():
|
|
"""Lädt data/projects.json oder gibt leere Struktur zurück."""
|
|
if DATA_FILE.exists():
|
|
with DATA_FILE.open(encoding="utf-8") as f:
|
|
return json.loads(f.read())
|
|
return {"version": "1.0.0", "projects": [], "daily_tasks": [], "recurring_tasks": []}
|
|
|
|
|
|
def _save(data):
|
|
"""Speichert data/projects.json."""
|
|
DATA_FILE.write_text(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
|
|
def list_projects():
|
|
"""Liest projects/ Ordner aus, synced mit data/projects.json.
|
|
|
|
Jeder Unterordner in ~/.hermes/projects/ wird als Projekt registriert.
|
|
Bereits existierende Projekte (nach folder) werden nicht dupliziert.
|
|
"""
|
|
data = _load()
|
|
|
|
# Sync: jede Folder in projects/ → Projekt-Eintrag wenn nicht vorhanden
|
|
for folder in sorted(PROJECTS_DIR.iterdir()):
|
|
if folder.is_dir() and not folder.name.startswith('.'):
|
|
exists = any(p.get('folder') == folder.name for p in data['projects'])
|
|
if not exists:
|
|
data['projects'].append({
|
|
"id": folder.name,
|
|
"name": folder.name.replace('-', ' ').replace('_', ' ').title(),
|
|
"description": "",
|
|
"folder": folder.name,
|
|
"category": "unknown",
|
|
"status": "active",
|
|
"created": datetime.now().date().isoformat(),
|
|
"updated": datetime.now().isoformat(),
|
|
"tasks": []
|
|
})
|
|
|
|
_save(data)
|
|
return data['projects']
|
|
|
|
|
|
def get_project(project_id):
|
|
"""Holt ein einzelnes Projekt nach ID."""
|
|
data = _load()
|
|
return next((p for p in data['projects'] if p['id'] == project_id), None)
|
|
|
|
|
|
def create_task(project_id, task):
|
|
"""Erstellt Task in Projekt oder als daily/recurring.
|
|
|
|
Args:
|
|
project_id: ID des Projekts (für project tasks) oder None
|
|
task: dict mit title, task_type, status, priority, due, tags
|
|
|
|
Returns:
|
|
Das erstellte Task-Objekt mit generierter ID
|
|
"""
|
|
data = _load()
|
|
|
|
# ID generieren basierend auf task_type
|
|
task_type = task.get('task_type', 'project')
|
|
if task_type == 'daily':
|
|
existing = len(data.get('daily_tasks', []))
|
|
task['id'] = f"daily-{existing + 1:03d}"
|
|
elif task_type == 'recurring':
|
|
existing = len(data.get('recurring_tasks', []))
|
|
task['id'] = f"recurring-{existing + 1:03d}"
|
|
else:
|
|
existing = sum(len(p.get('tasks', [])) for p in data['projects'])
|
|
task['id'] = f"project-{existing + 1:03d}"
|
|
|
|
task['created'] = datetime.now().isoformat()
|
|
task['completed'] = None
|
|
|
|
if task_type == 'project' and project_id:
|
|
for p in data['projects']:
|
|
if p['id'] == project_id:
|
|
if 'tasks' not in p:
|
|
p['tasks'] = []
|
|
p['tasks'].append(task)
|
|
p['updated'] = datetime.now().isoformat()
|
|
break
|
|
elif task_type == 'project':
|
|
# Unassigned project task → find or create Inbox project
|
|
inbox = next((p for p in data['projects'] if p['id'] == 'inbox'), None)
|
|
if not inbox:
|
|
inbox = {
|
|
'id': 'inbox',
|
|
'name': '📥 Inbox',
|
|
'color': '#6366f1',
|
|
'tasks': [],
|
|
'created': datetime.now().isoformat(),
|
|
'updated': datetime.now().isoformat()
|
|
}
|
|
data['projects'].insert(0, inbox)
|
|
inbox['tasks'].append(task)
|
|
inbox['updated'] = datetime.now().isoformat()
|
|
elif task_type == 'daily':
|
|
data['daily_tasks'].append(task)
|
|
elif task_type == 'recurring':
|
|
data['recurring_tasks'].append(task)
|
|
|
|
_save(data)
|
|
return task
|
|
|
|
|
|
def update_task(task_id, updates):
|
|
"""Updated Task (status, priority, due, etc.).
|
|
|
|
Sucht Task in allen drei Listen (projects.tasks, daily_tasks, recurring_tasks).
|
|
"""
|
|
data = _load()
|
|
|
|
# Search in project tasks
|
|
for p in data['projects']:
|
|
for t in p.get('tasks', []):
|
|
if t['id'] == task_id:
|
|
t.update(updates)
|
|
p['updated'] = datetime.now().isoformat()
|
|
_save(data)
|
|
return t
|
|
|
|
# Search in daily tasks
|
|
for t in data.get('daily_tasks', []):
|
|
if t['id'] == task_id:
|
|
t.update(updates)
|
|
_save(data)
|
|
return t
|
|
|
|
# Search in recurring tasks
|
|
for t in data.get('recurring_tasks', []):
|
|
if t['id'] == task_id:
|
|
t.update(updates)
|
|
_save(data)
|
|
return t
|
|
|
|
return None
|
|
|
|
|
|
def delete_task(task_id):
|
|
"""Löscht Task aus allen drei Listen."""
|
|
data = _load()
|
|
|
|
# Remove from project tasks
|
|
for p in data['projects']:
|
|
p['tasks'] = [t for t in p.get('tasks', []) if t['id'] != task_id]
|
|
|
|
# Remove from daily tasks
|
|
data['daily_tasks'] = [t for t in data.get('daily_tasks', []) if t['id'] != task_id]
|
|
|
|
# Remove from recurring tasks
|
|
data['recurring_tasks'] = [t for t in data.get('recurring_tasks', []) if t['id'] != task_id]
|
|
|
|
_save(data)
|
|
return True
|
|
|
|
|
|
def get_all_tasks():
|
|
"""Holt alle Tasks für Kanban-View.
|
|
|
|
Fügt project_name hinzu für Project-Tasks.
|
|
Setzt Defaults für fehlende Felder (defensive).
|
|
"""
|
|
data = _load()
|
|
tasks = []
|
|
|
|
# Defaults für alle Tasks
|
|
DEFAULT_FIELDS = {
|
|
'title': 'Untitled Task',
|
|
'task_type': 'project',
|
|
'status': 'todo',
|
|
'priority': 'p2',
|
|
'due': None,
|
|
'tags': [],
|
|
'project_id': None,
|
|
'project_name': None,
|
|
'completed': None,
|
|
}
|
|
|
|
for p in data['projects']:
|
|
for t in p.get('tasks', []):
|
|
t = dict(t) # Copy to avoid mutating original
|
|
t['project_name'] = p.get('name')
|
|
# Apply defaults for missing fields
|
|
for k, v in DEFAULT_FIELDS.items():
|
|
t.setdefault(k, v)
|
|
tasks.append(t)
|
|
|
|
for t in data.get('daily_tasks', []):
|
|
t = dict(t)
|
|
for k, v in DEFAULT_FIELDS.items():
|
|
t.setdefault(k, v)
|
|
t.setdefault('status', 'pending')
|
|
tasks.append(t)
|
|
|
|
for t in data.get('recurring_tasks', []):
|
|
t = dict(t)
|
|
for k, v in DEFAULT_FIELDS.items():
|
|
t.setdefault(k, v)
|
|
t.setdefault('status', 'pending')
|
|
tasks.append(t)
|
|
|
|
return tasks
|
|
|
|
|
|
def get_stats():
|
|
"""Statistiken für Projects Tab.
|
|
|
|
Returns:
|
|
dict mit total_tasks, done, today_completed, active_projects,
|
|
streak, by_priority, by_type, overdue
|
|
"""
|
|
from datetime import date, timedelta
|
|
|
|
data = _load()
|
|
all_tasks = get_all_tasks()
|
|
|
|
done = [t for t in all_tasks if t.get('status') == 'done']
|
|
today = date.today().isoformat()
|
|
today_done = [t for t in done if (t.get('completed') or '').startswith(today)]
|
|
|
|
# Streak: consecutive days with completions going backwards
|
|
streak = 0
|
|
check_date = date.today()
|
|
done_dates = set()
|
|
for t in done:
|
|
c = t.get('completed')
|
|
if c:
|
|
done_dates.add(c[:10])
|
|
while True:
|
|
d = check_date.isoformat()
|
|
if d in done_dates:
|
|
streak += 1
|
|
check_date -= timedelta(days=1)
|
|
else:
|
|
break
|
|
|
|
# By priority
|
|
by_priority = {'p1': 0, 'p2': 0, 'p3': 0}
|
|
for t in all_tasks:
|
|
p = t.get('priority', 'p2')
|
|
if p in by_priority:
|
|
by_priority[p] += 1
|
|
|
|
# By type
|
|
by_type = {'project': 0, 'daily': 0, 'recurring': 0}
|
|
for t in all_tasks:
|
|
by_type[t.get('task_type', 'project')] += 1
|
|
|
|
# Overdue
|
|
overdue = [
|
|
t for t in all_tasks
|
|
if t.get('due') and t['due'] < today and t.get('status') != 'done'
|
|
]
|
|
|
|
return {
|
|
"total_tasks": len(all_tasks),
|
|
"done": len(done),
|
|
"today_completed": len(today_done),
|
|
"active_projects": len([p for p in data['projects'] if p.get('status') == 'active']),
|
|
"streak": streak,
|
|
"by_priority": by_priority,
|
|
"by_type": by_type,
|
|
"overdue": len(overdue),
|
|
}
|