A R R R Associates
1177 lines
49 KiB
Python
1177 lines
49 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
|
|
from app.core.security.session_auth import get_current_user
|
|
from app.core.templating import templates
|
|
from app.modules.services.models import ClientServiceSubscription, ClientServiceTaskInstance
|
|
from app.modules.services.due_dates import apply_due_date_rule_to_subscription
|
|
from app.modules.services.execution import (
|
|
aqmm_task_summary_for_subscription,
|
|
update_engagement_closure_from_sources,
|
|
save_engagement_closure_confirmations,
|
|
approve_engagement_closure,
|
|
reopen_engagement_closure,
|
|
)
|
|
from app.modules.clients.models import Client
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.services.client_services import (
|
|
SUBSCRIPTION_STATUSES,
|
|
assessment_year_from_financial_year,
|
|
current_financial_year,
|
|
get_enabled_firm_service,
|
|
get_existing_subscription,
|
|
get_subscription,
|
|
normalize_financial_year,
|
|
normalize_period_label,
|
|
period_choices_for_service,
|
|
recurrence_requires_period,
|
|
list_assignable_users,
|
|
list_clients_for_assignment,
|
|
list_enabled_services_for_assignment,
|
|
list_subscription_payload,
|
|
list_review_partners,
|
|
parse_date,
|
|
review_partner_required_for_engagement,
|
|
ensure_engagement_quality_workflow,
|
|
enforce_quality_gate_on_subscription,
|
|
request_engagement_quality_declarations,
|
|
list_engagement_quality_declarations,
|
|
respond_engagement_quality_declaration,
|
|
verify_engagement_kyc_from_permanent_documents,
|
|
get_latest_engagement_kyc_verification,
|
|
get_current_engagement_letter,
|
|
mark_engagement_letter_completed,
|
|
approve_engagement_quality_workflow,
|
|
)
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.core.rbac.permission_guard import require_permission
|
|
from app.modules.core.tenancy.year_control import redirect_if_financial_year_locked, is_row_financial_year_locked
|
|
|
|
router = APIRouter(prefix="/services/engagements", tags=["services-engagements-ui"])
|
|
|
|
|
|
def _base_ctx(request: Request, user, db, **ctx):
|
|
base = {
|
|
"request": request,
|
|
"current_user": user,
|
|
"current_user_roles": get_user_roles(db, user.id),
|
|
"current_user_permissions": get_user_permissions(db, user.id),
|
|
"csrf_token": get_or_create_csrf_token(request),
|
|
"subscription_statuses": SUBSCRIPTION_STATUSES,
|
|
}
|
|
base.update(ctx)
|
|
return base
|
|
|
|
|
|
def _render(request: Request, template: str, db, user, **ctx):
|
|
return templates.TemplateResponse(template, _base_ctx(request, user, db, **ctx))
|
|
|
|
|
|
def _redirect_denied():
|
|
from app.core.http_responses import ui_access_denied
|
|
return ui_access_denied()
|
|
|
|
|
|
def _has_perm(db, user, code: str) -> bool:
|
|
try:
|
|
require_permission(db, user, code)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _active_tenant_id(request: Request, user) -> int:
|
|
return int(
|
|
request.session.get("active_tenant_id")
|
|
or request.session.get("selected_tenant_id")
|
|
or request.session.get("tenant_id")
|
|
or user.tenant_id
|
|
)
|
|
|
|
|
|
def _active_branch_id(request: Request, user, db) -> int | None:
|
|
value = request.session.get("active_branch_id")
|
|
if value in (None, "", 0, "0"):
|
|
if _has_perm(db, user, "clients.cross_branch"):
|
|
return None
|
|
return int(getattr(user, "branch_id", 0) or 0) or None
|
|
return int(value)
|
|
|
|
|
|
def _active_financial_year(request: Request) -> str:
|
|
return normalize_financial_year(
|
|
request.session.get("active_financial_year")
|
|
or getattr(request.state, "year_code", None)
|
|
)
|
|
|
|
|
|
def _locked_partner_id(db, user) -> int | None:
|
|
return int(user.id) if _has_perm(db, user, "clients.view.own_only") else None
|
|
|
|
|
|
def _can_manage_client_services(db, user) -> bool:
|
|
return _has_perm(db, user, "clients.edit")
|
|
|
|
|
|
def _can_lock_engagements(db, user) -> bool:
|
|
roles = set(get_user_roles(db, user.id))
|
|
return bool(roles.intersection({"Firm Admin", "Partner"}))
|
|
|
|
|
|
def _user_can_lock_subscription(db, user, row: ClientServiceSubscription) -> bool:
|
|
roles = set(get_user_roles(db, user.id))
|
|
if "Firm Admin" in roles:
|
|
return True
|
|
if "Partner" in roles:
|
|
client = getattr(row, "client", None)
|
|
return (
|
|
getattr(row, "assigned_partner_user_id", None) == user.id
|
|
or getattr(client, "partner_id", None) == user.id
|
|
)
|
|
return False
|
|
|
|
|
|
def _lock_subscription_row(row: ClientServiceSubscription, user) -> bool:
|
|
if getattr(row, "is_locked", False):
|
|
return False
|
|
from datetime import datetime, timezone
|
|
|
|
row.is_locked = True
|
|
row.status = "completed" if row.status == "active" else row.status
|
|
row.locked_at_utc = datetime.now(timezone.utc)
|
|
row.locked_by_user_id = user.id
|
|
row.updated_by_user_id = user.id
|
|
return True
|
|
|
|
|
|
@router.get("")
|
|
def subscription_list(
|
|
request: Request,
|
|
q: str = "",
|
|
financial_year: str = "",
|
|
include_inactive: bool = True,
|
|
locked: int = 0,
|
|
skipped: int = 0,
|
|
bulk_created: int = 0,
|
|
bulk_existing: int = 0,
|
|
bulk_skipped: int = 0,
|
|
):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.view")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
branch_id = _active_branch_id(request, user, db)
|
|
selected_financial_year = normalize_financial_year(financial_year or _active_financial_year(request))
|
|
rows = list_subscription_payload(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
financial_year=selected_financial_year,
|
|
q=q,
|
|
include_inactive=include_inactive,
|
|
)
|
|
return _render(
|
|
request,
|
|
"modules/services/templates/services/engagements/list.html",
|
|
db,
|
|
user,
|
|
title="Engagement Subscriptions",
|
|
rows=rows,
|
|
q=q,
|
|
financial_year=selected_financial_year,
|
|
include_inactive=include_inactive,
|
|
locked_count=locked,
|
|
skipped_count=skipped,
|
|
bulk_created_count=bulk_created,
|
|
bulk_existing_count=bulk_existing,
|
|
bulk_skipped_count=bulk_skipped,
|
|
can_manage=_can_manage_client_services(db, user),
|
|
can_lock_engagements=_can_lock_engagements(db, user),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/new")
|
|
def subscription_create_page(request: Request, client_id: int | None = None):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
branch_id = _active_branch_id(request, user, db)
|
|
clients = list_clients_for_assignment(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
partner_id=_locked_partner_id(db, user),
|
|
)
|
|
enabled_services = list_enabled_services_for_assignment(db, tenant_id=tenant_id)
|
|
assignable_users = list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
performing_partners = list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id, role_names=("Partner",))
|
|
review_partners = list_review_partners(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
|
|
return _render(
|
|
request,
|
|
"modules/services/templates/services/engagements/form.html",
|
|
db,
|
|
user,
|
|
title="Assign Service to Client",
|
|
mode="create",
|
|
subscription=None,
|
|
clients=clients,
|
|
enabled_services=enabled_services,
|
|
assignable_users=assignable_users,
|
|
performing_partners=performing_partners,
|
|
review_partners=review_partners,
|
|
selected_client_id=client_id,
|
|
financial_year=_active_financial_year(request),
|
|
period_choices=[],
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/new")
|
|
def subscription_create_submit(
|
|
request: Request,
|
|
client_id: int = Form(...),
|
|
service_catalogue_id: int = Form(...),
|
|
assigned_partner_user_id: str = Form(""),
|
|
assigned_manager_user_id: str = Form(""),
|
|
assigned_staff_user_id: str = Form(""),
|
|
performing_partner_user_id: str = Form(""),
|
|
review_partner_user_id: str = Form(""),
|
|
financial_year: str = Form(""),
|
|
period_label: str = Form(""),
|
|
generate_all_periods: str | None = Form(None),
|
|
start_date: str = Form(""),
|
|
end_date: str = Form(""),
|
|
expiry_date: str = Form(""),
|
|
status: str = Form("active"),
|
|
remarks: str = Form(""),
|
|
is_active: str | None = Form(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
selected_financial_year = normalize_financial_year(financial_year or _active_financial_year(request))
|
|
locked_response = redirect_if_financial_year_locked(db, tenant_id=tenant_id, year_code=selected_financial_year, redirect_url=f"/services/engagements?financial_year={selected_financial_year}")
|
|
if locked_response:
|
|
return locked_response
|
|
firm_selection = get_enabled_firm_service(db, tenant_id=tenant_id, service_catalogue_id=service_catalogue_id)
|
|
if not firm_selection:
|
|
return RedirectResponse(url="/services/engagements/new", status_code=303)
|
|
|
|
recurrence_type = getattr(firm_selection.catalogue, "recurrence_type", None)
|
|
if recurrence_requires_period(recurrence_type):
|
|
available_periods = period_choices_for_service(selected_financial_year, recurrence_type)
|
|
if generate_all_periods is not None:
|
|
requested_periods = [code for code, _label in available_periods]
|
|
else:
|
|
try:
|
|
requested_periods = [normalize_period_label(period_label, financial_year=selected_financial_year, recurrence_type=recurrence_type)]
|
|
except ValueError:
|
|
return RedirectResponse(url="/services/engagements/new?error=period", status_code=303)
|
|
else:
|
|
requested_periods = [""]
|
|
|
|
client = db.get(Client, client_id)
|
|
if not client or client.tenant_id != tenant_id:
|
|
return RedirectResponse(url="/services/engagements/new", status_code=303)
|
|
|
|
created_rows = []
|
|
existing_rows = []
|
|
for requested_period in requested_periods:
|
|
existing = get_existing_subscription(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
client_id=client_id,
|
|
service_catalogue_id=service_catalogue_id,
|
|
financial_year=selected_financial_year,
|
|
period_label=requested_period,
|
|
)
|
|
if existing:
|
|
existing_rows.append(existing)
|
|
continue
|
|
row = ClientServiceSubscription(
|
|
tenant_id=tenant_id,
|
|
client_id=client_id,
|
|
service_catalogue_id=service_catalogue_id,
|
|
financial_year=selected_financial_year,
|
|
period_label=requested_period,
|
|
assessment_year=assessment_year_from_financial_year(selected_financial_year),
|
|
engagement_type=getattr(firm_selection.catalogue, "engagement_type", "non_assurance") or "non_assurance",
|
|
created_by_user_id=user.id,
|
|
)
|
|
db.add(row)
|
|
created_rows.append(row)
|
|
|
|
if not created_rows:
|
|
row = existing_rows[0]
|
|
else:
|
|
row = created_rows[0]
|
|
rows_to_update = created_rows or [row]
|
|
for row in rows_to_update:
|
|
if getattr(row, "is_locked", False) or is_row_financial_year_locked(db, row):
|
|
return RedirectResponse(url=f"/services/engagements/{row.id}?year_locked=1", status_code=303)
|
|
|
|
row.financial_year = selected_financial_year
|
|
row.assessment_year = assessment_year_from_financial_year(selected_financial_year)
|
|
if not getattr(row, "engagement_type", None):
|
|
row.engagement_type = getattr(firm_selection.catalogue, "engagement_type", "non_assurance") or "non_assurance"
|
|
row.branch_id = firm_selection.default_branch_id or getattr(user, "branch_id", None)
|
|
row.firm_service_selection_id = firm_selection.id
|
|
row.assigned_partner_user_id = int(assigned_partner_user_id) if assigned_partner_user_id.strip() else None
|
|
requested_performing_partner_id = int(performing_partner_user_id) if performing_partner_user_id.strip() else None
|
|
row.performing_partner_user_id = requested_performing_partner_id or getattr(client, "default_performing_partner_user_id", None) or row.assigned_partner_user_id
|
|
row.assigned_manager_user_id = int(assigned_manager_user_id) if assigned_manager_user_id.strip() else None
|
|
row.assigned_staff_user_id = int(assigned_staff_user_id) if assigned_staff_user_id.strip() else None
|
|
if review_partner_required_for_engagement(db, tenant_id=tenant_id, engagement_type=row.engagement_type):
|
|
row.review_partner_user_id = int(review_partner_user_id) if review_partner_user_id.strip() else getattr(client, "default_review_partner_user_id", None)
|
|
else:
|
|
row.review_partner_user_id = None
|
|
row.start_date = parse_date(start_date)
|
|
row.end_date = parse_date(end_date)
|
|
row.expiry_date = parse_date(expiry_date)
|
|
row.status = status or "active"
|
|
row.remarks = remarks.strip() or None
|
|
row.is_active = is_active is not None
|
|
row.updated_by_user_id = user.id
|
|
apply_due_date_rule_to_subscription(db, row, force=True)
|
|
ensure_engagement_quality_workflow(db, subscription=row, actor_user_id=user.id, create_declarations=False)
|
|
enforce_quality_gate_on_subscription(row)
|
|
|
|
db.commit()
|
|
db.refresh(row)
|
|
if len(created_rows) > 1:
|
|
return RedirectResponse(url=f"/services/engagements?financial_year={selected_financial_year}&bulk_created={len(created_rows)}", status_code=303)
|
|
return RedirectResponse(url=f"/services/engagements/{row.id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
|
|
@router.get("/bulk-new")
|
|
def subscription_bulk_create_page(request: Request, error: str = ""):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
branch_id = _active_branch_id(request, user, db)
|
|
clients = list_clients_for_assignment(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
partner_id=_locked_partner_id(db, user),
|
|
)
|
|
enabled_services = list_enabled_services_for_assignment(db, tenant_id=tenant_id)
|
|
partners = list_assignable_users(db, tenant_id=tenant_id, role_names=("Partner",))
|
|
performing_partners = partners
|
|
managers = list_assignable_users(db, tenant_id=tenant_id, role_names=("Branch Manager",))
|
|
staff_users = list_assignable_users(db, tenant_id=tenant_id, role_names=("Staff",))
|
|
review_partners = list_review_partners(db, tenant_id=tenant_id)
|
|
|
|
return _render(
|
|
request,
|
|
"modules/services/templates/services/engagements/bulk_form.html",
|
|
db,
|
|
user,
|
|
title="Bulk Engagement Setup",
|
|
clients=clients,
|
|
enabled_services=enabled_services,
|
|
partners=partners,
|
|
performing_partners=performing_partners,
|
|
client_partner_names={row.id: (row.full_name or row.email) for row in partners},
|
|
managers=managers,
|
|
staff_users=staff_users,
|
|
review_partners=review_partners,
|
|
financial_year=_active_financial_year(request),
|
|
error_message={
|
|
"service": "Select a valid enabled firm service.",
|
|
"partner": "Select a valid active Partner.",
|
|
"performing_partner": "The selected Performing Partner is not available for this firm.",
|
|
"manager": "The selected Manager is not available for this firm.",
|
|
"staff": "The selected Staff member is not available for this firm.",
|
|
"review_partner": "The selected Review Partner is not available for this firm.",
|
|
"review_partner_required": "Review Partner is mandatory for an assurance engagement.",
|
|
"clients": "Select at least one permitted client.",
|
|
"period": "Select a valid month or quarter for the recurring service.",
|
|
}.get(error),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/bulk-new")
|
|
def subscription_bulk_create_submit(
|
|
request: Request,
|
|
client_ids: list[int] = Form([]),
|
|
service_catalogue_id: int = Form(...),
|
|
assigned_partner_user_id: int = Form(...),
|
|
assigned_manager_user_id: str = Form(""),
|
|
assigned_staff_user_id: str = Form(""),
|
|
performing_partner_user_id: str = Form(""),
|
|
review_partner_user_id: str = Form(""),
|
|
financial_year: str = Form(""),
|
|
period_label: str = Form(""),
|
|
generate_all_periods: str | None = Form(None),
|
|
remarks: str = Form(""),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
created_count = 0
|
|
existing_count = 0
|
|
skipped_count = 0
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
selected_financial_year = normalize_financial_year(financial_year or _active_financial_year(request))
|
|
locked_response = redirect_if_financial_year_locked(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
year_code=selected_financial_year,
|
|
redirect_url=f"/services/engagements?financial_year={selected_financial_year}",
|
|
)
|
|
if locked_response:
|
|
return locked_response
|
|
|
|
firm_selection = get_enabled_firm_service(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
service_catalogue_id=service_catalogue_id,
|
|
)
|
|
if not firm_selection:
|
|
return RedirectResponse(url="/services/engagements/bulk-new?error=service", status_code=303)
|
|
|
|
recurrence_type = getattr(firm_selection.catalogue, "recurrence_type", None)
|
|
if recurrence_requires_period(recurrence_type):
|
|
available_periods = period_choices_for_service(selected_financial_year, recurrence_type)
|
|
if generate_all_periods is not None:
|
|
requested_periods = [code for code, _label in available_periods]
|
|
else:
|
|
try:
|
|
requested_periods = [normalize_period_label(period_label, financial_year=selected_financial_year, recurrence_type=recurrence_type)]
|
|
except ValueError:
|
|
return RedirectResponse(url="/services/engagements/bulk-new?error=period", status_code=303)
|
|
else:
|
|
requested_periods = [""]
|
|
|
|
partners = list_assignable_users(db, tenant_id=tenant_id, role_names=("Partner",))
|
|
managers = list_assignable_users(db, tenant_id=tenant_id, role_names=("Branch Manager",))
|
|
staff_users = list_assignable_users(db, tenant_id=tenant_id, role_names=("Staff",))
|
|
review_partners = list_review_partners(db, tenant_id=tenant_id)
|
|
|
|
partner_ids = {row.id for row in partners}
|
|
performing_partner_ids = partner_ids
|
|
manager_ids = {row.id for row in managers}
|
|
staff_ids = {row.id for row in staff_users}
|
|
review_partner_ids = {row.id for row in review_partners}
|
|
|
|
if assigned_partner_user_id not in partner_ids:
|
|
return RedirectResponse(url="/services/engagements/bulk-new?error=partner", status_code=303)
|
|
|
|
performing_partner_id = int(performing_partner_user_id) if performing_partner_user_id.strip() else assigned_partner_user_id
|
|
manager_id = int(assigned_manager_user_id) if assigned_manager_user_id.strip() else None
|
|
staff_id = int(assigned_staff_user_id) if assigned_staff_user_id.strip() else None
|
|
review_partner_id = int(review_partner_user_id) if review_partner_user_id.strip() else None
|
|
if performing_partner_id not in performing_partner_ids:
|
|
return RedirectResponse(url="/services/engagements/bulk-new?error=performing_partner", status_code=303)
|
|
if manager_id is not None and manager_id not in manager_ids:
|
|
return RedirectResponse(url="/services/engagements/bulk-new?error=manager", status_code=303)
|
|
if staff_id is not None and staff_id not in staff_ids:
|
|
return RedirectResponse(url="/services/engagements/bulk-new?error=staff", status_code=303)
|
|
if review_partner_id is not None and review_partner_id not in review_partner_ids:
|
|
return RedirectResponse(url="/services/engagements/bulk-new?error=review_partner", status_code=303)
|
|
|
|
partner = db.get(User, assigned_partner_user_id)
|
|
if not partner or partner.tenant_id != tenant_id or not partner.is_active:
|
|
return RedirectResponse(url="/services/engagements/bulk-new?error=partner", status_code=303)
|
|
engagement_branch_id = partner.branch_id
|
|
|
|
allowed_clients = list_clients_for_assignment(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
branch_id=_active_branch_id(request, user, db),
|
|
partner_id=_locked_partner_id(db, user),
|
|
)
|
|
allowed_client_ids = {row.id for row in allowed_clients}
|
|
selected_client_ids = []
|
|
seen_client_ids = set()
|
|
for value in client_ids:
|
|
client_id = int(value)
|
|
if client_id in seen_client_ids:
|
|
continue
|
|
seen_client_ids.add(client_id)
|
|
if client_id in allowed_client_ids:
|
|
selected_client_ids.append(client_id)
|
|
else:
|
|
skipped_count += 1
|
|
|
|
if not selected_client_ids:
|
|
return RedirectResponse(url="/services/engagements/bulk-new?error=clients", status_code=303)
|
|
|
|
engagement_type = getattr(firm_selection.catalogue, "engagement_type", "non_assurance") or "non_assurance"
|
|
normalized_engagement_type = str(engagement_type).strip().lower().replace("-", "_").replace(" ", "_")
|
|
assurance_review_partner_required = normalized_engagement_type == "assurance"
|
|
workflow_review_partner_required = review_partner_required_for_engagement(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
engagement_type=engagement_type,
|
|
)
|
|
requires_review_partner = assurance_review_partner_required or workflow_review_partner_required
|
|
if assurance_review_partner_required and review_partner_id is None:
|
|
return RedirectResponse(
|
|
url="/services/engagements/bulk-new?error=review_partner_required",
|
|
status_code=303,
|
|
)
|
|
|
|
for client_id in selected_client_ids:
|
|
client = db.get(Client, client_id)
|
|
if not client or client.tenant_id != tenant_id:
|
|
skipped_count += len(requested_periods)
|
|
continue
|
|
|
|
for requested_period in requested_periods:
|
|
existing = get_existing_subscription(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
client_id=client_id,
|
|
service_catalogue_id=service_catalogue_id,
|
|
financial_year=selected_financial_year,
|
|
period_label=requested_period,
|
|
)
|
|
if existing:
|
|
existing_count += 1
|
|
continue
|
|
|
|
row = ClientServiceSubscription(
|
|
tenant_id=tenant_id,
|
|
branch_id=engagement_branch_id,
|
|
client_id=client_id,
|
|
service_catalogue_id=service_catalogue_id,
|
|
firm_service_selection_id=firm_selection.id,
|
|
assigned_partner_user_id=assigned_partner_user_id,
|
|
performing_partner_user_id=performing_partner_id or getattr(client, "default_performing_partner_user_id", None) or assigned_partner_user_id,
|
|
assigned_manager_user_id=manager_id,
|
|
assigned_staff_user_id=staff_id,
|
|
review_partner_user_id=(
|
|
review_partner_id
|
|
if review_partner_id is not None
|
|
else (
|
|
getattr(client, "default_review_partner_user_id", None)
|
|
if requires_review_partner
|
|
else None
|
|
)
|
|
),
|
|
financial_year=selected_financial_year,
|
|
period_label=requested_period,
|
|
assessment_year=assessment_year_from_financial_year(selected_financial_year),
|
|
engagement_type=engagement_type,
|
|
status="active",
|
|
remarks=remarks.strip() or None,
|
|
is_active=True,
|
|
created_by_user_id=user.id,
|
|
updated_by_user_id=user.id,
|
|
)
|
|
db.add(row)
|
|
db.flush()
|
|
apply_due_date_rule_to_subscription(db, row, force=True)
|
|
ensure_engagement_quality_workflow(db, subscription=row, actor_user_id=user.id, create_declarations=False)
|
|
enforce_quality_gate_on_subscription(row)
|
|
created_count += 1
|
|
|
|
db.commit()
|
|
return RedirectResponse(
|
|
url=(
|
|
f"/services/engagements?financial_year={selected_financial_year}"
|
|
f"&bulk_created={created_count}&bulk_existing={existing_count}&bulk_skipped={skipped_count}"
|
|
),
|
|
status_code=303,
|
|
)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/bulk-lock")
|
|
def subscription_bulk_lock(
|
|
request: Request,
|
|
subscription_ids: list[int] = Form([]),
|
|
financial_year: str = Form(""),
|
|
q: str = Form(""),
|
|
include_inactive: str | None = Form(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
locked_count = 0
|
|
skipped_count = 0
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_lock_engagements(db, user):
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
selected_ids = [int(value) for value in subscription_ids if value]
|
|
for subscription_id in selected_ids:
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if not row or not _user_can_lock_subscription(db, user, row):
|
|
skipped_count += 1
|
|
continue
|
|
if _lock_subscription_row(row, user):
|
|
locked_count += 1
|
|
else:
|
|
skipped_count += 1
|
|
|
|
if locked_count:
|
|
db.commit()
|
|
|
|
params = []
|
|
fy = normalize_financial_year(financial_year or _active_financial_year(request))
|
|
if fy:
|
|
params.append(f"financial_year={fy}")
|
|
if q.strip():
|
|
from urllib.parse import quote_plus
|
|
params.append(f"q={quote_plus(q.strip())}")
|
|
if include_inactive:
|
|
params.append("include_inactive=true")
|
|
params.append(f"locked={locked_count}")
|
|
params.append(f"skipped={skipped_count}")
|
|
suffix = "?" + "&".join(params) if params else ""
|
|
return RedirectResponse(url=f"/services/engagements{suffix}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/{subscription_id}")
|
|
def subscription_detail(request: Request, subscription_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.view")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if not row:
|
|
return RedirectResponse(url="/services/engagements", status_code=303)
|
|
active_fy = _active_financial_year(request)
|
|
if row.financial_year != active_fy:
|
|
return RedirectResponse(url=f"/services/engagements?financial_year={active_fy}", status_code=303)
|
|
|
|
ensure_engagement_quality_workflow(db, subscription=row, actor_user_id=user.id, create_declarations=False)
|
|
db.flush()
|
|
declarations = list_engagement_quality_declarations(db, subscription_id=row.id)
|
|
my_pending_declarations = [d for d in declarations if d.requested_user_id == user.id and d.status == "pending"]
|
|
kyc_verification = get_latest_engagement_kyc_verification(db, subscription_id=row.id)
|
|
engagement_letter = get_current_engagement_letter(db, subscription_id=row.id)
|
|
aqmm_task_summary = aqmm_task_summary_for_subscription(db, subscription_id=row.id)
|
|
closure_checklist, closure_summary = update_engagement_closure_from_sources(db, subscription=row, actor_user_id=user.id)
|
|
|
|
tasks = db.execute(
|
|
select(ClientServiceTaskInstance)
|
|
.where(ClientServiceTaskInstance.subscription_id == row.id)
|
|
.order_by(ClientServiceTaskInstance.sequence_no.asc(), ClientServiceTaskInstance.id.asc())
|
|
).scalars().all()
|
|
|
|
return _render(
|
|
request,
|
|
"modules/services/templates/services/engagements/detail.html",
|
|
db,
|
|
user,
|
|
title="Engagement Subscription",
|
|
row=row,
|
|
tasks=tasks,
|
|
declarations=declarations,
|
|
my_pending_declarations=my_pending_declarations,
|
|
kyc_verification=kyc_verification,
|
|
engagement_letter=engagement_letter,
|
|
aqmm_task_summary=aqmm_task_summary,
|
|
closure_checklist=closure_checklist,
|
|
closure_summary=closure_summary,
|
|
can_manage=_can_manage_client_services(db, user),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/aqmm/initiate")
|
|
def subscription_aqmm_initiate(request: Request, subscription_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row:
|
|
ensure_engagement_quality_workflow(db, subscription=row, actor_user_id=user.id, create_declarations=True)
|
|
enforce_quality_gate_on_subscription(row)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/aqmm/request-declarations")
|
|
def subscription_aqmm_request_declarations(request: Request, subscription_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row:
|
|
request_engagement_quality_declarations(db, subscription=row, actor_user_id=user.id)
|
|
enforce_quality_gate_on_subscription(row)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/aqmm/declarations/{declaration_id}/respond")
|
|
def subscription_aqmm_declaration_respond(
|
|
request: Request,
|
|
subscription_id: int,
|
|
declaration_id: int,
|
|
status: str = Form(...),
|
|
notes: 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(url="/login", status_code=303)
|
|
respond_engagement_quality_declaration(
|
|
db,
|
|
declaration_id=declaration_id,
|
|
current_user_id=user.id,
|
|
status=status,
|
|
notes=notes,
|
|
request=request,
|
|
)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/aqmm/kyc/verify")
|
|
def subscription_aqmm_kyc_verify(request: Request, subscription_id: int, notes: 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(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row:
|
|
verify_engagement_kyc_from_permanent_documents(db, subscription=row, actor_user_id=user.id, notes=notes)
|
|
enforce_quality_gate_on_subscription(row)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/aqmm/engagement-letter/complete")
|
|
def subscription_aqmm_engagement_letter_complete(
|
|
request: Request,
|
|
subscription_id: int,
|
|
acceptance_mode: str = Form(...),
|
|
notes: 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(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row:
|
|
mark_engagement_letter_completed(db, subscription=row, actor_user_id=user.id, acceptance_mode=acceptance_mode, notes=notes, request=request)
|
|
enforce_quality_gate_on_subscription(row)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/aqmm/approve")
|
|
def subscription_aqmm_approve(request: Request, subscription_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row:
|
|
try:
|
|
approve_engagement_quality_workflow(db, subscription=row, actor_user_id=user.id)
|
|
db.commit()
|
|
except ValueError:
|
|
db.rollback()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/closure/sync")
|
|
def subscription_closure_sync(request: Request, subscription_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.view")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row:
|
|
update_engagement_closure_from_sources(db, subscription=row, actor_user_id=user.id)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/closure/update")
|
|
def subscription_closure_update(
|
|
request: Request,
|
|
subscription_id: int,
|
|
deliverables_sent_to_client: str | None = Form(None),
|
|
billing_reviewed: str | None = Form(None),
|
|
open_points_closed: str | None = Form(None),
|
|
client_communication_completed: str | None = Form(None),
|
|
closure_note: 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(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row:
|
|
try:
|
|
save_engagement_closure_confirmations(
|
|
db,
|
|
subscription=row,
|
|
deliverables_sent_to_client=deliverables_sent_to_client is not None,
|
|
billing_reviewed=billing_reviewed is not None,
|
|
open_points_closed=open_points_closed is not None,
|
|
client_communication_completed=client_communication_completed is not None,
|
|
closure_note=closure_note,
|
|
actor_user_id=user.id,
|
|
)
|
|
db.commit()
|
|
except ValueError:
|
|
db.rollback()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/closure/approve")
|
|
def subscription_closure_approve(request: Request, subscription_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_lock_engagements(db, user):
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row:
|
|
try:
|
|
approve_engagement_closure(db, subscription=row, actor_user_id=user.id)
|
|
db.commit()
|
|
except ValueError:
|
|
db.rollback()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/closure/reopen")
|
|
def subscription_closure_reopen(
|
|
request: Request,
|
|
subscription_id: int,
|
|
reason: 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(url="/login", status_code=303)
|
|
if not _can_lock_engagements(db, user):
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row:
|
|
try:
|
|
reopen_engagement_closure(db, subscription=row, reason=reason, actor_user_id=user.id)
|
|
db.commit()
|
|
except ValueError:
|
|
db.rollback()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/{subscription_id}/edit")
|
|
def subscription_edit_page(request: Request, subscription_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
branch_id = _active_branch_id(request, user, db)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if not row:
|
|
return RedirectResponse(url="/services/engagements", status_code=303)
|
|
active_fy = _active_financial_year(request)
|
|
if row.financial_year != active_fy:
|
|
return RedirectResponse(url=f"/services/engagements?financial_year={active_fy}", status_code=303)
|
|
if getattr(row, "is_locked", False) or is_row_financial_year_locked(db, row):
|
|
return RedirectResponse(url=f"/services/engagements/{row.id}?year_locked=1", status_code=303)
|
|
|
|
clients = list_clients_for_assignment(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
partner_id=_locked_partner_id(db, user),
|
|
)
|
|
enabled_services = list_enabled_services_for_assignment(db, tenant_id=tenant_id)
|
|
assignable_users = list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
performing_partners = list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id, role_names=("Partner",))
|
|
review_partners = list_review_partners(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
return _render(
|
|
request,
|
|
"modules/services/templates/services/engagements/form.html",
|
|
db,
|
|
user,
|
|
title="Edit Engagement Subscription",
|
|
mode="edit",
|
|
subscription=row,
|
|
clients=clients,
|
|
enabled_services=enabled_services,
|
|
assignable_users=assignable_users,
|
|
performing_partners=performing_partners,
|
|
review_partners=review_partners,
|
|
selected_client_id=row.client_id,
|
|
financial_year=row.financial_year,
|
|
period_choices=period_choices_for_service(row.financial_year, getattr(row.catalogue, "recurrence_type", None)),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/edit")
|
|
def subscription_edit_submit(
|
|
request: Request,
|
|
subscription_id: int,
|
|
assigned_partner_user_id: str = Form(""),
|
|
assigned_manager_user_id: str = Form(""),
|
|
assigned_staff_user_id: str = Form(""),
|
|
performing_partner_user_id: str = Form(""),
|
|
review_partner_user_id: str = Form(""),
|
|
period_label: str = Form(""),
|
|
start_date: str = Form(""),
|
|
end_date: str = Form(""),
|
|
expiry_date: str = Form(""),
|
|
status: str = Form("active"),
|
|
remarks: str = Form(""),
|
|
is_active: str | None = Form(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if not row:
|
|
return RedirectResponse(url="/services/engagements", status_code=303)
|
|
active_fy = _active_financial_year(request)
|
|
if row.financial_year != active_fy:
|
|
return RedirectResponse(url=f"/services/engagements?financial_year={active_fy}", status_code=303)
|
|
if getattr(row, "is_locked", False) or is_row_financial_year_locked(db, row):
|
|
return RedirectResponse(url=f"/services/engagements/{row.id}?year_locked=1", status_code=303)
|
|
|
|
try:
|
|
new_period_label = normalize_period_label(
|
|
period_label,
|
|
financial_year=row.financial_year,
|
|
recurrence_type=getattr(row.catalogue, "recurrence_type", None),
|
|
)
|
|
except ValueError:
|
|
return RedirectResponse(url=f"/services/engagements/{row.id}/edit?error=period", status_code=303)
|
|
duplicate = get_existing_subscription(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
client_id=row.client_id,
|
|
service_catalogue_id=row.service_catalogue_id,
|
|
financial_year=row.financial_year,
|
|
period_label=new_period_label,
|
|
)
|
|
if duplicate and duplicate.id != row.id:
|
|
return RedirectResponse(url=f"/services/engagements/{row.id}/edit?error=duplicate_period", status_code=303)
|
|
row.period_label = new_period_label
|
|
row.assigned_partner_user_id = int(assigned_partner_user_id) if assigned_partner_user_id.strip() else None
|
|
requested_performing_partner_id = int(performing_partner_user_id) if performing_partner_user_id.strip() else None
|
|
row.performing_partner_user_id = requested_performing_partner_id or getattr(row.client, "default_performing_partner_user_id", None) or row.assigned_partner_user_id
|
|
row.assigned_manager_user_id = int(assigned_manager_user_id) if assigned_manager_user_id.strip() else None
|
|
row.assigned_staff_user_id = int(assigned_staff_user_id) if assigned_staff_user_id.strip() else None
|
|
if review_partner_required_for_engagement(db, tenant_id=tenant_id, engagement_type=row.engagement_type):
|
|
row.review_partner_user_id = int(review_partner_user_id) if review_partner_user_id.strip() else getattr(row.client, "default_review_partner_user_id", None)
|
|
else:
|
|
row.review_partner_user_id = None
|
|
row.start_date = parse_date(start_date)
|
|
row.end_date = parse_date(end_date)
|
|
row.expiry_date = parse_date(expiry_date)
|
|
row.status = status or "active"
|
|
row.remarks = remarks.strip() or None
|
|
row.is_active = is_active is not None
|
|
row.updated_by_user_id = user.id
|
|
apply_due_date_rule_to_subscription(db, row, force=True)
|
|
ensure_engagement_quality_workflow(db, subscription=row, actor_user_id=user.id, create_declarations=False)
|
|
enforce_quality_gate_on_subscription(row)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/services/engagements/{row.id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{subscription_id}/lock")
|
|
def subscription_lock(request: Request, subscription_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_lock_engagements(db, user):
|
|
return _redirect_denied()
|
|
tenant_id = _active_tenant_id(request, user)
|
|
row = get_subscription(db, subscription_id=subscription_id, tenant_id=tenant_id)
|
|
if row and row.financial_year != _active_financial_year(request):
|
|
return RedirectResponse(url=f"/services/engagements?financial_year={_active_financial_year(request)}", status_code=303)
|
|
if row and is_row_financial_year_locked(db, row):
|
|
return RedirectResponse(url=f"/services/engagements/{row.id}?year_locked=1", status_code=303)
|
|
if row and _user_can_lock_subscription(db, user, row):
|
|
if _lock_subscription_row(row, user):
|
|
db.commit()
|
|
return RedirectResponse(url=f"/services/engagements/{subscription_id}", status_code=303)
|
|
finally:
|
|
db.close()
|