A R R R Associates
1229 lines
54 KiB
Python
1229 lines
54 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import re
|
|
import secrets
|
|
import shutil
|
|
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",
|
|
]
|
|
|
|
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 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 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,
|
|
) -> EngagementDocument:
|
|
original_filename = Path(upload_file.filename or "document.bin").name
|
|
document_type = document_type if document_type in DOCUMENT_TYPES else "GENERAL"
|
|
|
|
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
|
|
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,
|
|
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)
|