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

Auto-create and repair subscription engagements

Browse Source
This commit is contained in:
A R R R Associates
2026-08-19 15:39:08 +05:30
parent fee4d6d235
commit 2f061d29b4
3 changed files with 396 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/services/subscriptions_ui.py
+325 -2
View File
@@ -224,8 +224,169 @@ def _engagement_is_effective(row: ClientServiceSubscription, effective_from: dat
return comparison_date is None or comparison_date >= effective_from
def _current_period_for_financial_year(financial_year: str, recurrence_type: str | None) -> str:
"""Return the current period inside the selected FY for recurring services.
For the active FY this means the current calendar month / FY quarter. The
caller still passes the value through normalize_period_label, so the same
validation used by the existing engagement creation flow remains in force.
"""
recurrence = (recurrence_type or "").strip().lower().replace("-", "_").replace(" ", "_")
today = date.today()
fy = normalize_financial_year(financial_year)
start_year = int(fy.split("-", 1)[0])
end_year = start_year + 1
if recurrence == "monthly":
if today.year < start_year or (today.year == start_year and today.month < 4):
year, month = start_year, 4
elif today.year > end_year or (today.year == end_year and today.month > 3):
year, month = end_year, 3
else:
year, month = today.year, today.month
return normalize_period_label(
f"{year:04d}-{month:02d}",
financial_year=fy,
recurrence_type=recurrence,
)
if recurrence == "quarterly":
if today.year < start_year or (today.year == start_year and today.month < 4):
quarter = "Q1"
elif today.year > end_year or (today.year == end_year and today.month > 3):
quarter = "Q4"
elif today.month in (4, 5, 6):
quarter = "Q1"
elif today.month in (7, 8, 9):
quarter = "Q2"
elif today.month in (10, 11, 12):
quarter = "Q3"
else:
quarter = "Q4"
return normalize_period_label(
quarter,
financial_year=fy,
recurrence_type=recurrence,
)
return ""
def _ensure_current_engagement_for_plan(
db,
*,
plan: ClientServicePlan,
tenant_id: int,
financial_year: str,
actor_user_id: int,
) -> tuple[ClientServiceSubscription, bool]:
"""Create the current engagement for one plan, reusing existing production logic.
Returns (engagement, created). Existing engagements are never duplicated or
overwritten. No task/review/evidence history is modified.
"""
selected_financial_year = normalize_financial_year(financial_year)
recurrence_type = getattr(plan, "recurrence_type", None) or getattr(plan.catalogue, "recurrence_type", None)
requested_period = _current_period_for_financial_year(selected_financial_year, recurrence_type)
existing = db.execute(
select(ClientServiceSubscription).where(
ClientServiceSubscription.tenant_id == tenant_id,
ClientServiceSubscription.service_catalogue_id == plan.service_catalogue_id,
ClientServiceSubscription.scope_key == plan.scope_key,
ClientServiceSubscription.financial_year == selected_financial_year,
ClientServiceSubscription.period_label == requested_period,
)
).scalar_one_or_none()
if existing:
if existing.service_plan_id is None:
existing.service_plan_id = plan.id
return existing, False
client = db.get(Client, plan.client_id)
if not client or client.tenant_id != tenant_id:
raise ValueError("client")
firm_selection = get_enabled_firm_service(
db,
tenant_id=tenant_id,
service_catalogue_id=plan.service_catalogue_id,
)
if not firm_selection:
raise ValueError("service")
engagement_type = getattr(firm_selection.catalogue, "engagement_type", None) or "non_assurance"
review_required = review_partner_required_for_engagement(
db,
tenant_id=tenant_id,
engagement_type=engagement_type,
)
review_partner_id = plan.default_review_partner_user_id if review_required else None
performing_partner_id = plan.default_performing_partner_user_id or plan.default_partner_user_id
if review_required and review_partner_id is None:
raise ValueError("review_partner_required")
if review_required and review_partner_id in {plan.default_partner_user_id, performing_partner_id}:
raise ValueError("review_partner_independence")
engagement = ClientServiceSubscription(
tenant_id=tenant_id,
branch_id=plan.branch_id or getattr(client, "branch_id", None),
service_plan_id=plan.id,
client_id=client.id,
scope_type=plan.scope_type,
scope_key=plan.scope_key,
business_unit_id=plan.business_unit_id,
client_branch_id=plan.client_branch_id,
registration_id=plan.registration_id,
service_catalogue_id=plan.service_catalogue_id,
firm_service_selection_id=firm_selection.id,
assigned_partner_user_id=plan.default_partner_user_id,
performing_partner_user_id=performing_partner_id,
assigned_manager_user_id=plan.default_manager_user_id,
assigned_staff_user_id=plan.default_staff_user_id,
review_partner_user_id=review_partner_id,
financial_year=selected_financial_year,
period_label=requested_period,
assessment_year=assessment_year_from_financial_year(selected_financial_year),
engagement_type=engagement_type,
start_date=plan.effective_from,
end_date=plan.effective_to,
status="active",
remarks=plan.remarks,
is_active=True,
created_by_user_id=actor_user_id,
updated_by_user_id=actor_user_id,
)
db.add(engagement)
db.flush()
apply_due_date_rule_to_subscription(db, engagement, force=True)
ensure_engagement_quality_workflow(
db,
subscription=engagement,
actor_user_id=actor_user_id,
create_declarations=False,
)
enforce_quality_gate_on_subscription(engagement)
generate_tasks_for_subscription_if_ready(
db,
subscription=engagement,
user_id=actor_user_id,
)
return engagement, True
@router.get("")
def subscription_master_list(request: Request, q: str = "", include_inactive: bool = False):
def subscription_master_list(
request: Request,
q: str = "",
include_inactive: bool = False,
repaired: int = 0,
skipped: int = 0,
repair_error: str = "",
):
db = CommonSessionLocal()
try:
user = get_current_user(request, db=db)
@@ -295,6 +456,10 @@ def subscription_master_list(request: Request, q: str = "", include_inactive: bo
q=q,
include_inactive=include_inactive,
can_edit_subscription="clients.edit" in set(get_user_permissions(db, user.id)),
repaired=repaired,
skipped=skipped,
repair_error=repair_error,
financial_year=_active_financial_year(request),
),
)
finally:
@@ -302,6 +467,85 @@ def subscription_master_list(request: Request, q: str = "", include_inactive: bo
@router.post("/generate-missing")
def subscription_generate_missing_submit(
request: Request,
financial_year: str = Form(""),
csrf_token: str = Form(...),
):
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
created_count = 0
skipped_count = 0
try:
user = get_current_user(request, db=db)
if not user:
return RedirectResponse("/login", 303)
try:
require_permission(db, user, "clients.edit")
except Exception:
return _denied()
tenant_id = _tenant_id(request, user)
branch_id = _branch_id(request, user)
selected_financial_year = normalize_financial_year(financial_year or _active_financial_year(request))
locked = redirect_if_financial_year_locked(
db,
tenant_id=tenant_id,
year_code=selected_financial_year,
redirect_url="/services/subscriptions?repair_error=financial_year",
)
if locked:
return locked
stmt = (
select(ClientServicePlan)
.options(
selectinload(ClientServicePlan.client),
selectinload(ClientServicePlan.catalogue),
)
.where(
ClientServicePlan.tenant_id == tenant_id,
ClientServicePlan.is_active.is_(True),
ClientServicePlan.status == "active",
)
.order_by(ClientServicePlan.id)
)
if branch_id:
stmt = stmt.where(ClientServicePlan.branch_id == branch_id)
plans = db.execute(stmt).scalars().all()
for plan in plans:
try:
_engagement, created = _ensure_current_engagement_for_plan(
db,
plan=plan,
tenant_id=tenant_id,
financial_year=selected_financial_year,
actor_user_id=user.id,
)
if created:
created_count += 1
else:
skipped_count += 1
except ValueError:
# A plan with an invalid/missing service setup or unmet assurance
# review-partner gate is left untouched; the rest continue safely.
skipped_count += 1
db.commit()
return RedirectResponse(
f"/services/subscriptions?repaired={created_count}&skipped={skipped_count}",
status_code=303,
)
except Exception:
db.rollback()
raise
finally:
db.close()
@router.get("/bulk")
def subscription_bulk_page(
request: Request,
@@ -1192,8 +1436,83 @@ def subscription_master_edit_submit(
db.close()
@router.post("/{plan_id}/generate-current")
def subscription_generate_current_submit(
request: Request,
plan_id: int,
financial_year: str = Form(""),
csrf_token: str = Form(...),
):
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
try:
user = get_current_user(request, db=db)
if not user:
return RedirectResponse("/login", 303)
try:
require_permission(db, user, "clients.edit")
except Exception:
return _denied()
tenant_id = _tenant_id(request, user)
branch_id = _branch_id(request, user)
plan = _load_plan(db, plan_id=plan_id, tenant_id=tenant_id, branch_id=branch_id)
if not plan:
return RedirectResponse("/services/subscriptions", 303)
if not plan.is_active or plan.status != "active":
return RedirectResponse(
f"/services/subscriptions/{plan.id}?generation_error=inactive",
status_code=303,
)
selected_financial_year = normalize_financial_year(financial_year or _active_financial_year(request))
locked = redirect_if_financial_year_locked(
db,
tenant_id=tenant_id,
year_code=selected_financial_year,
redirect_url=f"/services/subscriptions/{plan.id}?generation_error=financial_year",
)
if locked:
return locked
try:
_engagement, created = _ensure_current_engagement_for_plan(
db,
plan=plan,
tenant_id=tenant_id,
financial_year=selected_financial_year,
actor_user_id=user.id,
)
except ValueError as exc:
db.rollback()
code = str(exc) or "validation"
return RedirectResponse(
f"/services/subscriptions/{plan.id}?generation_error={code}",
status_code=303,
)
db.commit()
flag = "generated=1" if created else "existing=1"
return RedirectResponse(
f"/services/subscriptions/{plan.id}?{flag}",
status_code=303,
)
except Exception:
db.rollback()
raise
finally:
db.close()
@router.get("/{plan_id}")
def subscription_master_detail(request: Request, plan_id: int):
def subscription_master_detail(
request: Request,
plan_id: int,
generated: int = 0,
existing: int = 0,
generation_error: str = "",
):
db = CommonSessionLocal()
try:
user = get_current_user(request, db=db)
@@ -1228,6 +1547,10 @@ def subscription_master_detail(request: Request, plan_id: int):
plan=plan,
engagements=engagements,
can_edit_subscription="clients.edit" in permissions,
generated=generated,
existing=existing,
generation_error=generation_error,
financial_year=_active_financial_year(request),
**_registration_scope_context(
db,
tenant_id=tenant_id,