from __future__ import annotations import re from pathlib import Path from typing import Any, Iterable from sqlalchemy import select from app.modules.documents.models import BranchStorageNode, DocumentStorageJob, EngagementDocument from app.modules.documents.services import client_folder_parts, get_active_storage_node_for_branch, sanitize_segment from app.modules.services.models import FirmServiceTaskTemplate, ServiceTaskCategory from app.modules.accounting.accounting_mirror_models import AccountingMirrorRegistry _TALLY_MARKER_PREFIX = "ERP_TALLY_DATA_FY:" def _previous_financial_year(financial_year: str) -> str: match = re.match(r"^(\d{4})-(\d{2}|\d{4})$", (financial_year or "").strip()) if not match: return "" start = int(match.group(1)) prev_start = start - 1 return f"{prev_start:04d}-{start % 100:02d}" def _tally_document_year(doc: Any) -> str: description = str(getattr(doc, "description", "") or "").strip() if description.startswith(_TALLY_MARKER_PREFIX): return description[len(_TALLY_MARKER_PREFIX):].strip() title = str(getattr(doc, "title", "") or "").strip() match = re.match(r"^Tally Data\s*-\s*FY\s*(\d{4}-\d{2,4})$", title, re.IGNORECASE) if match: return match.group(1) return "" 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 _latest_local_final_path(db, version: Any) -> str: """Return the branch-agent acknowledged final path for a document version. For ordinary files this is the file path. For ZIP/TAR uploads handled by Local Agent 1.22.1+ this is the extracted directory path, while the original archive continues to remain the versioned source file for secure download/history. """ version_id = int(getattr(version, "id", 0) or 0) if not version_id: return "" job = db.execute( select(DocumentStorageJob).where( DocumentStorageJob.version_id == version_id, DocumentStorageJob.status == "completed", ).order_by(DocumentStorageJob.completed_at_utc.desc(), DocumentStorageJob.id.desc()).limit(1) ).scalar_one_or_none() return str(getattr(job, "local_final_path", "") or "").strip() if job else "" def _current_mirror_for_year(db, *, tenant_id: int, client_id: int, financial_year: str): if not tenant_id or not client_id or not financial_year: return None return db.execute( select(AccountingMirrorRegistry).where( AccountingMirrorRegistry.tenant_id == int(tenant_id), AccountingMirrorRegistry.client_id == int(client_id), AccountingMirrorRegistry.financial_year == str(financial_year).strip(), AccountingMirrorRegistry.is_current.is_(True), AccountingMirrorRegistry.is_active.is_(True), AccountingMirrorRegistry.status == "active", ).order_by(AccountingMirrorRegistry.version_no.desc(), AccountingMirrorRegistry.id.desc()).limit(1) ).scalar_one_or_none() def _mirror_row(db, *, tenant_id: int, client_id: int, financial_year: str) -> dict[str, Any]: mirror = _current_mirror_for_year( db, tenant_id=tenant_id, client_id=client_id, financial_year=financial_year ) if not mirror: return { "available": False, "financial_year": financial_year, "local_path": "", "relative_path": "", "version_no": 0, "company_name": "", "last_synced_at_utc": None, } relative_path = (Path(str(mirror.accounting_relative_dir or "")) / str(mirror.mirror_file_name or "")).as_posix() local_path = "" node = db.get(BranchStorageNode, int(mirror.storage_node_id)) if getattr(mirror, "storage_node_id", None) else None if node: root = str(getattr(node, "storage_root_path", "") or "").strip() local_path = _join_local_path(root, relative_path) if root else "" return { "available": True, "financial_year": financial_year, "local_path": local_path, "relative_path": relative_path, "version_no": int(getattr(mirror, "version_no", 0) or 0), "company_name": str(getattr(mirror, "company_name", "") or ""), "last_synced_at_utc": getattr(mirror, "last_synced_at_utc", None), "storage_node_name": str(getattr(node, "node_name", "") or "") if node else "", } def _tally_row_from_document(db, doc: Any, storage_root_path: str) -> dict[str, Any] | None: tally_year = _tally_document_year(doc) if not tally_year: return None versions = list(getattr(doc, "versions", None) or []) latest = versions[0] if versions else None latest_relative_path = str(getattr(latest, "local_relative_path", "") or "") if latest else "" local_final_path = _latest_local_final_path(db, latest) if latest else "" # local_final_path is already the workstation/storage-node acknowledged final # destination. For extracted ZIP/TAR Tally uploads it is the directory that # can be pasted into TallyPrime, so never prepend storage_root_path to it. local_path = local_final_path or (_join_local_path(storage_root_path, latest_relative_path) if latest_relative_path else "") return { "id": int(getattr(doc, "id", 0) or 0), "engagement_id": int(getattr(doc, "engagement_id", 0) or 0), "title": str(getattr(doc, "title", "") or "Tally Data"), "financial_year": tally_year, "version_no": int(getattr(doc, "current_version_no", 0) or 0), "latest_filename": str(getattr(latest, "original_filename", "") or "") if latest else "", "latest_relative_path": latest_relative_path, "local_final_path": local_final_path, "local_path": local_path, "storage_status": str(getattr(latest, "storage_status", "") or "") if latest else "", } def _task_category_options(db, engagement, tasks: list[Any]) -> list[dict[str, Any]]: """Return the service's real task-category master mapped to this engagement's tasks. Older task instances may not contain the task_category snapshot even though their FirmServiceTaskTemplate has since been linked to ServiceTaskCategory. Therefore category resolution deliberately uses both the generated task snapshot and the current template/category master, without changing any historical task rows. """ catalogue_id = int(getattr(engagement, "service_catalogue_id", 0) or 0) tenant_id = int(getattr(engagement, "tenant_id", 0) or 0) if not catalogue_id: return [] template_ids = { int(getattr(task, "firm_task_template_id", 0) or 0) for task in tasks if int(getattr(task, "firm_task_template_id", 0) or 0) } templates_by_id: dict[int, Any] = {} if template_ids: rows = db.execute( select(FirmServiceTaskTemplate).where(FirmServiceTaskTemplate.id.in_(template_ids)) ).scalars().all() templates_by_id = {int(row.id): row for row in rows} task_by_category_id: dict[int, int] = {} task_by_category_name: dict[str, int] = {} legacy_names: dict[str, tuple[str, int]] = {} for task in sorted(tasks, key=lambda x: (int(getattr(x, "sequence_no", 0) or 0), int(getattr(x, "id", 0) or 0))): task_id = int(getattr(task, "id", 0) or 0) if not task_id: continue template_id = int(getattr(task, "firm_task_template_id", 0) or 0) template = templates_by_id.get(template_id) category_id = int(getattr(template, "task_category_id", 0) or 0) if template else 0 snapshot_name = str(getattr(task, "task_category", "") or "").strip() template_name = str(getattr(template, "task_category", "") or "").strip() if template else "" category_name = snapshot_name or template_name if category_id and category_id not in task_by_category_id: task_by_category_id[category_id] = task_id if category_name: key = category_name.casefold() task_by_category_name.setdefault(key, task_id) legacy_names.setdefault(key, (category_name, task_id)) # Firm categories are authoritative for a firm engagement. If none exist, # retain support for the system/default category master. firm_rows = db.execute( select(ServiceTaskCategory).where( ServiceTaskCategory.service_catalogue_id == catalogue_id, ServiceTaskCategory.tenant_id == tenant_id, ServiceTaskCategory.is_active.is_(True), ).order_by(ServiceTaskCategory.sort_order.asc(), ServiceTaskCategory.name.asc(), ServiceTaskCategory.id.asc()) ).scalars().all() master_rows = list(firm_rows) if not master_rows: master_rows = db.execute( select(ServiceTaskCategory).where( ServiceTaskCategory.service_catalogue_id == catalogue_id, ServiceTaskCategory.tenant_id.is_(None), ServiceTaskCategory.is_active.is_(True), ).order_by(ServiceTaskCategory.sort_order.asc(), ServiceTaskCategory.name.asc(), ServiceTaskCategory.id.asc()) ).scalars().all() options: list[dict[str, Any]] = [] seen_names: set[str] = set() for category in master_rows: name = str(getattr(category, "name", "") or "").strip() if not name: continue key = name.casefold() task_id = task_by_category_id.get(int(category.id)) or task_by_category_name.get(key) options.append({ "id": int(category.id), "code": str(getattr(category, "code", "") or ""), "name": name, "task_id": task_id, "available": bool(task_id), "source": "master", }) seen_names.add(key) # Do not hide historical categories if a legacy task still has one that is no # longer present in the active master. for key, (name, task_id) in legacy_names.items(): if key in seen_names: continue options.append({ "id": None, "code": "", "name": name, "task_id": task_id, "available": True, "source": "legacy", }) seen_names.add(key) # Preserve the old behaviour only as a final fallback for genuinely # uncategorised engagements. if not options and tasks: first_task_id = int(getattr(tasks[0], "id", 0) or 0) if first_task_id: options.append({ "id": None, "code": "", "name": "General Workflow", "task_id": first_task_id, "available": True, "source": "fallback", }) return options 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. Accounting paths continue to follow the same FY/client convention used by the ERP Local Agent. Category uploads use existing task instances so evidence/AQMM controls remain authoritative. Tally Data is an engagement-level document destination and therefore is intentionally not forced into an unrelated task category. """ 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() prior_fy = _previous_financial_year(fy) tally_year_options = [year for year in (fy, prior_fy) if year] accounting_relative_dir = "" accounting_relative_path = "" accounting_local_path = "" storage_root_path = "" storage_node_name = "" tenant_id = int(getattr(engagement, "tenant_id", 0) or 0) 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, tenant_id, 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) # Client/FY accounting resources are shared across engagements for the same # client. This lets Tax Audit, Statutory Audit, ITR, etc. reuse the same Tally # dataset and registered mirror instead of requiring duplicate uploads. client_tally_documents: list[Any] = [] if tenant_id and client_id and tally_year_options: client_tally_documents = db.execute( select(EngagementDocument).where( EngagementDocument.tenant_id == tenant_id, EngagementDocument.client_id == client_id, EngagementDocument.is_deleted.is_(False), EngagementDocument.description.like(f"{_TALLY_MARKER_PREFIX}%"), EngagementDocument.financial_year.in_(tally_year_options), ).order_by(EngagementDocument.updated_at_utc.desc(), EngagementDocument.id.desc()) ).scalars().unique().all() client_tally_by_fy: dict[str, dict[str, Any]] = {} for tally_doc in client_tally_documents: row = _tally_row_from_document(db, tally_doc, storage_root_path) if row and row["financial_year"] not in client_tally_by_fy: client_tally_by_fy[row["financial_year"]] = row accounting_years: list[dict[str, Any]] = [] for year in tally_year_options: accounting_years.append({ "financial_year": year, "is_engagement_year": year == fy, "tally": client_tally_by_fy.get(year), "mirror": _mirror_row(db, tenant_id=tenant_id, client_id=client_id, financial_year=year), }) categories = _task_category_options(db, engagement, tasks) default_upload_task_id = next((row["task_id"] for row in categories if row.get("task_id")), None) document_options = [] tally_data_files: list[dict[str, Any]] = [] for doc in documents: versions = list(getattr(doc, "versions", None) or []) latest = versions[0] if versions else None latest_relative_path = str(getattr(latest, "local_relative_path", "") or "") if latest else "" local_final_path = _latest_local_final_path(db, latest) if latest else "" displayed_relative_path = local_final_path or latest_relative_path tally_year = _tally_document_year(doc) row = { "id": int(getattr(doc, "id", 0) or 0), "title": str(getattr(doc, "title", "") or "Document"), "tally_financial_year": tally_year, "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": latest_relative_path, "local_final_path": local_final_path, "displayed_relative_path": displayed_relative_path, "storage_status": str(getattr(latest, "storage_status", "") or "") if latest else "", } document_options.append(row) if tally_year: tally_local_path = local_final_path or (_join_local_path(storage_root_path, latest_relative_path) if latest_relative_path else "") tally_data_files.append({ **row, "financial_year": tally_year, "local_path": tally_local_path, }) tally_data_files.sort(key=lambda row: row.get("financial_year", ""), reverse=True) 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, "tally_year_options": tally_year_options, "accounting_years": accounting_years, "tally_data_files": tally_data_files, "task_categories": categories, "default_upload_task_id": default_upload_task_id, "document_options": document_options, }