A R R R Associates
1458 lines
60 KiB
Python
1458 lines
60 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, timezone
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import JSONResponse, RedirectResponse
|
|
from sqlalchemy import func, or_, select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
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.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
|
|
from app.modules.services.client_services import (
|
|
SUBSCRIPTION_STATUSES,
|
|
assessment_year_from_financial_year,
|
|
attach_engagement_to_plan,
|
|
ensure_engagement_quality_workflow,
|
|
enforce_quality_gate_on_subscription,
|
|
get_enabled_firm_service,
|
|
get_existing_subscription,
|
|
get_or_create_client_service_plan,
|
|
list_assignable_users,
|
|
list_clients_for_assignment,
|
|
list_enabled_services_for_assignment,
|
|
list_review_partners,
|
|
normalize_financial_year,
|
|
normalize_period_label,
|
|
parse_date,
|
|
period_choices_for_service,
|
|
recurrence_requires_period,
|
|
review_partner_required_for_engagement,
|
|
)
|
|
from app.modules.clients.models import Client, ClientBusinessUnit, ClientBranch
|
|
from app.modules.registrations.models import ClientRegistration, RegistrationType
|
|
from app.modules.services.due_dates import apply_due_date_rule_to_subscription
|
|
from app.modules.services.scope_targets import list_scope_targets, resolve_scope_target
|
|
from app.modules.services.models import (
|
|
ClientServicePlan,
|
|
ClientServiceSubscription,
|
|
ClientServiceTaskInstance,
|
|
)
|
|
|
|
router = APIRouter(prefix="/services/subscriptions", tags=["client-service-subscriptions-ui"])
|
|
|
|
|
|
def _tenant_id(request, user):
|
|
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 _branch_id(request, user):
|
|
value = request.session.get("active_branch_id")
|
|
if value in (None, "", 0, "0"):
|
|
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 _partner_scope_id(db, user) -> int | None:
|
|
roles = set(get_user_roles(db, user.id))
|
|
return int(user.id) if "Partner" in roles else None
|
|
|
|
|
|
def _registration_scope_context(db, *, tenant_id: int, plan: ClientServicePlan):
|
|
businesses = db.execute(
|
|
select(ClientBusinessUnit).where(
|
|
ClientBusinessUnit.tenant_id == tenant_id,
|
|
ClientBusinessUnit.client_id == plan.client_id,
|
|
ClientBusinessUnit.is_active.is_(True),
|
|
).order_by(
|
|
ClientBusinessUnit.is_primary.desc(),
|
|
ClientBusinessUnit.business_name,
|
|
)
|
|
).scalars().all()
|
|
|
|
branches = db.execute(
|
|
select(ClientBranch).where(
|
|
ClientBranch.tenant_id == tenant_id,
|
|
ClientBranch.client_id == plan.client_id,
|
|
ClientBranch.is_active.is_(True),
|
|
).order_by(
|
|
ClientBranch.is_primary.desc(),
|
|
ClientBranch.branch_name,
|
|
)
|
|
).scalars().all()
|
|
|
|
registration_rows = db.execute(
|
|
select(ClientRegistration, RegistrationType)
|
|
.join(
|
|
RegistrationType,
|
|
RegistrationType.id == ClientRegistration.registration_type_id,
|
|
)
|
|
.where(
|
|
ClientRegistration.tenant_id == tenant_id,
|
|
ClientRegistration.client_id == plan.client_id,
|
|
ClientRegistration.status.in_(("active", "valid", "registered")),
|
|
RegistrationType.is_active.is_(True),
|
|
)
|
|
.order_by(
|
|
RegistrationType.code,
|
|
ClientRegistration.registration_number,
|
|
)
|
|
).all()
|
|
|
|
registrations = []
|
|
registration_type_by_id = {}
|
|
for registration, registration_type in registration_rows:
|
|
registrations.append(registration)
|
|
registration_type_by_id[registration.id] = registration_type
|
|
|
|
business_by_id = {row.id: row for row in businesses}
|
|
branch_by_id = {row.id: row for row in branches}
|
|
registration_by_id = {row.id: row for row in registrations}
|
|
|
|
return {
|
|
"businesses": businesses,
|
|
"branches": branches,
|
|
"registrations": registrations,
|
|
"registration_type_by_id": registration_type_by_id,
|
|
"scope_business": business_by_id.get(plan.business_unit_id),
|
|
"scope_branch": branch_by_id.get(plan.client_branch_id),
|
|
"scope_registration": registration_by_id.get(plan.registration_id),
|
|
"scope_registration_type": (
|
|
registration_type_by_id.get(plan.registration_id)
|
|
if plan.registration_id
|
|
else None
|
|
),
|
|
}
|
|
|
|
|
|
|
|
def _subscription_scope_display_map(
|
|
db,
|
|
*,
|
|
tenant_id: int,
|
|
plans: list[ClientServicePlan],
|
|
) -> dict[int, dict[str, str]]:
|
|
"""Resolve human-friendly subscription names without changing stored scope data."""
|
|
if not plans:
|
|
return {}
|
|
|
|
business_ids = {int(row.business_unit_id) for row in plans if row.business_unit_id}
|
|
branch_ids = {int(row.client_branch_id) for row in plans if row.client_branch_id}
|
|
registration_ids = {int(row.registration_id) for row in plans if row.registration_id}
|
|
|
|
businesses = {}
|
|
if business_ids:
|
|
businesses = {
|
|
row.id: row
|
|
for row in db.execute(
|
|
select(ClientBusinessUnit).where(
|
|
ClientBusinessUnit.tenant_id == tenant_id,
|
|
ClientBusinessUnit.id.in_(business_ids),
|
|
)
|
|
).scalars().all()
|
|
}
|
|
|
|
branches = {}
|
|
if branch_ids:
|
|
branches = {
|
|
row.id: row
|
|
for row in db.execute(
|
|
select(ClientBranch).where(
|
|
ClientBranch.tenant_id == tenant_id,
|
|
ClientBranch.id.in_(branch_ids),
|
|
)
|
|
).scalars().all()
|
|
}
|
|
|
|
registrations = {}
|
|
registration_types = {}
|
|
if registration_ids:
|
|
registration_rows = db.execute(
|
|
select(ClientRegistration, RegistrationType)
|
|
.join(
|
|
RegistrationType,
|
|
RegistrationType.id == ClientRegistration.registration_type_id,
|
|
)
|
|
.where(
|
|
ClientRegistration.tenant_id == tenant_id,
|
|
ClientRegistration.id.in_(registration_ids),
|
|
)
|
|
).all()
|
|
for registration, registration_type in registration_rows:
|
|
registrations[registration.id] = registration
|
|
registration_types[registration.id] = registration_type
|
|
|
|
# A mapped registration can carry its own Business Unit / Client Branch.
|
|
# Load those as fallbacks even when the plan's denormalised IDs are absent.
|
|
extra_business_ids = {
|
|
int(row.business_unit_id)
|
|
for row in registrations.values()
|
|
if row.business_unit_id and row.business_unit_id not in businesses
|
|
}
|
|
if extra_business_ids:
|
|
businesses.update({
|
|
row.id: row
|
|
for row in db.execute(
|
|
select(ClientBusinessUnit).where(
|
|
ClientBusinessUnit.tenant_id == tenant_id,
|
|
ClientBusinessUnit.id.in_(extra_business_ids),
|
|
)
|
|
).scalars().all()
|
|
})
|
|
|
|
extra_branch_ids = {
|
|
int(row.client_branch_id)
|
|
for row in registrations.values()
|
|
if row.client_branch_id and row.client_branch_id not in branches
|
|
}
|
|
if extra_branch_ids:
|
|
branches.update({
|
|
row.id: row
|
|
for row in db.execute(
|
|
select(ClientBranch).where(
|
|
ClientBranch.tenant_id == tenant_id,
|
|
ClientBranch.id.in_(extra_branch_ids),
|
|
)
|
|
).scalars().all()
|
|
})
|
|
|
|
result: dict[int, dict[str, str]] = {}
|
|
for plan in plans:
|
|
client = plan.client
|
|
client_name = (getattr(client, "client_name", None) or "").strip()
|
|
client_code = (getattr(client, "client_code", None) or "").strip()
|
|
client_trade_name = (getattr(client, "trade_name", None) or "").strip()
|
|
scope_type = (getattr(plan, "scope_type", None) or "client").strip().lower()
|
|
|
|
primary = client_name or client_trade_name or client_code or f"Client {plan.client_id}"
|
|
secondary = client_code
|
|
context = ""
|
|
scope_label = "Client"
|
|
|
|
if scope_type == "business_unit":
|
|
business = businesses.get(plan.business_unit_id)
|
|
if business:
|
|
primary = (
|
|
(getattr(business, "trade_name", None) or "").strip()
|
|
or (getattr(business, "business_name", None) or "").strip()
|
|
or primary
|
|
)
|
|
secondary = (getattr(business, "business_code", None) or "").strip()
|
|
context = client_name if client_name and client_name != primary else ""
|
|
scope_label = "Business Unit"
|
|
|
|
elif scope_type == "client_branch":
|
|
branch = branches.get(plan.client_branch_id)
|
|
business = businesses.get(
|
|
getattr(branch, "business_unit_id", None) if branch else plan.business_unit_id
|
|
)
|
|
if branch:
|
|
primary = (getattr(branch, "branch_name", None) or "").strip() or primary
|
|
secondary = (getattr(branch, "branch_code", None) or "").strip()
|
|
business_name = ""
|
|
if business:
|
|
business_name = (
|
|
(getattr(business, "trade_name", None) or "").strip()
|
|
or (getattr(business, "business_name", None) or "").strip()
|
|
)
|
|
context_parts = []
|
|
if business_name and business_name != primary:
|
|
context_parts.append(business_name)
|
|
if client_name and client_name not in {primary, business_name}:
|
|
context_parts.append(f"Client: {client_name}")
|
|
context = " · ".join(context_parts)
|
|
scope_label = "Client Branch"
|
|
|
|
elif scope_type == "registration":
|
|
registration = registrations.get(plan.registration_id)
|
|
registration_type = registration_types.get(plan.registration_id)
|
|
if registration:
|
|
business = businesses.get(
|
|
getattr(registration, "business_unit_id", None) or plan.business_unit_id
|
|
)
|
|
branch = branches.get(
|
|
getattr(registration, "client_branch_id", None) or plan.client_branch_id
|
|
)
|
|
|
|
business_trade_name = (
|
|
(getattr(business, "trade_name", None) or "").strip()
|
|
if business else ""
|
|
)
|
|
business_name = (
|
|
(getattr(business, "business_name", None) or "").strip()
|
|
if business else ""
|
|
)
|
|
branch_name = (
|
|
(getattr(branch, "branch_name", None) or "").strip()
|
|
if branch else ""
|
|
)
|
|
|
|
# Registration-level subscriptions are identified by the
|
|
# registration's own trade name first. This prevents one
|
|
# individual/legal client with multiple registrations from
|
|
# appearing as several indistinguishable rows.
|
|
primary = (
|
|
(getattr(registration, "trade_name", None) or "").strip()
|
|
or business_trade_name
|
|
or business_name
|
|
or branch_name
|
|
or (getattr(registration, "legal_name", None) or "").strip()
|
|
or client_trade_name
|
|
or client_name
|
|
or primary
|
|
)
|
|
registration_number = (
|
|
getattr(registration, "registration_number", None) or ""
|
|
).strip()
|
|
registration_code = (
|
|
getattr(registration_type, "code", None) or ""
|
|
).strip()
|
|
secondary = (
|
|
f"{registration_code} · {registration_number}"
|
|
if registration_code and registration_number
|
|
else registration_number or registration_code
|
|
)
|
|
context_parts = []
|
|
if branch_name and branch_name != primary:
|
|
context_parts.append(branch_name)
|
|
if client_name and client_name != primary:
|
|
context_parts.append(f"Client: {client_name}")
|
|
context = " · ".join(context_parts)
|
|
scope_label = "Registration"
|
|
else:
|
|
# Legacy/unmapped rows retain their existing client display.
|
|
# No stored subscription or engagement data is altered.
|
|
scope_label = "Registration"
|
|
|
|
result[int(plan.id)] = {
|
|
"primary": primary,
|
|
"secondary": secondary,
|
|
"context": context,
|
|
"scope_label": scope_label,
|
|
"client_name": client_name,
|
|
"client_code": client_code,
|
|
}
|
|
|
|
return result
|
|
|
|
|
|
def _subscription_scope_display(
|
|
db,
|
|
*,
|
|
tenant_id: int,
|
|
plan: ClientServicePlan,
|
|
) -> dict[str, str]:
|
|
return _subscription_scope_display_map(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
plans=[plan],
|
|
).get(
|
|
int(plan.id),
|
|
{
|
|
"primary": plan.client.client_name,
|
|
"secondary": plan.client.client_code or "",
|
|
"context": "",
|
|
"scope_label": "Client",
|
|
"client_name": plan.client.client_name,
|
|
"client_code": plan.client.client_code or "",
|
|
},
|
|
)
|
|
|
|
def _scope_key_for(scope_type: str, target_id: int) -> str:
|
|
prefixes = {
|
|
"client": "CLIENT",
|
|
"business_unit": "BUSINESS",
|
|
"client_branch": "BRANCH",
|
|
"registration": "REGISTRATION",
|
|
}
|
|
if scope_type not in prefixes:
|
|
raise ValueError("scope_type")
|
|
return f"{prefixes[scope_type]}:{int(target_id)}"
|
|
|
|
def _ctx(request, db, user, **extra):
|
|
data = {
|
|
"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,
|
|
}
|
|
data.update(extra)
|
|
return data
|
|
|
|
|
|
def _denied():
|
|
from app.core.http_responses import ui_access_denied
|
|
return ui_access_denied()
|
|
|
|
|
|
def _load_plan(db, *, plan_id: int, tenant_id: int, branch_id: int | None):
|
|
stmt = (
|
|
select(ClientServicePlan)
|
|
.options(
|
|
selectinload(ClientServicePlan.client),
|
|
selectinload(ClientServicePlan.catalogue),
|
|
selectinload(ClientServicePlan.default_partner),
|
|
selectinload(ClientServicePlan.default_performing_partner),
|
|
selectinload(ClientServicePlan.default_manager),
|
|
selectinload(ClientServicePlan.default_staff),
|
|
selectinload(ClientServicePlan.default_review_partner),
|
|
)
|
|
.where(
|
|
ClientServicePlan.id == plan_id,
|
|
ClientServicePlan.tenant_id == tenant_id,
|
|
)
|
|
)
|
|
if branch_id:
|
|
stmt = stmt.where(ClientServicePlan.branch_id == branch_id)
|
|
return db.execute(stmt).scalar_one_or_none()
|
|
|
|
|
|
def _optional_int(value: str) -> int | None:
|
|
value = (value or "").strip()
|
|
return int(value) if value else None
|
|
|
|
|
|
def _allowed_id(value: int | None, allowed: set[int], field_name: str) -> int | None:
|
|
if value is not None and value not in allowed:
|
|
raise ValueError(field_name)
|
|
return value
|
|
|
|
|
|
def _engagement_is_open(row: ClientServiceSubscription) -> bool:
|
|
return (
|
|
not bool(getattr(row, "is_locked", False))
|
|
and (getattr(row, "status", "") or "").lower() not in {"completed", "cancelled", "inactive"}
|
|
)
|
|
|
|
|
|
def _engagement_is_effective(row: ClientServiceSubscription, effective_from: date | None) -> bool:
|
|
if not effective_from:
|
|
return True
|
|
comparison_date = (
|
|
getattr(row, "start_date", None)
|
|
or getattr(row, "current_due_date", None)
|
|
or getattr(row, "original_due_date", None)
|
|
)
|
|
return comparison_date is None or comparison_date >= effective_from
|
|
|
|
|
|
@router.get("")
|
|
def subscription_master_list(request: Request, q: str = "", include_inactive: bool = False):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse("/login", 303)
|
|
try:
|
|
require_permission(db, user, "clients.view")
|
|
except Exception:
|
|
return _denied()
|
|
|
|
tenant_id = _tenant_id(request, user)
|
|
branch_id = _branch_id(request, user)
|
|
counts = (
|
|
select(
|
|
ClientServiceSubscription.service_plan_id,
|
|
func.count(ClientServiceSubscription.id).label("engagement_count"),
|
|
func.max(ClientServiceSubscription.current_due_date).label("latest_due_date"),
|
|
)
|
|
.where(ClientServiceSubscription.service_plan_id.is_not(None))
|
|
.group_by(ClientServiceSubscription.service_plan_id)
|
|
.subquery()
|
|
)
|
|
stmt = (
|
|
select(ClientServicePlan, counts.c.engagement_count, counts.c.latest_due_date)
|
|
.options(
|
|
selectinload(ClientServicePlan.client),
|
|
selectinload(ClientServicePlan.catalogue),
|
|
selectinload(ClientServicePlan.default_partner),
|
|
selectinload(ClientServicePlan.default_performing_partner),
|
|
selectinload(ClientServicePlan.default_manager),
|
|
selectinload(ClientServicePlan.default_staff),
|
|
selectinload(ClientServicePlan.default_review_partner),
|
|
)
|
|
.outerjoin(counts, counts.c.service_plan_id == ClientServicePlan.id)
|
|
.where(ClientServicePlan.tenant_id == tenant_id)
|
|
)
|
|
if branch_id:
|
|
stmt = stmt.where(ClientServicePlan.branch_id == branch_id)
|
|
if not include_inactive:
|
|
stmt = stmt.where(ClientServicePlan.is_active.is_(True))
|
|
if q.strip():
|
|
from app.modules.clients.models import Client
|
|
from app.modules.services.models import ServiceCatalogue
|
|
term = f"%{q.strip()}%"
|
|
stmt = (
|
|
stmt.join(Client, Client.id == ClientServicePlan.client_id)
|
|
.join(ServiceCatalogue, ServiceCatalogue.id == ClientServicePlan.service_catalogue_id)
|
|
.where(
|
|
(Client.client_name.ilike(term))
|
|
| (Client.client_code.ilike(term))
|
|
| (ServiceCatalogue.service_name.ilike(term))
|
|
| (ServiceCatalogue.service_code.ilike(term))
|
|
)
|
|
)
|
|
rows = db.execute(
|
|
stmt.order_by(ClientServicePlan.is_active.desc(), ClientServicePlan.id.desc())
|
|
).all()
|
|
scope_displays = _subscription_scope_display_map(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
plans=[row[0] for row in rows],
|
|
)
|
|
return templates.TemplateResponse(
|
|
"modules/services/templates/services/subscriptions/list.html",
|
|
_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Client Service Subscriptions",
|
|
rows=rows,
|
|
scope_displays=scope_displays,
|
|
q=q,
|
|
include_inactive=include_inactive,
|
|
can_edit_subscription="clients.edit" in set(get_user_permissions(db, user.id)),
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
|
|
@router.get("/bulk")
|
|
def subscription_bulk_page(
|
|
request: Request,
|
|
error: str = "",
|
|
subscriptions_created: int = 0,
|
|
subscriptions_reused: int = 0,
|
|
engagements_created: int = 0,
|
|
engagements_skipped: int = 0,
|
|
):
|
|
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)
|
|
enabled_services = list_enabled_services_for_assignment(db, tenant_id=tenant_id)
|
|
partners = list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id, role_names=("Partner",))
|
|
managers = list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id, role_names=("Branch Manager",))
|
|
staff_users = list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id, role_names=("Staff",))
|
|
review_partners = list_review_partners(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
error_messages = {
|
|
"targets": "Select at least one permitted Client, Business Unit, Client Branch or Registration.",
|
|
"service": "Select a valid enabled firm service.",
|
|
"partner": "Select a valid Engagement Partner.",
|
|
"performing_partner": "Select a valid Performing Partner.",
|
|
"manager": "Select a valid Manager.",
|
|
"staff": "Select a valid Staff member.",
|
|
"review_partner": "Select a valid Review Partner.",
|
|
"review_partner_required": "Review Partner is mandatory for this assurance service.",
|
|
"review_partner_independence": "Review Partner must differ from the Engagement and Performing Partners.",
|
|
"period": "Select a valid month or quarter.",
|
|
"generation": "Select a valid engagement generation option.",
|
|
"financial_year": "The selected financial year is locked.",
|
|
}
|
|
return templates.TemplateResponse(
|
|
"modules/services/templates/services/subscriptions/bulk.html",
|
|
_ctx(
|
|
request, db, user, title="Bulk Client Subscriptions",
|
|
enabled_services=enabled_services, partners=partners, managers=managers,
|
|
staff_users=staff_users, review_partners=review_partners,
|
|
financial_year=_active_financial_year(request),
|
|
error_message=error_messages.get(error, ""),
|
|
subscriptions_created=subscriptions_created,
|
|
subscriptions_reused=subscriptions_reused,
|
|
engagements_created=engagements_created,
|
|
engagements_skipped=engagements_skipped,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/bulk/targets")
|
|
def subscription_bulk_targets(request: Request, service_catalogue_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return JSONResponse({"error": "Authentication required."}, status_code=401)
|
|
try:
|
|
require_permission(db, user, "clients.edit")
|
|
except Exception:
|
|
return JSONResponse({"error": "Access denied."}, status_code=403)
|
|
|
|
tenant_id = _tenant_id(request, user)
|
|
branch_id = _branch_id(request, user)
|
|
firm_selection = get_enabled_firm_service(db, tenant_id=tenant_id, service_catalogue_id=service_catalogue_id)
|
|
if not firm_selection:
|
|
return JSONResponse({"error": "Enabled firm service not found."}, status_code=404)
|
|
|
|
clients = list_clients_for_assignment(
|
|
db, tenant_id=tenant_id, branch_id=branch_id, partner_id=_partner_scope_id(db, user)
|
|
)
|
|
targets = list_scope_targets(db, tenant_id=tenant_id, clients=clients, catalogue=firm_selection.catalogue)
|
|
scope_type = getattr(firm_selection.catalogue, "service_scope_type", "client") or "client"
|
|
registration_type = getattr(firm_selection.catalogue, "required_registration_type", None)
|
|
return {
|
|
"scope_type": scope_type,
|
|
"registration_type": registration_type,
|
|
"targets": [
|
|
{
|
|
"token": row.token,
|
|
"client_code": row.client_code,
|
|
"client_name": row.client_name,
|
|
"pan": row.pan,
|
|
"business_unit": row.business_unit,
|
|
"client_branch": row.client_branch,
|
|
"registration_type": row.registration_type,
|
|
"registration_number": row.registration_number,
|
|
"trade_name": row.trade_name,
|
|
"state": row.state,
|
|
"entity_type": row.entity_type,
|
|
"subscription_status": row.existing_plan_status or "not_subscribed",
|
|
}
|
|
for row in targets
|
|
],
|
|
}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/bulk")
|
|
def subscription_bulk_submit(
|
|
request: Request,
|
|
scope_targets: list[str] = Form([]),
|
|
service_catalogue_id: int = Form(...),
|
|
default_partner_user_id: int = Form(...),
|
|
default_performing_partner_user_id: str = Form(""),
|
|
default_manager_user_id: str = Form(""),
|
|
default_staff_user_id: str = Form(""),
|
|
default_review_partner_user_id: str = Form(""),
|
|
effective_from: str = Form(""),
|
|
effective_to: str = Form(""),
|
|
auto_generate_periods: str | None = Form(None),
|
|
remarks: str = Form(""),
|
|
generation_mode: str = Form("subscription_only"),
|
|
financial_year: str = Form(""),
|
|
period_label: str = Form(""),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
subscriptions_created = subscriptions_reused = engagements_created = engagements_skipped = 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)
|
|
firm_selection = get_enabled_firm_service(db, tenant_id=tenant_id, service_catalogue_id=service_catalogue_id)
|
|
if not firm_selection:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=service", 303)
|
|
|
|
permitted_clients = list_clients_for_assignment(
|
|
db, tenant_id=tenant_id, branch_id=branch_id, partner_id=_partner_scope_id(db, user)
|
|
)
|
|
clients_by_id = {int(row.id): row for row in permitted_clients}
|
|
selected_tokens = list(dict.fromkeys(value for value in scope_targets if value))
|
|
if not selected_tokens:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=targets", 303)
|
|
|
|
plan_branch_id = branch_id or getattr(firm_selection, "default_branch_id", None) or getattr(user, "branch_id", None)
|
|
partners = list_assignable_users(db, tenant_id=tenant_id, branch_id=plan_branch_id, role_names=("Partner",))
|
|
managers = list_assignable_users(db, tenant_id=tenant_id, branch_id=plan_branch_id, role_names=("Branch Manager",))
|
|
staff_users = list_assignable_users(db, tenant_id=tenant_id, branch_id=plan_branch_id, role_names=("Staff",))
|
|
review_partners = list_review_partners(db, tenant_id=tenant_id, branch_id=plan_branch_id)
|
|
|
|
if default_partner_user_id not in {row.id for row in partners}:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=partner", 303)
|
|
performing_partner_id = _optional_int(default_performing_partner_user_id) or default_partner_user_id
|
|
if performing_partner_id not in {row.id for row in partners}:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=performing_partner", 303)
|
|
manager_id = _optional_int(default_manager_user_id)
|
|
if manager_id is not None and manager_id not in {row.id for row in managers}:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=manager", 303)
|
|
staff_id = _optional_int(default_staff_user_id)
|
|
if staff_id is not None and staff_id not in {row.id for row in staff_users}:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=staff", 303)
|
|
review_partner_id = _optional_int(default_review_partner_user_id)
|
|
if review_partner_id is not None and review_partner_id not in {row.id for row in review_partners}:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=review_partner", 303)
|
|
|
|
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)
|
|
if review_required and review_partner_id is None:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=review_partner_required", 303)
|
|
if review_required and review_partner_id in {default_partner_user_id, performing_partner_id}:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=review_partner_independence", 303)
|
|
|
|
if generation_mode not in {"subscription_only", "current_period", "all_periods"}:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=generation", 303)
|
|
selected_financial_year = normalize_financial_year(financial_year or _active_financial_year(request))
|
|
if generation_mode != "subscription_only":
|
|
locked = redirect_if_financial_year_locked(
|
|
db, tenant_id=tenant_id, year_code=selected_financial_year,
|
|
redirect_url="/services/subscriptions/bulk?error=financial_year",
|
|
)
|
|
if locked:
|
|
return locked
|
|
|
|
recurrence_type = getattr(firm_selection.catalogue, "recurrence_type", None)
|
|
if generation_mode == "subscription_only":
|
|
requested_periods = []
|
|
elif recurrence_requires_period(recurrence_type):
|
|
if generation_mode == "all_periods":
|
|
requested_periods = [code for code, _ in period_choices_for_service(selected_financial_year, recurrence_type)]
|
|
else:
|
|
try:
|
|
requested_periods = [normalize_period_label(
|
|
period_label, financial_year=selected_financial_year, recurrence_type=recurrence_type
|
|
)]
|
|
except ValueError:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=period", 303)
|
|
else:
|
|
requested_periods = [""]
|
|
|
|
plan_effective_from = parse_date(effective_from)
|
|
plan_effective_to = parse_date(effective_to)
|
|
if plan_effective_from and plan_effective_to and plan_effective_to < plan_effective_from:
|
|
return RedirectResponse("/services/subscriptions/bulk?error=period", 303)
|
|
|
|
for token in selected_tokens:
|
|
try:
|
|
client, scope_type, scope_key, business_unit_id, client_branch_id, registration_id = resolve_scope_target(
|
|
db, tenant_id=tenant_id, clients_by_id=clients_by_id, token=token, actor_user_id=user.id
|
|
)
|
|
except ValueError:
|
|
db.rollback()
|
|
return RedirectResponse("/services/subscriptions/bulk?error=targets", 303)
|
|
|
|
plan = db.execute(select(ClientServicePlan).where(
|
|
ClientServicePlan.tenant_id == tenant_id,
|
|
ClientServicePlan.service_catalogue_id == service_catalogue_id,
|
|
ClientServicePlan.scope_key == scope_key,
|
|
)).scalar_one_or_none()
|
|
|
|
if plan:
|
|
subscriptions_reused += 1
|
|
else:
|
|
plan = ClientServicePlan(
|
|
tenant_id=tenant_id, branch_id=plan_branch_id or client.branch_id,
|
|
client_id=client.id, scope_type=scope_type, scope_key=scope_key,
|
|
business_unit_id=business_unit_id, client_branch_id=client_branch_id,
|
|
registration_id=registration_id, service_catalogue_id=service_catalogue_id,
|
|
firm_service_selection_id=firm_selection.id,
|
|
default_partner_user_id=default_partner_user_id,
|
|
default_performing_partner_user_id=performing_partner_id,
|
|
default_manager_user_id=manager_id, default_staff_user_id=staff_id,
|
|
default_review_partner_user_id=review_partner_id,
|
|
recurrence_type=(recurrence_type or "one_time"),
|
|
effective_from=plan_effective_from, effective_to=plan_effective_to,
|
|
auto_generate_periods=auto_generate_periods is not None,
|
|
status="active", is_active=True, remarks=remarks.strip() or None,
|
|
created_by_user_id=user.id, updated_by_user_id=user.id,
|
|
)
|
|
db.add(plan); db.flush()
|
|
subscriptions_created += 1
|
|
|
|
for requested_period in requested_periods:
|
|
existing = db.execute(select(ClientServiceSubscription).where(
|
|
ClientServiceSubscription.tenant_id == tenant_id,
|
|
ClientServiceSubscription.service_catalogue_id == service_catalogue_id,
|
|
ClientServiceSubscription.scope_key == 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
|
|
engagements_skipped += 1
|
|
continue
|
|
|
|
engagement = ClientServiceSubscription(
|
|
tenant_id=tenant_id, branch_id=plan.branch_id, service_plan_id=plan.id,
|
|
client_id=client.id, scope_type=scope_type, scope_key=scope_key,
|
|
business_unit_id=business_unit_id, client_branch_id=client_branch_id,
|
|
registration_id=registration_id, service_catalogue_id=service_catalogue_id,
|
|
firm_service_selection_id=firm_selection.id,
|
|
assigned_partner_user_id=default_partner_user_id,
|
|
performing_partner_user_id=performing_partner_id,
|
|
assigned_manager_user_id=manager_id, assigned_staff_user_id=staff_id,
|
|
review_partner_user_id=review_partner_id if review_required 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, start_date=plan_effective_from,
|
|
end_date=plan_effective_to, status="active", remarks=remarks.strip() or None,
|
|
is_active=True, created_by_user_id=user.id, updated_by_user_id=user.id,
|
|
)
|
|
db.add(engagement); db.flush()
|
|
attach_engagement_to_plan(
|
|
db, engagement=engagement, client=client, catalogue=firm_selection.catalogue,
|
|
firm_selection=firm_selection, actor_user_id=user.id,
|
|
)
|
|
apply_due_date_rule_to_subscription(db, engagement, force=True)
|
|
ensure_engagement_quality_workflow(db, subscription=engagement, actor_user_id=user.id, create_declarations=False)
|
|
enforce_quality_gate_on_subscription(engagement)
|
|
engagements_created += 1
|
|
|
|
db.commit()
|
|
return RedirectResponse(
|
|
"/services/subscriptions/bulk"
|
|
f"?subscriptions_created={subscriptions_created}&subscriptions_reused={subscriptions_reused}"
|
|
f"&engagements_created={engagements_created}&engagements_skipped={engagements_skipped}",
|
|
303,
|
|
)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
|
|
@router.get("/{plan_id}/scope")
|
|
def subscription_scope_edit_page(
|
|
request: Request,
|
|
plan_id: int,
|
|
saved: int = 0,
|
|
updated_engagements: int = 0,
|
|
error: str = "",
|
|
):
|
|
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)
|
|
|
|
scope_context = _registration_scope_context(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
plan=plan,
|
|
)
|
|
error_messages = {
|
|
"scope_type": "Select a valid subscription scope.",
|
|
"business_unit": "Select a valid Business Unit belonging to this client.",
|
|
"client_branch": "Select a valid Client Branch belonging to this client.",
|
|
"registration": "Select a valid active registration belonging to this client.",
|
|
"registration_type": "The selected registration type does not match the Service Catalogue requirement.",
|
|
"duplicate_plan": "Another subscription already exists for this service and selected scope.",
|
|
"duplicate_engagement": "An engagement already exists for the selected scope, financial year and period.",
|
|
"propagation": "Select a valid propagation option.",
|
|
}
|
|
return templates.TemplateResponse(
|
|
"modules/services/templates/services/subscriptions/scope.html",
|
|
_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Map Subscription Scope",
|
|
plan=plan,
|
|
saved=saved,
|
|
updated_engagements=updated_engagements,
|
|
error_message=error_messages.get(error, ""),
|
|
scope_display=_subscription_scope_display(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
plan=plan,
|
|
),
|
|
**scope_context,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{plan_id}/scope")
|
|
def subscription_scope_edit_submit(
|
|
request: Request,
|
|
plan_id: int,
|
|
scope_type: str = Form(...),
|
|
business_unit_id: str = Form(""),
|
|
client_branch_id: str = Form(""),
|
|
registration_id: str = Form(""),
|
|
propagation_mode: str = Form("open_engagements"),
|
|
scope_change_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("/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)
|
|
|
|
normalized_scope = (
|
|
scope_type or ""
|
|
).strip().lower().replace("-", "_").replace(" ", "_")
|
|
if normalized_scope not in {
|
|
"client",
|
|
"business_unit",
|
|
"client_branch",
|
|
"registration",
|
|
}:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=scope_type",
|
|
303,
|
|
)
|
|
if propagation_mode not in {"subscription_only", "open_engagements"}:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=propagation",
|
|
303,
|
|
)
|
|
|
|
selected_business_id = _optional_int(business_unit_id)
|
|
selected_branch_id = _optional_int(client_branch_id)
|
|
selected_registration_id = _optional_int(registration_id)
|
|
|
|
business = None
|
|
client_branch = None
|
|
registration = None
|
|
registration_type = None
|
|
|
|
if selected_business_id:
|
|
business = db.get(ClientBusinessUnit, selected_business_id)
|
|
if (
|
|
not business
|
|
or business.tenant_id != tenant_id
|
|
or business.client_id != plan.client_id
|
|
or not business.is_active
|
|
):
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=business_unit",
|
|
303,
|
|
)
|
|
|
|
if selected_branch_id:
|
|
client_branch = db.get(ClientBranch, selected_branch_id)
|
|
if (
|
|
not client_branch
|
|
or client_branch.tenant_id != tenant_id
|
|
or client_branch.client_id != plan.client_id
|
|
or not client_branch.is_active
|
|
):
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=client_branch",
|
|
303,
|
|
)
|
|
selected_business_id = client_branch.business_unit_id
|
|
business = db.get(ClientBusinessUnit, selected_business_id)
|
|
|
|
if selected_registration_id:
|
|
registration = db.get(ClientRegistration, selected_registration_id)
|
|
if (
|
|
not registration
|
|
or registration.tenant_id != tenant_id
|
|
or registration.client_id != plan.client_id
|
|
or registration.status not in {"active", "valid", "registered"}
|
|
):
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=registration",
|
|
303,
|
|
)
|
|
registration_type = db.get(
|
|
RegistrationType,
|
|
registration.registration_type_id,
|
|
)
|
|
selected_business_id = (
|
|
registration.business_unit_id or selected_business_id
|
|
)
|
|
selected_branch_id = (
|
|
registration.client_branch_id or selected_branch_id
|
|
)
|
|
if selected_business_id:
|
|
business = db.get(ClientBusinessUnit, selected_business_id)
|
|
if selected_branch_id:
|
|
client_branch = db.get(ClientBranch, selected_branch_id)
|
|
|
|
if normalized_scope == "client":
|
|
target_id = plan.client_id
|
|
selected_business_id = None
|
|
selected_branch_id = None
|
|
selected_registration_id = None
|
|
elif normalized_scope == "business_unit":
|
|
if not business:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=business_unit",
|
|
303,
|
|
)
|
|
target_id = business.id
|
|
selected_branch_id = None
|
|
selected_registration_id = None
|
|
elif normalized_scope == "client_branch":
|
|
if not client_branch:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=client_branch",
|
|
303,
|
|
)
|
|
target_id = client_branch.id
|
|
selected_registration_id = None
|
|
else:
|
|
if not registration or not registration_type:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=registration",
|
|
303,
|
|
)
|
|
required_type = (
|
|
getattr(plan.catalogue, "required_registration_type", None)
|
|
or ""
|
|
).strip().upper()
|
|
actual_type = (registration_type.code or "").strip().upper()
|
|
if required_type and required_type != actual_type:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=registration_type",
|
|
303,
|
|
)
|
|
target_id = registration.id
|
|
|
|
new_scope_key = _scope_key_for(normalized_scope, target_id)
|
|
|
|
conflicting_plan = db.execute(
|
|
select(ClientServicePlan).where(
|
|
ClientServicePlan.tenant_id == tenant_id,
|
|
ClientServicePlan.service_catalogue_id
|
|
== plan.service_catalogue_id,
|
|
ClientServicePlan.scope_key == new_scope_key,
|
|
ClientServicePlan.id != plan.id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if conflicting_plan:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=duplicate_plan",
|
|
303,
|
|
)
|
|
|
|
open_engagements = []
|
|
if propagation_mode == "open_engagements":
|
|
open_engagements = db.execute(
|
|
select(ClientServiceSubscription).where(
|
|
ClientServiceSubscription.tenant_id == tenant_id,
|
|
ClientServiceSubscription.service_plan_id == plan.id,
|
|
)
|
|
).scalars().all()
|
|
open_engagements = [
|
|
row for row in open_engagements if _engagement_is_open(row)
|
|
]
|
|
|
|
for row in open_engagements:
|
|
conflict = db.execute(
|
|
select(ClientServiceSubscription.id).where(
|
|
ClientServiceSubscription.tenant_id == tenant_id,
|
|
ClientServiceSubscription.service_catalogue_id
|
|
== plan.service_catalogue_id,
|
|
ClientServiceSubscription.scope_key == new_scope_key,
|
|
ClientServiceSubscription.financial_year
|
|
== row.financial_year,
|
|
ClientServiceSubscription.period_label
|
|
== row.period_label,
|
|
ClientServiceSubscription.id != row.id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if conflict:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?error=duplicate_engagement",
|
|
303,
|
|
)
|
|
|
|
old_scope = plan.scope_key or f"CLIENT:{plan.client_id}"
|
|
plan.scope_type = normalized_scope
|
|
plan.scope_key = new_scope_key
|
|
plan.business_unit_id = selected_business_id
|
|
plan.client_branch_id = selected_branch_id
|
|
plan.registration_id = selected_registration_id
|
|
plan.updated_by_user_id = user.id
|
|
plan.updated_at_utc = datetime.now(timezone.utc)
|
|
|
|
updated_engagements = 0
|
|
if propagation_mode == "open_engagements":
|
|
for row in open_engagements:
|
|
row.scope_type = normalized_scope
|
|
row.scope_key = new_scope_key
|
|
row.business_unit_id = selected_business_id
|
|
row.client_branch_id = selected_branch_id
|
|
row.registration_id = selected_registration_id
|
|
row.updated_by_user_id = user.id
|
|
row.updated_at_utc = datetime.now(timezone.utc)
|
|
if scope_change_reason.strip():
|
|
note = (
|
|
f"Subscription scope changed from {old_scope} "
|
|
f"to {new_scope_key} on "
|
|
f"{datetime.now(timezone.utc).date().isoformat()} "
|
|
f"by user {user.id}: "
|
|
f"{scope_change_reason.strip()}"
|
|
)
|
|
row.remarks = (
|
|
f"{row.remarks.strip()}\n{note}"
|
|
if row.remarks
|
|
else note
|
|
)
|
|
updated_engagements += 1
|
|
|
|
db.commit()
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/scope?saved=1"
|
|
f"&updated_engagements={updated_engagements}",
|
|
303,
|
|
)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
@router.get("/{plan_id}/edit")
|
|
def subscription_master_edit_page(
|
|
request: Request,
|
|
plan_id: int,
|
|
saved: int = 0,
|
|
updated_engagements: int = 0,
|
|
updated_tasks: int = 0,
|
|
error: str = "",
|
|
):
|
|
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)
|
|
|
|
partners = list_assignable_users(
|
|
db, tenant_id=tenant_id, branch_id=plan.branch_id, role_names=("Partner",)
|
|
)
|
|
managers = list_assignable_users(
|
|
db, tenant_id=tenant_id, branch_id=plan.branch_id, role_names=("Branch Manager",)
|
|
)
|
|
staff_users = list_assignable_users(
|
|
db, tenant_id=tenant_id, branch_id=plan.branch_id, role_names=("Staff",)
|
|
)
|
|
review_partners = list_review_partners(
|
|
db, tenant_id=tenant_id, branch_id=plan.branch_id
|
|
)
|
|
return templates.TemplateResponse(
|
|
"modules/services/templates/services/subscriptions/edit.html",
|
|
_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Edit Client Service Subscription",
|
|
plan=plan,
|
|
partners=partners,
|
|
managers=managers,
|
|
staff_users=staff_users,
|
|
review_partners=review_partners,
|
|
saved=saved,
|
|
updated_engagements=updated_engagements,
|
|
updated_tasks=updated_tasks,
|
|
error=error,
|
|
scope_display=_subscription_scope_display(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
plan=plan,
|
|
),
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{plan_id}/edit")
|
|
def subscription_master_edit_submit(
|
|
request: Request,
|
|
plan_id: int,
|
|
default_partner_user_id: str = Form(""),
|
|
default_performing_partner_user_id: str = Form(""),
|
|
default_manager_user_id: str = Form(""),
|
|
default_staff_user_id: str = Form(""),
|
|
default_review_partner_user_id: str = Form(""),
|
|
effective_from: str = Form(""),
|
|
effective_to: str = Form(""),
|
|
status: str = Form("active"),
|
|
auto_generate_periods: str | None = Form(None),
|
|
remarks: str = Form(""),
|
|
propagation_mode: str = Form("future_only"),
|
|
reassignment_effective_from: str = Form(""),
|
|
reassignment_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("/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)
|
|
|
|
partners = list_assignable_users(
|
|
db, tenant_id=tenant_id, branch_id=plan.branch_id, role_names=("Partner",)
|
|
)
|
|
managers = list_assignable_users(
|
|
db, tenant_id=tenant_id, branch_id=plan.branch_id, role_names=("Branch Manager",)
|
|
)
|
|
staff_users = list_assignable_users(
|
|
db, tenant_id=tenant_id, branch_id=plan.branch_id, role_names=("Staff",)
|
|
)
|
|
review_partners = list_review_partners(
|
|
db, tenant_id=tenant_id, branch_id=plan.branch_id
|
|
)
|
|
|
|
try:
|
|
new_partner_id = _allowed_id(
|
|
_optional_int(default_partner_user_id), {row.id for row in partners}, "partner"
|
|
)
|
|
requested_performing_id = _allowed_id(
|
|
_optional_int(default_performing_partner_user_id),
|
|
{row.id for row in partners},
|
|
"performing_partner",
|
|
)
|
|
new_performing_id = requested_performing_id or new_partner_id
|
|
new_manager_id = _allowed_id(
|
|
_optional_int(default_manager_user_id), {row.id for row in managers}, "manager"
|
|
)
|
|
new_staff_id = _allowed_id(
|
|
_optional_int(default_staff_user_id), {row.id for row in staff_users}, "staff"
|
|
)
|
|
new_review_partner_id = _allowed_id(
|
|
_optional_int(default_review_partner_user_id),
|
|
{row.id for row in review_partners},
|
|
"review_partner",
|
|
)
|
|
plan_effective_from = parse_date(effective_from)
|
|
plan_effective_to = parse_date(effective_to)
|
|
change_effective_from = parse_date(reassignment_effective_from)
|
|
except (TypeError, ValueError) as exc:
|
|
code = str(exc) if str(exc) else "validation"
|
|
db.rollback()
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/edit?error={code}", status_code=303
|
|
)
|
|
|
|
allowed_statuses = {item[0] for item in SUBSCRIPTION_STATUSES}
|
|
normalized_status = (status or "active").strip().lower()
|
|
if normalized_status not in allowed_statuses:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/edit?error=status", status_code=303
|
|
)
|
|
if plan_effective_from and plan_effective_to and plan_effective_to < plan_effective_from:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/edit?error=dates", status_code=303
|
|
)
|
|
if propagation_mode not in {"future_only", "open_engagements"}:
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/edit?error=propagation", status_code=303
|
|
)
|
|
|
|
old_staff_id = plan.default_staff_user_id
|
|
plan.default_partner_user_id = new_partner_id
|
|
plan.default_performing_partner_user_id = new_performing_id
|
|
plan.default_manager_user_id = new_manager_id
|
|
plan.default_staff_user_id = new_staff_id
|
|
plan.default_review_partner_user_id = new_review_partner_id
|
|
plan.effective_from = plan_effective_from
|
|
plan.effective_to = plan_effective_to
|
|
plan.auto_generate_periods = auto_generate_periods is not None
|
|
plan.status = normalized_status
|
|
plan.is_active = normalized_status not in {"inactive", "cancelled"}
|
|
plan.remarks = remarks.strip() or None
|
|
plan.updated_by_user_id = user.id
|
|
plan.updated_at_utc = datetime.now(timezone.utc)
|
|
|
|
updated_engagement_count = 0
|
|
updated_task_count = 0
|
|
if propagation_mode == "open_engagements":
|
|
engagements = db.execute(
|
|
select(ClientServiceSubscription).where(
|
|
ClientServiceSubscription.service_plan_id == plan.id,
|
|
ClientServiceSubscription.tenant_id == tenant_id,
|
|
)
|
|
).scalars().all()
|
|
|
|
affected_ids: list[int] = []
|
|
for row in engagements:
|
|
if not _engagement_is_open(row):
|
|
continue
|
|
if not _engagement_is_effective(row, change_effective_from):
|
|
continue
|
|
row.assigned_partner_user_id = new_partner_id
|
|
row.performing_partner_user_id = new_performing_id
|
|
row.assigned_manager_user_id = new_manager_id
|
|
row.assigned_staff_user_id = new_staff_id
|
|
row.review_partner_user_id = new_review_partner_id
|
|
row.updated_by_user_id = user.id
|
|
row.updated_at_utc = datetime.now(timezone.utc)
|
|
if reassignment_reason.strip():
|
|
note = (
|
|
f"Subscription team updated on {datetime.now(timezone.utc).date().isoformat()} "
|
|
f"by user {user.id}: {reassignment_reason.strip()}"
|
|
)
|
|
row.remarks = f"{row.remarks.strip()}\n{note}" if row.remarks else note
|
|
affected_ids.append(row.id)
|
|
updated_engagement_count += 1
|
|
|
|
if affected_ids and old_staff_id != new_staff_id:
|
|
task_stmt = select(ClientServiceTaskInstance).where(
|
|
ClientServiceTaskInstance.subscription_id.in_(affected_ids),
|
|
ClientServiceTaskInstance.is_active.is_(True),
|
|
ClientServiceTaskInstance.is_locked.is_(False),
|
|
ClientServiceTaskInstance.status.notin_(("completed", "cancelled")),
|
|
)
|
|
if old_staff_id is None:
|
|
task_stmt = task_stmt.where(
|
|
ClientServiceTaskInstance.assigned_to_user_id.is_(None),
|
|
func.lower(func.coalesce(ClientServiceTaskInstance.default_role_name, "")) == "staff",
|
|
)
|
|
else:
|
|
task_stmt = task_stmt.where(
|
|
ClientServiceTaskInstance.assigned_to_user_id == old_staff_id
|
|
)
|
|
tasks = db.execute(task_stmt).scalars().all()
|
|
for task in tasks:
|
|
task.assigned_to_user_id = new_staff_id
|
|
task.updated_by_user_id = user.id
|
|
task.updated_at_utc = datetime.now(timezone.utc)
|
|
updated_task_count += 1
|
|
|
|
db.commit()
|
|
return RedirectResponse(
|
|
f"/services/subscriptions/{plan.id}/edit?saved=1"
|
|
f"&updated_engagements={updated_engagement_count}"
|
|
f"&updated_tasks={updated_task_count}",
|
|
status_code=303,
|
|
)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/{plan_id}")
|
|
def subscription_master_detail(request: Request, plan_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse("/login", 303)
|
|
try:
|
|
require_permission(db, user, "clients.view")
|
|
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)
|
|
engagements = db.execute(
|
|
select(ClientServiceSubscription)
|
|
.where(ClientServiceSubscription.service_plan_id == plan.id)
|
|
.order_by(
|
|
ClientServiceSubscription.financial_year.desc(),
|
|
ClientServiceSubscription.period_label.asc(),
|
|
ClientServiceSubscription.id.desc(),
|
|
)
|
|
).scalars().all()
|
|
permissions = set(get_user_permissions(db, user.id))
|
|
return templates.TemplateResponse(
|
|
"modules/services/templates/services/subscriptions/detail.html",
|
|
_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Client Service Subscription",
|
|
plan=plan,
|
|
engagements=engagements,
|
|
can_edit_subscription="clients.edit" in permissions,
|
|
scope_display=_subscription_scope_display(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
plan=plan,
|
|
),
|
|
**_registration_scope_context(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
plan=plan,
|
|
),
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|