PK! _4GGblockbuster/core/__init__.py# pylint: disable=C0111 __version__ = "0.1.0" DATE_FORMAT = "%Y-%m-%d" PK!$g g blockbuster/core/factory.py"""Functions to create a Task instance from text in todo.txt format""" import re from blockbuster.core.model import Task def _done(todotxt): """ Returns ------- tuple boolean indicating whether the task is complete the todo.txt string stripped of any completed portion """ done = todotxt.startswith("x") if done: todotxt = todotxt[1:].strip() return done, todotxt def _priority(todotxt): """ Returns ------- tuple Any priority character The todo.txt string stripped of any priority character """ regex = re.compile(r"\s*\((\S)\)") match = regex.search(todotxt) priority = None if match: priority = match.group(0).strip().lstrip("(").rstrip(")") todotxt = regex.sub("", todotxt).strip() return priority, todotxt def _dates(todotxt): regex = re.compile(r"\s*(\d{4}-\d{2}-\d{2})") match = regex.search(todotxt) dates = {"completed_at": None, "created_at": None} if match: matches = [item for item in regex.finditer(todotxt)] if len(matches) == 2: dates["completed_at"] = matches[0].group().strip() dates["created_at"] = matches[1].group().strip() else: dates["created_at"] = matches[0].group().strip() todotxt = regex.sub("", todotxt).strip() return dates, todotxt def _projects(todotxt): regex = re.compile(r"\s*(\+\w+)") match = regex.search(todotxt) projects = None if match: projects = [ item.group().strip().lstrip("+") for item in regex.finditer(todotxt) ] todotxt = regex.sub("", todotxt).strip() return projects, todotxt def _contexts(todotxt): regex = re.compile(r"\s+(\@\w+)") match = regex.search(todotxt) contexts = None if match: contexts = [ item.group().strip().lstrip("@") for item in regex.finditer(todotxt) ] todotxt = regex.sub("", todotxt).strip() return contexts, todotxt def _tags(todotxt): regex = re.compile(r"\s+(\w+\:\w+)") match = regex.search(todotxt) tags = None if match: items = (item.group().strip() for item in regex.finditer(todotxt)) tags = {item.split(":")[0]: item.split(":")[1] for item in items} todotxt = regex.sub("", todotxt).strip() return tags, todotxt def create_task(todotxt): """Create a Task instance from a string in todo.txt format""" todotxt = todotxt.strip() task = { "description": None, "done": False, "priority": None, "completed_at": None, "created_at": None, "projects": None, "contexts": None, "tags": None, } task["done"], todotxt = _done(todotxt) task["priority"], todotxt = _priority(todotxt) dates, todotxt = _dates(todotxt) task["completed_at"] = dates["completed_at"] task["created_at"] = dates["created_at"] task["projects"], todotxt = _projects(todotxt) task["contexts"], todotxt = _contexts(todotxt) task["tags"], todotxt = _tags(todotxt) task["description"] = todotxt.strip() return Task(**task) PK!% blockbuster/core/io.py"""CRUD operations on todo.txt files""" from hashlib import sha256 from blockbuster.core.factory import create_task from blockbuster.core.model import TasksAdded, TasksDeleted, TasksUpdated def read_tasks(file): """Create Tasks instances from a todo.txt file Parameters ---------- file A Path instance Returns ------- list of Task instances """ with file.open("r") as f: tasks_raw = f.readlines() return [create_task(todotxt) for todotxt in tasks_raw] def add_tasks(additions, file): """Add tasks to a todo.txt file Parameters ---------- additions A list or tuple of strings in todo.txt format file A Path instance Returns ------- TasksAdded An instance of blockbuster.core.model.Event """ with file.open("r+") as f: prior_hash = sha256(f.read().encode(encoding="UTF-8")).hexdigest() for task in additions: f.write(f"{task}\n") f.seek(0) new_hash = sha256(f.read().encode(encoding="UTF-8")).hexdigest() return TasksAdded( tasks=additions, file=file, prior_hash=prior_hash, new_hash=new_hash ) def delete_tasks(deletions, file): """Delete lines from a todo.txt file Parameters ---------- deletions A list or tuple of index numbers indicating which tasks to delete by their position in the file file A Path instance Returns ------- TasksDeleted An instance of blockbuster.core.model.Event """ with file.open("r+") as f: prior_hash = sha256(f.read().encode(encoding="UTF-8")).hexdigest() f.seek(0) tasks = f.readlines() keep_ids = [i for i in range(len(tasks)) if i not in deletions] f.seek(0) for task in [tasks[i] for i in keep_ids]: f.write(task) f.truncate() f.seek(0) new_hash = sha256(f.read().encode(encoding="UTF-8")).hexdigest() deleted_tasks = [tasks[i] for i in deletions] return TasksDeleted( tasks=deleted_tasks, file=file, prior_hash=prior_hash, new_hash=new_hash ) def update_tasks(updates, file): """Update lines in a todo.txt file Parameters ---------- updates A dictionary mapping the index number of the task within the file to a string of its updated content file A Path instance Returns ------- TasksUpdated An instance of blockbuster.core.model.Event """ with file.open("r+") as f: prior_hash = sha256(f.read().encode(encoding="UTF-8")).hexdigest() f.seek(0) tasks = f.readlines() new_tasks = [ updates[item[0]] if item[0] in updates else tasks[item[0]] for item in enumerate(tasks) ] print(new_tasks) f.seek(0) for task in new_tasks: f.write(f"{task.strip()}\n") f.truncate() f.seek(0) new_hash = sha256(f.read().encode(encoding="UTF-8")).hexdigest() updated_tasks = [value for value in updates.values()] return TasksUpdated( tasks=updated_tasks, file=file, prior_hash=prior_hash, new_hash=new_hash ) PK!J3blockbuster/core/model.py"""Classes to represent tasks and the events which cause them to change state. These classes have no dependencies on any other component of the blockbuster namespace. """ from abc import ABCMeta from datetime import datetime from blockbuster.core import DATE_FORMAT from marshmallow import Schema, fields, pre_dump class Task: """ Parameters ---------- description: Text to describe the task done: True if the task has been completed priority: By convention, a single upper case character to indicate priority level completed_at: The date on which the task was completed created_at: The date on which the task was created projects: A list of project tags (denoted by a + prefix in the text definition) contexts: A list of context tags (denoted by a @ prefix in the text definition) tags: A dict of user defined tag keys and values """ def __init__( self, description, done=False, priority=None, completed_at=None, created_at=None, projects=None, contexts=None, tags=None, ): self.description = description self.done = done self.priority = priority self.completed_at = completed_at self.created_at = created_at or datetime.now().date() self.projects = projects or [] self.contexts = contexts or [] self.tags = tags or {} def __repr__(self): result = "blockbuster.core.model.Task(" result += f'description="{self.description}", ' result += f"done={self.done}, " result += f'priority="{self.priority}", ' result += f"completed_at={repr(self.completed_at)}, " result += f"created_at={repr(self.created_at)}, " result += f"projects={self.projects}, " result += f"contexts={self.contexts}, " result += f"tags={self.tags})" return result def __str__(self): optional_prefixes = "" minimal_text = f"{self.created_at.strftime(DATE_FORMAT)} {self.description}" optional_suffixes = "" if self.done: optional_prefixes += "x " if self.priority: optional_prefixes += f"({self.priority}) " if self.completed_at: optional_prefixes += f"{self.completed_at.strftime(DATE_FORMAT)} " for project in self.projects: if project: optional_suffixes += f" +{project}" for context in self.contexts: if context: optional_suffixes += f" @{context}" for key, value in self.tags.items(): optional_suffixes += f" {key}:{value}" return optional_prefixes + minimal_text + optional_suffixes class EventSchema(Schema): """A marshmallow schema to serialise an Event instance. The schema provides the the 'dump' and 'dumps' methods to serialise event objects to a Python dictionary and JSON string respectively. """ event_type = fields.Str() occurred_at = fields.DateTime() tasks = fields.List(fields.Str()) file = fields.Str() prior_hash = fields.Str() new_hash = fields.Str() @pre_dump def add_event_type(self, item): """Add the fully qualified name of the Event class to the object being serialised.""" item.event_type = f"{__name__}.{item.__class__.__qualname__}" return item class Event: __metaclass__ = ABCMeta schema = EventSchema(strict=True) def __init__(self, tasks, file, prior_hash, new_hash): self.occurred_at = datetime.now() self.tasks = tasks self.file = file self.prior_hash = prior_hash self.new_hash = new_hash def __str__(self): return self.schema.dumps(self).data def to_dict(self): return self.schema.dump(self).data class TasksAdded(Event): pass class TasksUpdated(Event): pass class TasksDeleted(Event): pass PK!HnHTU&blockbuster_core-0.1.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H1  )blockbuster_core-0.1.0.dist-info/METADATAV]S6}ׯP:I!dBPOA J.=Wb-!)ں{FQ*ɟvv&Du4qU9H~R9Oi[<gC)?Τx:ɇO'⣮|8:8ك[gӒfKtm[PyG}rl$'tE!u6>̦ MB_L(z]6:C9HŒOw'<-UP+[C!,*PN|'}W &.o~aM;x6ljsU:8=ޖjW(_|1׆B!ıCųG|nuFh?ʥ6FΝdlI* 2P0hqHP)UaXJi] ɛ|[S*| ișWT*ѸIb-;Hr]g3a<= :|Hq3S; Ix1*n <B\L'=m3@x?7#BiJZ΢jO$\V-3tiG}JI$NaVUݒ^v(ae0(- KĈΙ9SJUn7(L )CB FG*UV~4J \A72s 2CMN6)r s2#Pw!$jZ`g!'3FT>pK*7@cT I V* `+F~~ tM6q*O7p+Zu n,'Q{zg1 uL6<5w=CnZ( D,ߢ0$m'̦۲V?ǯ}!\X@uYIg4,Zʇ#D ДG&ڣt\)JJ?H