from __future__ import annotations import hashlib import hmac import os import re import secrets import shutil from datetime import date, datetime, timezone from decimal import Decimal, InvalidOperation from dataclasses import dataclass from pathlib import Path from typing import BinaryIO from sqlalchemy import Select, func, or_, select from sqlalchemy.orm import Session, joinedload from app.modules.clients.models import Client from app.modules.core.rbac.permission_guard import require_permission from app.modules.core.tenancy.models import Branch, Tenant from app.modules.documents.models import ( BranchStorageNode, DocumentAccessLog, DocumentDownloadRequest, DocumentStorageJob, EngagementDocument, EngagementDocumentVersion, PermanentClientDocument, PermanentClientDocumentVersion, PermanentDocumentDownloadRequest, PermanentDocumentStorageJob, ) from app.modules.services.models import ClientServiceSubscription, ServiceCatalogue DOCUMENT_TYPES = [ "GENERAL", "CLIENT_DOCUMENT", "WORKING_PAPER", "BANK_STATEMENT", "TRIAL_BALANCE", "LEDGER", "GST_RETURN", "INCOME_TAX", "ROC", "AUDIT_REPORT", "SIGNED_OUTPUT", "ACKNOWLEDGEMENT", ] UDIN_STATUSES = ["not_required", "pending", "generated", "cancelled"] FINAL_RELEASE_STATUSES = ["draft", "released"] UDIN_REQUIRED_DOCUMENT_TYPES = {"AUDIT_REPORT", "SIGNED_OUTPUT", "ACKNOWLEDGEMENT"} MAX_UPLOAD_BYTES = int(os.getenv("DOCUMENT_MAX_UPLOAD_MB", "50")) * 1024 * 1024 def _resolve_document_storage_root(env_name: str, default_leaf: str) -> Path: """Return a short absolute storage root. Windows has a practical path-length limit in many Python/OS operations. The project is often run from a very deep development folder, so using a relative default like ``documents/engagement_documents`` can exceed that limit after adding audit-firm/branch/year/client/engagement folders. If an environment variable is provided, it is respected. Otherwise, on Windows we use a short drive-root path such as ``D:/AuditFirmERPDocuments/engagement_documents``. """ configured = (os.getenv(env_name) or "").strip() if configured: configured_path = Path(configured).expanduser() if configured_path.is_absolute(): return configured_path return (Path.cwd() / configured_path).resolve() if os.name == "nt": anchor = Path.cwd().anchor or (os.environ.get("SystemDrive", "C:") + "\\") return Path(anchor) / "AuditFirmERPDocuments" / default_leaf return (Path.cwd() / "documents" / default_leaf).resolve() DEFAULT_STORAGE_ROOT = _resolve_document_storage_root("DOCUMENT_STORAGE_ROOT", "engagement_documents") DOWNLOAD_CACHE_ROOT = _resolve_document_storage_root("DOCUMENT_DOWNLOAD_CACHE_ROOT", "download_cache") _SAFE_CHARS = re.compile(r"[^A-Za-z0-9._ -]+") @dataclass class DocumentScope: tenant_id: int | None branch_id: int | None role_names: set[str] is_system_admin: bool is_firm_admin: bool is_partner: bool is_branch_manager: bool is_staff: bool def sanitize_segment(value: object, default: str = "NA") -> str: text = str(value or default).strip() text = _SAFE_CHARS.sub("_", text) text = text.strip("._-") return text[:80] or default def _has_perm(db: Session, user, code: str) -> bool: try: require_permission(db, user, code) return True except Exception: return False def _now_utc() -> datetime: return datetime.now(timezone.utc) def _clean_udin_number(value: str | None) -> str | None: text = (value or "").strip().upper().replace(" ", "") return text or None def _to_decimal(value: str | Decimal | int | float | None) -> Decimal | None: if value in (None, ""): return None try: return Decimal(str(value).replace(",", "")).quantize(Decimal("0.01")) except (InvalidOperation, ValueError): raise ValueError("Invalid UDIN amount.") def _normalize_udin_required(document_type: str | None, explicit_value: bool | str | None = None) -> bool: if isinstance(explicit_value, str): lowered = explicit_value.strip().lower() if lowered in {"1", "true", "yes", "on", "required"}: return True if lowered in {"0", "false", "no", "off", "not_required"}: return False if explicit_value is not None: return bool(explicit_value) return (document_type or "").strip().upper() in UDIN_REQUIRED_DOCUMENT_TYPES def user_can_manage_udin(db: Session, user, document: EngagementDocument, scope: DocumentScope) -> bool: if not _has_perm(db, user, "udin.manage"): return False engagement = getattr(document, "engagement", None) or db.get(ClientServiceSubscription, document.engagement_id) return bool(engagement and user_can_view_engagement(db, user, engagement, scope)) def user_can_view_udin_register(db: Session, user, scope: DocumentScope) -> bool: return _has_perm(db, user, "udin.view") def build_document_scope(request, db: Session, user) -> DocumentScope: from app.modules.core.rbac.deps import get_user_roles roles = set(get_user_roles(db, user.id)) active_tenant = request.session.get("active_tenant_id") or request.session.get("selected_tenant_id") or getattr(user, "tenant_id", None) active_branch = request.session.get("active_branch_id") if active_branch in (None, "", 0, "0"): active_branch = getattr(user, "branch_id", None) return DocumentScope( tenant_id=int(active_tenant) if active_tenant else None, branch_id=int(active_branch) if active_branch else None, role_names=roles, is_system_admin="System Admin" in roles, is_firm_admin="Firm Admin" in roles, is_partner="Partner" in roles, is_branch_manager="Branch Manager" in roles, is_staff="Staff" in roles, ) def user_can_view_engagement(db: Session, user, engagement: ClientServiceSubscription, scope: DocumentScope) -> bool: if not _has_perm(db, user, "documents.view"): return False # Professional document contents are not exposed to System Admin merely by SaaS support role. if scope.is_system_admin and not (scope.is_firm_admin or scope.is_partner or scope.is_branch_manager or scope.is_staff): return False if not scope.is_system_admin and scope.tenant_id and engagement.tenant_id != scope.tenant_id: return False if scope.is_firm_admin: return engagement.tenant_id == getattr(user, "tenant_id", engagement.tenant_id) if scope.is_partner: client = getattr(engagement, "client", None) return ( engagement.tenant_id == getattr(user, "tenant_id", engagement.tenant_id) and ( engagement.assigned_partner_user_id == user.id or getattr(client, "partner_id", None) == user.id ) ) if scope.is_branch_manager: return ( engagement.tenant_id == getattr(user, "tenant_id", engagement.tenant_id) and (engagement.branch_id is None or engagement.branch_id == getattr(user, "branch_id", None)) ) if scope.is_staff: return ( engagement.tenant_id == getattr(user, "tenant_id", engagement.tenant_id) and (engagement.branch_id is None or engagement.branch_id == getattr(user, "branch_id", None)) and engagement.assigned_staff_user_id == user.id ) return False def user_can_upload_to_engagement(db: Session, user, engagement: ClientServiceSubscription, scope: DocumentScope) -> bool: return _has_perm(db, user, "documents.upload") and user_can_view_engagement(db, user, engagement, scope) def user_can_delete_document(db: Session, user, document: EngagementDocument, scope: DocumentScope) -> bool: if not _has_perm(db, user, "documents.delete"): return False if scope.is_firm_admin: return document.tenant_id == getattr(user, "tenant_id", document.tenant_id) if scope.is_partner: engagement = db.get(ClientServiceSubscription, document.engagement_id) return bool(engagement and user_can_view_engagement(db, user, engagement, scope)) return False def _engagement_query_base() -> Select: return ( select(ClientServiceSubscription) .options( joinedload(ClientServiceSubscription.client), joinedload(ClientServiceSubscription.catalogue), joinedload(ClientServiceSubscription.assigned_partner), joinedload(ClientServiceSubscription.assigned_staff), ) .order_by(ClientServiceSubscription.financial_year.desc(), ClientServiceSubscription.id.desc()) ) def list_visible_engagements(db: Session, user, scope: DocumentScope, q: str = "", financial_year: str | None = None, limit: int = 200): stmt = _engagement_query_base() if scope.is_firm_admin: stmt = stmt.where(ClientServiceSubscription.tenant_id == getattr(user, "tenant_id", None)) elif scope.is_partner: stmt = stmt.outerjoin(Client, Client.id == ClientServiceSubscription.client_id).where( ClientServiceSubscription.tenant_id == getattr(user, "tenant_id", None), or_(ClientServiceSubscription.assigned_partner_user_id == user.id, Client.partner_id == user.id), ) elif scope.is_branch_manager: stmt = stmt.where( ClientServiceSubscription.tenant_id == getattr(user, "tenant_id", None), or_(ClientServiceSubscription.branch_id == getattr(user, "branch_id", None), ClientServiceSubscription.branch_id.is_(None)), ) elif scope.is_staff: stmt = stmt.where( ClientServiceSubscription.tenant_id == getattr(user, "tenant_id", None), or_(ClientServiceSubscription.branch_id == getattr(user, "branch_id", None), ClientServiceSubscription.branch_id.is_(None)), ClientServiceSubscription.assigned_staff_user_id == user.id, ) else: return [] if financial_year: stmt = stmt.where(ClientServiceSubscription.financial_year == financial_year.strip()) if q: pattern = f"%{q.strip()}%" stmt = stmt.join(Client, Client.id == ClientServiceSubscription.client_id).join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceSubscription.service_catalogue_id).where( or_(Client.client_name.ilike(pattern), Client.client_code.ilike(pattern), ServiceCatalogue.service_name.ilike(pattern), ServiceCatalogue.service_code.ilike(pattern)) ) return db.execute(stmt.limit(limit)).unique().scalars().all() def list_documents_for_engagement(db: Session, engagement_id: int): return db.execute( select(EngagementDocument) .options(joinedload(EngagementDocument.versions), joinedload(EngagementDocument.client), joinedload(EngagementDocument.engagement)) .where(EngagementDocument.engagement_id == engagement_id, EngagementDocument.is_deleted.is_(False)) .order_by(EngagementDocument.updated_at_utc.desc()) ).unique().scalars().all() def get_document(db: Session, document_id: int) -> EngagementDocument | None: return db.execute( select(EngagementDocument) .options(joinedload(EngagementDocument.versions), joinedload(EngagementDocument.client), joinedload(EngagementDocument.engagement)) .where(EngagementDocument.id == document_id, EngagementDocument.is_deleted.is_(False)) ).unique().scalar_one_or_none() def get_latest_version(document: EngagementDocument) -> EngagementDocumentVersion | None: return document.versions[0] if document.versions else None def get_version(db: Session, version_id: int) -> EngagementDocumentVersion | None: return db.execute( select(EngagementDocumentVersion) .options(joinedload(EngagementDocumentVersion.document).joinedload(EngagementDocument.engagement)) .where(EngagementDocumentVersion.id == version_id) ).unique().scalar_one_or_none() def client_folder_parts(client: Client | None, client_id: int | None) -> tuple[str, str]: """Return A-Z bucket and readable Client Name_Code folder for local storage.""" client_name = sanitize_segment(getattr(client, "client_name", None) or getattr(client, "name", None) or f"Client {client_id}", "Client") client_code = sanitize_segment(getattr(client, "client_code", None) or client_id or "NA", "NA") first = (client_name[:1] or "#").upper() if not first.isalpha(): first = "#" return first, f"{client_name}_{client_code}" def _clean_original_name_parts(original_filename: str) -> tuple[str, str]: """Return Windows-safe file stem and extension for readable local storage names.""" original = Path(original_filename or "document.bin").name suffix = Path(original).suffix stem = original[:-len(suffix)] if suffix else original safe_stem = sanitize_segment(stem, "document") safe_suffix = re.sub(r"[^A-Za-z0-9.]", "", suffix)[:20] return safe_stem, safe_suffix or ".bin" def build_versioned_filename(original_filename: str, *, document_id: int, version_no: int, engagement_code: str | None = None) -> str: """Build readable traceable file names without extra DOC/Versions folders. Engagement document: OriginalFile_ENG-1-GST02_DOC000001_v001.xlsx Permanent document: OriginalFile_DOC000001_v001.pdf """ stem, suffix = _clean_original_name_parts(original_filename) doc_id = sanitize_segment(f"DOC{document_id:06d}", "DOC") version = sanitize_segment(f"v{version_no:03d}", "v001") if engagement_code: return f"{stem}_{sanitize_segment(engagement_code, 'ENGAGEMENT')}_{doc_id}_{version}{suffix}" return f"{stem}_{doc_id}_{version}{suffix}" def _first_text_value(*values: object) -> str | None: for value in values: if value is None: continue text = str(value).strip() if text: return text return None def infer_service_category(service: ServiceCatalogue | None) -> str: """Return a clean service category folder such as GST, Income Tax, ROC, Audit.""" category_obj = getattr(service, "category", None) if service is not None else None category_text = _first_text_value( getattr(category_obj, "name", None), getattr(category_obj, "category_name", None), getattr(service, "category_name", None) if service is not None else None, getattr(service, "service_category", None) if service is not None else None, getattr(service, "category", None) if service is not None and not hasattr(getattr(service, "category", None), "name") else None, ) if category_text: return sanitize_segment(category_text, "Other") name_text = (_first_text_value( getattr(service, "service_name", None) if service is not None else None, getattr(service, "name", None) if service is not None else None, getattr(service, "title", None) if service is not None else None, getattr(service, "service_code", None) if service is not None else None, ) or "").lower() if "gst" in name_text or "gstr" in name_text: return "GST" if "roc" in name_text or "mca" in name_text or "aoc" in name_text or "mgt" in name_text: return "ROC" if "income" in name_text or "itr" in name_text or "tax audit" in name_text or "3cd" in name_text: return "Income_Tax" if "tds" in name_text or "tcs" in name_text: return "TDS_TCS" if "audit" in name_text: return "Audit" if "pf" in name_text or "esi" in name_text: return "PF_ESI" if "account" in name_text or "book" in name_text: return "Accounts" return "Other" def infer_engagement_period(engagement: ClientServiceSubscription, service: ServiceCatalogue | None) -> str: """Return period folder for recurring/annual services, with safe fallback.""" candidate = _first_text_value( getattr(engagement, "period_label", None), getattr(engagement, "return_period", None), getattr(engagement, "compliance_period", None), getattr(engagement, "filing_period", None), getattr(engagement, "month_label", None), getattr(engagement, "quarter_label", None), getattr(engagement, "period", None), ) if candidate: return sanitize_segment(candidate, "General") recurrence = (_first_text_value( getattr(engagement, "recurrence_type", None), getattr(service, "recurrence_type", None) if service is not None else None, getattr(service, "frequency", None) if service is not None else None, ) or "").lower() fy = sanitize_segment(f"FY{getattr(engagement, 'financial_year', '') or 'General'}", "FY") ay = sanitize_segment(f"AY{getattr(engagement, 'assessment_year', '')}", "AY") if getattr(engagement, "assessment_year", None) else None if "month" in recurrence: return "Monthly" if "quarter" in recurrence: return "Quarterly" if "annual" in recurrence or "year" in recurrence: return ay or fy return fy or "General" def build_year_wise_relative_path(db: Session, engagement: ClientServiceSubscription, document: EngagementDocument, version_no: int, original_filename: str) -> Path: client = getattr(engagement, "client", None) or db.get(Client, engagement.client_id) service = getattr(engagement, "catalogue", None) or db.get(ServiceCatalogue, engagement.service_catalogue_id) fy = sanitize_segment(f"FY{engagement.financial_year}", "FY") letter, client_folder = client_folder_parts(client, engagement.client_id) service_code = sanitize_segment(getattr(service, "service_code", None) or engagement.service_catalogue_id, "ENG") engagement_code = sanitize_segment(f"ENG-{engagement.id}-{service_code}", "ENGAGEMENT") service_category = infer_service_category(service) period = infer_engagement_period(engagement, service) doc_type = sanitize_segment(document.document_type, "GENERAL") stored_filename = build_versioned_filename(original_filename, document_id=document.id, version_no=version_no, engagement_code=engagement_code) return Path(fy) / "Clients" / letter / client_folder / service_category / period / doc_type / stored_filename def build_permanent_relative_path(db: Session, client: Client, document: PermanentClientDocument, version_no: int, original_filename: str) -> Path: letter, client_folder = client_folder_parts(client, document.client_id) category = sanitize_segment(document.category, "Other Permanent Documents") stored_filename = build_versioned_filename(original_filename, document_id=document.id, version_no=version_no) return Path("Permanent") / "Clients" / letter / client_folder / category / stored_filename def create_document_code(db: Session, tenant_id: int, engagement_id: int, current_document_id: int | None = None) -> str: filters = [ EngagementDocument.tenant_id == tenant_id, EngagementDocument.engagement_id == engagement_id, ] if current_document_id: filters.append(EngagementDocument.id != current_document_id) count = db.execute(select(func.count(EngagementDocument.id)).where(*filters)).scalar_one() return f"ENG{engagement_id}-DOC{int(count) + 1:04d}" def log_document_access(db: Session, *, action: str, result: str, user, request=None, document=None, version=None, message: str | None = None): doc = document or getattr(version, "document", None) db.add(DocumentAccessLog( tenant_id=getattr(doc, "tenant_id", None), branch_id=getattr(doc, "branch_id", None), client_id=getattr(doc, "client_id", None), engagement_id=getattr(doc, "engagement_id", None), document_id=getattr(doc, "id", None), version_id=getattr(version, "id", None), action=action, result=result, user_id=getattr(user, "id", None), ip_address=(request.client.host if request and request.client else None), user_agent=(request.headers.get("user-agent")[:500] if request else None), message=message, )) def update_document_udin( db: Session, *, document: EngagementDocument, udin_required: bool | str | None, udin_number: str | None, udin_date: date | None, udin_document_date: date | None, udin_document_type: str | None, udin_financial_year: str | None, udin_amount: str | Decimal | int | float | None, udin_notes: str | None, user, ) -> EngagementDocument: required = _normalize_udin_required(document.document_type, udin_required) number = _clean_udin_number(udin_number) document.udin_required = required document.udin_number = number document.udin_date = udin_date document.udin_document_date = udin_document_date document.udin_document_type = (udin_document_type or "").strip()[:120] or None document.udin_financial_year = (udin_financial_year or document.financial_year or "").strip()[:9] or None document.udin_amount = _to_decimal(udin_amount) document.udin_notes = (udin_notes or "").strip() or None document.updated_by_user_id = getattr(user, "id", None) if required: if number: document.udin_status = "generated" document.udin_generated_by_user_id = getattr(user, "id", None) document.udin_verified_at_utc = _now_utc() else: document.udin_status = "pending" document.udin_generated_by_user_id = None document.udin_verified_at_utc = None else: document.udin_status = "not_required" if not number: document.udin_generated_by_user_id = None document.udin_verified_at_utc = None db.flush() return document def release_document_final( db: Session, *, document: EngagementDocument, release_notes: str | None, user, ) -> EngagementDocument: if not get_latest_version(document): raise ValueError("Cannot release a document without an uploaded version.") if document.udin_required and not _clean_udin_number(document.udin_number): raise ValueError("UDIN is required before final release for this document.") if document.udin_required and document.udin_status != "generated": raise ValueError("UDIN status must be Generated before final release.") # Phase 4 AQMM final release gate. Assurance engagements can release final # documents only after the AQMM-tagged task evidence/review trail is complete. engagement = document.engagement or db.get(ClientServiceSubscription, document.engagement_id) if engagement is not None: from app.modules.services.client_services import quality_required_for_engagement from app.modules.services.execution import assert_aqmm_quality_tasks_complete if quality_required_for_engagement(getattr(engagement, "engagement_type", None)): assert_aqmm_quality_tasks_complete(db, subscription_id=int(engagement.id)) document.final_release_status = "released" document.status = "final" document.released_by_user_id = getattr(user, "id", None) document.released_at_utc = _now_utc() document.release_notes = (release_notes or "").strip() or None document.updated_by_user_id = getattr(user, "id", None) db.flush() return document def reopen_final_document( db: Session, *, document: EngagementDocument, release_notes: str | None, user, ) -> EngagementDocument: document.final_release_status = "draft" document.status = "active" document.release_notes = (release_notes or "").strip() or document.release_notes document.updated_by_user_id = getattr(user, "id", None) db.flush() return document def list_udin_records( db: Session, user, scope: DocumentScope, *, financial_year: str | None = None, status: str | None = None, partner_user_id: int | None = None, q: str | None = None, limit: int = 500, ) -> list[EngagementDocument]: stmt = ( select(EngagementDocument) .options( joinedload(EngagementDocument.client), joinedload(EngagementDocument.engagement).joinedload(ClientServiceSubscription.assigned_partner), joinedload(EngagementDocument.versions), ) .where(EngagementDocument.is_deleted.is_(False)) .order_by(EngagementDocument.udin_date.desc().nullslast(), EngagementDocument.updated_at_utc.desc()) ) if scope.is_firm_admin: stmt = stmt.where(EngagementDocument.tenant_id == getattr(user, "tenant_id", None)) elif scope.is_partner: stmt = stmt.join(ClientServiceSubscription, ClientServiceSubscription.id == EngagementDocument.engagement_id).outerjoin(Client, Client.id == EngagementDocument.client_id).where( EngagementDocument.tenant_id == getattr(user, "tenant_id", None), or_(ClientServiceSubscription.assigned_partner_user_id == user.id, Client.partner_id == user.id, EngagementDocument.udin_generated_by_user_id == user.id), ) elif scope.is_branch_manager: stmt = stmt.where( EngagementDocument.tenant_id == getattr(user, "tenant_id", None), or_(EngagementDocument.branch_id == getattr(user, "branch_id", None), EngagementDocument.branch_id.is_(None)), ) elif scope.is_staff: stmt = stmt.join(ClientServiceSubscription, ClientServiceSubscription.id == EngagementDocument.engagement_id).where( EngagementDocument.tenant_id == getattr(user, "tenant_id", None), ClientServiceSubscription.assigned_staff_user_id == user.id, ) else: return [] if financial_year: stmt = stmt.where(EngagementDocument.financial_year == financial_year.strip()) if status: stmt = stmt.where(EngagementDocument.udin_status == status.strip()) if partner_user_id: stmt = stmt.join(ClientServiceSubscription, ClientServiceSubscription.id == EngagementDocument.engagement_id).where( or_(ClientServiceSubscription.assigned_partner_user_id == int(partner_user_id), EngagementDocument.udin_generated_by_user_id == int(partner_user_id)) ) if q: pattern = f"%{q.strip()}%" stmt = stmt.where(or_(EngagementDocument.title.ilike(pattern), EngagementDocument.document_code.ilike(pattern), EngagementDocument.udin_number.ilike(pattern))) return db.execute(stmt.limit(limit)).unique().scalars().all() def hash_storage_secret(secret: str) -> str: return hashlib.sha256((secret or "").encode("utf-8")).hexdigest() def generate_storage_secret() -> str: return secrets.token_urlsafe(32) def _same_branch_node_filter(stmt, tenant_id: int, branch_id: int | None): stmt = stmt.where(BranchStorageNode.tenant_id == int(tenant_id)) if branch_id is None: stmt = stmt.where(BranchStorageNode.branch_id.is_(None)) else: stmt = stmt.where(BranchStorageNode.branch_id == int(branch_id)) return stmt def get_canonical_storage_node(db: Session, tenant_id: int, branch_id: int | None) -> BranchStorageNode | None: """Return the permanent storage node for an audit-firm/branch pair. The oldest row is canonical so repeated package generation never changes the node code already written in the branch storage identity file. """ stmt = select(BranchStorageNode).where(BranchStorageNode.tenant_id == int(tenant_id)) if branch_id is None: stmt = stmt.where(BranchStorageNode.branch_id.is_(None)) else: stmt = stmt.where(BranchStorageNode.branch_id == int(branch_id)) return db.execute(stmt.order_by(BranchStorageNode.id.asc()).limit(1)).scalar_one_or_none() def deactivate_duplicate_storage_nodes(db: Session, *, keep_node: BranchStorageNode) -> int: """Disable duplicate nodes for the same audit-firm/branch. Business rule: one branch must have only one active Local Storage Agent. Callers should normally pass the canonical/oldest node as keep_node. """ stmt = select(BranchStorageNode).where( BranchStorageNode.tenant_id == int(keep_node.tenant_id), BranchStorageNode.id != int(keep_node.id), BranchStorageNode.is_active.is_(True), ) if keep_node.branch_id is None: stmt = stmt.where(BranchStorageNode.branch_id.is_(None)) else: stmt = stmt.where(BranchStorageNode.branch_id == int(keep_node.branch_id)) duplicates = db.execute(stmt).scalars().all() for dup in duplicates: dup.is_active = False dup.status = "disabled_duplicate" return len(duplicates) def create_branch_storage_node( db: Session, *, tenant_id: int, branch_id: int | None, node_code: str, node_name: str, connector_url: str | None, storage_root_path: str | None, quota_limit_gb: int | None, user, ) -> tuple[BranchStorageNode, str]: raw_secret = generate_storage_secret() existing = get_canonical_storage_node(db, int(tenant_id), branch_id) if existing: # Reuse the original node code forever. This prevents new -002/-003 # packages for the same branch and keeps the local storage identity # file valid. Package regeneration rotates only the secret. existing.node_name = existing.node_name or (node_name or existing.node_code or "Branch Storage Node")[:200] existing.connector_url = (connector_url or "").strip() or existing.connector_url existing.storage_root_path = existing.storage_root_path or ((storage_root_path or "").strip() or None) existing.quota_limit_bytes = (int(quota_limit_gb) * 1024 * 1024 * 1024 if quota_limit_gb else existing.quota_limit_bytes) existing.secret_key_hash = hash_storage_secret(raw_secret) existing.is_active = True existing.status = "active" deactivate_duplicate_storage_nodes(db, keep_node=existing) db.flush() return existing, raw_secret node = BranchStorageNode( tenant_id=tenant_id, branch_id=branch_id, node_code=sanitize_segment(node_code or node_name, "NODE"), node_name=(node_name or node_code or "Branch Storage Node")[:200], connector_url=(connector_url or "").strip() or None, storage_root_path=(storage_root_path or "").strip() or None, quota_limit_bytes=(int(quota_limit_gb) * 1024 * 1024 * 1024 if quota_limit_gb else None), secret_key_hash=hash_storage_secret(raw_secret), created_by_user_id=getattr(user, "id", None), is_active=True, status="active", ) db.add(node) db.flush() deactivate_duplicate_storage_nodes(db, keep_node=node) return node, raw_secret def list_storage_nodes(db: Session, tenant_id: int | None = None, branch_id: int | None = None): stmt = select(BranchStorageNode).order_by(BranchStorageNode.tenant_id, BranchStorageNode.branch_id, BranchStorageNode.node_name) if tenant_id: stmt = stmt.where(BranchStorageNode.tenant_id == tenant_id) if branch_id: stmt = stmt.where(BranchStorageNode.branch_id == branch_id) return db.execute(stmt).scalars().all() def get_active_storage_node_for_branch(db: Session, tenant_id: int, branch_id: int | None) -> BranchStorageNode | None: stmt = ( select(BranchStorageNode) .where( BranchStorageNode.tenant_id == tenant_id, BranchStorageNode.is_active.is_(True), BranchStorageNode.status == "active", ) .order_by(BranchStorageNode.branch_id.desc(), BranchStorageNode.id.asc()) ) if branch_id: stmt = stmt.where(or_(BranchStorageNode.branch_id == branch_id, BranchStorageNode.branch_id.is_(None))) else: stmt = stmt.where(BranchStorageNode.branch_id.is_(None)) return db.execute(stmt.limit(1)).scalar_one_or_none() def create_storage_job_for_version(db: Session, *, version: EngagementDocumentVersion, user) -> DocumentStorageJob | None: node = get_active_storage_node_for_branch(db, version.tenant_id, version.branch_id) if not node: return None job = DocumentStorageJob( tenant_id=version.tenant_id, branch_id=version.branch_id, storage_node_id=node.id, document_id=version.document_id, version_id=version.id, staging_relative_path=version.local_relative_path, target_relative_path=version.local_relative_path, file_size_bytes=version.file_size_bytes, expected_hash_sha256=version.file_hash_sha256, created_by_user_id=getattr(user, "id", None), ) version.storage_backend = "BRANCH_STORAGE_NODE" version.storage_status = "pending_local_sync" db.add(job) return job def authenticate_storage_node(db: Session, node_code: str | None, secret: str | None, request=None) -> BranchStorageNode | None: if not node_code or not secret: return None node = db.execute( select(BranchStorageNode).where(BranchStorageNode.node_code == node_code, BranchStorageNode.is_active.is_(True)) ).scalar_one_or_none() if not node or not hmac.compare_digest(node.secret_key_hash, hash_storage_secret(secret)): return None node.last_seen_at_utc = __import__("datetime").datetime.now(__import__("datetime").timezone.utc) node.last_seen_ip = request.client.host if request and request.client else None return node def list_pending_storage_jobs(db: Session, node: BranchStorageNode, limit: int = 20): return db.execute( select(DocumentStorageJob) .where(DocumentStorageJob.storage_node_id == node.id, DocumentStorageJob.status.in_(["pending", "retry"])) .order_by(DocumentStorageJob.priority.asc(), DocumentStorageJob.created_at_utc.asc()) .limit(limit) ).scalars().all() def get_storage_job_for_node(db: Session, node: BranchStorageNode, job_id: int) -> DocumentStorageJob | None: return db.execute( select(DocumentStorageJob).where(DocumentStorageJob.id == job_id, DocumentStorageJob.storage_node_id == node.id) ).scalar_one_or_none() def acknowledge_storage_job(db: Session, *, node: BranchStorageNode, job: DocumentStorageJob, acknowledged_hash: str, local_final_path: str | None, success: bool, error: str | None = None) -> bool: from datetime import datetime, timezone job.attempts = int(job.attempts or 0) + 1 if success and acknowledged_hash and acknowledged_hash.lower() == job.expected_hash_sha256.lower(): job.status = "completed" job.completed_at_utc = datetime.now(timezone.utc) job.acknowledged_hash_sha256 = acknowledged_hash.lower() job.local_final_path = local_final_path version = db.get(EngagementDocumentVersion, job.version_id) if version: version.storage_status = "stored_on_branch_node" node.used_storage_bytes = int(node.used_storage_bytes or 0) + int(job.file_size_bytes or 0) return True job.status = "failed" if job.attempts >= 3 else "retry" job.last_error = error or "Hash mismatch or local storage acknowledgement failed." return False def list_storage_jobs(db: Session, tenant_id: int | None = None, branch_id: int | None = None, status: str | None = None, limit: int = 200): stmt = select(DocumentStorageJob).order_by(DocumentStorageJob.created_at_utc.desc()) if tenant_id: stmt = stmt.where(DocumentStorageJob.tenant_id == tenant_id) if branch_id: stmt = stmt.where(DocumentStorageJob.branch_id == branch_id) if status: stmt = stmt.where(DocumentStorageJob.status == status) return db.execute(stmt.limit(limit)).scalars().all() def _now_utc(): from datetime import datetime, timezone return datetime.now(timezone.utc) def get_completed_storage_job_for_version(db: Session, version_id: int) -> DocumentStorageJob | None: """Return the latest completed local-storage job for a document version.""" return 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() def create_download_request_for_version(db: Session, *, version: EngagementDocumentVersion, user, request=None) -> DocumentDownloadRequest | None: """Queue a secure local-to-cloud download request for the branch storage agent. If the staged ERP file is already available, callers should stream that file directly and avoid creating a request. This request is used when the version is stored on the branch node and the ERP copy is unavailable. """ completed_job = get_completed_storage_job_for_version(db, version.id) if not completed_job or not completed_job.storage_node_id: return None # Reuse a recent pending/ready request for the same user/version to avoid duplicate queues. existing = db.execute( select(DocumentDownloadRequest) .where( DocumentDownloadRequest.version_id == version.id, DocumentDownloadRequest.requested_by_user_id == getattr(user, "id", None), DocumentDownloadRequest.request_status.in_(["pending", "picked", "ready"]), ) .order_by(DocumentDownloadRequest.created_at_utc.desc()) .limit(1) ).scalar_one_or_none() if existing: return existing download_request = DocumentDownloadRequest( tenant_id=version.tenant_id, branch_id=version.branch_id, storage_node_id=completed_job.storage_node_id, document_id=version.document_id, version_id=version.id, local_relative_path=completed_job.local_final_path or completed_job.target_relative_path, expected_hash_sha256=version.file_hash_sha256, file_size_bytes=version.file_size_bytes, requested_by_user_id=getattr(user, "id", None), requested_ip=(request.client.host if request and request.client else None), requested_user_agent=(request.headers.get("user-agent")[:500] if request else None), ) db.add(download_request) return download_request def list_pending_download_requests(db: Session, node: BranchStorageNode, limit: int = 20): return db.execute( select(DocumentDownloadRequest) .where( DocumentDownloadRequest.storage_node_id == node.id, DocumentDownloadRequest.request_status.in_(["pending", "retry"]), ) .order_by(DocumentDownloadRequest.created_at_utc.asc()) .limit(limit) ).scalars().all() def get_download_request_for_node(db: Session, node: BranchStorageNode, request_id: int) -> DocumentDownloadRequest | None: return db.execute( select(DocumentDownloadRequest).where( DocumentDownloadRequest.id == request_id, DocumentDownloadRequest.storage_node_id == node.id, ) ).scalar_one_or_none() def get_download_request(db: Session, request_id: int) -> DocumentDownloadRequest | None: return db.execute( select(DocumentDownloadRequest) .options(joinedload(DocumentDownloadRequest.document), joinedload(DocumentDownloadRequest.version)) .where(DocumentDownloadRequest.id == request_id) ).unique().scalar_one_or_none() def list_download_requests(db: Session, tenant_id: int | None = None, branch_id: int | None = None, status: str | None = None, limit: int = 200): stmt = ( select(DocumentDownloadRequest) .options(joinedload(DocumentDownloadRequest.document), joinedload(DocumentDownloadRequest.version)) .order_by(DocumentDownloadRequest.created_at_utc.desc()) ) if tenant_id: stmt = stmt.where(DocumentDownloadRequest.tenant_id == tenant_id) if branch_id: stmt = stmt.where(DocumentDownloadRequest.branch_id == branch_id) if status: stmt = stmt.where(DocumentDownloadRequest.request_status == status) return db.execute(stmt.limit(limit)).scalars().all() def list_recent_download_requests_for_engagement(db: Session, engagement_id: int, user_id: int | None = None, limit: int = 10): stmt = ( select(DocumentDownloadRequest) .options(joinedload(DocumentDownloadRequest.document), joinedload(DocumentDownloadRequest.version)) .join(EngagementDocumentVersion, EngagementDocumentVersion.id == DocumentDownloadRequest.version_id) .where(EngagementDocumentVersion.engagement_id == engagement_id) .order_by(DocumentDownloadRequest.created_at_utc.desc()) .limit(limit) ) if user_id: stmt = stmt.where(DocumentDownloadRequest.requested_by_user_id == user_id) return db.execute(stmt).scalars().all() def download_request_cache_path(download_request: DocumentDownloadRequest) -> Path | None: if not download_request.cached_relative_path: return None return DOWNLOAD_CACHE_ROOT / Path(download_request.cached_relative_path) def fulfill_download_request_from_upload(db: Session, *, node: BranchStorageNode, download_request: DocumentDownloadRequest, upload_file) -> bool: """Store the file uploaded by local agent into ERP download cache and verify hash.""" if download_request.storage_node_id != node.id: return False version = db.get(EngagementDocumentVersion, download_request.version_id) if not version: download_request.request_status = "failed" download_request.failed_at_utc = _now_utc() download_request.last_error = "Document version not found." return False safe_name = sanitize_segment(version.original_filename or f"version_{version.id}.bin", "document.bin") cache_rel = Path(f"request_{download_request.id}") / f"v{version.version_no:03d}_{safe_name}" cache_abs = DOWNLOAD_CACHE_ROOT / cache_rel cache_abs.parent.mkdir(parents=True, exist_ok=True) hasher = hashlib.sha256() total = 0 with cache_abs.open("wb") as out: while True: chunk = upload_file.file.read(1024 * 1024) if not chunk: break total += len(chunk) if total > MAX_UPLOAD_BYTES: out.close() cache_abs.unlink(missing_ok=True) download_request.request_status = "failed" download_request.failed_at_utc = _now_utc() download_request.last_error = f"Uploaded stream exceeds maximum limit of {MAX_UPLOAD_BYTES // (1024 * 1024)} MB." return False hasher.update(chunk) out.write(chunk) actual_hash = hasher.hexdigest() download_request.attempts = int(download_request.attempts or 0) + 1 if actual_hash.lower() != (download_request.expected_hash_sha256 or "").lower(): cache_abs.unlink(missing_ok=True) download_request.request_status = "failed" if download_request.attempts >= 3 else "retry" download_request.last_error = "Uploaded file hash does not match the original document version." if download_request.request_status == "failed": download_request.failed_at_utc = _now_utc() return False download_request.request_status = "ready" download_request.cached_relative_path = str(cache_rel).replace("\\", "/") download_request.cached_hash_sha256 = actual_hash.lower() download_request.file_size_bytes = total download_request.fulfilled_at_utc = _now_utc() download_request.last_error = None return True def save_uploaded_revision( db: Session, *, engagement: ClientServiceSubscription, upload_file, title: str, document_type: str, description: str | None, remarks: str | None, user, existing_document_id: int | None = None, udin_required: bool | str | None = None, ) -> EngagementDocument: original_filename = Path(upload_file.filename or "document.bin").name document_type = document_type if document_type in DOCUMENT_TYPES else "GENERAL" resolved_udin_required = _normalize_udin_required(document_type, udin_required) if existing_document_id: document = db.get(EngagementDocument, existing_document_id) if not document or document.is_deleted or document.engagement_id != engagement.id: raise ValueError("Invalid document selected for new revision.") if title: document.title = title.strip()[:255] if description is not None: document.description = description.strip() or None document.document_type = document_type document.udin_required = resolved_udin_required if resolved_udin_required and (document.udin_status or "not_required") == "not_required": document.udin_status = "pending" if not resolved_udin_required and not document.udin_number: document.udin_status = "not_required" else: document = EngagementDocument( tenant_id=engagement.tenant_id, branch_id=engagement.branch_id, client_id=engagement.client_id, engagement_id=engagement.id, financial_year=engagement.financial_year, assessment_year=engagement.assessment_year, document_code="PENDING", document_type=document_type, udin_required=resolved_udin_required, udin_status="pending" if resolved_udin_required else "not_required", final_release_status="draft", title=(title.strip()[:255] if title else original_filename[:255]), description=description.strip() if description else None, created_by_user_id=user.id, updated_by_user_id=user.id, ) db.add(document) db.flush() document.document_code = create_document_code(db, engagement.tenant_id, engagement.id, document.id) next_version_no = int(document.current_version_no or 0) + 1 rel_path = build_year_wise_relative_path(db, engagement, document, next_version_no, original_filename) abs_path = DEFAULT_STORAGE_ROOT / rel_path try: abs_path.parent.mkdir(parents=True, exist_ok=True) except OSError as exc: raise RuntimeError( "Unable to create ERP document staging folder. " f"storage_root='{DEFAULT_STORAGE_ROOT}', relative_path='{rel_path}'. " "On Windows, set DOCUMENT_STORAGE_ROOT to a short absolute path like " "D:\\AuditFirmERPDocuments\\engagement_documents." ) from exc hasher = hashlib.sha256() total = 0 with abs_path.open("wb") as out: while True: chunk = upload_file.file.read(1024 * 1024) if not chunk: break total += len(chunk) if total > MAX_UPLOAD_BYTES: out.close() abs_path.unlink(missing_ok=True) raise ValueError(f"File exceeds maximum upload limit of {MAX_UPLOAD_BYTES // (1024 * 1024)} MB.") hasher.update(chunk) out.write(chunk) version = EngagementDocumentVersion( document_id=document.id, tenant_id=document.tenant_id, branch_id=document.branch_id, client_id=document.client_id, engagement_id=document.engagement_id, version_no=next_version_no, original_filename=original_filename, stored_filename=abs_path.name, content_type=getattr(upload_file, "content_type", None), file_size_bytes=total, file_hash_sha256=hasher.hexdigest(), local_relative_path=str(rel_path).replace("\\", "/"), remarks=remarks.strip() if remarks else None, uploaded_by_user_id=user.id, ) document.current_version_no = next_version_no document.updated_by_user_id = user.id db.add(version) db.flush() create_storage_job_for_version(db, version=version, user=user) return document PERMANENT_DOCUMENT_CATEGORIES = [ "Company Registration", "GST Registration", "Income Tax", "Bank", "KYC", "Agreements", "Licenses", "DSC", "ROC Master Data", "Other Permanent Documents", ] def user_can_view_client_documents(db: Session, user, client: Client, scope: DocumentScope) -> bool: if not _has_perm(db, user, "documents.view"): return False if scope.is_system_admin and not (scope.is_firm_admin or scope.is_partner or scope.is_branch_manager or scope.is_staff): return False if scope.is_firm_admin: return client.tenant_id == getattr(user, "tenant_id", client.tenant_id) if scope.is_partner: return client.tenant_id == getattr(user, "tenant_id", client.tenant_id) and getattr(client, "partner_id", None) == user.id if scope.is_branch_manager or scope.is_staff: return client.tenant_id == getattr(user, "tenant_id", client.tenant_id) and (getattr(client, "branch_id", None) in (None, getattr(user, "branch_id", None))) return False def user_can_upload_client_documents(db: Session, user, client: Client, scope: DocumentScope) -> bool: return _has_perm(db, user, "documents.upload") and user_can_view_client_documents(db, user, client, scope) def list_visible_clients_for_permanent_documents(db: Session, user, scope: DocumentScope, q: str = "", limit: int = 200): stmt = select(Client).order_by(Client.client_name.asc()).limit(limit) if scope.is_firm_admin: stmt = stmt.where(Client.tenant_id == getattr(user, "tenant_id", None)) elif scope.is_partner: stmt = stmt.where(Client.tenant_id == getattr(user, "tenant_id", None), Client.partner_id == user.id) elif scope.is_branch_manager or scope.is_staff: stmt = stmt.where(Client.tenant_id == getattr(user, "tenant_id", None), or_(Client.branch_id == getattr(user, "branch_id", None), Client.branch_id.is_(None))) else: return [] if q: pattern = f"%{q.strip()}%" stmt = stmt.where(or_(Client.client_name.ilike(pattern), Client.client_code.ilike(pattern))) return db.execute(stmt).scalars().all() def list_permanent_documents_for_client(db: Session, client_id: int): return db.execute( select(PermanentClientDocument) .options(joinedload(PermanentClientDocument.versions), joinedload(PermanentClientDocument.client)) .where(PermanentClientDocument.client_id == client_id, PermanentClientDocument.is_deleted.is_(False)) .order_by(PermanentClientDocument.category.asc(), PermanentClientDocument.updated_at_utc.desc()) ).unique().scalars().all() def get_permanent_document(db: Session, document_id: int) -> PermanentClientDocument | None: return db.execute( select(PermanentClientDocument) .options(joinedload(PermanentClientDocument.versions), joinedload(PermanentClientDocument.client)) .where(PermanentClientDocument.id == document_id, PermanentClientDocument.is_deleted.is_(False)) ).unique().scalar_one_or_none() def get_permanent_version(db: Session, version_id: int) -> PermanentClientDocumentVersion | None: return db.execute( select(PermanentClientDocumentVersion) .options(joinedload(PermanentClientDocumentVersion.document).joinedload(PermanentClientDocument.client)) .where(PermanentClientDocumentVersion.id == version_id) ).unique().scalar_one_or_none() def get_latest_permanent_version(document: PermanentClientDocument) -> PermanentClientDocumentVersion | None: return document.versions[0] if document.versions else None def create_permanent_document_code(db: Session, tenant_id: int, client_id: int, current_document_id: int | None = None) -> str: filters = [PermanentClientDocument.tenant_id == tenant_id, PermanentClientDocument.client_id == client_id] if current_document_id: filters.append(PermanentClientDocument.id != current_document_id) count = db.execute(select(func.count(PermanentClientDocument.id)).where(*filters)).scalar_one() return f"PERM{client_id}-DOC{int(count) + 1:04d}" def get_active_storage_node_for_permanent_client(db: Session, tenant_id: int, branch_id: int | None) -> BranchStorageNode | None: stmt = select(BranchStorageNode).where( BranchStorageNode.tenant_id == tenant_id, BranchStorageNode.is_active.is_(True), BranchStorageNode.status == "active", ) if branch_id is None: stmt = stmt.where(BranchStorageNode.branch_id.is_(None)) else: stmt = stmt.where(or_(BranchStorageNode.branch_id == branch_id, BranchStorageNode.branch_id.is_(None))) return db.execute(stmt.order_by(BranchStorageNode.branch_id.desc(), BranchStorageNode.id.asc()).limit(1)).scalar_one_or_none() def create_permanent_storage_job_for_version(db: Session, *, version: PermanentClientDocumentVersion, user) -> PermanentDocumentStorageJob | None: node = get_active_storage_node_for_permanent_client(db, version.tenant_id, version.branch_id) if not node: return None job = PermanentDocumentStorageJob( tenant_id=version.tenant_id, branch_id=version.branch_id, storage_node_id=node.id, document_id=version.document_id, version_id=version.id, staging_relative_path=version.local_relative_path, target_relative_path=version.local_relative_path, file_size_bytes=version.file_size_bytes, expected_hash_sha256=version.file_hash_sha256, created_by_user_id=getattr(user, "id", None), ) db.add(job) return job def list_pending_permanent_storage_jobs(db: Session, node: BranchStorageNode, limit: int = 20): return db.execute( select(PermanentDocumentStorageJob) .where(PermanentDocumentStorageJob.storage_node_id == node.id, PermanentDocumentStorageJob.status.in_(["pending", "retry"])) .order_by(PermanentDocumentStorageJob.priority.asc(), PermanentDocumentStorageJob.created_at_utc.asc()) .limit(limit) ).scalars().all() def get_permanent_storage_job_for_node(db: Session, node: BranchStorageNode, job_id: int) -> PermanentDocumentStorageJob | None: return db.execute(select(PermanentDocumentStorageJob).where(PermanentDocumentStorageJob.id == job_id, PermanentDocumentStorageJob.storage_node_id == node.id)).scalar_one_or_none() def acknowledge_permanent_storage_job(db: Session, *, node: BranchStorageNode, job: PermanentDocumentStorageJob, acknowledged_hash: str, local_final_path: str | None, success: bool, error: str | None = None) -> bool: if job.storage_node_id != node.id: return False job.attempts = int(job.attempts or 0) + 1 if not success: job.status = "failed" if job.attempts >= 3 else "retry" job.last_error = error or "Local storage agent reported failure." return False if acknowledged_hash.lower() != (job.expected_hash_sha256 or "").lower(): job.status = "failed" if job.attempts >= 3 else "retry" job.last_error = "Hash mismatch after local storage write." return False job.status = "completed" job.completed_at_utc = _now_utc() job.acknowledged_hash_sha256 = acknowledged_hash.lower() job.local_final_path = local_final_path or job.target_relative_path job.last_error = None version = db.get(PermanentClientDocumentVersion, job.version_id) if version: version.storage_status = "synced_local" return True def get_completed_permanent_storage_job_for_version(db: Session, version_id: int) -> PermanentDocumentStorageJob | None: return db.execute( select(PermanentDocumentStorageJob) .where(PermanentDocumentStorageJob.version_id == version_id, PermanentDocumentStorageJob.status == "completed") .order_by(PermanentDocumentStorageJob.completed_at_utc.desc(), PermanentDocumentStorageJob.id.desc()) .limit(1) ).scalar_one_or_none() def create_permanent_download_request_for_version(db: Session, *, version: PermanentClientDocumentVersion, user, request=None) -> PermanentDocumentDownloadRequest | None: completed_job = get_completed_permanent_storage_job_for_version(db, version.id) if not completed_job: return None existing = db.execute( select(PermanentDocumentDownloadRequest) .where( PermanentDocumentDownloadRequest.version_id == version.id, PermanentDocumentDownloadRequest.requested_by_user_id == getattr(user, "id", None), PermanentDocumentDownloadRequest.request_status.in_(["pending", "picked", "ready"]), ) .order_by(PermanentDocumentDownloadRequest.created_at_utc.desc()) .limit(1) ).scalar_one_or_none() if existing: return existing item = PermanentDocumentDownloadRequest( tenant_id=version.tenant_id, branch_id=version.branch_id, storage_node_id=completed_job.storage_node_id, document_id=version.document_id, version_id=version.id, local_relative_path=completed_job.local_final_path or completed_job.target_relative_path, expected_hash_sha256=version.file_hash_sha256, file_size_bytes=version.file_size_bytes, requested_by_user_id=getattr(user, "id", None), requested_ip=(request.client.host if request and request.client else None), requested_user_agent=(request.headers.get("user-agent")[:500] if request else None), ) db.add(item) return item def list_pending_permanent_download_requests(db: Session, node: BranchStorageNode, limit: int = 20): return db.execute( select(PermanentDocumentDownloadRequest) .where(PermanentDocumentDownloadRequest.storage_node_id == node.id, PermanentDocumentDownloadRequest.request_status.in_(["pending", "retry"])) .order_by(PermanentDocumentDownloadRequest.created_at_utc.asc()) .limit(limit) ).scalars().all() def get_permanent_download_request_for_node(db: Session, node: BranchStorageNode, request_id: int) -> PermanentDocumentDownloadRequest | None: return db.execute(select(PermanentDocumentDownloadRequest).where(PermanentDocumentDownloadRequest.id == request_id, PermanentDocumentDownloadRequest.storage_node_id == node.id)).scalar_one_or_none() def permanent_download_request_cache_path(download_request: PermanentDocumentDownloadRequest) -> Path | None: if not download_request.cached_relative_path: return None return DOWNLOAD_CACHE_ROOT / Path(download_request.cached_relative_path) def fulfill_permanent_download_request_from_upload(db: Session, *, node: BranchStorageNode, download_request: PermanentDocumentDownloadRequest, upload_file) -> bool: if download_request.storage_node_id != node.id: return False version = db.get(PermanentClientDocumentVersion, download_request.version_id) if not version: download_request.request_status = "failed" download_request.failed_at_utc = _now_utc() download_request.last_error = "Permanent document version not found." return False safe_name = sanitize_segment(version.original_filename or f"version_{version.id}.bin", "document.bin") cache_rel = Path(f"permanent_request_{download_request.id}") / f"v{version.version_no:03d}_{safe_name}" cache_abs = DOWNLOAD_CACHE_ROOT / cache_rel cache_abs.parent.mkdir(parents=True, exist_ok=True) hasher = hashlib.sha256() total = 0 with cache_abs.open("wb") as out: while True: chunk = upload_file.file.read(1024 * 1024) if not chunk: break total += len(chunk) if total > MAX_UPLOAD_BYTES: out.close(); cache_abs.unlink(missing_ok=True) download_request.request_status = "failed"; download_request.failed_at_utc = _now_utc(); download_request.last_error = "Uploaded stream exceeds maximum limit." return False hasher.update(chunk); out.write(chunk) actual_hash = hasher.hexdigest() download_request.attempts = int(download_request.attempts or 0) + 1 if actual_hash.lower() != (download_request.expected_hash_sha256 or "").lower(): cache_abs.unlink(missing_ok=True) download_request.request_status = "failed" if download_request.attempts >= 3 else "retry" download_request.last_error = "Uploaded file hash does not match the permanent document version." if download_request.request_status == "failed": download_request.failed_at_utc = _now_utc() return False download_request.request_status = "ready" download_request.cached_relative_path = str(cache_rel).replace("\\", "/") download_request.cached_hash_sha256 = actual_hash.lower() download_request.file_size_bytes = total download_request.fulfilled_at_utc = _now_utc() download_request.last_error = None return True def save_uploaded_permanent_revision(db: Session, *, client: Client, upload_file, title: str, category: str, description: str | None, remarks: str | None, user, existing_document_id: int | None = None) -> PermanentClientDocument: original_filename = Path(upload_file.filename or "document.bin").name category = category if category in PERMANENT_DOCUMENT_CATEGORIES else "Other Permanent Documents" if existing_document_id: document = db.get(PermanentClientDocument, existing_document_id) if not document or document.is_deleted or document.client_id != client.id: raise ValueError("Invalid permanent document selected for new revision.") if title: document.title = title.strip()[:255] if description is not None: document.description = description.strip() or None document.category = category else: document = PermanentClientDocument( tenant_id=client.tenant_id, branch_id=getattr(client, "branch_id", None), client_id=client.id, document_code="PENDING", category=category, title=(title.strip()[:255] if title else original_filename[:255]), description=description.strip() if description else None, created_by_user_id=getattr(user, "id", None), updated_by_user_id=getattr(user, "id", None), ) db.add(document); db.flush() document.document_code = create_permanent_document_code(db, client.tenant_id, client.id, document.id) next_version_no = int(document.current_version_no or 0) + 1 rel_path = build_permanent_relative_path(db, client, document, next_version_no, original_filename) abs_path = DEFAULT_STORAGE_ROOT / rel_path abs_path.parent.mkdir(parents=True, exist_ok=True) hasher = hashlib.sha256(); total = 0 with abs_path.open("wb") as out: while True: chunk = upload_file.file.read(1024 * 1024) if not chunk: break total += len(chunk) if total > MAX_UPLOAD_BYTES: out.close(); abs_path.unlink(missing_ok=True) raise ValueError(f"File exceeds maximum upload limit of {MAX_UPLOAD_BYTES // (1024 * 1024)} MB.") hasher.update(chunk); out.write(chunk) version = PermanentClientDocumentVersion( document_id=document.id, tenant_id=document.tenant_id, branch_id=document.branch_id, client_id=document.client_id, version_no=next_version_no, original_filename=original_filename, stored_filename=abs_path.name, content_type=getattr(upload_file, "content_type", None), file_size_bytes=total, file_hash_sha256=hasher.hexdigest(), local_relative_path=str(rel_path).replace("\\", "/"), remarks=remarks.strip() if remarks else None, uploaded_by_user_id=getattr(user, "id", None), ) document.current_version_no = next_version_no document.updated_by_user_id = getattr(user, "id", None) db.add(version); db.flush() create_permanent_storage_job_for_version(db, version=version, user=user) return document def permanent_version_absolute_path(version: PermanentClientDocumentVersion) -> Path: return DEFAULT_STORAGE_ROOT / Path(version.local_relative_path) def version_absolute_path(version: EngagementDocumentVersion) -> Path: return DEFAULT_STORAGE_ROOT / Path(version.local_relative_path)