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 _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"): return RedirectResponse(url=default_url, status_code=303) 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 # ----------------------------- @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("Partner,System Admin"), default_storage_mode: str = Form("local_only"), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) if not _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 "Partner,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("Partner,System Admin"), default_storage_mode: str = Form("local_only"), is_active: str | None = Form(None), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) 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 "Partner,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(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) 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("Partner,System Admin"), session_duration_minutes: str = Form("480"), lockout_attempts: str = Form("5"), lockout_minutes: str = Form("15"), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) 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 "Partner,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/ui/static/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 "/static/uploads/branding/" + f"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(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) if not _can_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(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) if not _can_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(...), ): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) if not _can_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(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) if not _can_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(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) if not _can_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(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) 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(...)): validate_csrf(request, csrf_token) db = CommonSessionLocal() try: user = get_current_user(request, db=db) if not user: return RedirectResponse(url="/login", status_code=303) if not _can_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()