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

Add digital client acceptance workflow

Browse Source
This commit is contained in:
A R R R Associates
2026-07-07 09:49:20 +05:30
parent 671adbd1a2
commit 1b5fec270b
6 changed files with 1103 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/clients/service.py
+436
View File
@@ -1,17 +1,23 @@
from __future__ import annotations
import csv
import hashlib
import io
import os
import re
from pathlib import Path
from datetime import datetime, timezone
from fastapi import HTTPException
from app.modules.clients import repository
from sqlalchemy import select, func, or_
from app.core.security.passwords import hash_password
from app.modules.clients.association_admin_service import (
ensure_active_association,
update_association_fields,
)
from app.modules.clients.models import ClientAcceptanceDeclaration, ClientKycVerification, ClientEngagementLetter
from app.modules.clients.constants import (
CLIENT_CATEGORY_OPTIONS,
CLIENT_STATUS,
@@ -19,6 +25,11 @@ from app.modules.clients.constants import (
RISK_CATEGORIES,
CLIENT_ACCEPTANCE_APPROVAL_REQUIRED_RISKS,
)
from app.modules.alerts.service import create_alert
from app.modules.core.iam.models import User
from app.modules.core.rbac.models import Role, UserRole
from app.modules.documents.models import PermanentClientDocument
def _payload_from_schema(data):
@@ -588,3 +599,428 @@ def mark_client_acceptance_pending_service(db, *, row, actor_user_id: int, revie
payload_json={"review_notes": review_notes},
)
return updated
# ---------------------------------------------------------------------------
# Phase 2.1 - digital client acceptance workflow
# ---------------------------------------------------------------------------
DECLARATION_TYPES = {"independence", "conflict"}
DECLARATION_CLEAR_STATUSES = {"declared_clear"}
ENGAGEMENT_LETTER_ACCEPTED_STATUSES = {"digitally_accepted", "manual_verified"}
KYC_REQUIRED_KEYWORDS = ["PAN", "GST", "Incorporation", "Partnership Deed", "Address Proof", "Authorisation"]
ACCEPTANCE_STORAGE_ROOT = Path(os.getenv("CLIENT_ACCEPTANCE_STORAGE_ROOT", "documents/client_acceptance")).resolve()
def _clean_text(value: str | None, max_len: int | None = None) -> str | None:
value = (value or "").strip()
if not value:
return None
return value[:max_len] if max_len else value
def _slug(value: str | None) -> str:
value = re.sub(r"[^A-Za-z0-9._-]+", "_", (value or "file").strip())
return value.strip("._-")[:80] or "file"
def _request_ip_and_agent(request):
ip = None
try:
ip = request.headers.get("x-forwarded-for", "").split(",")[0].strip() or request.client.host
except Exception:
ip = None
try:
agent = request.headers.get("user-agent")
except Exception:
agent = None
return ip, agent
def _declaration_text(client, declaration_type: str) -> str:
if declaration_type == "conflict":
return (
f"I confirm that I have reviewed possible conflict of interest for {client.client_name}. "
"I do not have any financial, business, family, employment, litigation or other relationship "
"that creates a conflict, except as specifically disclosed in my remarks."
)
return (
f"I confirm that I am independent for the proposed/continuing engagement of {client.client_name}. "
"I will immediately report any independence threat or safeguard requirement to the engagement partner."
)
def _role_name_for_user(db, user_id: int) -> str:
names = db.execute(
select(Role.name)
.join(UserRole, UserRole.role_id == Role.id)
.where(UserRole.user_id == user_id)
.order_by(Role.name.asc())
).scalars().all()
return ",".join(names) if names else ""
def _applicable_acceptance_user_ids(db, client) -> list[int]:
ids = []
for uid in [getattr(client, "partner_id", None), getattr(client, "default_review_partner_user_id", None), getattr(client, "created_by_user_id", None)]:
if uid and uid not in ids:
ids.append(int(uid))
if not ids and getattr(client, "updated_by_user_id", None):
ids.append(int(client.updated_by_user_id))
return ids
def _safe_create_alert(db, *, user_id: int, title: str, message: str, client, actor_user_id: int | None = None, target_url: str | None = None):
try:
create_alert(
db,
user_id=user_id,
tenant_id=client.tenant_id,
branch_id=client.branch_id,
role_context="client_acceptance",
alert_type="client",
priority="high",
title=title,
message=message,
target_url=target_url or f"/clients/{client.id}",
created_by_user_id=actor_user_id,
commit=False,
)
except Exception:
pass
def list_client_acceptance_declarations(db, *, client_id: int):
return db.execute(
select(ClientAcceptanceDeclaration)
.where(ClientAcceptanceDeclaration.client_id == client_id)
.order_by(ClientAcceptanceDeclaration.declaration_type.asc(), ClientAcceptanceDeclaration.assigned_user_id.asc())
).scalars().all()
def list_my_pending_acceptance_declarations(db, *, user_id: int, limit: int = 25):
return db.execute(
select(ClientAcceptanceDeclaration)
.where(ClientAcceptanceDeclaration.assigned_user_id == user_id, ClientAcceptanceDeclaration.status == "pending")
.order_by(ClientAcceptanceDeclaration.requested_at_utc.desc())
.limit(limit)
).scalars().all()
def request_client_acceptance_declarations_service(db, *, row, actor_user_id: int):
assigned_ids = _applicable_acceptance_user_ids(db, row)
if not assigned_ids:
raise HTTPException(status_code=400, detail="No applicable partner/review user is mapped to this client.")
created = 0
for uid in assigned_ids:
for dtype in sorted(DECLARATION_TYPES):
existing = db.execute(
select(ClientAcceptanceDeclaration).where(
ClientAcceptanceDeclaration.client_id == row.id,
ClientAcceptanceDeclaration.declaration_type == dtype,
ClientAcceptanceDeclaration.assigned_user_id == uid,
)
).scalar_one_or_none()
if not existing:
db.add(ClientAcceptanceDeclaration(
client_id=row.id,
tenant_id=row.tenant_id,
branch_id=row.branch_id,
declaration_type=dtype,
assigned_user_id=uid,
assigned_role_snapshot=_role_name_for_user(db, uid),
declaration_text=_declaration_text(row, dtype),
requested_by_user_id=actor_user_id,
))
created += 1
_safe_create_alert(
db,
user_id=uid,
title=f"{dtype.title()} declaration required",
message=f"Please submit your {dtype} declaration for {row.client_name}.",
client=row,
actor_user_id=actor_user_id,
)
repository.write_audit_log(
db,
client_id=row.id,
tenant_id=row.tenant_id,
branch_id=row.branch_id,
actor_user_id=actor_user_id,
action="acceptance_declarations_requested",
summary="Digital independence/conflict declarations requested.",
payload_json={"created": created, "assigned_user_ids": assigned_ids},
)
db.commit()
return list_client_acceptance_declarations(db, client_id=row.id)
def submit_client_acceptance_declaration_service(db, *, declaration_id: int, current_user, clear: bool, notes: str | None, issue_details: str | None, request=None):
decl = db.get(ClientAcceptanceDeclaration, declaration_id)
if not decl:
raise HTTPException(status_code=404, detail="Declaration not found.")
if decl.assigned_user_id != current_user.id:
raise HTTPException(status_code=403, detail="This declaration is assigned to another user.")
ip, agent = _request_ip_and_agent(request)
decl.status = "declared_clear" if clear else "declared_issue"
decl.response_notes = _clean_text(notes)
decl.issue_details = _clean_text(issue_details)
decl.responded_at_utc = datetime.now(timezone.utc)
decl.response_ip = ip
decl.response_user_agent = agent
db.add(decl)
db.flush()
_refresh_acceptance_derived_statuses(db, client_id=decl.client_id, actor_user_id=current_user.id)
db.commit()
return decl
def get_client_kyc_verification(db, *, client_id: int) -> ClientKycVerification | None:
return db.execute(select(ClientKycVerification).where(ClientKycVerification.client_id == client_id)).scalar_one_or_none()
def _permanent_docs_summary(db, *, client_id: int) -> tuple[str, int]:
docs = db.execute(
select(PermanentClientDocument)
.where(PermanentClientDocument.client_id == client_id, PermanentClientDocument.is_deleted.is_(False))
.order_by(PermanentClientDocument.category.asc(), PermanentClientDocument.title.asc())
).scalars().all()
parts = []
for d in docs:
parts.append(f"{d.category}: {d.title} ({d.current_version_no} version(s))")
return "\n".join(parts), len(docs)
def sync_client_kyc_from_permanent_documents_service(db, *, row, actor_user_id: int):
summary, count = _permanent_docs_summary(db, client_id=row.id)
kyc = get_client_kyc_verification(db, client_id=row.id)
status = "pending_verification" if count else "pending_documents"
required_summary = "\n".join(f"- {x}" for x in KYC_REQUIRED_KEYWORDS)
if not kyc:
kyc = ClientKycVerification(client_id=row.id, tenant_id=row.tenant_id, branch_id=row.branch_id)
if kyc.status != "verified":
kyc.status = status
kyc.required_document_summary = required_summary
kyc.available_document_summary = summary or "No permanent client documents uploaded yet."
db.add(kyc)
repository.write_audit_log(
db,
client_id=row.id,
tenant_id=row.tenant_id,
branch_id=row.branch_id,
actor_user_id=actor_user_id,
action="kyc_synced_from_permanent_documents",
summary="KYC status refreshed from permanent client documents.",
payload_json={"permanent_document_count": count, "status": kyc.status},
)
db.commit()
return kyc
def verify_client_kyc_service(db, *, row, actor_user_id: int, notes: str | None = None):
kyc = sync_client_kyc_from_permanent_documents_service(db, row=row, actor_user_id=actor_user_id)
if not kyc.available_document_summary or kyc.available_document_summary.startswith("No permanent"):
raise HTTPException(status_code=400, detail="KYC cannot be verified until permanent client documents are uploaded.")
now = datetime.now(timezone.utc)
kyc.status = "verified"
kyc.verified_by_user_id = actor_user_id
kyc.verified_at_utc = now
kyc.rejected_by_user_id = None
kyc.rejected_at_utc = None
kyc.verification_notes = _clean_text(notes)
db.add(kyc)
updated = repository.update_client(db, row, {"kyc_completed": True})
repository.write_audit_log(db, client_id=row.id, tenant_id=row.tenant_id, branch_id=row.branch_id, actor_user_id=actor_user_id, action="kyc_verified", summary="KYC verified using permanent client documents.", payload_json={"notes": notes})
db.commit()
return updated, kyc
def reject_client_kyc_service(db, *, row, actor_user_id: int, notes: str | None = None):
kyc = get_client_kyc_verification(db, client_id=row.id) or ClientKycVerification(client_id=row.id, tenant_id=row.tenant_id, branch_id=row.branch_id)
now = datetime.now(timezone.utc)
kyc.status = "rejected"
kyc.rejected_by_user_id = actor_user_id
kyc.rejected_at_utc = now
kyc.verification_notes = _clean_text(notes)
db.add(kyc)
updated = repository.update_client(db, row, {"kyc_completed": False})
_safe_create_alert(db, user_id=row.portal_user_id, title="KYC documents require resubmission", message=notes or "Please update the permanent documents requested by the firm.", client=row, actor_user_id=actor_user_id, target_url="/client/documents") if getattr(row, "portal_user_id", None) else None
repository.write_audit_log(db, client_id=row.id, tenant_id=row.tenant_id, branch_id=row.branch_id, actor_user_id=actor_user_id, action="kyc_rejected", summary="KYC rejected / resubmission requested.", payload_json={"notes": notes})
db.commit()
return updated, kyc
def get_current_engagement_letter(db, *, client_id: int) -> ClientEngagementLetter | None:
return db.execute(
select(ClientEngagementLetter)
.where(ClientEngagementLetter.client_id == client_id)
.order_by(ClientEngagementLetter.version_no.desc(), ClientEngagementLetter.id.desc())
).scalars().first()
def _default_engagement_letter_body(row) -> str:
return f"""Dear {row.client_name},
We are pleased to confirm our understanding of the terms and objectives of our professional engagement. The services will be performed subject to applicable laws, professional standards, ICAI requirements, management responsibilities, timely submission of records and payment of agreed fees.
Management is responsible for the completeness and accuracy of records, explanations and representations provided to the firm. Our responsibility is limited to the scope agreed for the relevant service/financial year. This engagement letter, once accepted digitally or by signed upload, will form part of the client acceptance and continuance evidence.
Regards,
A R R R & Associates"""
def draft_client_engagement_letter_service(db, *, row, actor_user_id: int, title: str | None = None, body_text: str | None = None):
latest = get_current_engagement_letter(db, client_id=row.id)
version = (latest.version_no + 1) if latest else 1
letter = ClientEngagementLetter(
client_id=row.id,
tenant_id=row.tenant_id,
branch_id=row.branch_id,
letter_code=f"ENG-{row.id}",
title=_clean_text(title, 255) or f"Engagement Letter - {row.client_name}",
body_text=_clean_text(body_text) or _default_engagement_letter_body(row),
version_no=version,
status="draft",
created_by_user_id=actor_user_id,
)
db.add(letter)
repository.write_audit_log(db, client_id=row.id, tenant_id=row.tenant_id, branch_id=row.branch_id, actor_user_id=actor_user_id, action="engagement_letter_drafted", summary="Engagement letter draft created.", payload_json={"version_no": version})
db.commit()
db.refresh(letter)
return letter
def approve_and_send_engagement_letter_service(db, *, row, letter_id: int, actor_user_id: int):
letter = db.get(ClientEngagementLetter, letter_id)
if not letter or letter.client_id != row.id:
raise HTTPException(status_code=404, detail="Engagement letter not found.")
now = datetime.now(timezone.utc)
letter.status = "sent_to_client"
letter.approved_by_user_id = actor_user_id
letter.approved_at_utc = now
letter.sent_to_client_at_utc = now
letter.pdf_hash_sha256 = hashlib.sha256(letter.body_text.encode("utf-8")).hexdigest()
db.add(letter)
if getattr(row, "portal_user_id", None):
_safe_create_alert(db, user_id=row.portal_user_id, title="Engagement letter ready for acceptance", message=f"Please accept the engagement letter for {row.client_name} digitally using OTP or upload the signed copy.", client=row, actor_user_id=actor_user_id, target_url="/client/engagement-letter")
repository.write_audit_log(db, client_id=row.id, tenant_id=row.tenant_id, branch_id=row.branch_id, actor_user_id=actor_user_id, action="engagement_letter_sent", summary="Engagement letter approved and sent to client.", payload_json={"letter_id": letter.id, "version_no": letter.version_no})
db.commit()
return letter
def render_engagement_letter_html(letter: ClientEngagementLetter, client) -> str:
import html
body = html.escape(letter.body_text).replace("\n", "<br>")
return f"""<!doctype html><html><head><meta charset='utf-8'><title>{html.escape(letter.title)}</title>
<style>body{{font-family:Arial,sans-serif;max-width:850px;margin:40px auto;line-height:1.55;color:#111827}}.meta{{color:#64748b;font-size:13px}}.box{{border:1px solid #cbd5e1;border-radius:12px;padding:24px}}</style></head><body>
<div class='box'><h1>{html.escape(letter.title)}</h1><p class='meta'>Client: {html.escape(client.client_name)} | Version: {letter.version_no} | Status: {html.escape(letter.status)}</p><p>{body}</p></div>
<script>window.print && setTimeout(function(){{window.print()}}, 300);</script></body></html>"""
def digitally_accept_engagement_letter_service(db, *, row, letter_id: int, current_user, declaration_text: str | None, request=None):
letter = db.get(ClientEngagementLetter, letter_id)
if not letter or letter.client_id != row.id:
raise HTTPException(status_code=404, detail="Engagement letter not found.")
if letter.status not in {"sent_to_client", "manual_uploaded", "rejected"}:
raise HTTPException(status_code=400, detail="Only partner-approved engagement letters sent to client can be accepted.")
ip, agent = _request_ip_and_agent(request)
now = datetime.now(timezone.utc)
declaration = _clean_text(declaration_text) or "I have read and accept the engagement letter for and on behalf of the client."
letter.status = "digitally_accepted"
letter.accepted_mode = "digital_otp"
letter.accepted_by_user_id = current_user.id
letter.accepted_at_utc = now
letter.accepted_ip = ip
letter.accepted_user_agent = agent
letter.acceptance_declaration_text = declaration
letter.otp_verified = True
db.add(letter)
updated = repository.update_client(db, row, {"engagement_letter_received": True})
repository.write_audit_log(db, client_id=row.id, tenant_id=row.tenant_id, branch_id=row.branch_id, actor_user_id=current_user.id, action="engagement_letter_digitally_accepted", summary="Engagement letter accepted digitally with OTP/session evidence.", payload_json={"letter_id": letter.id, "ip": ip})
db.commit()
return updated, letter
def save_manual_signed_engagement_letter_service(db, *, row, letter_id: int, upload_file, current_user, request=None):
letter = db.get(ClientEngagementLetter, letter_id)
if not letter or letter.client_id != row.id:
raise HTTPException(status_code=404, detail="Engagement letter not found.")
if letter.status not in {"sent_to_client", "manual_uploaded", "rejected"}:
raise HTTPException(status_code=400, detail="Signed copy can be uploaded only after partner approval and sending to client.")
ACCEPTANCE_STORAGE_ROOT.mkdir(parents=True, exist_ok=True)
folder = ACCEPTANCE_STORAGE_ROOT / str(row.tenant_id) / str(row.id)
folder.mkdir(parents=True, exist_ok=True)
filename = _slug(getattr(upload_file, "filename", None) or "signed_engagement_letter.pdf")
target = folder / f"letter_{letter.id}_signed_{filename}"
content = upload_file.file.read()
if not content:
raise HTTPException(status_code=400, detail="Uploaded signed engagement letter is empty.")
target.write_bytes(content)
letter.status = "manual_uploaded"
letter.accepted_mode = "manual_signed_upload"
letter.accepted_by_user_id = current_user.id
letter.accepted_at_utc = datetime.now(timezone.utc)
letter.manual_signed_file_path = str(target)
letter.manual_signed_file_hash_sha256 = hashlib.sha256(content).hexdigest()
db.add(letter)
repository.write_audit_log(db, client_id=row.id, tenant_id=row.tenant_id, branch_id=row.branch_id, actor_user_id=current_user.id, action="engagement_letter_signed_uploaded", summary="Client uploaded manually signed engagement letter.", payload_json={"letter_id": letter.id, "filename": filename})
if row.partner_id:
_safe_create_alert(db, user_id=row.partner_id, title="Signed engagement letter uploaded", message=f"Please verify signed engagement letter for {row.client_name}.", client=row, actor_user_id=current_user.id)
db.commit()
return letter
def verify_manual_engagement_letter_service(db, *, row, letter_id: int, actor_user_id: int):
letter = db.get(ClientEngagementLetter, letter_id)
if not letter or letter.client_id != row.id:
raise HTTPException(status_code=404, detail="Engagement letter not found.")
if letter.status != "manual_uploaded":
raise HTTPException(status_code=400, detail="Only manually uploaded engagement letters can be verified.")
now = datetime.now(timezone.utc)
letter.status = "manual_verified"
letter.manual_verified_by_user_id = actor_user_id
letter.manual_verified_at_utc = now
db.add(letter)
updated = repository.update_client(db, row, {"engagement_letter_received": True})
repository.write_audit_log(db, client_id=row.id, tenant_id=row.tenant_id, branch_id=row.branch_id, actor_user_id=actor_user_id, action="engagement_letter_manual_verified", summary="Manual signed engagement letter verified.", payload_json={"letter_id": letter.id})
db.commit()
return updated, letter
def _refresh_acceptance_derived_statuses(db, *, client_id: int, actor_user_id: int | None = None):
client = repository.get_client_by_id(db, client_id)
if not client:
return None
declarations = list_client_acceptance_declarations(db, client_id=client_id)
independence = [d for d in declarations if d.declaration_type == "independence"]
conflict = [d for d in declarations if d.declaration_type == "conflict"]
independence_ok = bool(independence) and all(d.status in DECLARATION_CLEAR_STATUSES for d in independence)
conflict_ok = bool(conflict) and all(d.status in DECLARATION_CLEAR_STATUSES for d in conflict)
any_issue = any(d.status == "declared_issue" for d in declarations)
kyc = get_client_kyc_verification(db, client_id=client_id)
letter = get_current_engagement_letter(db, client_id=client_id)
payload = {
"independence_check_completed": independence_ok,
"conflict_check_completed": conflict_ok,
"kyc_completed": bool(kyc and kyc.status == "verified"),
"engagement_letter_received": bool(letter and letter.status in ENGAGEMENT_LETTER_ACCEPTED_STATUSES),
}
if any_issue:
payload["acceptance_status"] = "pending_review"
updated = repository.update_client(db, client, payload)
return updated
def build_client_acceptance_workflow_payload(db, *, client_id: int) -> dict:
client = repository.get_client_by_id(db, client_id)
declarations = list_client_acceptance_declarations(db, client_id=client_id)
kyc = get_client_kyc_verification(db, client_id=client_id)
letter = get_current_engagement_letter(db, client_id=client_id)
permanent_summary, permanent_count = _permanent_docs_summary(db, client_id=client_id)
return {
"acceptance_declarations": declarations,
"kyc_verification": kyc,
"engagement_letter": letter,
"permanent_document_summary": permanent_summary,
"permanent_document_count": permanent_count,
"workflow_ready": bool(client and client.independence_check_completed and client.conflict_check_completed and client.kyc_completed and (not client.engagement_letter_required or client.engagement_letter_received)),
}