org-editor/task_manager.py

import logging
import os
import sys
import threading
from datetime import datetime
from typing import List

import org_rw
from org_rw import OrgDoc, OrgTime

EXTENSIONS = (".org", ".org.txt")

from gi.repository import GObject


def is_today(ot: OrgTime):
    now = datetime.now()
    return (
        (ot.time.year == now.year)
        and (ot.time.month == now.month)
        and (ot.time.day == now.day)
    )


class Agenda:
    def __init__(
        self,
        /,
        with_hour: List[org_rw.Headline],
        no_hour: List[org_rw.Headline],
    ):
        self.with_hour = with_hour
        self.no_hour = no_hour

    def print(self):
        for item in self.with_hour:
            print(item.scheduled.time, item.state, item.title)

        if len(self.with_hour) > 0:
            print("--------")

        for item in self.no_hour:
            print(item.scheduled.time, item.state, item.title)

class TaskManager:
    docs: List[OrgDoc]
    threads: List[threading.Thread]

    def __init__(self, docs_path: os.PathLike):
        self.docs_path = docs_path
        self.threads = []
        self.docs = None

    def load(self):
        t0 = datetime.now()
        top = os.path.abspath(self.docs_path)

        docs = []

        for root, dirs, files in os.walk(top):
            # Prune dirs
            i = 0
            while i < len(dirs):
                if dirs[i].startswith(".git"):
                    del dirs[i]
                else:
                    i += 1

            # Process files
            for name in files:
                if all(map(lambda ext: not name.endswith(ext), EXTENSIONS)):
                    continue

                path = os.path.join(root, name)

                try:
                    doc = org_rw.load(open(path), extra_cautious=True)
                    docs.append(doc)
                except Exception as err:
                    import traceback

                    traceback.print_exc()
                    print(f"== On {path}")
                    sys.exit(1)

        t1 = datetime.now()
        logging.info("Loaded {} files in {}s".format(len(docs), t1 - t0))

        self.docs = docs

    def get_task_list(self, callback):
        def aux():
            if self.docs is None:
                self.load()
            result = self.get_agenda()
            print("Result", result)
            GObject.idle_add(callback, result)

        thread = threading.Thread(target=aux)
        thread.start()
        self.threads.append(thread)

    def get_agenda(self) -> Agenda:
        headline_count = 0
        items_in_agenda = []
        now = datetime.now()

        for doc in self.docs:
            for hl in doc.getAllHeadlines():
                headline_count += 1

                if (
                    hl.scheduled
                    and isinstance(hl.scheduled, OrgTime)
                    and hl.scheduled.time.active
                ):
                    if is_today(hl.scheduled):
                        items_in_agenda.append(hl)
                    elif (hl.scheduled.time.to_datetime() < now) and hl.is_todo:
                        items_in_agenda.append(hl)

        logging.info("Read {} items".format(headline_count))
        logging.info("{} items in agenda today".format(len(items_in_agenda)))

        items_with_hour = [
            item
            for item in items_in_agenda
            if item.scheduled and is_today(item.scheduled) and item.scheduled.time.hour
        ]
        other_items = [
            item
            for item in items_in_agenda
            if not (
                item.scheduled and is_today(item.scheduled) and item.scheduled.time.hour
            )
        ]

        logging.info("{} items today for a specific hour".format(len(items_with_hour)))

        return Agenda(
            with_hour=sorted(items_with_hour, key=lambda x: x.scheduled.time),
            no_hour=other_items,
        )