from urllib.parse import parse_qsl, urlencode from fastapi.templating import Jinja2Templates from sqlalchemy import select templates = Jinja2Templates(directory="app") from app.core.db.common import CommonSessionLocal from app.modules.core.iam.scope import build_scope, list_visible_branches, list_visible_tenants from app.modules.core.rbac.ui_permissions import ( can_change_branch_tenant, can_export_clients, can_import_service_tasks, can_import_services, can_manage_branches, can_manage_clients, can_manage_rbac, can_manage_service_tasks, can_manage_services, can_manage_settings, can_manage_tenants, can_manage_users, can_view_employee_dashboard, can_view_employees, can_manage_employees, can_change_employee_status, can_switch_employee_tenant, can_switch_employee_branch, can_view_employee_portal, can_edit_own_employee_profile, can_view_own_employee_work, can_manage_employee_work, can_view_employee_progress, can_request_employee_registration, can_approve_employee_registrations, can_punch_employee_attendance, can_view_own_employee_attendance, can_view_all_employee_attendance, can_approve_employee_attendance, can_apply_employee_leave, can_view_own_employee_leave, can_view_all_employee_leave, can_approve_employee_leave, can_manage_employee_leave_types, can_manage_employee_leave_balances, can_view_own_employee_documents, can_upload_own_employee_documents, can_view_all_employee_documents, can_manage_employee_documents, can_verify_employee_documents, can_manage_employee_document_types, can_view_employee_onboarding, can_manage_employee_onboarding, can_approve_employee_onboarding, can_view_employee_offboarding, can_manage_employee_offboarding, can_approve_employee_offboarding, can_request_own_employee_offboarding, can_import_employee_hr, can_manage_employee_payroll_structures, can_run_employee_payroll, can_view_employee_payroll, can_view_own_employee_payslips, can_approve_employee_payroll, can_view_consultants, can_manage_consultants, can_link_consultant_clients, can_manage_consultant_service_requests, can_manage_consultant_conversions, can_view_consultant_portal, can_manage_own_consultant_workspace, can_switch_client_branch, can_switch_client_tenant, can_switch_service_branch, can_switch_service_tenant, can_view_audit, can_view_branches, can_view_clients, can_view_billing, can_create_billing, can_generate_billing_invoices, can_view_billing_fee_structure, can_import_billing_fee_structure, can_view_platform_billing, can_manage_platform_billing, can_generate_platform_billing, can_manage_platform_plans, can_manage_platform_subscriptions, can_view_marketplace_leads, can_create_marketplace_leads, can_assign_marketplace_leads, can_update_marketplace_leads, can_convert_marketplace_leads, can_view_documents, can_upload_documents, can_download_documents, can_delete_documents, can_view_rbac, can_view_services, can_view_settings, can_view_tenants, can_view_users, can_view_own_alerts, can_manage_alerts, can_view_notice_cases, can_manage_notice_cases, can_upload_notice_case_documents, can_download_notice_case_documents, can_delete_notice_case_documents, ) def build_page_url(base_url: str, page: int, query: str | None = None) -> str: params = dict(parse_qsl((query or "").lstrip("?"), keep_blank_values=True)) params["page"] = str(page) qs = urlencode(params) return f"{base_url}?{qs}" if qs else base_url def get_active_tenant_id(request, current_user=None): if not current_user: return None return request.session.get("active_tenant_id") or getattr(current_user, "tenant_id", None) def get_active_branch_id(request, current_user=None): if not current_user: return None # return None when "all branches" context is active val = request.session.get("active_branch_id") if val in (None, "", 0, "0"): return None return val def get_active_tenant_code(request, current_user=None): if not current_user: return None return request.session.get("active_tenant_code") or request.session.get("tenant_code") or getattr(request.state, "tenant_code", None) def get_active_branch_code(request, current_user=None): if not current_user: return None val = request.session.get("active_branch_code") or request.session.get("branch_code") return val or getattr(request.state, "branch_code", None) def get_active_financial_year(request, current_user=None): if not current_user: return None return request.session.get("active_financial_year") or getattr(request.state, "year_code", None) def get_active_assessment_year(request, current_user=None): if not current_user: return None fy_code = get_active_financial_year(request, current_user) if not fy_code: return None db = CommonSessionLocal() try: from app.modules.core.tenancy.models import FinancialYear tenant_id = get_active_tenant_id(request, current_user) or getattr(current_user, "tenant_id", None) fy = db.execute( select(FinancialYear).where( FinancialYear.tenant_id == tenant_id, FinancialYear.year_code == fy_code, ) ).scalar_one_or_none() return fy.assessment_year if fy else None except Exception: return None finally: db.close() def get_unread_alert_count(request, current_user=None): if not current_user: return 0 try: from app.modules.alerts.service import count_unread_alerts except Exception: return 0 db = CommonSessionLocal() try: return count_unread_alerts(db, current_user) except Exception: return 0 finally: db.close() def _safe_static_path(path: str | None) -> str | None: path = (path or "").strip() if not path: return None if path.startswith("/static/"): return path if path.startswith("app/ui/static/"): return "/static/" + path.split("app/ui/static/", 1)[1] return path def get_domain_context(request) -> dict: """Return safe domain context populated by Phase 7T.2 middleware.""" try: ctx = getattr(request.state, "domain_context", None) return ctx if isinstance(ctx, dict) else {"is_resolved": False} except Exception: return {"is_resolved": False} def _branding_default() -> dict: return { "firm_name": "Audit Firm ERP", "branch_name": "", "logo_url": None, "favicon_url": None, "primary_color": "#2563eb", "accent_color": "#0f172a", "contact_email": None, "contact_mobile": None, "website_url": None, "domain_name": None, "domain_type": None, "domain_resolved": False, "is_marketplace_domain": False, "is_consultant_domain": False, "consultant_name": None, "consultant_firm_name": None, } def _tenant_branding_from_row(tenant, branch=None, default: dict | None = None) -> dict: default = default or _branding_default() if not tenant: return default.copy() return { **default, "firm_name": getattr(tenant, "display_name", None) or getattr(tenant, "name", None) or default["firm_name"], "branch_name": getattr(branch, "name", None) if branch else "All Branches", "logo_url": _safe_static_path(getattr(tenant, "logo_path", None)), "favicon_url": _safe_static_path(getattr(tenant, "favicon_path", None)), "primary_color": getattr(tenant, "primary_color", None) or default["primary_color"], "accent_color": getattr(tenant, "accent_color", None) or default["accent_color"], "contact_email": getattr(tenant, "contact_email", None), "contact_mobile": getattr(tenant, "contact_mobile", None), "website_url": getattr(tenant, "website_url", None), } def _domain_branding(request, default: dict | None = None) -> dict: default = default or _branding_default() ctx = get_domain_context(request) if not ctx.get("is_resolved"): return default.copy() db = CommonSessionLocal() try: from app.modules.core.tenancy.models import Branch, Tenant from app.modules.consultants.models import ConsultantProfile from app.modules.core.iam.models import User domain_type = ctx.get("domain_type") tenant_id = ctx.get("tenant_id") or ctx.get("parent_tenant_id") branch_id = ctx.get("branch_id") consultant_id = ctx.get("consultant_id") tenant = db.execute(select(Tenant).where(Tenant.id == tenant_id)).scalar_one_or_none() if tenant_id else None branch = db.execute(select(Branch).where(Branch.id == branch_id)).scalar_one_or_none() if branch_id else None branding = _tenant_branding_from_row(tenant, branch, default) branding.update({ "domain_name": ctx.get("domain_name") or ctx.get("host"), "domain_type": domain_type, "domain_resolved": True, "is_marketplace_domain": domain_type == "marketplace", "is_consultant_domain": str(domain_type or "").startswith("consultant_"), }) if domain_type == "marketplace": branding["firm_name"] = "FilingABC" branding["branch_name"] = "Marketplace" return branding if consultant_id: consultant = db.execute(select(ConsultantProfile).where(ConsultantProfile.id == consultant_id)).scalar_one_or_none() if consultant: consultant_name = getattr(consultant, "contact_person", None) or getattr(consultant, "firm_name", None) or "Consultant" consultant_firm_name = getattr(consultant, "firm_name", None) or consultant_name branding["consultant_name"] = consultant_name branding["consultant_firm_name"] = consultant_firm_name branding["firm_name"] = consultant_firm_name branding["branch_name"] = "Consultant Workspace" branding["contact_email"] = getattr(consultant, "email", None) or branding.get("contact_email") branding["contact_mobile"] = getattr(consultant, "mobile", None) or branding.get("contact_mobile") # If the consultant user has a profile photo, use it as the domain logo. user_id = getattr(consultant, "user_id", None) if user_id: user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none() logo = _safe_static_path(getattr(user, "profile_photo_path", None)) if logo: branding["logo_url"] = logo return branding except Exception: return default.copy() finally: db.close() def get_current_tenant_name(request, current_user=None): if not current_user: return _domain_branding(request).get("firm_name") or "Audit Firm" db = CommonSessionLocal() try: from app.modules.core.tenancy.models import Tenant tenant_id = get_active_tenant_id(request, current_user) tenant = db.execute(select(Tenant).where(Tenant.id == tenant_id)).scalar_one_or_none() if not tenant: return _domain_branding(request).get("firm_name") or "Audit Firm" return getattr(tenant, "display_name", None) or tenant.name or "Audit Firm" except Exception: return _domain_branding(request).get("firm_name") or "Audit Firm" finally: db.close() def get_current_branch_name(request, current_user=None): if not current_user: return _domain_branding(request).get("branch_name") or "-" db = CommonSessionLocal() try: from app.modules.core.tenancy.models import Branch branch_id = get_active_branch_id(request, current_user) or getattr(current_user, "branch_id", None) if not branch_id: return "All Branches" branch = db.execute(select(Branch).where(Branch.id == branch_id)).scalar_one_or_none() return branch.name if branch else "-" except Exception: return "-" finally: db.close() def get_current_firm_branding(request, current_user=None): default = _branding_default() # Before login, domain branding is the only safe branding source. This supports # arrr.associates, auditfirm.filingabc.com, filingabc.com and consultant domains. if not current_user: return _domain_branding(request, default) db = CommonSessionLocal() try: from app.modules.core.tenancy.models import Branch, Tenant tenant_id = get_active_tenant_id(request, current_user) branch_id = get_active_branch_id(request, current_user) or getattr(current_user, "branch_id", None) tenant = db.execute(select(Tenant).where(Tenant.id == tenant_id)).scalar_one_or_none() branch = db.execute(select(Branch).where(Branch.id == branch_id)).scalar_one_or_none() if branch_id else None if not tenant: return _domain_branding(request, default) branding = _tenant_branding_from_row(tenant, branch, default) ctx = get_domain_context(request) if ctx.get("is_resolved"): branding.update({ "domain_name": ctx.get("domain_name") or ctx.get("host"), "domain_type": ctx.get("domain_type"), "domain_resolved": True, "is_marketplace_domain": ctx.get("domain_type") == "marketplace", "is_consultant_domain": str(ctx.get("domain_type") or "").startswith("consultant_"), }) return branding except Exception: return _domain_branding(request, default) finally: db.close() def get_user_profile_photo_url(current_user=None): if not current_user: return None try: from app.modules.core.iam.profile_service import profile_photo_url return profile_photo_url(current_user) except Exception: return None def get_user_initials(current_user=None): try: from app.modules.core.iam.profile_service import user_initials return user_initials(current_user) except Exception: return "U" def get_client_sidebar_auditor_card(request, current_user=None): """Return the client-facing auditor card for the logged-in client user. This is used only by the sidebar. It reuses Phase 7Q.5 auditor_service and does not create or alter any business workflow. """ if not current_user: return None db = CommonSessionLocal() try: from app.modules.clients.auditor_service import build_client_auditor_card from app.modules.clients.models import Client from app.modules.core.tenancy.models import Branch, Tenant tenant_id = get_active_tenant_id(request, current_user) or getattr(current_user, "tenant_id", None) email = (getattr(current_user, "email", None) or "").strip().lower() stmt = ( select(Client, Tenant.name.label("tenant_name"), Branch.name.label("branch_name")) .join(Tenant, Tenant.id == Client.tenant_id, isouter=True) .join(Branch, Branch.id == Client.branch_id, isouter=True) .where(Client.is_active.is_(True), Client.is_archived.is_(False)) ) if tenant_id: stmt = stmt.where(Client.tenant_id == int(tenant_id)) if email: stmt = stmt.where((Client.portal_user_id == current_user.id) | (Client.email == email) | (Client.alternate_email == email)) else: stmt = stmt.where(Client.portal_user_id == current_user.id) result = db.execute(stmt.order_by(Client.id.desc())).first() if not result: return None client, tenant_name, branch_name = result client_row = { "id": client.id, "tenant_id": client.tenant_id, "branch_id": client.branch_id, "tenant_name": tenant_name, "branch_name": branch_name, "partner_id": client.partner_id, "default_review_partner_user_id": client.default_review_partner_user_id, } return build_client_auditor_card(db, client_row) except Exception: return None finally: db.close() def get_context_tenants(request, current_user=None, permissions=None, role_names=None): if not current_user: return [] if not ( can_switch_service_tenant(current_user, permissions, role_names) or can_switch_client_tenant(current_user, permissions, role_names) or can_switch_employee_tenant(current_user, permissions, role_names) ): return [] db = CommonSessionLocal() try: scope = build_scope(db, current_user) return list_visible_tenants(db, scope) finally: db.close() def get_context_branches(request, current_user=None, permissions=None, role_names=None): if not current_user: return [] if not ( can_switch_service_branch(current_user, permissions, role_names) or can_switch_client_branch(current_user, permissions, role_names) or can_switch_employee_branch(current_user, permissions, role_names) ): return [] db = CommonSessionLocal() try: scope = build_scope(db, current_user) tenant_id = int(get_active_tenant_id(request, current_user) or current_user.tenant_id) return list_visible_branches(db, scope, tenant_id=tenant_id) finally: db.close() def get_context_financial_years(request, current_user=None, permissions=None, role_names=None): if not current_user: return [] db = CommonSessionLocal() try: from app.modules.core.tenancy.models import FinancialYear tenant_id = int(get_active_tenant_id(request, current_user) or current_user.tenant_id) return db.execute( select(FinancialYear) .where(FinancialYear.tenant_id == tenant_id) .order_by(FinancialYear.start_date.desc(), FinancialYear.year_code.desc()) ).scalars().all() finally: db.close() templates.env.globals.update( can_view_users=can_view_users, can_view_own_alerts=can_view_own_alerts, can_manage_alerts=can_manage_alerts, can_view_notice_cases=can_view_notice_cases, can_manage_notice_cases=can_manage_notice_cases, can_upload_notice_case_documents=can_upload_notice_case_documents, can_download_notice_case_documents=can_download_notice_case_documents, can_delete_notice_case_documents=can_delete_notice_case_documents, can_view_employee_dashboard=can_view_employee_dashboard, can_view_employees=can_view_employees, can_manage_employees=can_manage_employees, can_change_employee_status=can_change_employee_status, can_switch_employee_tenant=can_switch_employee_tenant, can_switch_employee_branch=can_switch_employee_branch, can_view_employee_portal=can_view_employee_portal, can_edit_own_employee_profile=can_edit_own_employee_profile, can_view_own_employee_work=can_view_own_employee_work, can_manage_employee_work=can_manage_employee_work, can_view_employee_progress=can_view_employee_progress, can_request_employee_registration=can_request_employee_registration, can_approve_employee_registrations=can_approve_employee_registrations, can_punch_employee_attendance=can_punch_employee_attendance, can_view_own_employee_attendance=can_view_own_employee_attendance, can_view_all_employee_attendance=can_view_all_employee_attendance, can_approve_employee_attendance=can_approve_employee_attendance, can_apply_employee_leave=can_apply_employee_leave, can_view_own_employee_leave=can_view_own_employee_leave, can_view_all_employee_leave=can_view_all_employee_leave, can_approve_employee_leave=can_approve_employee_leave, can_manage_employee_leave_types=can_manage_employee_leave_types, can_manage_employee_leave_balances=can_manage_employee_leave_balances, can_view_own_employee_documents=can_view_own_employee_documents, can_upload_own_employee_documents=can_upload_own_employee_documents, can_view_all_employee_documents=can_view_all_employee_documents, can_manage_employee_documents=can_manage_employee_documents, can_verify_employee_documents=can_verify_employee_documents, can_manage_employee_document_types=can_manage_employee_document_types, can_view_employee_onboarding=can_view_employee_onboarding, can_manage_employee_onboarding=can_manage_employee_onboarding, can_approve_employee_onboarding=can_approve_employee_onboarding, can_view_employee_offboarding=can_view_employee_offboarding, can_manage_employee_offboarding=can_manage_employee_offboarding, can_approve_employee_offboarding=can_approve_employee_offboarding, can_request_own_employee_offboarding=can_request_own_employee_offboarding, can_import_employee_hr=can_import_employee_hr, can_manage_employee_payroll_structures=can_manage_employee_payroll_structures, can_run_employee_payroll=can_run_employee_payroll, can_view_employee_payroll=can_view_employee_payroll, can_view_own_employee_payslips=can_view_own_employee_payslips, can_approve_employee_payroll=can_approve_employee_payroll, can_manage_users=can_manage_users, can_view_consultants=can_view_consultants, can_manage_consultants=can_manage_consultants, can_link_consultant_clients=can_link_consultant_clients, can_manage_consultant_service_requests=can_manage_consultant_service_requests, can_manage_consultant_conversions=can_manage_consultant_conversions, can_view_consultant_portal=can_view_consultant_portal, can_manage_own_consultant_workspace=can_manage_own_consultant_workspace, can_view_settings=can_view_settings, can_manage_settings=can_manage_settings, can_view_rbac=can_view_rbac, can_manage_rbac=can_manage_rbac, can_view_audit=can_view_audit, can_view_tenants=can_view_tenants, can_manage_tenants=can_manage_tenants, can_view_branches=can_view_branches, can_manage_branches=can_manage_branches, can_change_branch_tenant=can_change_branch_tenant, can_view_services=can_view_services, can_manage_services=can_manage_services, can_manage_service_tasks=can_manage_service_tasks, can_import_services=can_import_services, can_import_service_tasks=can_import_service_tasks, can_switch_service_tenant=can_switch_service_tenant, can_switch_service_branch=can_switch_service_branch, can_view_clients=can_view_clients, can_view_billing=can_view_billing, can_create_billing=can_create_billing, can_generate_billing_invoices=can_generate_billing_invoices, can_view_billing_fee_structure=can_view_billing_fee_structure, can_import_billing_fee_structure=can_import_billing_fee_structure, can_view_platform_billing=can_view_platform_billing, can_manage_platform_billing=can_manage_platform_billing, can_generate_platform_billing=can_generate_platform_billing, can_manage_platform_plans=can_manage_platform_plans, can_manage_platform_subscriptions=can_manage_platform_subscriptions, can_view_marketplace_leads=can_view_marketplace_leads, can_create_marketplace_leads=can_create_marketplace_leads, can_assign_marketplace_leads=can_assign_marketplace_leads, can_update_marketplace_leads=can_update_marketplace_leads, can_convert_marketplace_leads=can_convert_marketplace_leads, can_view_documents=can_view_documents, can_upload_documents=can_upload_documents, can_download_documents=can_download_documents, can_delete_documents=can_delete_documents, can_manage_clients=can_manage_clients, can_export_clients=can_export_clients, can_switch_client_tenant=can_switch_client_tenant, can_switch_client_branch=can_switch_client_branch, get_unread_alert_count=get_unread_alert_count, get_current_tenant_name=get_current_tenant_name, get_current_branch_name=get_current_branch_name, get_current_firm_branding=get_current_firm_branding, get_domain_context=get_domain_context, get_user_profile_photo_url=get_user_profile_photo_url, get_user_initials=get_user_initials, get_client_sidebar_auditor_card=get_client_sidebar_auditor_card, get_context_tenants=get_context_tenants, get_context_branches=get_context_branches, get_context_financial_years=get_context_financial_years, get_active_tenant_id=get_active_tenant_id, get_active_tenant_code=get_active_tenant_code, get_active_branch_id=get_active_branch_id, get_active_branch_code=get_active_branch_code, get_active_financial_year=get_active_financial_year, get_active_assessment_year=get_active_assessment_year, build_page_url=build_page_url, )