Timer.py

import os
import sys
import tempfile
import time as pytime
import sys as pysystem
import os as pyos
import shutil as pyshutil
import datetime as pydate
import json as pyjson
from _pyrepl import readline

from dataclasses import dataclass as pydataclass
import shlex
from typing import Optional, Dict


class Utils:
    _POSIX_fd = None
    _POSIX_old_term_settings = None

    @staticmethod
    def scan_for_all_json_names(custom_dir: str):
        json_files = []
        for filename in os.listdir(custom_dir):
            if filename.endswith(".json"):
                json_files.append(filename)
        return json_files

    @staticmethod
    def auto_complete(tasks: list, cmd_text: str, state):
        # @TODO -- implement this!
        buffer = readline.get_line_buffer()
        tokens = buffer.strip().split()

        if len(tokens) >= 1 and tokens[0] == "start":
            options = [t for t in tasks if t.startswith(cmd_text)]
        else:
            options = []

        try:
            return options[state]
        except IndexError:
            return []

    @staticmethod
    def format_time(seconds) -> str:

        weeks = seconds // 604800
        seconds %= 604800

        days = seconds // 86400
        seconds %= 86400

        hours = seconds // 3600
        seconds %= 3600

        minutes = seconds // 60
        seconds %= 60

        return f"{weeks:02}:{days:02}:{hours:02}:{minutes:02}:{seconds:02}"

    @staticmethod
    def parse_command(line: str) -> (str, str):
        try:
            parts = shlex.split(line)
        except Exception:
            return None, None
        if not parts:
            return None, None
        cmd = parts[0].lower()
        arg = " ".join(parts[1:]) if len(parts) > 1 else ""
        return cmd, arg

    @staticmethod
    def clear_line():
        try:
            cols = pyshutil.get_terminal_size().columns
            if not cols or cols < 1:
                cols = ASSUMED_MAX_LINE_SIZE
        except Exception:
            cols = ASSUMED_MAX_LINE_SIZE
        pysystem.stdout.write("\r" + " " * cols + "\r")
        pysystem.stdout.flush()

    @staticmethod
    def get_key_nonblocking():
        if pyos.name == WINDOWS_SYSTEM:
            if msvcrt.kbhit():
                return msvcrt.getwch()
            return None
        else:
            dr, _, _ = select.select([sys.stdin], [], [], 0)
            if dr:
                return sys.stdin.read(1)
            return None

    @staticmethod
    def flush_input():
        if pyos.name == WINDOWS_SYSTEM:
            while msvcrt.kbhit():
                msvcrt.getwch()
        else:
            dr, _, _ = select.select([sys.stdin], [], [], 0)
            while dr:
                pysystem.stdin.read(1)
                dr, _, _ = select.select([sys.stdin], [], [], 0)

    @staticmethod
    def posix_terminal_setup():
        if pyos.name == WINDOWS_SYSTEM:
            return
        else:
            fd = pysystem.stdin.fileno()
            Utils._POSIX_fd = fd
            Utils._POSIX_old_term_settings = termios.tcgetattr(fd)
            new = termios.tcgetattr(fd)
            new[3] = new[3] & ~termios.ICANON & ~termios.ECHO
            termios.tcsetattr(fd, termios.TCSADRAIN, new)

    @staticmethod
    def restore_posix_terminal():
        if pyos.name == WINDOWS_SYSTEM:
            return
        else:
            if Utils._POSIX_fd is not None and Utils._POSIX_old_term_settings is not None:
                try:
                    termios.tcsetattr(Utils._POSIX_fd, termios.TCSADRAIN, Utils._POSIX_old_term_settings)
                except Exception:
                    pass


AUTO_ARCHIVE: bool = True

FILE_MAIN_DIR: str = "active"
FILE_ARCHIVE_DIR: str = "archive"
FILE_NAME: str = "tasks_"

CURRENT_DEV_WEEK: str = pydate.datetime.now(pydate.timezone.utc).strftime("%Y-%m-%W")

if AUTO_ARCHIVE:
    try:
        print("Auto-Archive -- Initiated")
        expected_name = FILE_MAIN_DIR + "/" + FILE_NAME + CURRENT_DEV_WEEK + ".json"
        possible_names = Utils.scan_for_all_json_names(FILE_MAIN_DIR)
        actual_name = FILE_MAIN_DIR + "/" + possible_names[0]
        if len(possible_names) > 1:
            print("Auto-Archive error:: Several possible files detected within:" + FILE_MAIN_DIR)
        elif expected_name != actual_name:
            print("Auto-Archive -- Attempting to archive file: " + actual_name)
            try:
                source_path = actual_name
                destination_path = pyos.path.join(FILE_ARCHIVE_DIR, possible_names[0])  # @TODO: FIX!
                pyos.makedirs(FILE_ARCHIVE_DIR, exist_ok=True)
                print(destination_path)
                pyshutil.copy(source_path, destination_path)
                print("Auto-Archive -- Copy successful")
                pyos.remove(source_path)
            except Exception:
                print("Auto-Archive error:: Failed to copy file... >:(")
        else:
            print("Auto-Archive -- No changes detected")
        print("\n\n\n")
    except Exception:
        pass

FILE_PATH: str = FILE_MAIN_DIR + "/" + FILE_NAME + CURRENT_DEV_WEEK + ".json"

WINDOWS_SYSTEM: str = "nt"
STOP_TIMER_KEY: chr = 's'
ASSUMED_MAX_LINE_SIZE: int = 80

CMD_HELP: str = "help"
CMD_CREATE: str = "create"
CMD_START: str = "start"
CMD_CLOSE: str = "close"
CMD_LIST: str = "list"
CMD_QUIT: str = "quit"

if pyos.name == WINDOWS_SYSTEM:
    import msvcrt
else:
    import select
    import termios


@pydataclass
class Task:
    name: str
    time: int
    status: bool
    date_created: str
    date_closed: Optional[str] = None

    def to_dict(self) -> dict:
        return {
            "name": self.name,
            "time": self.time,
            "status": self.status,
            "date_created": self.date_created,
            "date_closed": self.date_closed,
        }

    @staticmethod
    def from_dict(data: dict) -> "Task":
        name = data.get("name", "")
        time = int(data.get("time", 0))
        status = bool(data.get("status", False))
        date_created = data.get("date_created", pydate.datetime.now(pydate.timezone.utc).isoformat())
        date_closed = data.get("date_closed", None)
        return Task(name=name, time=time, status=status, date_created=date_created, date_closed=date_closed)


class TaskInfo:
    def __init__(self, file_path: str):
        self.file_path = file_path
        self.tasks: Dict[str, Task] = {}

    def _load(self) -> Dict[str, Task]:
        if not pyos.path.exists(self.file_path):
            return {}
        try:
            with open(self.file_path, "r", encoding="utf-8") as file:
                data = pyjson.load(file)
            if not isinstance(data, dict):
                return {}
            loaded: Dict[str, Task] = {}
            for k, value in data.items():
                if not isinstance(k, str):
                    continue
                if isinstance(value, dict):
                    try:
                        loaded[k] = Task.from_dict(value)
                    except Exception:
                        continue
            return loaded
        except Exception:
            return {}

    def load(self) -> None:
        self.tasks = self._load()

    def _save(self) -> None:
        parent = pyos.path.dirname(self.file_path)
        if parent and not pyos.path.exists(parent):
            try:
                pyos.makedirs(parent, exist_ok=True)
            except Exception:
                raise
        serializable = {tid: task.to_dict() for tid, task in self.tasks.items()}
        dir_for_temp = parent if parent else "."
        temp_fd, temp_path = tempfile.mkstemp(prefix="tasks_", suffix=".tmp", dir=dir_for_temp)
        try:
            with pyos.fdopen(temp_fd, "w", encoding="utf-8") as file:
                pyjson.dump(serializable, file, ensure_ascii=False, indent=2)
                file.flush()
                try:
                    pyos.fsync(file.fileno())
                except Exception:
                    pass
            try:
                pyos.replace(temp_path, self.file_path)
            except Exception:
                pyos.rename(temp_path, self.file_path)
        finally:
            if pyos.path.exists(temp_path):
                try:
                    pyos.remove(temp_path)
                except Exception:
                    pass

    def save(self) -> None:
        self._save()

    def all_tasks(self) -> Dict[str, Task]:
        return dict(self.tasks)

    def get_task(self, task_name: str) -> Task | None:
        task = self.tasks.get(task_name)
        if task is None:
            return None
        if task.status is False:
            return None
        return task

    def make_new_task(self, task_name: str) -> None:
        if self.get_task(task_name) is not None:
            return
        task_id = task_name
        task: Task = Task(task_name, 0, True,
                          pydate.datetime.now(pydate.timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"), None)
        self.tasks[task_id] = task
        self.save()

    def update_task(self, task_name: str, new_time: int, new_status: bool):
        task = self.get_task(task_name)
        if task is None:
            return
        task.time = new_time
        if not new_status:
            task.status = new_status
            task.date_closed = pydate.datetime.now(pydate.timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
        self.save()


def help_cmd():
    print("Commands:")
    print("\t " + CMD_HELP + ":: displays commands")
    print("\t " + CMD_CREATE + ":: creates a new task, if the name is available")
    print("\t " + CMD_START + ":: starts the timer for a task")
    print("\t " + CMD_LIST + ":: lists all tasks")
    print("\t " + CMD_CLOSE + ":: closes a task, keeping it on record while making it unavailable")
    print("\t " + CMD_QUIT + ":: closes the application")


def create_task(task_info: TaskInfo, task_name: str) -> None:
    task_info.load()
    task = task_info.get_task(task_name)
    if task is not None:
        print("Task already exists!")
        return
    task_info.make_new_task(task_name)


def start_task(task_info: TaskInfo, task_name: str) -> None:
    timing: bool = True
    task_info.load()
    task = task_info.get_task(task_name)
    if task is None:
        print("Task not found")
        return

    print("Starting Task: " + str(task_name) + "\n\tPress " + STOP_TIMER_KEY + " to stop the timer...")
    time = task.time

    if pyos.name != WINDOWS_SYSTEM:
        try:
            Utils.posix_terminal_setup()
        except Exception:
            print("Warning")

    try:
        Utils.flush_input()
        try:
            while timing:
                key = Utils.get_key_nonblocking()
                if key:
                    if pyos.name == WINDOWS_SYSTEM:
                        if key in ('\x00', '\xe0'):
                            if msvcrt.kbhit():
                                msvcrt.getwch()
                        else:
                            if key.lower() == STOP_TIMER_KEY:
                                Utils.clear_line()
                                print("Stopping...")
                                timing = False
                    else:
                        if key.lower() == STOP_TIMER_KEY:
                            Utils.clear_line()
                            print("Stopping...")
                            timing = False

                pysystem.stdout.write("\rTime:" + Utils.format_time(time))
                pysystem.stdout.flush()
                pytime.sleep(1)
                time += 1
        except KeyboardInterrupt:
            pass
    finally:
        if pyos.name != WINDOWS_SYSTEM:
            try:
                Utils.restore_posix_terminal()
            except Exception:
                pass
    task_info.update_task(task.name, time, True)
    print("Stopped!")


def close_task(task_info: TaskInfo, task_name: str) -> None:
    task_info.load()
    task = task_info.get_task(task_name)
    if task is None:
        print("Task not found")
    print("Are you sure you want to close Task:" + str(task_name) + "?")
    print("Type 'yes' to close...")
    try:
        line = input("> ")
    except (KeyboardInterrupt, EOFError):
        print("\nAborting close operation...")
        return
    cmd_a = Utils.parse_command(line)
    if cmd_a[0].lower() == "yes":
        print("Closing Task" + str(task_name) + "...")
        task_info.update_task(task_name, task.time, False)
        print("Task closed")


def list_tasks(task_info: TaskInfo):
    task_info.load()
    for task_name in task_info.tasks:
        task = task_info.tasks[task_name]
        """
        name: str
        time: int
        status: bool
        date_created: str
        date_closed: Optional[str] = None
        """
        print("\tName: " + str(task.name))
        print("\tTime: " + str(Utils.format_time(task.time)))
        print("\tStatus: " + ("ACTIVE" if task.status else "CLOSED"))
        print("\tDate Created: " + str(task.date_created))
        print("\tDate Closed: " + str(task.date_closed))


if __name__ == '__main__':
    running: bool = True

    tasks = TaskInfo(FILE_PATH)
    tasks.load()

    print("===| Timer --- App |===")
    while running:
        print("\nType '" + str(CMD_HELP) + "' for a list of commands...")
        try:
            line = input("> ")
        except (KeyboardInterrupt, EOFError):
            print("\nAborting...")
            running = False
            continue

        cmd, arg = Utils.parse_command(line)

        if cmd == CMD_HELP:
            help_cmd()
        elif cmd == CMD_CREATE:
            create_task(tasks, arg)
        elif cmd == CMD_START:
            start_task(tasks, arg)
        elif cmd == CMD_CLOSE:
            close_task(tasks, arg)
        elif cmd == CMD_LIST:
            list_tasks(tasks)
        elif cmd == CMD_QUIT:
            running = False
        else:
            print("Unknown command: " + cmd)

    tasks.save()
    exit(0)