A R R R Associates
470 lines
21 KiB
Python
470 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
from urllib.parse import quote
|
|
|
|
from fastapi import APIRouter, File, Form, Request, UploadFile
|
|
from fastapi.responses import RedirectResponse, StreamingResponse
|
|
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.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.core.rbac.permission_guard import require_permission
|
|
from app.modules.notice_cases.service import (
|
|
CASE_DOCUMENT_TYPES,
|
|
CASE_PRIORITIES,
|
|
CASE_STATUSES,
|
|
CASE_TYPES,
|
|
DEPARTMENTS,
|
|
case_dashboard_summary,
|
|
EVENT_TYPES,
|
|
HEARING_STATUSES,
|
|
ORDER_TYPES,
|
|
add_event,
|
|
add_hearing,
|
|
add_order,
|
|
can_delete_notice_case_documents,
|
|
can_manage_notice_cases,
|
|
can_upload_notice_case_documents,
|
|
case_document_absolute_path,
|
|
create_case,
|
|
get_case,
|
|
list_assignable_users,
|
|
list_cases,
|
|
list_clients_for_case,
|
|
list_engagements_for_client,
|
|
save_case_document,
|
|
update_case,
|
|
user_can_access_case,
|
|
)
|
|
from app.modules.notice_cases.models import NoticeCaseDocument
|
|
from app.modules.core.tenancy.models import FinancialYear
|
|
from app.modules.core.tenancy.year_control import redirect_if_financial_year_locked, is_row_financial_year_locked
|
|
|
|
router = APIRouter(prefix="/notice-cases", tags=["notice-cases-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),
|
|
"departments": DEPARTMENTS,
|
|
"case_types": CASE_TYPES,
|
|
"case_statuses": CASE_STATUSES,
|
|
"case_priorities": CASE_PRIORITIES,
|
|
"event_types": EVENT_TYPES,
|
|
"hearing_statuses": HEARING_STATUSES,
|
|
"order_types": ORDER_TYPES,
|
|
"case_document_types": CASE_DOCUMENT_TYPES,
|
|
}
|
|
base.update(ctx)
|
|
return base
|
|
|
|
|
|
def _render(request: Request, template_name: str, db, user, **ctx):
|
|
return templates.TemplateResponse(template_name, _base_ctx(request, user, db, **ctx))
|
|
|
|
|
|
def _redirect_denied():
|
|
return RedirectResponse(url="/system-settings", status_code=303)
|
|
|
|
|
|
def _require_user(request: Request, db, permission: str):
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return None, RedirectResponse(url="/login", status_code=303)
|
|
try:
|
|
require_permission(db, user, permission)
|
|
except Exception:
|
|
return user, _redirect_denied()
|
|
return user, None
|
|
|
|
|
|
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")
|
|
perms = set(get_user_permissions(db, user.id))
|
|
if value in (None, "", 0, "0"):
|
|
if "notice_cases.cross_branch" in perms or "notice_cases.cross_tenant" in perms:
|
|
return None
|
|
return int(getattr(user, "branch_id", 0) or 0) or None
|
|
return int(value)
|
|
|
|
|
|
def _active_financial_year(request: Request) -> str | None:
|
|
value = request.session.get("active_financial_year") or getattr(request.state, "year_code", None)
|
|
value = (value or "").strip()
|
|
return value or None
|
|
|
|
|
|
def _active_assessment_year(db, tenant_id: int, financial_year: str | None) -> str | None:
|
|
fy_code = (financial_year or "").strip()
|
|
if not fy_code:
|
|
return None
|
|
row = db.execute(
|
|
select(FinancialYear).where(
|
|
FinancialYear.tenant_id == tenant_id,
|
|
FinancialYear.year_code == fy_code,
|
|
)
|
|
).scalar_one_or_none()
|
|
return row.assessment_year if row else None
|
|
|
|
|
|
def _financial_year_options(db, tenant_id: int):
|
|
return list(
|
|
db.execute(
|
|
select(FinancialYear)
|
|
.where(FinancialYear.tenant_id == tenant_id)
|
|
.order_by(FinancialYear.start_date.desc(), FinancialYear.year_code.desc())
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def _selected_financial_year(request: Request, query_financial_year: str | None) -> str | None:
|
|
value = (query_financial_year or "").strip()
|
|
if value.lower() == "all":
|
|
return None
|
|
return value or _active_financial_year(request)
|
|
|
|
|
|
def _case_access_allowed(request: Request, db, user, case) -> bool:
|
|
return user_can_access_case(
|
|
db,
|
|
user,
|
|
case,
|
|
active_tenant_id=_active_tenant_id(request, user),
|
|
active_branch_id=_active_branch_id(request, user, db),
|
|
active_financial_year=_active_financial_year(request),
|
|
active_assessment_year=_active_assessment_year(db, _active_tenant_id(request, user), _active_financial_year(request)),
|
|
)
|
|
|
|
|
|
def _form_data(**kwargs):
|
|
return {k: v for k, v in kwargs.items()}
|
|
|
|
|
|
@router.get("")
|
|
def cases_list(
|
|
request: Request,
|
|
q: str = "",
|
|
department: str = "",
|
|
status: str = "",
|
|
financial_year: str = "",
|
|
assessment_year: str = "",
|
|
include_archived: bool = False,
|
|
):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.view")
|
|
if response:
|
|
return response
|
|
tenant_id = _active_tenant_id(request, user)
|
|
branch_id = _active_branch_id(request, user, db)
|
|
selected_fy = _selected_financial_year(request, financial_year)
|
|
selected_ay = (assessment_year or "").strip() or None
|
|
if selected_fy:
|
|
selected_ay = None
|
|
rows = list_cases(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
q=q,
|
|
department=department,
|
|
status=status,
|
|
include_archived=include_archived,
|
|
financial_year=selected_fy,
|
|
assessment_year=selected_ay,
|
|
)
|
|
return _render(
|
|
request,
|
|
"modules/notice_cases/templates/notice_cases/list.html",
|
|
db,
|
|
user,
|
|
title="Notices & Cases",
|
|
rows=rows,
|
|
summary=case_dashboard_summary(rows),
|
|
q=q,
|
|
department=department,
|
|
status=status,
|
|
financial_year=selected_fy or "all",
|
|
assessment_year=selected_ay or "",
|
|
active_financial_year=_active_financial_year(request),
|
|
financial_year_options=_financial_year_options(db, tenant_id),
|
|
include_archived=include_archived,
|
|
can_manage=can_manage_notice_cases(db, user),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/new")
|
|
def case_create_page(request: Request, client_id: int | None = None):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.create")
|
|
if response:
|
|
return response
|
|
tenant_id = _active_tenant_id(request, user)
|
|
branch_id = _active_branch_id(request, user, db)
|
|
active_fy = _active_financial_year(request)
|
|
active_ay = _active_assessment_year(db, tenant_id, active_fy)
|
|
clients = list_clients_for_case(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
engagements = list_engagements_for_client(db, tenant_id=tenant_id, client_id=client_id, financial_year=active_fy) if client_id else []
|
|
users = list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
return _render(request, "modules/notice_cases/templates/notice_cases/form.html", db, user, title="Create Notice / Case", mode="create", case=None, clients=clients, engagements=engagements, assignable_users=users, selected_client_id=client_id, form_error="", active_financial_year=active_fy, active_assessment_year=active_ay)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/new")
|
|
def case_create_submit(
|
|
request: Request,
|
|
client_id: int = Form(...),
|
|
engagement_id: str = Form(""),
|
|
department: str = Form("GST"),
|
|
case_type: str = Form("Notice"),
|
|
title: str = Form(""),
|
|
reference_no: str = Form(""),
|
|
din_ack_no: str = Form(""),
|
|
notice_date: str = Form(""),
|
|
due_date: str = Form(""),
|
|
financial_year: str = Form(""),
|
|
assessment_year: str = Form(""),
|
|
period_label: str = Form(""),
|
|
status: str = Form("open"),
|
|
priority: str = Form("normal"),
|
|
assigned_partner_user_id: str = Form(""),
|
|
assigned_manager_user_id: str = Form(""),
|
|
assigned_staff_user_id: str = Form(""),
|
|
issue_summary: str = Form(""),
|
|
remarks: str = Form(""),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.create")
|
|
if response:
|
|
return response
|
|
tenant_id = _active_tenant_id(request, user)
|
|
branch_id = _active_branch_id(request, user, db)
|
|
active_fy = _active_financial_year(request)
|
|
active_ay = _active_assessment_year(db, tenant_id, active_fy)
|
|
locked_response = redirect_if_financial_year_locked(db, tenant_id=tenant_id, year_code=active_fy, redirect_url=f"/notice-cases?financial_year={active_fy or ''}")
|
|
if locked_response:
|
|
return locked_response
|
|
data = locals().copy(); data.pop("request", None); data.pop("db", None); data.pop("user", None); data.pop("response", None); data.pop("csrf_token", None); data.pop("tenant_id", None); data.pop("branch_id", None)
|
|
data["active_financial_year"] = active_fy
|
|
data["active_assessment_year"] = active_ay
|
|
try:
|
|
row = create_case(db, tenant_id=tenant_id, branch_id=branch_id, user=user, data=data)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/notice-cases/{row.id}?created=1", status_code=303)
|
|
except Exception as exc:
|
|
db.rollback()
|
|
clients = list_clients_for_case(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
engagements = list_engagements_for_client(db, tenant_id=tenant_id, client_id=client_id, financial_year=active_fy) if client_id else []
|
|
users = list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
return _render(request, "modules/notice_cases/templates/notice_cases/form.html", db, user, title="Create Notice / Case", mode="create", case=data, clients=clients, engagements=engagements, assignable_users=users, selected_client_id=client_id, form_error=str(exc), active_financial_year=active_fy, active_assessment_year=active_ay)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/{case_id}")
|
|
def case_detail(request: Request, case_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.view")
|
|
if response:
|
|
return response
|
|
case = get_case(db, case_id)
|
|
if not case or not _case_access_allowed(request, db, user, case):
|
|
return _redirect_denied()
|
|
return _render(request, "modules/notice_cases/templates/notice_cases/detail.html", db, user, title=case.case_code, case=case, can_manage=can_manage_notice_cases(db, user), can_upload=can_upload_notice_case_documents(db, user), can_delete_documents=can_delete_notice_case_documents(db, user))
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/{case_id}/edit")
|
|
def case_edit_page(request: Request, case_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.edit")
|
|
if response:
|
|
return response
|
|
case = get_case(db, case_id)
|
|
if not case or not _case_access_allowed(request, db, user, case):
|
|
return _redirect_denied()
|
|
clients = [case.client]
|
|
active_fy = _active_financial_year(request)
|
|
active_ay = _active_assessment_year(db, case.tenant_id, active_fy)
|
|
engagements = list_engagements_for_client(db, tenant_id=case.tenant_id, client_id=case.client_id, financial_year=active_fy)
|
|
users = list_assignable_users(db, tenant_id=case.tenant_id, branch_id=case.branch_id)
|
|
return _render(request, "modules/notice_cases/templates/notice_cases/form.html", db, user, title="Edit Notice / Case", mode="edit", case=case, clients=clients, engagements=engagements, assignable_users=users, selected_client_id=case.client_id, form_error="", active_financial_year=active_fy, active_assessment_year=active_ay)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{case_id}/edit")
|
|
def case_edit_submit(request: Request, case_id: int, department: str = Form("GST"), case_type: str = Form("Notice"), title: str = Form(""), reference_no: str = Form(""), din_ack_no: str = Form(""), notice_date: str = Form(""), due_date: str = Form(""), financial_year: str = Form(""), assessment_year: str = Form(""), period_label: str = Form(""), status: str = Form("open"), priority: str = Form("normal"), assigned_partner_user_id: str = Form(""), assigned_manager_user_id: str = Form(""), assigned_staff_user_id: str = Form(""), issue_summary: str = Form(""), remarks: str = Form(""), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.edit")
|
|
if response:
|
|
return response
|
|
case = get_case(db, case_id)
|
|
if not case or not _case_access_allowed(request, db, user, case):
|
|
return _redirect_denied()
|
|
if is_row_financial_year_locked(db, case):
|
|
return RedirectResponse(url=f"/notice-cases/{case.id}?year_locked=1", status_code=303)
|
|
active_fy = _active_financial_year(request)
|
|
active_ay = _active_assessment_year(db, case.tenant_id, active_fy)
|
|
data = locals().copy(); [data.pop(k, None) for k in ["request","db","user","response","case","csrf_token","case_id"]]
|
|
data["active_financial_year"] = active_fy
|
|
data["active_assessment_year"] = active_ay
|
|
try:
|
|
update_case(db, case=case, user=user, data=data)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/notice-cases/{case.id}?updated=1", status_code=303)
|
|
except Exception as exc:
|
|
db.rollback()
|
|
clients = [case.client]
|
|
engagements = list_engagements_for_client(db, tenant_id=case.tenant_id, client_id=case.client_id, financial_year=active_fy)
|
|
users = list_assignable_users(db, tenant_id=case.tenant_id, branch_id=case.branch_id)
|
|
return _render(request, "modules/notice_cases/templates/notice_cases/form.html", db, user, title="Edit Notice / Case", mode="edit", case=case, clients=clients, engagements=engagements, assignable_users=users, selected_client_id=case.client_id, form_error=str(exc), active_financial_year=active_fy, active_assessment_year=active_ay)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{case_id}/events")
|
|
def case_add_event(request: Request, case_id: int, event_type: str = Form("Internal Note"), event_date: str = Form(""), description: str = Form(""), next_due_date: str = Form(""), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.events.manage")
|
|
if response:
|
|
return response
|
|
case = get_case(db, case_id)
|
|
if not case or not _case_access_allowed(request, db, user, case):
|
|
return _redirect_denied()
|
|
if is_row_financial_year_locked(db, case):
|
|
return RedirectResponse(url=f"/notice-cases/{case.id}?year_locked=1", status_code=303)
|
|
add_event(db, case=case, user=user, data=locals())
|
|
db.commit()
|
|
return RedirectResponse(url=f"/notice-cases/{case_id}?event_added=1", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{case_id}/hearings")
|
|
def case_add_hearing(request: Request, case_id: int, hearing_date: str = Form(""), hearing_time: str = Form(""), venue_or_mode: str = Form(""), officer_name: str = Form(""), agenda: str = Form(""), outcome: str = Form(""), status: str = Form("scheduled"), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.hearings.manage")
|
|
if response:
|
|
return response
|
|
case = get_case(db, case_id)
|
|
if not case or not _case_access_allowed(request, db, user, case):
|
|
return _redirect_denied()
|
|
if is_row_financial_year_locked(db, case):
|
|
return RedirectResponse(url=f"/notice-cases/{case.id}?year_locked=1", status_code=303)
|
|
add_hearing(db, case=case, user=user, data=locals())
|
|
db.commit()
|
|
return RedirectResponse(url=f"/notice-cases/{case_id}?hearing_added=1", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{case_id}/orders")
|
|
def case_add_order(request: Request, case_id: int, order_type: str = Form("Other"), order_no: str = Form(""), order_date: str = Form(""), demand_amount: str = Form("0"), tax_amount: str = Form("0"), interest_amount: str = Form("0"), penalty_amount: str = Form("0"), summary: str = Form(""), appeal_due_date: str = Form(""), appeal_filed: str | None = Form(None), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.orders.manage")
|
|
if response:
|
|
return response
|
|
case = get_case(db, case_id)
|
|
if not case or not _case_access_allowed(request, db, user, case):
|
|
return _redirect_denied()
|
|
if is_row_financial_year_locked(db, case):
|
|
return RedirectResponse(url=f"/notice-cases/{case.id}?year_locked=1", status_code=303)
|
|
add_order(db, case=case, user=user, data=locals())
|
|
db.commit()
|
|
return RedirectResponse(url=f"/notice-cases/{case_id}?order_added=1", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{case_id}/documents/upload")
|
|
def case_document_upload(request: Request, case_id: int, title: str = Form(""), document_type: str = Form("OTHER"), description: str = Form(""), event_id: str = Form(""), file: UploadFile = File(...), csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.documents.upload")
|
|
if response:
|
|
return response
|
|
case = get_case(db, case_id)
|
|
if not case or not _case_access_allowed(request, db, user, case):
|
|
return _redirect_denied()
|
|
if is_row_financial_year_locked(db, case):
|
|
return RedirectResponse(url=f"/notice-cases/{case.id}?year_locked=1", status_code=303)
|
|
if not file or not file.filename:
|
|
return RedirectResponse(url=f"/notice-cases/{case_id}?error=missing_file", status_code=303)
|
|
save_case_document(db, case=case, upload_file=file, user=user, title=title, document_type=document_type, description=description, event_id=int(event_id) if event_id else None)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/notice-cases/{case_id}?uploaded=1", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/documents/{document_id}/download")
|
|
def case_document_download(request: Request, document_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.documents.download")
|
|
if response:
|
|
return response
|
|
doc = db.execute(select(NoticeCaseDocument).where(NoticeCaseDocument.id == document_id, NoticeCaseDocument.is_deleted.is_(False))).scalar_one_or_none()
|
|
case = get_case(db, doc.case_id) if doc else None
|
|
if not doc or not case or not _case_access_allowed(request, db, user, case):
|
|
return _redirect_denied()
|
|
path = case_document_absolute_path(doc)
|
|
if not path.exists():
|
|
return RedirectResponse(url=f"/notice-cases/{doc.case_id}?error=file_missing", status_code=303)
|
|
quoted = quote(doc.original_filename)
|
|
headers = {"Content-Disposition": f"attachment; filename*=UTF-8''{quoted}"}
|
|
return StreamingResponse(path.open("rb"), media_type=doc.content_type or "application/octet-stream", headers=headers)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/documents/{document_id}/delete")
|
|
def case_document_delete(request: Request, document_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db, "notice_cases.documents.delete")
|
|
if response:
|
|
return response
|
|
doc = db.execute(select(NoticeCaseDocument).where(NoticeCaseDocument.id == document_id, NoticeCaseDocument.is_deleted.is_(False))).scalar_one_or_none()
|
|
case = get_case(db, doc.case_id) if doc else None
|
|
if not doc or not case or not _case_access_allowed(request, db, user, case):
|
|
return _redirect_denied()
|
|
doc.is_deleted = True
|
|
doc.deleted_at_utc = __import__('datetime').datetime.now(__import__('datetime').timezone.utc)
|
|
doc.deleted_by_user_id = user.id
|
|
db.commit()
|
|
return RedirectResponse(url=f"/notice-cases/{doc.case_id}?deleted=1", status_code=303)
|
|
finally:
|
|
db.close()
|