Move AQMM controls to assurance engagement workflow
This commit is contained in:
A R R R Associates
@@ -13,6 +13,7 @@ from app.modules.services.models import ClientServiceSubscription, FirmServiceSe
|
||||
|
||||
SUBSCRIPTION_STATUSES = [
|
||||
("draft", "Draft"),
|
||||
("pending_acceptance", "Pending AQMM Acceptance"),
|
||||
("active", "Active"),
|
||||
("on_hold", "On Hold"),
|
||||
("completed", "Completed"),
|
||||
@@ -204,3 +205,392 @@ def review_partner_required_for_engagement(db: Session, *, tenant_id: int, engag
|
||||
|
||||
def list_review_partners(db: Session, *, tenant_id: int, branch_id: int | None = None):
|
||||
return list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id, role_names=("Partner",))
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# AQMM engagement-level quality workflow helpers
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from app.modules.alerts.service import create_alert
|
||||
from app.modules.services.models import EngagementQualityDeclaration, EngagementKycVerification, EngagementLetter
|
||||
|
||||
QUALITY_NOT_REQUIRED = "not_required"
|
||||
QUALITY_PENDING = "pending_acceptance"
|
||||
QUALITY_IN_PROGRESS = "in_progress"
|
||||
QUALITY_READY = "ready_for_approval"
|
||||
QUALITY_APPROVED = "approved"
|
||||
QUALITY_REJECTED = "rejected"
|
||||
|
||||
DECLARATION_PENDING = "pending"
|
||||
DECLARATION_CLEAR = "declared_clear"
|
||||
DECLARATION_CONFLICT = "conflict_declared"
|
||||
DECLARATION_NOT_APPLICABLE = "not_applicable"
|
||||
|
||||
|
||||
def is_assurance_engagement(engagement_type: str | None) -> bool:
|
||||
return (engagement_type or "").strip().lower() == "assurance"
|
||||
|
||||
|
||||
def quality_required_for_engagement(engagement_type: str | None) -> bool:
|
||||
return is_assurance_engagement(engagement_type)
|
||||
|
||||
|
||||
def _engagement_team_user_ids(subscription: ClientServiceSubscription) -> list[int]:
|
||||
ids = [
|
||||
subscription.assigned_partner_user_id,
|
||||
subscription.assigned_manager_user_id,
|
||||
subscription.assigned_staff_user_id,
|
||||
subscription.review_partner_user_id,
|
||||
]
|
||||
seen: set[int] = set()
|
||||
result: list[int] = []
|
||||
for uid in ids:
|
||||
if uid and int(uid) not in seen:
|
||||
seen.add(int(uid))
|
||||
result.append(int(uid))
|
||||
return result
|
||||
|
||||
|
||||
def _role_for_engagement_user(subscription: ClientServiceSubscription, user_id: int) -> str | None:
|
||||
if subscription.assigned_partner_user_id == user_id:
|
||||
return "Assigned Partner"
|
||||
if subscription.review_partner_user_id == user_id:
|
||||
return "Review Partner"
|
||||
if subscription.assigned_manager_user_id == user_id:
|
||||
return "Assigned Manager"
|
||||
if subscription.assigned_staff_user_id == user_id:
|
||||
return "Assigned Staff"
|
||||
return None
|
||||
|
||||
|
||||
def ensure_engagement_quality_workflow(
|
||||
db: Session,
|
||||
*,
|
||||
subscription: ClientServiceSubscription,
|
||||
actor_user_id: int | None = None,
|
||||
create_declarations: bool = False,
|
||||
) -> ClientServiceSubscription:
|
||||
"""Initialise or sync AQMM status for an engagement/subscription.
|
||||
|
||||
Full quality workflow is mandatory only for assurance engagements. For
|
||||
non-assurance engagements, the quality fields are reset to not_required.
|
||||
"""
|
||||
required = quality_required_for_engagement(subscription.engagement_type)
|
||||
subscription.quality_workflow_required = required
|
||||
|
||||
if not required:
|
||||
subscription.quality_workflow_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_acceptance_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_independence_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_conflict_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_kyc_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_engagement_letter_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_block_reason = None
|
||||
return subscription
|
||||
|
||||
if subscription.quality_workflow_status in (None, "", QUALITY_NOT_REQUIRED):
|
||||
subscription.quality_workflow_status = QUALITY_PENDING
|
||||
if subscription.quality_acceptance_status in (None, "", QUALITY_NOT_REQUIRED):
|
||||
subscription.quality_acceptance_status = QUALITY_PENDING
|
||||
if subscription.quality_independence_status in (None, "", QUALITY_NOT_REQUIRED):
|
||||
subscription.quality_independence_status = "pending_declarations"
|
||||
if subscription.quality_conflict_status in (None, "", QUALITY_NOT_REQUIRED):
|
||||
subscription.quality_conflict_status = "pending_declarations"
|
||||
if subscription.quality_kyc_status in (None, "", QUALITY_NOT_REQUIRED):
|
||||
subscription.quality_kyc_status = "pending_verification"
|
||||
if subscription.quality_engagement_letter_status in (None, "", QUALITY_NOT_REQUIRED):
|
||||
subscription.quality_engagement_letter_status = "pending_client_acceptance"
|
||||
|
||||
if create_declarations:
|
||||
request_engagement_quality_declarations(db, subscription=subscription, actor_user_id=actor_user_id)
|
||||
|
||||
update_engagement_quality_summary(db, subscription=subscription)
|
||||
return subscription
|
||||
|
||||
|
||||
def request_engagement_quality_declarations(
|
||||
db: Session,
|
||||
*,
|
||||
subscription: ClientServiceSubscription,
|
||||
actor_user_id: int | None = None,
|
||||
) -> list[EngagementQualityDeclaration]:
|
||||
ensure_engagement_quality_workflow(db, subscription=subscription, actor_user_id=actor_user_id, create_declarations=False)
|
||||
created_or_existing: list[EngagementQualityDeclaration] = []
|
||||
target_url = f"/services/engagements/{subscription.id}"
|
||||
|
||||
for user_id in _engagement_team_user_ids(subscription):
|
||||
role = _role_for_engagement_user(subscription, user_id)
|
||||
for declaration_type in ("independence", "conflict"):
|
||||
existing = db.execute(
|
||||
select(EngagementQualityDeclaration).where(
|
||||
EngagementQualityDeclaration.subscription_id == subscription.id,
|
||||
EngagementQualityDeclaration.declaration_type == declaration_type,
|
||||
EngagementQualityDeclaration.requested_user_id == user_id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if existing:
|
||||
created_or_existing.append(existing)
|
||||
continue
|
||||
row = EngagementQualityDeclaration(
|
||||
tenant_id=subscription.tenant_id,
|
||||
branch_id=subscription.branch_id,
|
||||
client_id=subscription.client_id,
|
||||
subscription_id=subscription.id,
|
||||
declaration_type=declaration_type,
|
||||
requested_user_id=user_id,
|
||||
requested_role=role,
|
||||
status=DECLARATION_PENDING,
|
||||
created_by_user_id=actor_user_id,
|
||||
)
|
||||
db.add(row)
|
||||
db.flush()
|
||||
created_or_existing.append(row)
|
||||
try:
|
||||
create_alert(
|
||||
db,
|
||||
user_id=user_id,
|
||||
tenant_id=subscription.tenant_id,
|
||||
branch_id=subscription.branch_id,
|
||||
alert_type="general",
|
||||
priority="high" if declaration_type == "conflict" else "normal",
|
||||
title=f"AQMM {declaration_type.title()} Declaration Required",
|
||||
message=f"Please submit your {declaration_type} declaration for this assurance engagement.",
|
||||
target_url=target_url,
|
||||
created_by_user_id=actor_user_id,
|
||||
commit=False,
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
update_engagement_quality_summary(db, subscription=subscription)
|
||||
return created_or_existing
|
||||
|
||||
|
||||
def list_engagement_quality_declarations(db: Session, *, subscription_id: int) -> list[EngagementQualityDeclaration]:
|
||||
return db.execute(
|
||||
select(EngagementQualityDeclaration)
|
||||
.where(EngagementQualityDeclaration.subscription_id == subscription_id)
|
||||
.order_by(EngagementQualityDeclaration.declaration_type.asc(), EngagementQualityDeclaration.id.asc())
|
||||
).scalars().all()
|
||||
|
||||
|
||||
def respond_engagement_quality_declaration(
|
||||
db: Session,
|
||||
*,
|
||||
declaration_id: int,
|
||||
current_user_id: int,
|
||||
status: str,
|
||||
notes: str | None = None,
|
||||
request=None,
|
||||
) -> EngagementQualityDeclaration:
|
||||
row = db.get(EngagementQualityDeclaration, declaration_id)
|
||||
if not row or int(row.requested_user_id) != int(current_user_id):
|
||||
raise ValueError("Declaration not found for current user.")
|
||||
if status not in {DECLARATION_CLEAR, DECLARATION_CONFLICT, DECLARATION_NOT_APPLICABLE}:
|
||||
raise ValueError("Invalid declaration status.")
|
||||
row.status = status
|
||||
row.response_notes = (notes or "").strip() or None
|
||||
row.responded_at_utc = datetime.now(timezone.utc)
|
||||
if request is not None:
|
||||
row.ip_address = getattr(getattr(request, "client", None), "host", None)
|
||||
row.user_agent = request.headers.get("user-agent") if hasattr(request, "headers") else None
|
||||
subscription = db.get(ClientServiceSubscription, row.subscription_id)
|
||||
if subscription:
|
||||
update_engagement_quality_summary(db, subscription=subscription)
|
||||
return row
|
||||
|
||||
|
||||
def get_latest_engagement_kyc_verification(db: Session, *, subscription_id: int) -> EngagementKycVerification | None:
|
||||
return db.execute(
|
||||
select(EngagementKycVerification)
|
||||
.where(EngagementKycVerification.subscription_id == subscription_id)
|
||||
.order_by(EngagementKycVerification.id.desc())
|
||||
).scalars().first()
|
||||
|
||||
|
||||
def verify_engagement_kyc_from_permanent_documents(
|
||||
db: Session,
|
||||
*,
|
||||
subscription: ClientServiceSubscription,
|
||||
actor_user_id: int,
|
||||
notes: str | None = None,
|
||||
) -> EngagementKycVerification:
|
||||
ensure_engagement_quality_workflow(db, subscription=subscription, actor_user_id=actor_user_id, create_declarations=False)
|
||||
row = get_latest_engagement_kyc_verification(db, subscription_id=subscription.id)
|
||||
if not row:
|
||||
row = EngagementKycVerification(
|
||||
tenant_id=subscription.tenant_id,
|
||||
branch_id=subscription.branch_id,
|
||||
client_id=subscription.client_id,
|
||||
subscription_id=subscription.id,
|
||||
created_by_user_id=actor_user_id,
|
||||
)
|
||||
db.add(row)
|
||||
row.status = "verified"
|
||||
row.source = "permanent_documents"
|
||||
row.verification_notes = (notes or "").strip() or "Verified from permanent document vault."
|
||||
row.verified_by_user_id = actor_user_id
|
||||
row.verified_at_utc = datetime.now(timezone.utc)
|
||||
subscription.quality_kyc_status = "verified"
|
||||
update_engagement_quality_summary(db, subscription=subscription)
|
||||
return row
|
||||
|
||||
|
||||
def get_current_engagement_letter(db: Session, *, subscription_id: int) -> EngagementLetter | None:
|
||||
return db.execute(
|
||||
select(EngagementLetter)
|
||||
.where(EngagementLetter.subscription_id == subscription_id)
|
||||
.order_by(EngagementLetter.version_no.desc(), EngagementLetter.id.desc())
|
||||
).scalars().first()
|
||||
|
||||
|
||||
def mark_engagement_letter_completed(
|
||||
db: Session,
|
||||
*,
|
||||
subscription: ClientServiceSubscription,
|
||||
actor_user_id: int,
|
||||
acceptance_mode: str,
|
||||
notes: str | None = None,
|
||||
request=None,
|
||||
) -> EngagementLetter:
|
||||
ensure_engagement_quality_workflow(db, subscription=subscription, actor_user_id=actor_user_id, create_declarations=False)
|
||||
if acceptance_mode not in {"digital_otp", "manual_signed_upload"}:
|
||||
raise ValueError("Invalid engagement letter acceptance mode.")
|
||||
letter = get_current_engagement_letter(db, subscription_id=subscription.id)
|
||||
if not letter:
|
||||
letter = EngagementLetter(
|
||||
tenant_id=subscription.tenant_id,
|
||||
branch_id=subscription.branch_id,
|
||||
client_id=subscription.client_id,
|
||||
subscription_id=subscription.id,
|
||||
title="Engagement Letter",
|
||||
version_no=1,
|
||||
status="draft_pending",
|
||||
created_by_user_id=actor_user_id,
|
||||
)
|
||||
db.add(letter)
|
||||
letter.status = "digitally_accepted" if acceptance_mode == "digital_otp" else "manual_signed_verified"
|
||||
letter.acceptance_mode = acceptance_mode
|
||||
letter.client_acceptance_declaration = (notes or "").strip() or None
|
||||
letter.client_accepted_by_user_id = actor_user_id if acceptance_mode == "digital_otp" else letter.client_accepted_by_user_id
|
||||
letter.client_accepted_at_utc = datetime.now(timezone.utc) if acceptance_mode == "digital_otp" else letter.client_accepted_at_utc
|
||||
letter.manual_verified_by_user_id = actor_user_id if acceptance_mode == "manual_signed_upload" else letter.manual_verified_by_user_id
|
||||
letter.manual_verified_at_utc = datetime.now(timezone.utc) if acceptance_mode == "manual_signed_upload" else letter.manual_verified_at_utc
|
||||
if request is not None:
|
||||
letter.ip_address = getattr(getattr(request, "client", None), "host", None)
|
||||
letter.user_agent = request.headers.get("user-agent") if hasattr(request, "headers") else None
|
||||
subscription.quality_engagement_letter_status = "completed"
|
||||
update_engagement_quality_summary(db, subscription=subscription)
|
||||
return letter
|
||||
|
||||
|
||||
def update_engagement_quality_summary(db: Session, *, subscription: ClientServiceSubscription) -> ClientServiceSubscription:
|
||||
if not quality_required_for_engagement(subscription.engagement_type):
|
||||
subscription.quality_workflow_required = False
|
||||
subscription.quality_workflow_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_acceptance_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_independence_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_conflict_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_kyc_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_engagement_letter_status = QUALITY_NOT_REQUIRED
|
||||
subscription.quality_block_reason = None
|
||||
return subscription
|
||||
|
||||
subscription.quality_workflow_required = True
|
||||
declarations = list_engagement_quality_declarations(db, subscription_id=subscription.id) if subscription.id else []
|
||||
independence = [d for d in declarations if d.declaration_type == "independence"]
|
||||
conflicts = [d for d in declarations if d.declaration_type == "conflict"]
|
||||
|
||||
if independence:
|
||||
if any(d.status == DECLARATION_PENDING for d in independence):
|
||||
subscription.quality_independence_status = "pending_declarations"
|
||||
elif any(d.status == DECLARATION_CONFLICT for d in independence):
|
||||
subscription.quality_independence_status = "issue_reported"
|
||||
else:
|
||||
subscription.quality_independence_status = "completed"
|
||||
else:
|
||||
subscription.quality_independence_status = "pending_declarations"
|
||||
|
||||
if conflicts:
|
||||
if any(d.status == DECLARATION_PENDING for d in conflicts):
|
||||
subscription.quality_conflict_status = "pending_declarations"
|
||||
elif any(d.status == DECLARATION_CONFLICT for d in conflicts):
|
||||
subscription.quality_conflict_status = "conflict_reported"
|
||||
else:
|
||||
subscription.quality_conflict_status = "clear"
|
||||
else:
|
||||
subscription.quality_conflict_status = "pending_declarations"
|
||||
|
||||
kyc = get_latest_engagement_kyc_verification(db, subscription_id=subscription.id) if subscription.id else None
|
||||
if kyc and kyc.status == "verified":
|
||||
subscription.quality_kyc_status = "verified"
|
||||
elif subscription.quality_kyc_status not in {"verified", "not_required"}:
|
||||
subscription.quality_kyc_status = "pending_verification"
|
||||
|
||||
letter = get_current_engagement_letter(db, subscription_id=subscription.id) if subscription.id else None
|
||||
if letter and letter.status in {"digitally_accepted", "manual_signed_verified", "completed"}:
|
||||
subscription.quality_engagement_letter_status = "completed"
|
||||
elif subscription.quality_engagement_letter_status not in {"completed", "not_required"}:
|
||||
subscription.quality_engagement_letter_status = "pending_client_acceptance"
|
||||
|
||||
ready = (
|
||||
subscription.quality_independence_status == "completed"
|
||||
and subscription.quality_conflict_status == "clear"
|
||||
and subscription.quality_kyc_status == "verified"
|
||||
and subscription.quality_engagement_letter_status == "completed"
|
||||
)
|
||||
if subscription.quality_acceptance_status == QUALITY_APPROVED:
|
||||
subscription.quality_workflow_status = QUALITY_APPROVED
|
||||
subscription.quality_block_reason = None
|
||||
elif ready:
|
||||
subscription.quality_workflow_status = QUALITY_READY
|
||||
subscription.quality_acceptance_status = QUALITY_READY
|
||||
subscription.quality_block_reason = None
|
||||
else:
|
||||
subscription.quality_workflow_status = QUALITY_IN_PROGRESS
|
||||
subscription.quality_acceptance_status = QUALITY_PENDING
|
||||
blockers = []
|
||||
if subscription.quality_independence_status != "completed":
|
||||
blockers.append("independence declarations pending/issue")
|
||||
if subscription.quality_conflict_status != "clear":
|
||||
blockers.append("conflict declarations pending/conflict")
|
||||
if subscription.quality_kyc_status != "verified":
|
||||
blockers.append("KYC verification pending")
|
||||
if subscription.quality_engagement_letter_status != "completed":
|
||||
blockers.append("engagement letter acceptance pending")
|
||||
subscription.quality_block_reason = "; ".join(blockers) or None
|
||||
return subscription
|
||||
|
||||
|
||||
def approve_engagement_quality_workflow(
|
||||
db: Session,
|
||||
*,
|
||||
subscription: ClientServiceSubscription,
|
||||
actor_user_id: int,
|
||||
) -> ClientServiceSubscription:
|
||||
ensure_engagement_quality_workflow(db, subscription=subscription, actor_user_id=actor_user_id, create_declarations=False)
|
||||
update_engagement_quality_summary(db, subscription=subscription)
|
||||
if subscription.quality_workflow_status != QUALITY_READY:
|
||||
raise ValueError(subscription.quality_block_reason or "AQMM workflow is not ready for approval.")
|
||||
subscription.quality_acceptance_status = QUALITY_APPROVED
|
||||
subscription.quality_workflow_status = QUALITY_APPROVED
|
||||
subscription.quality_approved_by_user_id = actor_user_id
|
||||
subscription.quality_approved_at_utc = datetime.now(timezone.utc)
|
||||
subscription.quality_block_reason = None
|
||||
subscription.status = "active"
|
||||
subscription.is_active = True
|
||||
return subscription
|
||||
|
||||
|
||||
def enforce_quality_gate_on_subscription(subscription: ClientServiceSubscription) -> None:
|
||||
"""Prevent assurance engagement from becoming active before AQMM approval."""
|
||||
if quality_required_for_engagement(subscription.engagement_type) and subscription.quality_acceptance_status != QUALITY_APPROVED:
|
||||
subscription.quality_workflow_required = True
|
||||
if subscription.quality_workflow_status in (None, "", QUALITY_NOT_REQUIRED):
|
||||
subscription.quality_workflow_status = QUALITY_PENDING
|
||||
if subscription.status == "active":
|
||||
subscription.status = "pending_acceptance"
|
||||
subscription.is_active = False
|
||||
if not subscription.quality_block_reason:
|
||||
subscription.quality_block_reason = "AQMM acceptance workflow pending for assurance engagement."
|
||||
|
||||
Reference in New Issue
Block a user