Add engagement financial year correction workflow
This commit is contained in:
A R R R Associates
@@ -0,0 +1,299 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from fastapi import Request
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.modules.core.audit.models import AuditLog
|
||||
from app.modules.core.tenancy.year_control import is_financial_year_locked, is_row_financial_year_locked
|
||||
from app.modules.documents.models import EngagementDocument
|
||||
from app.modules.services.client_services import assessment_year_from_financial_year, period_choices_for_service
|
||||
from app.modules.services.due_dates import apply_due_date_rule_to_subscription
|
||||
from app.modules.services.models import (
|
||||
ClientServiceSubscription,
|
||||
ClientServiceTaskInstance,
|
||||
EngagementClosureChecklist,
|
||||
)
|
||||
|
||||
_FINANCIAL_YEAR_RE = re.compile(r"^(\d{4})-(\d{2})$")
|
||||
_CLOSED_ENGAGEMENT_STATUSES = {"completed", "cancelled", "inactive"}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EngagementFYCorrectionResult:
|
||||
subscription_id: int
|
||||
old_financial_year: str
|
||||
new_financial_year: str
|
||||
old_period_label: str
|
||||
new_period_label: str
|
||||
task_count: int
|
||||
document_count: int
|
||||
|
||||
|
||||
def validate_financial_year(value: str | None) -> str:
|
||||
raw = (value or "").strip()
|
||||
match = _FINANCIAL_YEAR_RE.fullmatch(raw)
|
||||
if not match:
|
||||
raise ValueError("Financial year must be in YYYY-YY format, for example 2025-26.")
|
||||
start_year = int(match.group(1))
|
||||
expected_suffix = str(start_year + 1)[-2:]
|
||||
if match.group(2) != expected_suffix:
|
||||
raise ValueError("Financial year end year does not match the start year.")
|
||||
return f"{start_year:04d}-{expected_suffix}"
|
||||
|
||||
|
||||
def _remap_period_label(row: ClientServiceSubscription, target_financial_year: str) -> str:
|
||||
recurrence = (getattr(getattr(row, "catalogue", None), "recurrence_type", None) or "one_time").strip().lower()
|
||||
current = (getattr(row, "period_label", None) or "").strip()
|
||||
choices = period_choices_for_service(target_financial_year, recurrence)
|
||||
valid_codes = [code for code, _label in choices]
|
||||
|
||||
if recurrence == "monthly":
|
||||
# Period codes are YYYY-MM. Keep the same month while moving it to the
|
||||
# corresponding month inside the corrected financial year.
|
||||
try:
|
||||
month = int(current.rsplit("-", 1)[1])
|
||||
except (ValueError, IndexError):
|
||||
raise ValueError("The existing monthly period is invalid and cannot be remapped automatically.")
|
||||
target_start = int(target_financial_year.split("-", 1)[0])
|
||||
target_year = target_start if month >= 4 else target_start + 1
|
||||
candidate = f"{target_year:04d}-{month:02d}"
|
||||
if candidate not in valid_codes:
|
||||
raise ValueError("The monthly period cannot be mapped to the corrected financial year.")
|
||||
return candidate
|
||||
|
||||
if recurrence == "quarterly":
|
||||
if current not in valid_codes:
|
||||
raise ValueError("The existing quarter is invalid for the corrected financial year.")
|
||||
return current
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
def _engagement_has_finalised_documents(db: Session, subscription_id: int) -> bool:
|
||||
rows = db.execute(
|
||||
select(EngagementDocument).where(
|
||||
EngagementDocument.engagement_id == subscription_id,
|
||||
EngagementDocument.is_deleted.is_(False),
|
||||
)
|
||||
).scalars().all()
|
||||
for document in rows:
|
||||
if (getattr(document, "final_release_status", None) or "draft").strip().lower() == "released":
|
||||
return True
|
||||
if (getattr(document, "udin_number", None) or "").strip():
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def validate_engagement_can_change_financial_year(
|
||||
db: Session,
|
||||
*,
|
||||
row: ClientServiceSubscription,
|
||||
target_financial_year: str,
|
||||
) -> None:
|
||||
if getattr(row, "is_locked", False):
|
||||
raise ValueError("Locked engagements cannot be moved to another financial year.")
|
||||
if is_row_financial_year_locked(db, row):
|
||||
raise ValueError(f"Source FY {row.financial_year} is locked and cannot be changed.")
|
||||
if is_financial_year_locked(db, tenant_id=row.tenant_id, year_code=target_financial_year):
|
||||
raise ValueError(f"Target FY {target_financial_year} is locked and cannot receive an engagement.")
|
||||
if not getattr(row, "is_active", True):
|
||||
raise ValueError("Inactive engagements cannot be moved to another financial year.")
|
||||
status = (getattr(row, "status", None) or "active").strip().lower()
|
||||
if status in _CLOSED_ENGAGEMENT_STATUSES:
|
||||
raise ValueError("Only open engagements can have their financial year corrected.")
|
||||
|
||||
closure = db.execute(
|
||||
select(EngagementClosureChecklist).where(
|
||||
EngagementClosureChecklist.subscription_id == row.id
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if closure and (closure.closure_status or "").strip().lower() == "approved":
|
||||
raise ValueError("The engagement closure is already approved. Reopen it before correcting the financial year.")
|
||||
if _engagement_has_finalised_documents(db, row.id):
|
||||
raise ValueError("A released final document or UDIN exists. Reopen/correct the finalisation workflow before changing FY.")
|
||||
|
||||
|
||||
def _duplicate_engagement(
|
||||
db: Session,
|
||||
*,
|
||||
row: ClientServiceSubscription,
|
||||
target_financial_year: str,
|
||||
target_period_label: str,
|
||||
) -> ClientServiceSubscription | None:
|
||||
return db.execute(
|
||||
select(ClientServiceSubscription).where(
|
||||
ClientServiceSubscription.tenant_id == row.tenant_id,
|
||||
ClientServiceSubscription.service_catalogue_id == row.service_catalogue_id,
|
||||
ClientServiceSubscription.scope_key == row.scope_key,
|
||||
ClientServiceSubscription.financial_year == target_financial_year,
|
||||
ClientServiceSubscription.period_label == target_period_label,
|
||||
ClientServiceSubscription.id != row.id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def _add_audit_log(
|
||||
db: Session,
|
||||
*,
|
||||
request: Request | None,
|
||||
row: ClientServiceSubscription,
|
||||
actor,
|
||||
old_financial_year: str,
|
||||
new_financial_year: str,
|
||||
old_period_label: str,
|
||||
new_period_label: str,
|
||||
reason: str,
|
||||
task_count: int,
|
||||
document_count: int,
|
||||
) -> None:
|
||||
ip_address = request.client.host if request and request.client else None
|
||||
user_agent = request.headers.get("user-agent") if request else None
|
||||
client_name = getattr(getattr(row, "client", None), "client_name", None) or f"Client {row.client_id}"
|
||||
service_name = getattr(getattr(row, "catalogue", None), "service_name", None) or f"Service {row.service_catalogue_id}"
|
||||
db.add(
|
||||
AuditLog(
|
||||
actor_user_id=getattr(actor, "id", None),
|
||||
actor_email=getattr(actor, "email", None),
|
||||
actor_tenant_id=getattr(actor, "tenant_id", None),
|
||||
actor_branch_id=getattr(actor, "branch_id", None),
|
||||
action="engagement.financial_year_corrected",
|
||||
entity_type="ClientServiceSubscription",
|
||||
entity_id=str(row.id),
|
||||
entity_name=f"{client_name} - {service_name}",
|
||||
status="success",
|
||||
target_tenant_id=row.tenant_id,
|
||||
target_branch_id=row.branch_id,
|
||||
ip_address=ip_address,
|
||||
user_agent=user_agent,
|
||||
details_json=json.dumps(
|
||||
{
|
||||
"old_financial_year": old_financial_year,
|
||||
"new_financial_year": new_financial_year,
|
||||
"old_assessment_year": assessment_year_from_financial_year(old_financial_year),
|
||||
"new_assessment_year": assessment_year_from_financial_year(new_financial_year),
|
||||
"old_period_label": old_period_label,
|
||||
"new_period_label": new_period_label,
|
||||
"reason": reason,
|
||||
"task_instances_updated": task_count,
|
||||
"engagement_documents_updated": document_count,
|
||||
"physical_document_paths_moved": False,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def correct_engagement_financial_year(
|
||||
db: Session,
|
||||
*,
|
||||
row: ClientServiceSubscription,
|
||||
target_financial_year: str,
|
||||
actor,
|
||||
reason: str,
|
||||
request: Request | None = None,
|
||||
) -> EngagementFYCorrectionResult:
|
||||
target_fy = validate_financial_year(target_financial_year)
|
||||
reason_text = (reason or "").strip()
|
||||
if not reason_text:
|
||||
raise ValueError("A correction reason is required for the audit trail.")
|
||||
|
||||
old_fy = validate_financial_year(row.financial_year)
|
||||
if target_fy == old_fy:
|
||||
raise ValueError("The corrected financial year is the same as the existing financial year.")
|
||||
|
||||
validate_engagement_can_change_financial_year(db, row=row, target_financial_year=target_fy)
|
||||
old_period = (row.period_label or "").strip()
|
||||
new_period = _remap_period_label(row, target_fy)
|
||||
|
||||
duplicate = _duplicate_engagement(
|
||||
db,
|
||||
row=row,
|
||||
target_financial_year=target_fy,
|
||||
target_period_label=new_period,
|
||||
)
|
||||
if duplicate:
|
||||
raise ValueError(
|
||||
f"A matching engagement already exists in FY {target_fy}"
|
||||
+ (f" for period {new_period}" if new_period else "")
|
||||
+ "."
|
||||
)
|
||||
|
||||
# Pre-flight the task unique key before mutating anything. A partially
|
||||
# inconsistent legacy subscription must be repaired manually rather than
|
||||
# losing execution/review history.
|
||||
target_task = db.execute(
|
||||
select(ClientServiceTaskInstance.id).where(
|
||||
ClientServiceTaskInstance.subscription_id == row.id,
|
||||
ClientServiceTaskInstance.financial_year == target_fy,
|
||||
).limit(1)
|
||||
).scalar_one_or_none()
|
||||
if target_task is not None:
|
||||
raise ValueError(
|
||||
f"This engagement already has a task instance tagged to FY {target_fy}. "
|
||||
"The mixed-year task data must be reviewed before FY correction."
|
||||
)
|
||||
|
||||
new_ay = assessment_year_from_financial_year(target_fy)
|
||||
tasks = db.execute(
|
||||
select(ClientServiceTaskInstance).where(
|
||||
ClientServiceTaskInstance.subscription_id == row.id
|
||||
)
|
||||
).scalars().all()
|
||||
documents = db.execute(
|
||||
select(EngagementDocument).where(
|
||||
EngagementDocument.engagement_id == row.id,
|
||||
EngagementDocument.is_deleted.is_(False),
|
||||
)
|
||||
).scalars().all()
|
||||
|
||||
row.financial_year = target_fy
|
||||
row.assessment_year = new_ay
|
||||
row.period_label = new_period
|
||||
row.updated_by_user_id = getattr(actor, "id", None)
|
||||
|
||||
for task in tasks:
|
||||
task.financial_year = target_fy
|
||||
task.assessment_year = new_ay
|
||||
task.period_label = new_period
|
||||
task.updated_by_user_id = getattr(actor, "id", None)
|
||||
|
||||
# Document metadata follows the corrected engagement. Existing physical
|
||||
# revision paths are deliberately preserved so no already-uploaded evidence
|
||||
# is orphaned; future uploads use the corrected engagement FY.
|
||||
for document in documents:
|
||||
document.financial_year = target_fy
|
||||
document.assessment_year = new_ay
|
||||
document.updated_by_user_id = getattr(actor, "id", None)
|
||||
|
||||
apply_due_date_rule_to_subscription(db, row, force=True)
|
||||
_add_audit_log(
|
||||
db,
|
||||
request=request,
|
||||
row=row,
|
||||
actor=actor,
|
||||
old_financial_year=old_fy,
|
||||
new_financial_year=target_fy,
|
||||
old_period_label=old_period,
|
||||
new_period_label=new_period,
|
||||
reason=reason_text,
|
||||
task_count=len(tasks),
|
||||
document_count=len(documents),
|
||||
)
|
||||
db.flush()
|
||||
|
||||
return EngagementFYCorrectionResult(
|
||||
subscription_id=row.id,
|
||||
old_financial_year=old_fy,
|
||||
new_financial_year=target_fy,
|
||||
old_period_label=old_period,
|
||||
new_period_label=new_period,
|
||||
task_count=len(tasks),
|
||||
document_count=len(documents),
|
||||
)
|
||||
Reference in New Issue
Block a user