Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add phase 2 consultant visibility communications and document requests

Browse Source
This commit is contained in:
A R R R Associates
2026-07-22 23:55:52 +05:30
parent 8f51d16c7d
commit e9db81b32f
8 changed files with 278 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/work_detail/service.py
+82 -28
View File
@@ -24,6 +24,7 @@ from app.modules.services.models import (
ClientServiceSubscription,
ClientServiceTaskInstance,
ServiceTaskComment,
ServiceTaskDocumentRequest,
)
MANAGEMENT_ROLES = {"System Admin", "Firm Admin"}
@@ -42,6 +43,12 @@ class WorkAccess:
allowed_comment_types: list[tuple[str, str]]
allowed_visibilities: list[tuple[str, str]]
back_url: str
can_view_assignee: bool = True
can_view_document_requests: bool = True
can_create_document_requests: bool = False
can_update_document_requests: bool = False
can_view_engagement_documents: bool = True
can_view_permanent_documents: bool = True
def _roles(db: Session, user) -> set[str]:
@@ -72,31 +79,24 @@ def _current_consultant(db: Session, user) -> ConsultantProfile | None:
).scalar_one_or_none()
def _consultant_can_view_engagement(db: Session, *, consultant: ConsultantProfile, engagement: ClientServiceSubscription) -> bool:
linked = db.execute(
select(ClientConsultantLink.id).where(
def _consultant_link_for_engagement(db: Session, *, consultant: ConsultantProfile, engagement: ClientServiceSubscription) -> ClientConsultantLink | None:
today = date.today()
return db.execute(
select(ClientConsultantLink).where(
ClientConsultantLink.tenant_id == consultant.tenant_id,
ClientConsultantLink.client_id == engagement.client_id,
ClientConsultantLink.consultant_id == consultant.id,
ClientConsultantLink.is_active.is_(True),
ClientConsultantLink.can_view_communications.is_(True),
ClientConsultantLink.can_view_engagements.is_(True),
or_(ClientConsultantLink.service_catalogue_id.is_(None), ClientConsultantLink.service_catalogue_id == engagement.service_catalogue_id),
or_(ClientConsultantLink.effective_from.is_(None), ClientConsultantLink.effective_from <= today),
or_(ClientConsultantLink.effective_to.is_(None), ClientConsultantLink.effective_to >= today),
)
).first()
if not linked:
return False
visible_comment = db.execute(
select(ServiceTaskComment.id)
.join(ClientServiceTaskInstance, ClientServiceTaskInstance.id == ServiceTaskComment.task_instance_id)
.where(
ServiceTaskComment.tenant_id == consultant.tenant_id,
ServiceTaskComment.subscription_id == engagement.id,
ServiceTaskComment.visibility == "consultant",
ServiceTaskComment.is_deleted.is_(False),
ClientServiceTaskInstance.is_active.is_(True),
)
.limit(1)
).first()
return bool(visible_comment)
).scalars().first()
def _consultant_can_view_engagement(db: Session, *, consultant: ConsultantProfile, engagement: ClientServiceSubscription) -> bool:
return _consultant_link_for_engagement(db, consultant=consultant, engagement=engagement) is not None
def _is_assigned_staff(engagement: ClientServiceSubscription, tasks: list[ClientServiceTaskInstance], user) -> bool:
@@ -148,7 +148,7 @@ def _load_tasks(db: Session, engagement: ClientServiceSubscription) -> list[Clie
).scalars().all()
def _load_documents(db: Session, engagement: ClientServiceSubscription) -> tuple[list[EngagementDocument], list[PermanentClientDocument]]:
def _load_documents(db: Session, engagement: ClientServiceSubscription, access: WorkAccess) -> tuple[list[EngagementDocument], list[PermanentClientDocument]]:
engagement_documents = db.execute(
select(EngagementDocument)
.options(selectinload(EngagementDocument.versions))
@@ -160,6 +160,8 @@ def _load_documents(db: Session, engagement: ClientServiceSubscription) -> tuple
)
.order_by(EngagementDocument.document_type.asc(), EngagementDocument.title.asc())
).unique().scalars().all()
if not access.can_view_engagement_documents:
engagement_documents = []
permanent_documents = db.execute(
select(PermanentClientDocument)
.options(selectinload(PermanentClientDocument.versions))
@@ -171,6 +173,8 @@ def _load_documents(db: Session, engagement: ClientServiceSubscription) -> tuple
.order_by(PermanentClientDocument.category.asc(), PermanentClientDocument.title.asc())
.limit(50)
).unique().scalars().all()
if not access.can_view_permanent_documents:
permanent_documents = []
return engagement_documents, permanent_documents
@@ -194,7 +198,7 @@ def _load_timeline(db: Session, engagement: ClientServiceSubscription, access: W
def _build_access(db: Session, *, user, engagement: ClientServiceSubscription, tasks: list[ClientServiceTaskInstance]) -> WorkAccess | None:
roles = _roles(db, user)
if roles.intersection(MANAGEMENT_ROLES):
return WorkAccess("admin", True, True, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, "/services/work-tracker")
return WorkAccess("admin", True, True, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, "/services/work-tracker", can_create_document_requests=True, can_update_document_requests=True)
if roles.intersection(CLIENT_ROLES):
client_row = _current_client_row(db, user)
@@ -204,20 +208,30 @@ def _build_access(db: Session, *, user, engagement: ClientServiceSubscription, t
if roles.intersection(CONSULTANT_ROLES):
consultant = _current_consultant(db, user)
if consultant and _consultant_can_view_engagement(db, consultant=consultant, engagement=engagement):
return WorkAccess("consultant", False, True, [("consultant_clarification", "Consultant Clarification")], [("consultant", "Consultant")], "/consultant/work")
if consultant:
link = _consultant_link_for_engagement(db, consultant=consultant, engagement=engagement)
if link:
return WorkAccess(
"consultant", False, bool(link.can_reply_to_clarifications),
[("consultant_clarification", "Consultant Clarification")], [("consultant", "Consultant")], "/consultant/work",
can_view_assignee=bool(link.can_view_assignee),
can_view_document_requests=bool(link.can_view_document_requests),
can_update_document_requests=bool(link.can_reply_to_clarifications or link.can_act_for_client),
can_view_engagement_documents=bool(link.can_view_final_documents or link.can_upload_documents),
can_view_permanent_documents=bool(link.can_view_permanent_documents),
)
return None
if roles.intersection(PARTNER_ROLES) and _is_partner_for_engagement(engagement, user):
return WorkAccess("partner", True, True, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, "/partner/reviews")
return WorkAccess("partner", True, True, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, "/partner/reviews", can_create_document_requests=True, can_update_document_requests=True)
if roles.intersection(MANAGER_ROLES):
if _is_manager_for_engagement(engagement, user) or (engagement.tenant_id == user.tenant_id and (engagement.branch_id in (None, user.branch_id))):
return WorkAccess("manager", True, True, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, "/manager/work")
return WorkAccess("manager", True, True, TASK_COMMENT_TYPES, TASK_COMMENT_VISIBILITIES, "/manager/work", can_create_document_requests=True, can_update_document_requests=True)
if roles.intersection(STAFF_ROLES) or roles.intersection({"Employee"}):
if _is_assigned_staff(engagement, tasks, user):
return WorkAccess("staff", True, True, [("internal_note", "Internal Note"), ("client_clarification", "Client Clarification")], [("internal", "Internal"), ("client", "Client")], "/employee/work")
return WorkAccess("staff", True, True, [("internal_note", "Internal Note"), ("client_clarification", "Client Clarification")], [("internal", "Internal"), ("client", "Client")], "/employee/work", can_create_document_requests=True, can_update_document_requests=True)
return None
@@ -235,7 +249,13 @@ def load_unified_engagement_detail(db: Session, *, request, user, engagement_id:
if not access:
return None
engagement_documents, permanent_documents = _load_documents(db, engagement)
engagement_documents, permanent_documents = _load_documents(db, engagement, access)
document_requests = db.execute(
select(ServiceTaskDocumentRequest)
.options(selectinload(ServiceTaskDocumentRequest.task), selectinload(ServiceTaskDocumentRequest.requested_by), selectinload(ServiceTaskDocumentRequest.responded_by))
.where(ServiceTaskDocumentRequest.tenant_id == engagement.tenant_id, ServiceTaskDocumentRequest.subscription_id == engagement.id, ServiceTaskDocumentRequest.is_active.is_(True))
.order_by(ServiceTaskDocumentRequest.status.asc(), ServiceTaskDocumentRequest.due_date.asc(), ServiceTaskDocumentRequest.id.desc())
).scalars().all() if access.can_view_document_requests else []
timeline = _load_timeline(db, engagement, access)
today = date.today()
for task in tasks:
@@ -256,6 +276,7 @@ def load_unified_engagement_detail(db: Session, *, request, user, engagement_id:
"engagement_documents": engagement_documents,
"permanent_documents": permanent_documents,
"timeline": timeline,
"document_requests": document_requests,
"access": access,
"role_context": access.role_context,
"task_statuses": TASK_STATUSES,
@@ -303,3 +324,36 @@ def save_task_comment(db: Session, *, task: ClientServiceTaskInstance, access: W
clean_visibility = visibility if visibility in allowed_visibilities else next(iter(allowed_visibilities), "internal")
row = add_task_comment(db, task=task, comment_type=clean_type, visibility=clean_visibility, message=message, user_id=user_id)
return row is not None
def create_document_request(db: Session, *, task: ClientServiceTaskInstance, access: WorkAccess, title: str, description: str, requested_from: str, due_date: date | None, request_type: str, user_id: int) -> ServiceTaskDocumentRequest | None:
if not access.can_create_document_requests or not (title or "").strip():
return None
allowed_from = {"client", "consultant", "client_and_consultant"}
allowed_types = {"document", "clarification", "approval", "information"}
row = ServiceTaskDocumentRequest(tenant_id=task.tenant_id, branch_id=task.branch_id, subscription_id=task.subscription_id, task_instance_id=task.id, client_id=task.client_id, request_type=request_type if request_type in allowed_types else "document", title=title.strip(), description=(description or "").strip() or None, requested_from=requested_from if requested_from in allowed_from else "client_and_consultant", due_date=due_date, status="pending", requested_by_user_id=user_id)
db.add(row)
visibility = "consultant" if row.requested_from == "consultant" else "client"
if row.requested_from == "client_and_consultant":
add_task_comment(db, task=task, comment_type="document_request", visibility="client", message=f"{row.title}: {row.description or ''}".strip(), user_id=user_id)
add_task_comment(db, task=task, comment_type="document_request", visibility="consultant", message=f"{row.title}: {row.description or ''}".strip(), user_id=user_id)
else:
add_task_comment(db, task=task, comment_type="document_request", visibility=visibility, message=f"{row.title}: {row.description or ''}".strip(), user_id=user_id)
return row
def update_document_request(db: Session, *, request_row: ServiceTaskDocumentRequest, access: WorkAccess, status: str, response_note: str, user_id: int) -> bool:
if not access.can_update_document_requests:
return False
allowed = {"pending", "received", "clarification_required", "verified", "rejected", "closed"}
clean = status if status in allowed else request_row.status
request_row.status = clean
request_row.response_note = (response_note or "").strip() or request_row.response_note
request_row.responded_by_user_id = user_id
now = datetime.now(timezone.utc)
if clean in {"received", "verified", "closed"} and not request_row.received_at_utc:
request_row.received_at_utc = now
if clean in {"verified", "closed"}:
request_row.verified_at_utc = now
request_row.verified_by_user_id = user_id
return True