Improve employee and partner engagement workspace

2026-09-02 12:31:27 +05:30
parent fbb9a62eb8
commit 2a4fd2bfdd
6 changed files with 259 additions and 1 deletions
@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any, Iterable
+
+from app.modules.documents.services import client_folder_parts, get_active_storage_node_for_branch, sanitize_segment
+
+
+def _join_local_path(root: str, relative: str) -> str:
+    root = (root or "").strip().rstrip("\\/")
+    if not root:
+        return ""
+    if re.match(r"^[A-Za-z]:", root) or "\\" in root:
+        return root + "\\" + relative.replace("/", "\\")
+    return root + "/" + relative.lstrip("/")
+
+
+def build_engagement_resource_context(db, engagement, tasks: Iterable[Any], documents: Iterable[Any]) -> dict[str, Any]:
+    """Presentation-only engagement resource data using existing storage/task models.
+
+    No new persistence is introduced here. Accounting paths follow the same FY/client
+    folder convention already used by the ERP Local Agent. Task-category uploads map
+    to an existing task so evidence/AQMM/document controls remain authoritative.
+    """
+    tasks = list(tasks or [])
+    documents = list(documents or [])
+    client = getattr(engagement, "client", None)
+    client_id = int(getattr(engagement, "client_id", 0) or 0)
+    fy = str(getattr(engagement, "financial_year", "") or "").strip()
+
+    accounting_relative_dir = ""
+    accounting_relative_path = ""
+    accounting_local_path = ""
+    storage_root_path = ""
+    storage_node_name = ""
+    if client and client_id and fy:
+        fy_folder = sanitize_segment(f"FY{fy}", "FY")
+        letter, client_folder = client_folder_parts(client, client_id)
+        relative_dir = Path(fy_folder) / "Clients" / letter / client_folder / "Accounting"
+        accounting_relative_dir = relative_dir.as_posix()
+        accounting_relative_path = (relative_dir / f"client_{client_id:08d}.act").as_posix()
+        node = get_active_storage_node_for_branch(
+            db,
+            int(getattr(engagement, "tenant_id", 0) or 0),
+            getattr(engagement, "branch_id", None),
+        )
+        if node:
+            storage_root_path = str(getattr(node, "storage_root_path", "") or "").strip()
+            storage_node_name = str(getattr(node, "node_name", "") or "").strip()
+            accounting_local_path = _join_local_path(storage_root_path, accounting_relative_path)
+
+    category_map: dict[str, int] = {}
+    for task in tasks:
+        category = str(getattr(task, "task_category", "") or "").strip() or "General Workflow"
+        task_id = int(getattr(task, "id", 0) or 0)
+        if task_id and category not in category_map:
+            category_map[category] = task_id
+    categories = [{"name": name, "task_id": task_id} for name, task_id in category_map.items()]
+
+    document_options = []
+    for doc in documents:
+        versions = list(getattr(doc, "versions", None) or [])
+        latest = versions[0] if versions else None
+        document_options.append({
+            "id": int(getattr(doc, "id", 0) or 0),
+            "title": str(getattr(doc, "title", "") or "Document"),
+            "document_type": str(getattr(doc, "document_type", "GENERAL") or "GENERAL"),
+            "version_no": int(getattr(doc, "current_version_no", 0) or 0),
+            "task_instance_id": getattr(doc, "task_instance_id", None),
+            "latest_filename": str(getattr(latest, "original_filename", "") or "") if latest else "",
+            "latest_relative_path": str(getattr(latest, "local_relative_path", "") or "") if latest else "",
+            "storage_status": str(getattr(latest, "storage_status", "") or "") if latest else "",
+        })
+
+    return {
+        "accounting_relative_dir": accounting_relative_dir,
+        "accounting_relative_path": accounting_relative_path,
+        "accounting_local_path": accounting_local_path,
+        "storage_root_path": storage_root_path,
+        "storage_node_name": storage_node_name,
+        "task_categories": categories,
+        "default_upload_task_id": categories[0]["task_id"] if categories else None,
+        "document_options": document_options,
+    }