filetime check

dmapp/dmcore/task.py

@@ -1,3 +1,4 @@
+import datetime
 import re
 from pathlib import Path
 from typing import Optional
@@ -27,6 +28,9 @@ def parse_line(line: str) -> tuple[Optional[str], Optional[str]]:
 
 def file_to_db(filepath: str):
     """Convert task file to MongoDB entries."""
+    if filepath is None:
+        filepath = task_file
+
     current_path = []
     tasks.delete_many({})
     seen_paths = set()
@@ -59,15 +63,6 @@ def file_to_db(filepath: str):
 
             seen_paths.add(full_path)
 
-    tasks.createIndex(
-        {"path": 1},
-        {
-            name: "idx_tasks_path_ci",
-            background: true,
-            collation: {locale: "es", strength: 1},
-        },
-    )
-
 
 def format_task_line(
     path_parts: list, indent_level: int, task_id: str, current_task: str
@@ -139,7 +134,6 @@ def extract(line: str) -> Optional[str]:
             # Extract everything between | and * and strip spaces
             id_part = line[pipe_index + 1 : -1].strip()
             if len(id_part) == 8:
-                state.save("current", task=id_part)
                 return id_part
     return None
 
@@ -149,9 +143,39 @@ def read_and_extract(filepath: str) -> Optional[str]:
     if filepath is None:
         filepath = task_file
 
+    mtime = get_file_mtime(filepath)
+    state.save("current", filetime=mtime)
+
     with open(filepath, "r") as file:
         for line in file:
             task_id = extract(line)
             if task_id:
                 return task_id
     return None
+
+
+def get_file_mtime(filepath: str) -> str:
+    """Get file modification time as ISO format string."""
+    if filepath is None:
+        filepath = task_file
+    return datetime.datetime.fromtimestamp(Path(filepath).stat().st_mtime).isoformat()
+
+
+def get_tasks_tree(root_path: str, only_with_ids: bool = True) -> list[dict]:
+    """
+    Get all tasks under a given path node, including the node itself.
+    """
+
+    query = {"path": {"$regex": root_path}}
+
+    # Add task_id filter if requested
+    if only_with_ids:
+        query["task_id"] = {"$exists": True}
+
+    # Query and sort by path to maintain hierarchy
+    cursor = tasks.find(
+        query,
+        {"path": 1, "task_id": 1, "_id": 0},
+    ).sort("path", 1)
+
+    return list(cursor)