tasks.py

#!/usr/bin/env python3
"""TaskManager model.

Pure data + persistence: it loads/saves the per-list task files and mutates
them. It performs NO terminal I/O -- mutators take their input as arguments,
honor ``dry_run`` (skip the write), and return the affected Task so the
dispatcher can print confirmations. Rendering lives in render.py.

Each list is a JSON file under ``lists/`` holding the four categories, each a
list of task objects ``{title, status, notes}``. Old plain-text ``.txt`` lists
(one title per line, sections split by ``==========``) are migrated to JSON
automatically the first time they are loaded.
"""
import os
import json

# Repo location is derived from this file so the tool works wherever the
# TaskManager checkout lives (previously hardcoded to ~/Dev/TaskManager).
HOME_DIR = os.path.dirname(os.path.realpath(__file__))
PREFS_PATH = '%s/prefs.txt' % HOME_DIR
LISTS_DIR = '%s/lists' % HOME_DIR
NAME = 'name'
LIST = 'list'
TXT_EXT = '.txt'
JSON_EXT = '.json'
TERM_DELIMITER = '=' * 10
FIRST = 'first'
LAST = 'last'
UP = -1
DOWN = 1
WAIT = 'wait'
BACKLOG = 'backlog'
SHORTTERM = 'short-term'
LONGTERM = 'long-term'

# Persisted section order in each list file. Do not reorder: it defines the
# on-disk layout (and the legacy .txt section order).
MODES = [SHORTTERM, LONGTERM, WAIT, BACKLOG]

# Category names/aliases accepted by resolve_mode() and the -L/--list flag.
_MODE_ALIASES = {
    'short': SHORTTERM, 'short-term': SHORTTERM, 'st': SHORTTERM, 's': SHORTTERM,
    'long': LONGTERM, 'long-term': LONGTERM, 'lt': LONGTERM, 'l': LONGTERM,
    'wait': WAIT, 'watch': WAIT, 'w': WAIT,
    'backlog': BACKLOG, 'bl': BACKLOG, 'b': BACKLOG,
}


def resolve_mode(name):
    """Map a user-supplied category name/alias to a mode constant.

    Returns None for None input; raises ValueError on anything unknown.
    """
    if name is None:
        return None
    key = name.strip().lower()
    if key in _MODE_ALIASES:
        return _MODE_ALIASES[key]
    raise ValueError("unknown list '%s' (use short|long|wait|backlog)" % name)


def list_display_name(filename):
    """User-facing name of a list file (drop directory and extension)."""
    return os.path.splitext(os.path.basename(filename))[0]


class Task(object):
    """A single task: a title plus optional free-form status and notes."""

    def __init__(self, title, status='', notes=None):
        self.title = title
        self.status = status or ''
        self.notes = list(notes) if notes else []

    @classmethod
    def from_raw(cls, raw):
        """Build a Task from JSON (dict), a legacy plain string, or a Task."""
        if isinstance(raw, Task):
            return raw
        if isinstance(raw, str):
            return cls(raw)
        return cls(raw.get('title', ''), raw.get('status', ''),
                   raw.get('notes') or [])

    def to_dict(self):
        return {'title': self.title, 'status': self.status, 'notes': self.notes}

    def __str__(self):
        # Lets existing '%s' % task call sites keep printing the title.
        return self.title


class TaskManager(object):
    def __init__(self):
        self._path = HOME_DIR

        if os.path.isfile(PREFS_PATH):
            with open(PREFS_PATH) as fs:
                self._prefs = fs.readlines()
        else:
            self._prefs = None

        self._mode = SHORTTERM
        self._active_name = ''          # base name, e.g. 'Today'
        self._task_file = ''            # absolute path to the active .json file
        self._task_lists = {m: [] for m in MODES}
        self._lists = self._get_lists()

        self._initialize_active_list()

    # --- read-only views -------------------------------------------------
    @property
    def st_tasks(self):
        return self._task_lists[SHORTTERM]

    @property
    def lt_tasks(self):
        return self._task_lists[LONGTERM]

    @property
    def wait_tasks(self):
        return self._task_lists[WAIT]

    @property
    def backlog_tasks(self):
        return self._task_lists[BACKLOG]

    @property
    def mode(self):
        return self._mode

    @property
    def active_list(self):
        """User-facing name of the active list (no extension)."""
        return self._active_name

    @property
    def list_names(self):
        return list(self._lists)

    @property
    def selected_list(self):
        return self._task_lists[self._mode]

    @property
    def lists(self):
        return {SHORTTERM: {NAME: 'Short-Term', LIST: self.st_tasks},
                LONGTERM: {NAME: 'Long-Term', LIST: self.lt_tasks},
                BACKLOG: {NAME: 'Backlog', LIST: self.backlog_tasks},
                WAIT: {NAME: 'Wait/Watch', LIST: self.wait_tasks}}

    def _list_for(self, mode=None):
        """The underlying list object for a category (defaults to active mode)."""
        return self._task_lists[mode or self._mode]

    # --- persistence -----------------------------------------------------
    def _initialize_active_list(self):
        if self._prefs and self._prefs[0].strip():
            self.change_default(self._prefs[0].strip())
        else:
            active_list = self._lists[0] if self._lists else 'new_list'
            self.change_default(active_list)

    def _write_to_file(self, fn, text):
        with open(fn, 'w') as f:
            f.write(text)

    def _read_json(self, path):
        with open(path) as f:
            try:
                data = json.load(f)
            except ValueError:
                data = {}
        return {m: [Task.from_raw(t) for t in data.get(m, [])] for m in MODES}

    def _read_legacy(self, path):
        """Parse an old plain-text list (titles only, sections split by
        TERM_DELIMITER) into Task lists."""
        task_types = {m: [] for m in MODES}
        with open(path) as fs:
            lines = [l for l in fs.read().split('\n') if l != '']
        index = 0
        for line in lines:
            if line == TERM_DELIMITER:
                index += 1
                continue
            if index < len(MODES):
                task_types[MODES[index]].append(Task(line))
        return task_types

    def _get_lists(self):
        """Sorted base names of all lists (either .txt or .json) on disk."""
        names = set()
        if os.path.isdir(LISTS_DIR):
            for f in os.listdir(LISTS_DIR):
                base, ext = os.path.splitext(f)
                if ext in (TXT_EXT, JSON_EXT):
                    names.add(base)
        return sorted(names)

    def _load_list(self):
        base = self._active_name
        json_path = '%s/%s%s' % (LISTS_DIR, base, JSON_EXT)
        txt_path = '%s/%s%s' % (LISTS_DIR, base, TXT_EXT)
        self._task_file = json_path
        if os.path.isfile(json_path):
            self._task_lists = self._read_json(json_path)
        elif os.path.isfile(txt_path):
            # One-time migration: read the old plaintext, write JSON, drop .txt.
            self._task_lists = self._read_legacy(txt_path)
            self._write_changes()
            os.remove(txt_path)
        else:
            self._task_lists = {m: [] for m in MODES}
        self._lists = self._get_lists()

    def set_mode(self, mode):
        self._mode = mode

    def change_default(self, new, dry_run=False):
        base = list_display_name(new)
        if dry_run:
            return base
        self._active_name = base
        self._write_to_file(PREFS_PATH, base)
        self._load_list()
        return base

    def _write_changes(self):
        data = {m: [t.to_dict() for t in self._task_lists[m]] for m in MODES}
        self._write_to_file(self._task_file, json.dumps(data, indent=2) + '\n')

    # --- task reads ------------------------------------------------------
    def peek_task(self, index, mode=None):
        """Return the Task at index without mutating. Raises IndexError."""
        return self._list_for(mode)[index]

    # --- task mutations (no terminal I/O; return the affected Task) ------
    def append_task(self, text, mode=None, dry_run=False):
        task = Task(text)
        if not dry_run:
            self._list_for(mode).append(task)
            self._write_changes()
        return task

    def prepend_task(self, text, mode=None, dry_run=False):
        task = Task(text)
        if not dry_run:
            self._list_for(mode).insert(0, task)
            self._write_changes()
        return task

    def delete_task(self, index, mode=None, dry_run=False):
        task = self._list_for(mode)[index]
        if not dry_run:
            del self._list_for(mode)[index]
            self._write_changes()
        return task

    def finish_task(self, index, mode=None, dry_run=False):
        # Logging the finished task to the devlog is the dispatcher's job.
        return self.delete_task(index, mode=mode, dry_run=dry_run)

    def move_task(self, index, new_slot=None, direction=None, mode=None,
                  dry_run=False):
        lst = self._list_for(mode)
        task = lst[index]
        if dry_run:
            return task
        if direction is not None:
            lst.insert(index + direction, lst.pop(index))
        elif new_slot is not None:
            if new_slot == FIRST:
                lst.insert(0, lst.pop(index))
            else:
                lst.append(lst.pop(index))
        self._write_changes()
        return task

    def change_type(self, index, new_type, mode=None, dry_run=False):
        src = mode or self._mode
        task = self._task_lists[src][index]
        if not dry_run:
            del self._task_lists[src][index]
            self._task_lists[new_type].append(task)
            self._write_changes()
        return task, new_type

    def set_title(self, index, title, mode=None, dry_run=False):
        task = self._list_for(mode)[index]
        if not dry_run:
            task.title = title
            self._write_changes()
        return task

    def set_status(self, index, status, mode=None, dry_run=False):
        task = self._list_for(mode)[index]
        if not dry_run:
            task.status = status or ''
            self._write_changes()
        return task

    def add_note(self, index, note, mode=None, dry_run=False):
        task = self._list_for(mode)[index]
        if not dry_run:
            task.notes.append(note)
            self._write_changes()
        return task

    def remove_note(self, index, note_index, mode=None, dry_run=False):
        task = self._list_for(mode)[index]
        note = task.notes[note_index]
        if not dry_run:
            del task.notes[note_index]
            self._write_changes()
        return note

    # --- list-level operations -------------------------------------------
    def list_exists(self, name):
        return list_display_name(name) in self._lists

    def move_list(self, new_name, dry_run=False):
        base = list_display_name(new_name)
        if dry_run:
            return base
        os.rename(self._task_file, '%s/%s%s' % (LISTS_DIR, base, JSON_EXT))
        self.change_default(base)
        return base

    def remove_list(self, name, dry_run=False):
        base = list_display_name(name)
        if dry_run:
            return base
        for ext in (JSON_EXT, TXT_EXT):
            path = '%s/%s%s' % (LISTS_DIR, base, ext)
            if os.path.isfile(path):
                os.remove(path)
        self._lists = self._get_lists()
        return base