A R R R Associates
1496 lines
60 KiB
Python
1496 lines
60 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, time, timezone
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
import re
|
|
|
|
from fastapi import APIRouter, File, Form, Request, UploadFile
|
|
from fastapi.responses import FileResponse, 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.core.audit.service import model_snapshot, pair_before_after, write_audit_log
|
|
from app.modules.core.tenancy.models import Branch, FinancialYear, Tenant, YearBackupExport
|
|
from app.modules.core.tenancy.settings_models import BranchSettings
|
|
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.iam.scope import build_scope, list_visible_branches, list_visible_tenants
|
|
from app.modules.core.tenancy.services import build_branches_payload, build_tenants_payload
|
|
from app.modules.system_settings.year_backup_service import build_year_backup_export
|
|
|
|
router = APIRouter(prefix="/system-settings", tags=["system-settings-ui"])
|
|
|
|
|
|
def _csrf_rejected(request: Request):
|
|
from app.core.http_responses import forbidden_response
|
|
return forbidden_response(request, "CSRF validation failed")
|
|
|
|
|
|
|
|
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),
|
|
}
|
|
base.update(ctx)
|
|
return base
|
|
|
|
|
|
def _redirect_denied(default_url: str = "/system-settings"):
|
|
from app.core.http_responses import ui_access_denied
|
|
return ui_access_denied()
|
|
|
|
|
|
def _render_with_user(request: Request, template: str, ctx: dict, status_code: int = 200):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
return templates.TemplateResponse(template, _base_ctx(request, user, db, **ctx), status_code=status_code)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _parse_time(s: str | None) -> time | None:
|
|
if not s:
|
|
return None
|
|
s = s.strip()
|
|
if not s:
|
|
return None
|
|
hh, mm = s.split(":")
|
|
return time(int(hh), int(mm))
|
|
|
|
def _parse_date(s: str | None) -> date | None:
|
|
s = (s or "").strip()
|
|
if not s:
|
|
return None
|
|
return date.fromisoformat(s)
|
|
|
|
|
|
def _assessment_year_from_fy(year_code: str) -> str:
|
|
try:
|
|
start_year = int((year_code or "").split("-", 1)[0])
|
|
except Exception:
|
|
return ""
|
|
end_year = start_year + 1
|
|
return f"{end_year}-{str(end_year + 1)[-2:]}"
|
|
|
|
|
|
def _default_dates_from_fy(year_code: str) -> tuple[date | None, date | None]:
|
|
try:
|
|
start_year = int((year_code or "").split("-", 1)[0])
|
|
return date(start_year, 4, 1), date(start_year + 1, 3, 31)
|
|
except Exception:
|
|
return None, None
|
|
|
|
|
|
def _active_tenant_id_for_settings(request: Request, user) -> int:
|
|
return int(request.session.get("active_tenant_id") or getattr(user, "tenant_id", 0) or 0)
|
|
|
|
|
|
def _store_active_tenant_context(request: Request, tenant: Tenant | None) -> None:
|
|
if not tenant:
|
|
request.session.pop("active_tenant_id", None)
|
|
request.session.pop("active_tenant_code", None)
|
|
return
|
|
request.session["active_tenant_id"] = int(tenant.id)
|
|
request.session["active_tenant_code"] = tenant.code
|
|
|
|
|
|
def _store_active_branch_context(request: Request, branch: Branch | None) -> None:
|
|
if not branch:
|
|
request.session.pop("active_branch_id", None)
|
|
request.session.pop("active_branch_code", None)
|
|
return
|
|
request.session["active_branch_id"] = int(branch.id)
|
|
request.session["active_branch_code"] = branch.code
|
|
|
|
|
|
def _can_manage_financial_years(db, user) -> bool:
|
|
roles = set(get_user_roles(db, user.id))
|
|
perms = set(get_user_permissions(db, user.id))
|
|
return "System Admin" in roles or "Firm Admin" in roles or "system.settings.edit" in perms
|
|
|
|
|
|
def _can_view_financial_years(db, user) -> bool:
|
|
roles = set(get_user_roles(db, user.id))
|
|
perms = set(get_user_permissions(db, user.id))
|
|
return bool(roles.intersection({"System Admin", "Firm Admin", "Partner", "Branch Manager"})) or "system.settings.view" in perms
|
|
|
|
|
|
def _visible_financial_year_tenant_ids(db, user) -> set[int]:
|
|
roles = set(get_user_roles(db, user.id))
|
|
if "System Admin" in roles:
|
|
scope = build_scope(db, user)
|
|
return {int(t.id) for t in list_visible_tenants(db, scope)}
|
|
return {int(user.tenant_id)} if getattr(user, "tenant_id", None) else set()
|
|
|
|
|
|
def _current_financial_year(db, tenant_id: int) -> FinancialYear | None:
|
|
return db.execute(
|
|
select(FinancialYear).where(
|
|
FinancialYear.tenant_id == tenant_id,
|
|
FinancialYear.is_current.is_(True),
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def _financial_year_redirect_back(request: Request) -> RedirectResponse:
|
|
return RedirectResponse(url=request.headers.get("referer") or "/system-settings/financial-years", status_code=303)
|
|
|
|
|
|
def _is_system_admin(db, user) -> bool:
|
|
return "System Admin" in get_user_roles(db, user.id)
|
|
|
|
|
|
def _is_firm_admin(db, user) -> bool:
|
|
return "Firm Admin" in get_user_roles(db, user.id)
|
|
|
|
|
|
@router.get("")
|
|
def dashboard(request: Request):
|
|
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, "system.settings.view")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
finally:
|
|
db.close()
|
|
return _render_with_user(
|
|
request,
|
|
"modules/system_settings/templates/dashboard.html",
|
|
{"title": "System Settings"},
|
|
)
|
|
|
|
|
|
# -----------------------------
|
|
# Tenant - System Admin only
|
|
# -----------------------------
|
|
|
|
|
|
# Phase 3 security hardening: list/index pages must not accept direct unsafe POSTs.
|
|
# Existing create/update features continue to use their dedicated /new or action routes.
|
|
@router.post("/tenants")
|
|
def tenants_list_post_rejected(request: Request):
|
|
return _csrf_rejected(request)
|
|
|
|
|
|
@router.post("/branches")
|
|
def branches_list_post_rejected(request: Request):
|
|
return _csrf_rejected(request)
|
|
|
|
|
|
@router.post("/financial-years")
|
|
def financial_years_list_post_rejected(request: Request):
|
|
return _csrf_rejected(request)
|
|
|
|
|
|
@router.get("/tenants")
|
|
def tenants_list(request: Request, q: str = "", page: int = 1, per_page: int = 10):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
|
|
payload = build_tenants_payload(db, build_scope(db, user), q=q, page=page, per_page=per_page)
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/tenants_list.html",
|
|
_base_ctx(request, user, db, title="Tenants", **payload),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/tenants/new")
|
|
def tenant_create_page(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
finally:
|
|
db.close()
|
|
return _render_with_user(
|
|
request,
|
|
"modules/system_settings/templates/tenant_create.html",
|
|
{"title": "Create Tenant"},
|
|
)
|
|
|
|
|
|
@router.post("/tenants/new")
|
|
def tenant_create_submit(
|
|
request: Request,
|
|
code: str = Form(...),
|
|
name: str = Form(...),
|
|
firm_type: str = Form("proprietorship"),
|
|
default_timezone: str = Form("Asia/Kolkata"),
|
|
default_session_duration_minutes: int = Form(480),
|
|
default_otp_required_roles_csv: str = Form("System Admin"),
|
|
default_storage_mode: str = Form("local_only"),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
|
|
tenant = Tenant(
|
|
code=code.strip(),
|
|
name=name.strip(),
|
|
is_active=True,
|
|
firm_type=(firm_type or "partnership").strip(),
|
|
default_timezone=(default_timezone or "Asia/Kolkata").strip(),
|
|
default_session_duration_minutes=default_session_duration_minutes or 480,
|
|
default_otp_required_roles_csv=(default_otp_required_roles_csv or "System Admin").strip(),
|
|
default_storage_mode=(default_storage_mode or "local_only").strip(),
|
|
)
|
|
db.add(tenant)
|
|
db.commit()
|
|
db.refresh(tenant)
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="tenant.create",
|
|
entity_type="tenant",
|
|
actor=user,
|
|
request=request,
|
|
entity_id=tenant.id,
|
|
entity_name=tenant.name,
|
|
target_tenant_id=tenant.id,
|
|
details={"after": model_snapshot(tenant, ["code", "name", "firm_type", "is_active"])},
|
|
)
|
|
return RedirectResponse(url="/system-settings/tenants", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/tenants/{tenant_id}/edit")
|
|
def tenant_edit_page(request: Request, tenant_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
|
|
tenant = db.execute(select(Tenant).where(Tenant.id == tenant_id)).scalar_one_or_none()
|
|
if not tenant:
|
|
return RedirectResponse(url="/system-settings/tenants", status_code=303)
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/tenant_edit.html",
|
|
_base_ctx(request, user, db, tenant=tenant, title="Edit Tenant"),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/tenants/{tenant_id}/edit")
|
|
def tenant_edit_submit(
|
|
request: Request,
|
|
tenant_id: int,
|
|
name: str = Form(...),
|
|
firm_type: str = Form("proprietorship"),
|
|
default_timezone: str = Form("Asia/Kolkata"),
|
|
default_session_duration_minutes: int = Form(480),
|
|
default_otp_required_roles_csv: str = Form("System Admin"),
|
|
default_storage_mode: str = Form("local_only"),
|
|
is_active: str | None = Form(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _is_system_admin(db, user):
|
|
return _redirect_denied()
|
|
|
|
tenant = db.execute(select(Tenant).where(Tenant.id == tenant_id)).scalar_one_or_none()
|
|
if not tenant:
|
|
return RedirectResponse(url="/system-settings/tenants", status_code=303)
|
|
|
|
before_snapshot = model_snapshot(tenant, ["code", "name", "firm_type", "default_timezone", "default_session_duration_minutes", "default_otp_required_roles_csv", "default_storage_mode", "is_active"])
|
|
tenant.name = name.strip()
|
|
tenant.firm_type = (firm_type or "partnership").strip()
|
|
tenant.default_timezone = (default_timezone or "Asia/Kolkata").strip()
|
|
tenant.default_session_duration_minutes = default_session_duration_minutes or 480
|
|
tenant.default_otp_required_roles_csv = (default_otp_required_roles_csv or "System Admin").strip()
|
|
tenant.default_storage_mode = (default_storage_mode or "local_only").strip()
|
|
tenant.is_active = is_active is not None
|
|
db.commit()
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="tenant.update",
|
|
entity_type="tenant",
|
|
actor=user,
|
|
request=request,
|
|
entity_id=tenant.id,
|
|
entity_name=tenant.name,
|
|
target_tenant_id=tenant.id,
|
|
details=pair_before_after(before_snapshot, model_snapshot(tenant, ["code", "name", "firm_type", "default_timezone", "default_session_duration_minutes", "default_otp_required_roles_csv", "default_storage_mode", "is_active"])),
|
|
)
|
|
return RedirectResponse(url="/system-settings/tenants", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# -----------------------------
|
|
# Branch
|
|
# -----------------------------
|
|
@router.get("/branches")
|
|
def branches_list(request: Request, q: str = "", page: int = 1, per_page: int = 10, tenant_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, "system.settings.view")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
scope = build_scope(db, user)
|
|
payload = build_branches_payload(db, scope, q=q, page=page, per_page=per_page, tenant_id=tenant_id)
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/branches_list.html",
|
|
_base_ctx(request, user, db, title="Branches", **payload),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/branches/new")
|
|
def branch_create_page(request: Request, tenant_id: int | None = None):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
roles = get_user_roles(db, user.id)
|
|
if "System Admin" not in roles and "Firm Admin" not in roles:
|
|
return _redirect_denied()
|
|
|
|
if "System Admin" in roles:
|
|
tenants = db.execute(select(Tenant).where(Tenant.is_active.is_(True)).order_by(Tenant.name)).scalars().all()
|
|
selected_tenant_id = tenant_id
|
|
can_change_tenant = True
|
|
else:
|
|
tenants = db.execute(select(Tenant).where(Tenant.id == user.tenant_id)).scalars().all()
|
|
selected_tenant_id = user.tenant_id
|
|
can_change_tenant = False
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/branch_create.html",
|
|
_base_ctx(
|
|
request,
|
|
user,
|
|
db,
|
|
tenants=tenants,
|
|
selected_tenant_id=selected_tenant_id,
|
|
can_change_tenant=can_change_tenant,
|
|
title="Create Branch",
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/branches/new")
|
|
def branch_create_submit(
|
|
request: Request,
|
|
tenant_id: int = Form(...),
|
|
code: str = Form(...),
|
|
name: str = Form(...),
|
|
timezone: str = Form("Asia/Kolkata"),
|
|
office_start_time: str = Form(""),
|
|
office_end_time: str = Form(""),
|
|
smtp_host: str = Form(""),
|
|
smtp_port: str = Form(""),
|
|
smtp_username: str = Form(""),
|
|
smtp_password: str = Form(""),
|
|
smtp_use_tls: str | None = Form(None),
|
|
local_storage_path: str = Form(""),
|
|
address_line1: str = Form(""),
|
|
address_line2: str = Form(""),
|
|
city: str = Form(""),
|
|
state: str = Form(""),
|
|
pin_code: str = Form(""),
|
|
gstin: str = Form(""),
|
|
pan: str = Form(""),
|
|
geo_address: str = Form(""),
|
|
latitude: str = Form(""),
|
|
longitude: str = Form(""),
|
|
attendance_geo_enabled: str | None = Form(None),
|
|
attendance_geo_radius_meters: str = Form("100"),
|
|
attendance_grace_minutes: str = Form("10"),
|
|
attendance_half_day_after_time: str = Form(""),
|
|
attendance_rule_enabled: str | None = Form(None),
|
|
attendance_ip_enabled: str | None = Form(None),
|
|
attendance_allowed_ip_csv: str = Form(""),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
roles = get_user_roles(db, user.id)
|
|
if "System Admin" not in roles and "Firm Admin" not in roles:
|
|
return _redirect_denied()
|
|
|
|
if "Firm Admin" in roles and tenant_id != user.tenant_id:
|
|
return _redirect_denied()
|
|
|
|
tenant = db.execute(select(Tenant).where(Tenant.id == tenant_id)).scalar_one_or_none()
|
|
if not tenant:
|
|
return RedirectResponse(url="/system-settings/branches", status_code=303)
|
|
|
|
existing = db.execute(
|
|
select(Branch).where(Branch.tenant_id == tenant_id, Branch.code == code.strip())
|
|
).scalar_one_or_none()
|
|
|
|
if existing:
|
|
if "System Admin" in roles:
|
|
tenants = db.execute(select(Tenant).where(Tenant.is_active.is_(True)).order_by(Tenant.name)).scalars().all()
|
|
can_change_tenant = True
|
|
else:
|
|
tenants = db.execute(select(Tenant).where(Tenant.id == user.tenant_id)).scalars().all()
|
|
can_change_tenant = False
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/branch_create.html",
|
|
_base_ctx(
|
|
request,
|
|
user,
|
|
db,
|
|
tenants=tenants,
|
|
selected_tenant_id=tenant_id,
|
|
can_change_tenant=can_change_tenant,
|
|
title="Create Branch",
|
|
flash="Branch code already exists for this tenant.",
|
|
),
|
|
status_code=400,
|
|
)
|
|
|
|
branch = Branch(
|
|
tenant_id=tenant_id,
|
|
code=code.strip(),
|
|
name=name.strip(),
|
|
timezone=(timezone or "Asia/Kolkata").strip(),
|
|
office_start_time=_parse_time(office_start_time),
|
|
office_end_time=_parse_time(office_end_time),
|
|
smtp_host=smtp_host or None,
|
|
smtp_port=int(smtp_port) if str(smtp_port).strip() else None,
|
|
smtp_username=smtp_username or None,
|
|
smtp_password=smtp_password or None,
|
|
smtp_use_tls=smtp_use_tls is not None,
|
|
local_storage_path=local_storage_path or None,
|
|
is_active=True,
|
|
)
|
|
db.add(branch)
|
|
db.commit()
|
|
db.refresh(branch)
|
|
|
|
settings = BranchSettings(
|
|
branch_id=branch.id,
|
|
address_line1=address_line1 or None,
|
|
address_line2=address_line2 or None,
|
|
city=city or None,
|
|
state=state or None,
|
|
pin_code=pin_code or None,
|
|
gstin=gstin or None,
|
|
pan=pan or None,
|
|
geo_address=geo_address or None,
|
|
latitude=float(latitude) if str(latitude).strip() else None,
|
|
longitude=float(longitude) if str(longitude).strip() else None,
|
|
attendance_geo_enabled=attendance_geo_enabled is not None,
|
|
attendance_geo_radius_meters=int(attendance_geo_radius_meters) if str(attendance_geo_radius_meters).strip() else 100,
|
|
attendance_grace_minutes=int(attendance_grace_minutes) if str(attendance_grace_minutes).strip() else 10,
|
|
attendance_half_day_after_time=_parse_time(attendance_half_day_after_time),
|
|
attendance_rule_enabled=attendance_rule_enabled is not None,
|
|
attendance_ip_enabled=attendance_ip_enabled is not None,
|
|
attendance_allowed_ip_csv=attendance_allowed_ip_csv or None,
|
|
)
|
|
db.add(settings)
|
|
db.commit()
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="branch.create",
|
|
entity_type="branch",
|
|
actor=user,
|
|
request=request,
|
|
entity_id=branch.id,
|
|
entity_name=branch.name,
|
|
target_tenant_id=branch.tenant_id,
|
|
target_branch_id=branch.id,
|
|
details={
|
|
"branch": model_snapshot(branch, ["tenant_id", "code", "name", "is_active", "timezone", "allow_login", "allow_new_assignments", "is_head_office"]),
|
|
"settings": model_snapshot(settings, ["city", "state", "pin_code", "gstin", "pan"]),
|
|
},
|
|
)
|
|
|
|
return RedirectResponse(url="/system-settings/branches", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/branches/{branch_id}/edit")
|
|
def branch_edit_page(request: Request, branch_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
roles = get_user_roles(db, user.id)
|
|
if "System Admin" not in roles and "Firm Admin" not in roles:
|
|
return _redirect_denied()
|
|
|
|
branch = db.execute(select(Branch).where(Branch.id == branch_id)).scalar_one_or_none()
|
|
if not branch:
|
|
return RedirectResponse(url="/system-settings/branches", status_code=303)
|
|
|
|
if "Firm Admin" in roles and branch.tenant_id != user.tenant_id:
|
|
return _redirect_denied()
|
|
|
|
settings = db.execute(select(BranchSettings).where(BranchSettings.branch_id == branch.id)).scalar_one_or_none()
|
|
if not settings:
|
|
settings = BranchSettings(branch_id=branch.id)
|
|
db.add(settings)
|
|
db.commit()
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/branch_edit.html",
|
|
_base_ctx(request, user, db, branch=branch, settings=settings, title="Edit Branch"),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/branches/{branch_id}/edit")
|
|
def branch_edit_submit(
|
|
request: Request,
|
|
branch_id: int,
|
|
csrf_token: str = Form(...),
|
|
name: str = Form(...),
|
|
timezone: str = Form("Asia/Kolkata"),
|
|
office_start_time: str = Form(""),
|
|
office_end_time: str = Form(""),
|
|
smtp_host: str = Form(""),
|
|
smtp_port: str = Form(""),
|
|
smtp_username: str = Form(""),
|
|
smtp_password: str = Form(""),
|
|
smtp_use_tls: str | None = Form(None),
|
|
local_storage_path: str = Form(""),
|
|
address_line1: str = Form(""),
|
|
address_line2: str = Form(""),
|
|
city: str = Form(""),
|
|
state: str = Form(""),
|
|
pin_code: str = Form(""),
|
|
gstin: str = Form(""),
|
|
pan: str = Form(""),
|
|
geo_address: str = Form(""),
|
|
latitude: str = Form(""),
|
|
longitude: str = Form(""),
|
|
attendance_geo_enabled: str | None = Form(None),
|
|
attendance_geo_radius_meters: str = Form("100"),
|
|
attendance_grace_minutes: str = Form("10"),
|
|
attendance_half_day_after_time: str = Form(""),
|
|
attendance_rule_enabled: str | None = Form(None),
|
|
attendance_ip_enabled: str | None = Form(None),
|
|
attendance_allowed_ip_csv: str = Form(""),
|
|
letterhead_logo_path: str = Form(""),
|
|
letterhead_signature_path: str = Form(""),
|
|
letterhead_stamp_path: str = Form(""),
|
|
working_days_csv: str = Form("MON,TUE,WED,THU,FRI,SAT"),
|
|
holidays_json: str = Form("[]"),
|
|
timezone_locked: str | None = Form(None),
|
|
email_from_name: str = Form(""),
|
|
email_from_email: str = Form(""),
|
|
email_reply_to: str = Form(""),
|
|
default_cc_csv: str = Form(""),
|
|
default_bcc_csv: str = Form(""),
|
|
email_signature_html: str = Form(""),
|
|
storage_mode: str = Form("local_only"),
|
|
folder_template: str = Form("{root}/Clients/{client_code}/{fy}/{service}/"),
|
|
max_file_mb: str = Form("25"),
|
|
allowed_ext_csv: str = Form("pdf,jpg,jpeg,png,xlsx,xls,docx,zip"),
|
|
retention_years: str = Form("8"),
|
|
otp_required_roles_csv: str = Form("System Admin"),
|
|
session_duration_minutes: str = Form("480"),
|
|
lockout_attempts: str = Form("5"),
|
|
lockout_minutes: str = Form("15"),
|
|
):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
roles = get_user_roles(db, user.id)
|
|
if "System Admin" not in roles and "Firm Admin" not in roles:
|
|
return _redirect_denied()
|
|
|
|
branch = db.execute(select(Branch).where(Branch.id == branch_id)).scalar_one_or_none()
|
|
if not branch:
|
|
return RedirectResponse(url="/system-settings/branches", status_code=303)
|
|
|
|
if "Firm Admin" in roles and branch.tenant_id != user.tenant_id:
|
|
return _redirect_denied()
|
|
|
|
bs = db.execute(select(BranchSettings).where(BranchSettings.branch_id == branch.id)).scalar_one_or_none()
|
|
|
|
tz_locked_now = bool(bs.timezone_locked) if bs else False
|
|
before_branch = model_snapshot(branch, ["tenant_id", "code", "name", "is_active", "timezone", "allow_login", "allow_new_assignments", "is_head_office", "smtp_host", "smtp_port", "smtp_username", "smtp_use_tls", "local_storage_path"])
|
|
before_settings = model_snapshot(bs, ["address_line1", "address_line2", "city", "state", "pin_code", "gstin", "pan", "geo_address", "latitude", "longitude", "attendance_geo_enabled", "attendance_geo_radius_meters", "attendance_grace_minutes", "attendance_half_day_after_time", "attendance_rule_enabled", "attendance_ip_enabled", "attendance_allowed_ip_csv", "working_days_csv", "holidays_json", "timezone_locked", "email_from_name", "email_from_email", "email_reply_to", "default_cc_csv", "default_bcc_csv", "storage_mode", "folder_template", "max_file_mb", "allowed_ext_csv", "retention_years", "otp_required_roles_csv", "session_duration_minutes", "lockout_attempts", "lockout_minutes"]) if bs else {}
|
|
|
|
branch.name = name.strip()
|
|
if not tz_locked_now:
|
|
branch.timezone = (timezone or "Asia/Kolkata").strip()
|
|
branch.office_start_time = _parse_time(office_start_time)
|
|
branch.office_end_time = _parse_time(office_end_time)
|
|
branch.smtp_host = smtp_host or None
|
|
branch.smtp_port = int(smtp_port) if str(smtp_port).strip() else None
|
|
branch.smtp_username = smtp_username or None
|
|
branch.smtp_password = smtp_password or None
|
|
branch.smtp_use_tls = smtp_use_tls is not None
|
|
branch.local_storage_path = local_storage_path or None
|
|
|
|
settings = bs
|
|
if not settings:
|
|
settings = BranchSettings(branch_id=branch.id)
|
|
db.add(settings)
|
|
|
|
settings.address_line1 = address_line1 or None
|
|
settings.address_line2 = address_line2 or None
|
|
settings.city = city or None
|
|
settings.state = state or None
|
|
settings.pin_code = pin_code or None
|
|
settings.gstin = gstin or None
|
|
settings.pan = pan or None
|
|
settings.geo_address = geo_address or None
|
|
settings.latitude = float(latitude) if str(latitude).strip() else None
|
|
settings.longitude = float(longitude) if str(longitude).strip() else None
|
|
settings.attendance_geo_enabled = attendance_geo_enabled is not None
|
|
settings.attendance_geo_radius_meters = int(attendance_geo_radius_meters) if str(attendance_geo_radius_meters).strip() else 100
|
|
settings.attendance_grace_minutes = int(attendance_grace_minutes) if str(attendance_grace_minutes).strip() else 10
|
|
settings.attendance_half_day_after_time = _parse_time(attendance_half_day_after_time)
|
|
settings.attendance_rule_enabled = attendance_rule_enabled is not None
|
|
settings.attendance_ip_enabled = attendance_ip_enabled is not None
|
|
settings.attendance_allowed_ip_csv = attendance_allowed_ip_csv or None
|
|
settings.letterhead_logo_path = letterhead_logo_path or None
|
|
settings.letterhead_signature_path = letterhead_signature_path or None
|
|
settings.letterhead_stamp_path = letterhead_stamp_path or None
|
|
settings.working_days_csv = (working_days_csv or "MON,TUE,WED,THU,FRI,SAT").strip()
|
|
settings.holidays_json = (holidays_json or "[]").strip()
|
|
settings.timezone_locked = timezone_locked is not None
|
|
settings.email_from_name = email_from_name or None
|
|
settings.email_from_email = email_from_email or None
|
|
settings.email_reply_to = email_reply_to or None
|
|
settings.default_cc_csv = default_cc_csv or None
|
|
settings.default_bcc_csv = default_bcc_csv or None
|
|
settings.email_signature_html = email_signature_html or None
|
|
settings.storage_mode = (storage_mode or "local_only").strip()
|
|
settings.folder_template = (folder_template or "{root}/Clients/{client_code}/{fy}/{service}/").strip()
|
|
settings.max_file_mb = int(max_file_mb) if str(max_file_mb).strip() else 25
|
|
settings.allowed_ext_csv = (allowed_ext_csv or "pdf,jpg,jpeg,png,xlsx,xls,docx,zip").strip()
|
|
settings.retention_years = int(retention_years) if str(retention_years).strip() else 8
|
|
settings.otp_required_roles_csv = (otp_required_roles_csv or "System Admin").strip()
|
|
settings.session_duration_minutes = int(session_duration_minutes) if str(session_duration_minutes).strip() else 480
|
|
settings.lockout_attempts = int(lockout_attempts) if str(lockout_attempts).strip() else 5
|
|
settings.lockout_minutes = int(lockout_minutes) if str(lockout_minutes).strip() else 15
|
|
|
|
db.commit()
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="branch.update",
|
|
entity_type="branch",
|
|
actor=user,
|
|
request=request,
|
|
entity_id=branch.id,
|
|
entity_name=branch.name,
|
|
target_tenant_id=branch.tenant_id,
|
|
target_branch_id=branch.id,
|
|
details={
|
|
"branch": pair_before_after(before_branch, model_snapshot(branch, ["tenant_id", "code", "name", "is_active", "timezone", "allow_login", "allow_new_assignments", "is_head_office", "smtp_host", "smtp_port", "smtp_username", "smtp_use_tls", "local_storage_path"])),
|
|
"settings": pair_before_after(before_settings, model_snapshot(settings, ["address_line1", "address_line2", "city", "state", "pin_code", "gstin", "pan", "geo_address", "latitude", "longitude", "attendance_geo_enabled", "attendance_geo_radius_meters", "attendance_grace_minutes", "attendance_half_day_after_time", "attendance_rule_enabled", "attendance_ip_enabled", "attendance_allowed_ip_csv", "working_days_csv", "holidays_json", "timezone_locked", "email_from_name", "email_from_email", "email_reply_to", "default_cc_csv", "default_bcc_csv", "storage_mode", "folder_template", "max_file_mb", "allowed_ext_csv", "retention_years", "otp_required_roles_csv", "session_duration_minutes", "lockout_attempts", "lockout_minutes"])),
|
|
},
|
|
)
|
|
return RedirectResponse(url="/system-settings/branches", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# -----------------------------
|
|
# Phase 7Q.2 - Firm Branding Settings
|
|
# -----------------------------
|
|
BRANDING_UPLOAD_ROOT = Path("/app/data/storage/uploads/branding")
|
|
BRANDING_PUBLIC_PREFIX = "/storage/uploads/branding"
|
|
_ALLOWED_BRANDING_SUFFIXES = {".png", ".jpg", ".jpeg", ".webp", ".ico", ".svg"}
|
|
_HEX_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{6}$")
|
|
|
|
|
|
def _clean_optional(value: str | None) -> str | None:
|
|
value = (value or "").strip()
|
|
return value or None
|
|
|
|
|
|
def _clean_hex_color(value: str | None, fallback: str | None = None) -> str | None:
|
|
value = (value or "").strip()
|
|
if not value:
|
|
return fallback
|
|
return value if _HEX_COLOR_RE.match(value) else fallback
|
|
|
|
|
|
def _can_manage_branding(db, user) -> bool:
|
|
roles = set(get_user_roles(db, user.id))
|
|
return "System Admin" in roles or "Firm Admin" in roles
|
|
|
|
|
|
def _resolve_branding_scope(db, user, tenant_id: int | None = None, branch_id: int | None = None):
|
|
roles = set(get_user_roles(db, user.id))
|
|
if "System Admin" in roles:
|
|
effective_tenant_id = tenant_id or user.tenant_id
|
|
else:
|
|
effective_tenant_id = user.tenant_id
|
|
|
|
tenant = db.execute(select(Tenant).where(Tenant.id == effective_tenant_id)).scalar_one_or_none()
|
|
if not tenant:
|
|
return None, None, None
|
|
|
|
if "System Admin" not in roles and tenant.id != user.tenant_id:
|
|
return None, None, None
|
|
|
|
if branch_id:
|
|
branch = db.execute(select(Branch).where(Branch.id == branch_id, Branch.tenant_id == tenant.id)).scalar_one_or_none()
|
|
else:
|
|
branch = None
|
|
|
|
if not branch and user.branch_id:
|
|
branch = db.execute(select(Branch).where(Branch.id == user.branch_id, Branch.tenant_id == tenant.id)).scalar_one_or_none()
|
|
|
|
if not branch:
|
|
branch = db.execute(
|
|
select(Branch).where(Branch.tenant_id == tenant.id, Branch.is_active.is_(True)).order_by(Branch.is_head_office.desc(), Branch.name)
|
|
).scalar_one_or_none()
|
|
|
|
settings = None
|
|
if branch:
|
|
settings = db.execute(select(BranchSettings).where(BranchSettings.branch_id == branch.id)).scalar_one_or_none()
|
|
if not settings:
|
|
settings = BranchSettings(branch_id=branch.id)
|
|
db.add(settings)
|
|
db.commit()
|
|
db.refresh(settings)
|
|
return tenant, branch, settings
|
|
|
|
|
|
async def _save_branding_upload(upload: UploadFile | None, *, tenant_id: int, kind: str) -> str | None:
|
|
if not upload or not upload.filename:
|
|
return None
|
|
original = Path(upload.filename).name
|
|
suffix = Path(original).suffix.lower()
|
|
if suffix not in _ALLOWED_BRANDING_SUFFIXES:
|
|
raise ValueError("Unsupported branding file type. Allowed: PNG, JPG, WEBP, ICO and SVG.")
|
|
data = await upload.read()
|
|
if not data:
|
|
return None
|
|
if len(data) > 2 * 1024 * 1024:
|
|
raise ValueError("Branding image size should not exceed 2 MB.")
|
|
folder = BRANDING_UPLOAD_ROOT / f"tenant_{tenant_id}"
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
filename = f"{kind}_{uuid4().hex}{suffix}"
|
|
path = folder / filename
|
|
path.write_bytes(data)
|
|
return f"{BRANDING_PUBLIC_PREFIX}/tenant_{tenant_id}/{filename}"
|
|
|
|
|
|
@router.get("/branding")
|
|
def branding_page(request: Request, tenant_id: int | None = None, branch_id: int | None = None, saved: int = 0):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_manage_branding(db, user):
|
|
return _redirect_denied()
|
|
|
|
roles = set(get_user_roles(db, user.id))
|
|
tenant, branch, settings = _resolve_branding_scope(db, user, tenant_id=tenant_id, branch_id=branch_id)
|
|
if not tenant:
|
|
return _redirect_denied()
|
|
|
|
if "System Admin" in roles:
|
|
tenants = db.execute(select(Tenant).where(Tenant.is_active.is_(True)).order_by(Tenant.name)).scalars().all()
|
|
else:
|
|
tenants = db.execute(select(Tenant).where(Tenant.id == user.tenant_id)).scalars().all()
|
|
branches = db.execute(select(Branch).where(Branch.tenant_id == tenant.id, Branch.is_active.is_(True)).order_by(Branch.name)).scalars().all()
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/branding.html",
|
|
_base_ctx(
|
|
request,
|
|
user,
|
|
db,
|
|
title="Firm Branding",
|
|
tenant=tenant,
|
|
branch=branch,
|
|
settings=settings,
|
|
tenants=tenants,
|
|
branches=branches,
|
|
can_change_tenant=("System Admin" in roles),
|
|
saved=bool(saved),
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/branding")
|
|
async def branding_save(
|
|
request: Request,
|
|
tenant_id: int = Form(...),
|
|
branch_id: int | None = Form(None),
|
|
display_name: str = Form(""),
|
|
primary_color: str = Form("#2563eb"),
|
|
accent_color: str = Form("#0f172a"),
|
|
website_url: str = Form(""),
|
|
contact_email: str = Form(""),
|
|
contact_mobile: str = Form(""),
|
|
address_line1: str = Form(""),
|
|
address_line2: str = Form(""),
|
|
city: str = Form(""),
|
|
state: str = Form(""),
|
|
pin_code: str = Form(""),
|
|
gstin: str = Form(""),
|
|
pan: str = Form(""),
|
|
invoice_footer_text: str = Form(""),
|
|
bank_name: str = Form(""),
|
|
bank_account_name: str = Form(""),
|
|
bank_account_number: str = Form(""),
|
|
bank_ifsc: str = Form(""),
|
|
upi_id: str = Form(""),
|
|
logo_file: UploadFile | None = File(None),
|
|
favicon_file: UploadFile | None = File(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_manage_branding(db, user):
|
|
return _redirect_denied()
|
|
|
|
tenant, branch, settings = _resolve_branding_scope(db, user, tenant_id=tenant_id, branch_id=branch_id)
|
|
if not tenant:
|
|
return _redirect_denied()
|
|
|
|
before_tenant = model_snapshot(tenant, ["name", "display_name", "logo_path", "favicon_path", "primary_color", "accent_color", "website_url", "contact_email", "contact_mobile"])
|
|
before_settings = model_snapshot(settings, ["address_line1", "address_line2", "city", "state", "pin_code", "gstin", "pan", "invoice_footer_text", "bank_name", "bank_account_name", "bank_account_number", "bank_ifsc", "upi_id"]) if settings else {}
|
|
|
|
tenant.display_name = _clean_optional(display_name)
|
|
tenant.primary_color = _clean_hex_color(primary_color, "#2563eb")
|
|
tenant.accent_color = _clean_hex_color(accent_color, "#0f172a")
|
|
tenant.website_url = _clean_optional(website_url)
|
|
tenant.contact_email = _clean_optional(contact_email)
|
|
tenant.contact_mobile = _clean_optional(contact_mobile)
|
|
|
|
logo_path = await _save_branding_upload(logo_file, tenant_id=tenant.id, kind="logo")
|
|
favicon_path = await _save_branding_upload(favicon_file, tenant_id=tenant.id, kind="favicon")
|
|
if logo_path:
|
|
tenant.logo_path = logo_path
|
|
if favicon_path:
|
|
tenant.favicon_path = favicon_path
|
|
|
|
if settings:
|
|
settings.address_line1 = _clean_optional(address_line1)
|
|
settings.address_line2 = _clean_optional(address_line2)
|
|
settings.city = _clean_optional(city)
|
|
settings.state = _clean_optional(state)
|
|
settings.pin_code = _clean_optional(pin_code)
|
|
settings.gstin = _clean_optional(gstin)
|
|
settings.pan = _clean_optional(pan)
|
|
settings.invoice_footer_text = _clean_optional(invoice_footer_text)
|
|
settings.bank_name = _clean_optional(bank_name)
|
|
settings.bank_account_name = _clean_optional(bank_account_name)
|
|
settings.bank_account_number = _clean_optional(bank_account_number)
|
|
settings.bank_ifsc = _clean_optional(bank_ifsc)
|
|
settings.upi_id = _clean_optional(upi_id)
|
|
|
|
db.commit()
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="firm_branding.update",
|
|
entity_type="tenant",
|
|
actor=user,
|
|
request=request,
|
|
entity_id=tenant.id,
|
|
entity_name=tenant.display_name or tenant.name,
|
|
target_tenant_id=tenant.id,
|
|
target_branch_id=branch.id if branch else None,
|
|
details={
|
|
"tenant": pair_before_after(before_tenant, model_snapshot(tenant, ["name", "display_name", "logo_path", "favicon_path", "primary_color", "accent_color", "website_url", "contact_email", "contact_mobile"])),
|
|
"settings": pair_before_after(before_settings, model_snapshot(settings, ["address_line1", "address_line2", "city", "state", "pin_code", "gstin", "pan", "invoice_footer_text", "bank_name", "bank_account_name", "bank_account_number", "bank_ifsc", "upi_id"]) if settings else {}),
|
|
},
|
|
)
|
|
suffix = f"?tenant_id={tenant.id}"
|
|
if branch:
|
|
suffix += f"&branch_id={branch.id}"
|
|
suffix += "&saved=1"
|
|
return RedirectResponse(url="/system-settings/branding" + suffix, status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
# -----------------------------
|
|
# Phase v2.0.4-A - Financial Year Master
|
|
# -----------------------------
|
|
@router.get("/financial-years")
|
|
def financial_years_list(request: Request, tenant_id: int | None = None):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_view_financial_years(db, user):
|
|
return _redirect_denied()
|
|
|
|
visible_tenant_ids = _visible_financial_year_tenant_ids(db, user)
|
|
active_tenant_id = tenant_id or _active_tenant_id_for_settings(request, user)
|
|
if active_tenant_id not in visible_tenant_ids:
|
|
active_tenant_id = int(user.tenant_id)
|
|
tenants = db.execute(
|
|
select(Tenant).where(Tenant.id.in_(visible_tenant_ids)).order_by(Tenant.name)
|
|
).scalars().all() if visible_tenant_ids else []
|
|
rows = db.execute(
|
|
select(FinancialYear)
|
|
.where(FinancialYear.tenant_id == active_tenant_id)
|
|
.order_by(FinancialYear.start_date.desc(), FinancialYear.year_code.desc())
|
|
).scalars().all()
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/financial_years_list.html",
|
|
_base_ctx(
|
|
request,
|
|
user,
|
|
db,
|
|
title="Financial Years",
|
|
financial_years=rows,
|
|
tenants=tenants,
|
|
selected_tenant_id=active_tenant_id,
|
|
can_manage_fy=_can_manage_financial_years(db, user),
|
|
latest_backups={row.financial_year_id: row for row in db.execute(select(YearBackupExport).where(YearBackupExport.tenant_id == active_tenant_id).order_by(YearBackupExport.generated_at_utc.desc())).scalars().all()},
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/financial-years/new")
|
|
def financial_year_create_page(request: Request, tenant_id: int | None = None, year_code: str = ""):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_manage_financial_years(db, user):
|
|
return _redirect_denied()
|
|
visible_tenant_ids = _visible_financial_year_tenant_ids(db, user)
|
|
active_tenant_id = tenant_id or _active_tenant_id_for_settings(request, user)
|
|
if active_tenant_id not in visible_tenant_ids:
|
|
active_tenant_id = int(user.tenant_id)
|
|
tenants = db.execute(
|
|
select(Tenant).where(Tenant.id.in_(visible_tenant_ids)).order_by(Tenant.name)
|
|
).scalars().all() if visible_tenant_ids else []
|
|
start_date, end_date = _default_dates_from_fy(year_code)
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/financial_year_create.html",
|
|
_base_ctx(
|
|
request,
|
|
user,
|
|
db,
|
|
title="Create Financial Year",
|
|
tenants=tenants,
|
|
selected_tenant_id=active_tenant_id,
|
|
form_data={
|
|
"year_code": year_code,
|
|
"assessment_year": _assessment_year_from_fy(year_code),
|
|
"start_date": start_date.isoformat() if start_date else "",
|
|
"end_date": end_date.isoformat() if end_date else "",
|
|
},
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/financial-years/new")
|
|
def financial_year_create_submit(
|
|
request: Request,
|
|
tenant_id: int = Form(...),
|
|
year_code: str = Form(...),
|
|
assessment_year: str = Form(""),
|
|
start_date: str = Form(...),
|
|
end_date: str = Form(...),
|
|
is_current: str | None = Form(None),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_manage_financial_years(db, user):
|
|
return _redirect_denied()
|
|
if tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
|
|
year_code_clean = year_code.strip()
|
|
assessment_year_clean = (assessment_year or _assessment_year_from_fy(year_code_clean)).strip()
|
|
start_date_value = _parse_date(start_date)
|
|
end_date_value = _parse_date(end_date)
|
|
if not year_code_clean or not assessment_year_clean or not start_date_value or not end_date_value or start_date_value > end_date_value:
|
|
return RedirectResponse(url=f"/system-settings/financial-years/new?tenant_id={tenant_id}", status_code=303)
|
|
|
|
exists = db.execute(
|
|
select(FinancialYear).where(
|
|
FinancialYear.tenant_id == tenant_id,
|
|
FinancialYear.year_code == year_code_clean,
|
|
)
|
|
).scalar_one_or_none()
|
|
if exists:
|
|
return RedirectResponse(url=f"/system-settings/financial-years?tenant_id={tenant_id}", status_code=303)
|
|
|
|
if is_current is not None:
|
|
for existing_current in db.execute(
|
|
select(FinancialYear).where(
|
|
FinancialYear.tenant_id == tenant_id,
|
|
FinancialYear.is_current.is_(True),
|
|
)
|
|
).scalars().all():
|
|
existing_current.is_current = False
|
|
|
|
now = datetime.now(timezone.utc)
|
|
fy = FinancialYear(
|
|
tenant_id=tenant_id,
|
|
year_code=year_code_clean,
|
|
assessment_year=assessment_year_clean,
|
|
start_date=start_date_value,
|
|
end_date=end_date_value,
|
|
is_current=is_current is not None or _current_financial_year(db, tenant_id) is None,
|
|
is_locked=False,
|
|
created_at_utc=now,
|
|
updated_at_utc=now,
|
|
)
|
|
db.add(fy)
|
|
db.commit()
|
|
db.refresh(fy)
|
|
write_audit_log(
|
|
db,
|
|
action="financial_year.create",
|
|
entity_type="financial_year",
|
|
actor=user,
|
|
request=request,
|
|
entity_id=fy.id,
|
|
entity_name=fy.year_code,
|
|
target_tenant_id=fy.tenant_id,
|
|
details={"after": model_snapshot(fy, ["tenant_id", "year_code", "assessment_year", "start_date", "end_date", "is_current", "is_locked"])},
|
|
)
|
|
return RedirectResponse(url=f"/system-settings/financial-years?tenant_id={tenant_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/financial-years/{fy_id}/edit")
|
|
def financial_year_edit_page(request: Request, fy_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_manage_financial_years(db, user):
|
|
return _redirect_denied()
|
|
fy = db.execute(select(FinancialYear).where(FinancialYear.id == fy_id)).scalar_one_or_none()
|
|
if not fy or fy.tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
tenant = db.execute(select(Tenant).where(Tenant.id == fy.tenant_id)).scalar_one_or_none()
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/financial_year_edit.html",
|
|
_base_ctx(request, user, db, title="Edit Financial Year", fy=fy, tenant=tenant),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/financial-years/{fy_id}/edit")
|
|
def financial_year_edit_submit(
|
|
request: Request,
|
|
fy_id: int,
|
|
assessment_year: str = Form(...),
|
|
start_date: str = Form(...),
|
|
end_date: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_manage_financial_years(db, user):
|
|
return _redirect_denied()
|
|
fy = db.execute(select(FinancialYear).where(FinancialYear.id == fy_id)).scalar_one_or_none()
|
|
if not fy or fy.tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
if fy.is_locked:
|
|
return RedirectResponse(url=f"/system-settings/financial-years?tenant_id={fy.tenant_id}", status_code=303)
|
|
|
|
before = model_snapshot(fy, ["assessment_year", "start_date", "end_date", "is_current", "is_locked"])
|
|
start_date_value = _parse_date(start_date)
|
|
end_date_value = _parse_date(end_date)
|
|
if not start_date_value or not end_date_value or start_date_value > end_date_value:
|
|
return RedirectResponse(url=f"/system-settings/financial-years/{fy.id}/edit", status_code=303)
|
|
fy.assessment_year = assessment_year.strip()
|
|
fy.start_date = start_date_value
|
|
fy.end_date = end_date_value
|
|
fy.updated_at_utc = datetime.now(timezone.utc)
|
|
db.commit()
|
|
write_audit_log(
|
|
db,
|
|
action="financial_year.update",
|
|
entity_type="financial_year",
|
|
actor=user,
|
|
request=request,
|
|
entity_id=fy.id,
|
|
entity_name=fy.year_code,
|
|
target_tenant_id=fy.tenant_id,
|
|
details=pair_before_after(before, model_snapshot(fy, ["assessment_year", "start_date", "end_date", "is_current", "is_locked"])),
|
|
)
|
|
return RedirectResponse(url=f"/system-settings/financial-years?tenant_id={fy.tenant_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/financial-years/{fy_id}/make-current")
|
|
def financial_year_make_current(request: Request, fy_id: int, csrf_token: str = Form(...)):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_manage_financial_years(db, user):
|
|
return _redirect_denied()
|
|
fy = db.execute(select(FinancialYear).where(FinancialYear.id == fy_id)).scalar_one_or_none()
|
|
if not fy or fy.tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
for existing_current in db.execute(
|
|
select(FinancialYear).where(FinancialYear.tenant_id == fy.tenant_id, FinancialYear.is_current.is_(True))
|
|
).scalars().all():
|
|
existing_current.is_current = False
|
|
fy.is_current = True
|
|
fy.updated_at_utc = datetime.now(timezone.utc)
|
|
db.commit()
|
|
if int(request.session.get("active_tenant_id") or user.tenant_id) == fy.tenant_id:
|
|
request.session["active_financial_year"] = fy.year_code
|
|
return RedirectResponse(url=f"/system-settings/financial-years?tenant_id={fy.tenant_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/financial-years/{fy_id}/lock")
|
|
def financial_year_lock(request: Request, fy_id: int, csrf_token: str = Form(...)):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_manage_financial_years(db, user):
|
|
return _redirect_denied()
|
|
fy = db.execute(select(FinancialYear).where(FinancialYear.id == fy_id)).scalar_one_or_none()
|
|
if not fy or fy.tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
fy.is_locked = True
|
|
fy.locked_at_utc = datetime.now(timezone.utc)
|
|
fy.locked_by_user_id = user.id
|
|
fy.updated_at_utc = datetime.now(timezone.utc)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/system-settings/financial-years?tenant_id={fy.tenant_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/financial-years/{fy_id}/unlock")
|
|
def financial_year_unlock(request: Request, fy_id: int, csrf_token: str = Form(...)):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
roles = set(get_user_roles(db, user.id))
|
|
if "System Admin" not in roles and "Firm Admin" not in roles:
|
|
return _redirect_denied()
|
|
fy = db.execute(select(FinancialYear).where(FinancialYear.id == fy_id)).scalar_one_or_none()
|
|
if not fy or fy.tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
fy.is_locked = False
|
|
fy.locked_at_utc = None
|
|
fy.locked_by_user_id = None
|
|
fy.updated_at_utc = datetime.now(timezone.utc)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/system-settings/financial-years?tenant_id={fy.tenant_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/financial-years/{fy_id}/backup")
|
|
def financial_year_backup_page(request: Request, fy_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_view_financial_years(db, user):
|
|
return _redirect_denied()
|
|
fy = db.execute(select(FinancialYear).where(FinancialYear.id == fy_id)).scalar_one_or_none()
|
|
if not fy or fy.tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
exports = db.execute(
|
|
select(YearBackupExport)
|
|
.where(YearBackupExport.financial_year_id == fy.id)
|
|
.order_by(YearBackupExport.generated_at_utc.desc(), YearBackupExport.id.desc())
|
|
).scalars().all()
|
|
return templates.TemplateResponse(
|
|
"modules/system_settings/templates/financial_year_backup.html",
|
|
_base_ctx(
|
|
request,
|
|
user,
|
|
db,
|
|
title=f"Backup Export - {fy.year_code}",
|
|
fy=fy,
|
|
exports=exports,
|
|
can_manage_fy=_can_manage_financial_years(db, user),
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/financial-years/{fy_id}/backup/export")
|
|
def financial_year_backup_export_submit(request: Request, fy_id: int, csrf_token: str = Form(...)):
|
|
try:
|
|
validate_csrf(request, csrf_token)
|
|
except PermissionError:
|
|
return _csrf_rejected(request)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_manage_financial_years(db, user):
|
|
return _redirect_denied()
|
|
fy = db.execute(select(FinancialYear).where(FinancialYear.id == fy_id)).scalar_one_or_none()
|
|
if not fy or fy.tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
export = build_year_backup_export(db, financial_year=fy, user_id=user.id)
|
|
db.commit()
|
|
write_audit_log(
|
|
db,
|
|
action="financial_year.backup_export",
|
|
entity_type="financial_year",
|
|
actor=user,
|
|
request=request,
|
|
entity_id=fy.id,
|
|
entity_name=fy.year_code,
|
|
target_tenant_id=fy.tenant_id,
|
|
details={"backup_export_id": export.id, "file_size_bytes": export.file_size_bytes},
|
|
)
|
|
return RedirectResponse(url=f"/system-settings/financial-years/{fy.id}/backup?exported=1", status_code=303)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/financial-years/backups/{export_id}/download")
|
|
def financial_year_backup_download(request: Request, export_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not _can_view_financial_years(db, user):
|
|
return _redirect_denied()
|
|
export = db.execute(select(YearBackupExport).where(YearBackupExport.id == export_id)).scalar_one_or_none()
|
|
if not export or export.tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
path = Path(export.export_file_path)
|
|
if not path.exists() or not path.is_file():
|
|
return RedirectResponse(url=f"/system-settings/financial-years/{export.financial_year_id}/backup?error=file_missing", status_code=303)
|
|
return FileResponse(path, filename=path.name, media_type="application/zip")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/context/financial-year/{year_code}")
|
|
def switch_active_financial_year(request: Request, year_code: str):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
active_tenant_id = int(request.session.get("active_tenant_id") or user.tenant_id)
|
|
if active_tenant_id not in _visible_financial_year_tenant_ids(db, user):
|
|
return _redirect_denied()
|
|
fy = db.execute(
|
|
select(FinancialYear).where(
|
|
FinancialYear.tenant_id == active_tenant_id,
|
|
FinancialYear.year_code == year_code.strip(),
|
|
)
|
|
).scalar_one_or_none()
|
|
if not fy:
|
|
return _redirect_denied()
|
|
request.session["active_financial_year"] = fy.year_code
|
|
return _financial_year_redirect_back(request)
|
|
finally:
|
|
db.close()
|
|
|
|
def _redirect_back(request: Request, default_url: str = "/services") -> RedirectResponse:
|
|
return RedirectResponse(url=request.headers.get("referer") or default_url, status_code=303)
|
|
|
|
|
|
@router.get("/context/tenant/{tenant_id}")
|
|
def switch_active_tenant(request: Request, tenant_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
roles = get_user_roles(db, user.id)
|
|
perms = set(get_user_permissions(db, user.id))
|
|
if "System Admin" not in roles or "services.cross_tenant" not in perms:
|
|
return _redirect_denied()
|
|
|
|
scope = build_scope(db, user)
|
|
visible_ids = {t.id for t in list_visible_tenants(db, scope)}
|
|
if tenant_id not in visible_ids:
|
|
return _redirect_denied()
|
|
|
|
tenant = db.get(Tenant, tenant_id)
|
|
if not tenant or not tenant.is_active:
|
|
return _redirect_denied()
|
|
|
|
_store_active_tenant_context(request, tenant)
|
|
_store_active_branch_context(request, None)
|
|
current_fy = _current_financial_year(db, tenant_id)
|
|
if current_fy:
|
|
request.session["active_financial_year"] = current_fy.year_code
|
|
else:
|
|
request.session.pop("active_financial_year", None)
|
|
return _redirect_back(request)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/context/branch/{branch_id}")
|
|
def switch_active_branch(request: Request, branch_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
perms = set(get_user_permissions(db, user.id))
|
|
roles = get_user_roles(db, user.id)
|
|
if "services.cross_branch" not in perms and not ("System Admin" in roles and "services.cross_tenant" in perms):
|
|
return _redirect_denied()
|
|
|
|
active_tenant_id = int(request.session.get("active_tenant_id") or user.tenant_id)
|
|
if branch_id == 0:
|
|
_store_active_branch_context(request, None)
|
|
return _redirect_back(request)
|
|
|
|
scope = build_scope(db, user)
|
|
visible_ids = {b.id for b in list_visible_branches(db, scope, tenant_id=active_tenant_id)}
|
|
if branch_id not in visible_ids:
|
|
return _redirect_denied()
|
|
|
|
branch = db.get(Branch, branch_id)
|
|
if not branch or int(branch.tenant_id) != int(active_tenant_id) or not branch.is_active:
|
|
return _redirect_denied()
|
|
|
|
_store_active_branch_context(request, branch)
|
|
return _redirect_back(request)
|
|
finally:
|
|
db.close()
|
|
|