worklog.py

"""The worklog text format: pure parsing and aggregation engine.

No filesystem access, no GTK -- see store.py for file handling.

Format:
# YYYY-MM-DD [comment]
H:MM (token) comment on opening a task
H:MM comment on continuing the open task
H:MM (*) comment on ending the task
[anything else] -- a free note attached to the current segment
"""
import re
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, date


@dataclass
class Segment:
    task: str
    start: datetime
    end: datetime | None
    comment: str
    notes: list[str] = field(default_factory=list)


def project(token):
    return token.split('-')[0]


DAY_RE = re.compile(r"^#\s*(\d{4})-(\d{2})-(\d{2})")
TIME_RE = re.compile(r"^(\d{1,2}):(\d{2})(?:\s*\(([^)]+)\))?\s*(.*)$")


def day_header(day):
    return f"# {day.isoformat()}"


def open_line(time, token, comment=""):
    text = f"{time.hour}:{time.minute:02} ({token})"
    return f"{text} {comment}" if comment else text


def stop_line(time, comment=""):
    text = f"{time.hour}:{time.minute:02} (*)"
    return f"{text} {comment}" if comment else text


def note_line(time, text):
    if text.startswith('('):
        text = '\\' + text
    return f"{time.hour}:{time.minute:02} {text}"


class Journal:
    def __init__(self, segments, open_segment, problems):
        self.segments = segments
        self.open_segment = open_segment
        self.problems = problems

    def current(self):
        return self.open_segment.task if self.open_segment else None

    def recent_tasks(self, limit=None):
        by_start = sorted(self.segments, key=lambda s: s.start, reverse=True)
        if self.open_segment:
            by_start = [self.open_segment, *by_start]
        seen = dict.fromkeys(s.task for s in by_start)
        return list(seen)[:limit]

    def by_task(self):
        groups = defaultdict(list)
        for s in self.segments:
            groups[s.task].append(s)
        return groups

    def by_day(self):
        groups = defaultdict(list)
        for s in self.segments:
            groups[s.start.date()].append(s)
        return groups

    def total_minutes(self, segments=None):
        segments = self.segments if segments is None else segments
        return sum((s.end - s.start).total_seconds() // 60 for s in segments)


def parse(text):
    lines = text.splitlines() if isinstance(text, str) else list(text)
    segments = []
    problems = []
    day = None
    open_seg = None
    unclosed_days = []

    def close(end):
        nonlocal open_seg
        if open_seg is not None:
            if open_seg.task != '*':
                segments.append(Segment(open_seg.task, open_seg.start, end, open_seg.comment, open_seg.notes))
            open_seg = None

    for lineno, line in enumerate(lines, start=1):
        if m := DAY_RE.match(line):
            if open_seg is not None:
                unclosed_days.append((day, lineno, open_seg))
                open_seg = None
            year, month, daynum = (int(x) for x in m.groups())
            try:
                day = date(year, month, daynum)
            except ValueError as e:
                problems.append((lineno, f"invalid date: {line!r}: {e}"))
                day = None
            continue
        if m := TIME_RE.match(line):
            hour, minute, token, comment = m.groups()
            hour, minute = int(hour), int(minute)
            if not (0 <= hour < 24 and 0 <= minute < 60):
                problems.append((lineno, f"invalid time: {line!r}"))
                continue
            if day is None:
                problems.append((lineno, f"time entry before any day header: {line!r}"))
                continue
            when = datetime(day.year, day.month, day.day, hour, minute)
            if token is not None and not token.strip():
                problems.append((lineno, f"empty task token: {line!r}"))
                continue
            token = token.strip() if token else token
            if token:
                if open_seg is not None and when < open_seg.start:
                    problems.append((lineno, f"time went backwards: {line!r}"))
                    continue
                if token == '*' and open_seg is not None and comment.strip():
                    open_seg.notes.append(comment.strip())
                close(when)
                if token != '*':
                    open_seg = Segment(token, when, None, comment.strip())
            elif open_seg is not None:
                if when < open_seg.start:
                    problems.append((lineno, f"time went backwards: {line!r}"))
                    continue
                note = comment.strip()
                if note.startswith('\\('):
                    note = note[1:]
                if note:
                    open_seg.notes.append(note)

    for bad_day, lineno, seg in unclosed_days:
        problems.append((lineno, f"day {bad_day} left unclosed by task {seg.task!r}"))

    return Journal(segments, open_seg, problems)


def test():
    import sys

    text = (
        "# 2026-03-01\n"
        "9:00 (proj-1) morning\n"
        "9:30 (proj-2) switch\n"
        "10:00 (*)\n"
        "# 2026-03-02\n"
        "8:00 (proj-3) next day\n"
        "8:45 (proj-1) back to one\n"
        "9:15 (*)\n"
    )
    journal = parse(text)

    if journal.problems:
        print(f"FAIL: unexpected problems parsing a clean worklog: {journal.problems}")
        sys.exit(1)
    if journal.current() is not None or journal.open_segment is not None:
        print("FAIL: a worklog whose last day is closed should not be running")
        sys.exit(1)

    recent = journal.recent_tasks(20)
    if len(recent) != len(set(recent)):
        print("FAIL: recent_tasks returned duplicates")
        sys.exit(1)
    starts = {s.task: s.start for s in journal.segments}
    for i in range(len(recent) - 1):
        if starts[recent[i]] < starts[recent[i + 1]]:
            print("FAIL: recent_tasks not ordered most-recent-first")
            sys.exit(1)

    j = parse("# 2026-03-01\n9:00 (proj-1) doing stuff\n")
    if j.current() != "proj-1" or j.open_segment is None or j.problems:
        print("FAIL: (a) latest open day should be current")
        sys.exit(1)

    j = parse(
        "# 2026-03-01\n"
        "9:00 (proj-1) doing stuff\n"
        "# 2026-03-02\n"
        "9:00 (proj-2) other stuff\n"
        "10:00 (*)\n"
    )
    if not j.problems:
        print("FAIL: (b) earlier unclosed day should be a problem")
        sys.exit(1)
    if j.current() is not None or j.open_segment is not None:
        print("FAIL: (b) latest day is closed, should not be running")
        sys.exit(1)
    if [s.task for s in j.segments] != ["proj-2"]:
        print(f"FAIL: (b) expected only proj-2 counted, got {[s.task for s in j.segments]}")
        sys.exit(1)

    j = parse(
        "# 2026-03-01\n"
        "10:00 (proj-1) start\n"
        "9:00 (proj-2) backwards\n"
    )
    if not j.problems:
        print("FAIL: (c) backwards time should be a problem")
        sys.exit(1)
    if j.current() != "proj-1":
        print("FAIL: (c) earlier good segment should survive as open segment")
        sys.exit(1)

    j = parse(
        "# 2026-03-01\n"
        "9:00 (proj-1) work\n"
        "9:30 (*)\n"
        "9:45 (proj-2) more work\n"
    )
    if j.problems:
        print(f"FAIL: (d) unexpected problems: {j.problems}")
        sys.exit(1)
    if any(s.task == '*' for s in j.segments):
        print("FAIL: (d) idle stretch should not be counted")
        sys.exit(1)
    first = [s for s in j.segments if s.task == 'proj-1']
    if len(first) != 1 or first[0].end.minute != 30:
        print("FAIL: (d) proj-1 should end at 9:30, idle stretch excluded")
        sys.exit(1)

    j = parse(
        "# 2026-03-01\n"
        "9:00 (proj-1) work\n"
        "9:15 a continuation note\n"
    )
    if j.segments:
        print("FAIL: (e) continuation note should not close the open segment")
        sys.exit(1)
    if j.open_segment is None or j.open_segment.notes != ["a continuation note"]:
        print(f"FAIL: (e) note should land in open segment notes, got {j.open_segment}")
        sys.exit(1)

    j = parse(
        "# 2026-03-15\n"
        + open_line(datetime(2026, 3, 15, 8, 5), "proj-1", "x") + "\n"
        + note_line(datetime(2026, 3, 15, 8, 5), "(paren) note") + "\n"
    )
    if j.current() != "proj-1" or j.open_segment.notes != ["(paren) note"]:
        print(f"FAIL: (f) note starting with parenthesized word did not round-trip, got {j.open_segment}")
        sys.exit(1)

    j = parse(
        "# 2026-03-15\n"
        + open_line(datetime(2026, 3, 15, 8, 5), "proj-1", "x") + "\n"
        + note_line(datetime(2026, 3, 15, 8, 6), "(*) note") + "\n"
    )
    if j.current() != "proj-1" or j.open_segment is None:
        print("FAIL: (f) note starting with (*) should not stop the running task")
        sys.exit(1)

    j = parse("# 2026-03-01\n9:00 ( ) huh\n10:00 (*)\n")
    if not j.problems or j.by_task():
        print("FAIL: (g) empty task token should be a problem, not a zero-name task")
        sys.exit(1)

    j = parse(
        "# 2026-03-01\n"
        + open_line(datetime(2026, 3, 1, 9, 0), "proj-1") + "\n"
        + stop_line(datetime(2026, 3, 1, 10, 0), "wrapped up") + "\n"
    )
    if len(j.segments) != 1 or j.segments[0].notes != ["wrapped up"]:
        print(f"FAIL: (h) stop with a comment should attach it as a note, got {j.segments}")
        sys.exit(1)

    print("PASS")


if __name__ == "__main__":
    test()