org-editor/task_manager.py

188 lines
5.4 KiB
Python
Raw Permalink Normal View History

2021-12-26 19:57:59 +00:00
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 = []
2022-03-25 18:26:53 +00:00
if self.docs is None:
self.docs = docs
2021-12-26 19:57:59 +00:00
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:
2022-03-25 18:26:53 +00:00
doc = org_rw.load(open(path), extra_cautious=False)
2021-12-26 19:57:59 +00:00
docs.append(doc)
2022-03-25 18:26:53 +00:00
yield doc
2021-12-26 19:57:59 +00:00
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, progress_callback, complete_callback):
2021-12-26 19:57:59 +00:00
def aux():
if self.docs is None:
2022-03-25 18:26:53 +00:00
last_result = None
# No docs read yet, load them iteratively
2022-03-25 18:26:53 +00:00
for doc in self.load():
result = self.get_agenda_from_doc(doc)
GObject.idle_add(progress_callback, result)
2022-03-25 18:26:53 +00:00
else:
result = self.get_agenda()
print("Result", result)
GObject.idle_add(progress_callback, result)
GObject.idle_add(complete_callback, True)
2021-12-26 19:57:59 +00:00
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,
)
def get_agenda_from_doc(self, doc: OrgDoc) -> Agenda:
headline_count = 0
items_in_agenda = []
now = datetime.now()
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)
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
)
]
return Agenda(
with_hour=sorted(items_with_hour, key=lambda x: x.scheduled.time),
no_hour=other_items,
)