A R R R Associates
469 lines
18 KiB
Python
469 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, timezone
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import 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.services.client_services import (
|
|
SUBSCRIPTION_STATUSES,
|
|
list_assignable_users,
|
|
list_review_partners,
|
|
parse_date,
|
|
)
|
|
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 _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()
|
|
return templates.TemplateResponse(
|
|
"modules/services/templates/services/subscriptions/list.html",
|
|
_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Client Service Subscriptions",
|
|
rows=rows,
|
|
q=q,
|
|
include_inactive=include_inactive,
|
|
),
|
|
)
|
|
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,
|
|
),
|
|
)
|
|
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,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|