96 lines
2.8 KiB
Python
96 lines
2.8 KiB
Python
from models.models import Task
|
|
|
|
|
|
def build_task_tree(tasks: list[Task]) -> list[dict]:
|
|
"""Build nested tree from flat DB task list."""
|
|
task_map = {}
|
|
for t in tasks:
|
|
task_map[t.id] = {
|
|
"id": t.id,
|
|
"plan_id": t.plan_id,
|
|
"parent_id": t.parent_id,
|
|
"task_id": t.task_id,
|
|
"title": t.title,
|
|
"description": t.description,
|
|
"dependencies": t.dependencies or [],
|
|
"duration_in_hours": t.duration_in_hours,
|
|
"priority": t.priority,
|
|
"due_date": t.due_date.isoformat() if t.due_date else None,
|
|
"status": t.status,
|
|
"sort_order": t.sort_order,
|
|
"children": [],
|
|
}
|
|
|
|
roots = []
|
|
for t in tasks:
|
|
node = task_map[t.id]
|
|
if t.parent_id is None:
|
|
roots.append(node)
|
|
else:
|
|
parent = task_map.get(t.parent_id)
|
|
if parent:
|
|
parent["children"].append(node)
|
|
else:
|
|
roots.append(node)
|
|
|
|
# Sort children by sort_order at every level
|
|
def sort_children(nodes: list[dict]) -> list[dict]:
|
|
nodes.sort(key=lambda n: n.get("sort_order", 0))
|
|
for n in nodes:
|
|
if n["children"]:
|
|
sort_children(n["children"])
|
|
return nodes
|
|
|
|
sort_children(roots)
|
|
return roots
|
|
|
|
|
|
def find_in_tree(nodes: list[dict], task_id: int) -> dict | None:
|
|
"""Find a task node in the tree by its DB id."""
|
|
for n in nodes:
|
|
if n["id"] == task_id:
|
|
return n
|
|
found = find_in_tree(n["children"], task_id)
|
|
if found:
|
|
return found
|
|
return None
|
|
|
|
|
|
def calculate_task_depths(tasks: list[Task]) -> dict[int, int]:
|
|
"""Calculate depth for all tasks in one pass using a parent_id -> id map.
|
|
Returns {task_id: depth}."""
|
|
task_by_id = {t.id: t for t in tasks}
|
|
depth_cache: dict[int, int] = {}
|
|
|
|
def _depth(task_id: int) -> int:
|
|
if task_id in depth_cache:
|
|
return depth_cache[task_id]
|
|
task = task_by_id.get(task_id)
|
|
if not task or task.parent_id is None:
|
|
depth_cache[task_id] = 0
|
|
return 0
|
|
d = _depth(task.parent_id) + 1
|
|
depth_cache[task_id] = d
|
|
return d
|
|
|
|
for t in tasks:
|
|
_depth(t.id)
|
|
|
|
return depth_cache
|
|
|
|
|
|
def collect_descendant_ids(tasks: list[Task], parent_id: int) -> list[int]:
|
|
"""Collect all descendant task IDs using a pre-loaded list (no N+1 queries)."""
|
|
children_map: dict[int, list[int]] = {}
|
|
for t in tasks:
|
|
if t.parent_id is not None:
|
|
children_map.setdefault(t.parent_id, []).append(t.id)
|
|
|
|
result = []
|
|
stack = children_map.get(parent_id, [])
|
|
while stack:
|
|
cid = stack.pop()
|
|
result.append(cid)
|
|
stack.extend(children_map.get(cid, []))
|
|
return result
|