auto task switch for work desktop (default)
This commit is contained in:
1
.gitignore
vendored
1
.gitignore
vendored
@@ -3,3 +3,4 @@ __pycache__
|
||||
nohup.out
|
||||
dm.out
|
||||
dm.err
|
||||
sample_task_file
|
||||
@@ -9,7 +9,13 @@ logging.basicConfig(
|
||||
format="%(asctime)s %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s",
|
||||
datefmt="%Y-%m-%d %H:%M:%S",
|
||||
)
|
||||
# 2) Get your module’s logger and bump it to DEBUG
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
# 3) Silence pymongo (and any other libs you find chatty)
|
||||
logging.getLogger("pymongo").setLevel(logging.WARNING)
|
||||
|
||||
|
||||
# MongoDB configuration
|
||||
client = MongoClient()
|
||||
|
||||
@@ -64,9 +64,14 @@ while True:
|
||||
last_doc = switches.find_one(sort=[("_id", -1)])
|
||||
|
||||
# work workflow
|
||||
|
||||
if current_workspace in work_desktops.keys():
|
||||
pass
|
||||
work_states = state.retrieve_work_state()
|
||||
current_work_task = work_states[work_desktops[current_workspace]]
|
||||
logger.debug(f"current_work_task: {current_work_task}")
|
||||
if current_task != current_work_task:
|
||||
state.save("current", task=current_work_task)
|
||||
task.db_to_file_as_is(None)
|
||||
current_task = current_work_task
|
||||
|
||||
# regular flow
|
||||
if (
|
||||
|
||||
@@ -45,3 +45,7 @@ def init_work_state(wd: dict):
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def retrieve_work_state():
|
||||
return states.find_one({"_id": "work"})
|
||||
|
||||
@@ -83,6 +83,9 @@ def format_task_line(
|
||||
|
||||
def db_to_file_as_is(filepath: str):
|
||||
"""Write tasks from MongoDB to file exactly as they were read."""
|
||||
if filepath is None:
|
||||
filepath = task_file
|
||||
|
||||
current_task = state.retrieve("current").get("task")
|
||||
all_tasks = list(tasks.find())
|
||||
|
||||
@@ -141,12 +144,12 @@ def extract(line: str) -> Optional[str]:
|
||||
return None
|
||||
|
||||
|
||||
def read_and_extract(file_path: str) -> Optional[str]:
|
||||
def read_and_extract(filepath: str) -> Optional[str]:
|
||||
"""Read file and update state if current task is found."""
|
||||
if file_path is None:
|
||||
file_path = task_file
|
||||
if filepath is None:
|
||||
filepath = task_file
|
||||
|
||||
with open(file_path, "r") as file:
|
||||
with open(filepath, "r") as file:
|
||||
for line in file:
|
||||
task_id = extract(line)
|
||||
if task_id:
|
||||
|
||||
Reference in New Issue
Block a user