"""RBAC-scoped HR data lookups exposed to the assistant as tool calls. Separate module from apps.chatbot.tools (projects/tasks) because HR data is a different sensitivity tier — salary and personal fields never appear here unless the scoping rule below explicitly allows it. Every function takes `user` first and applies the same scoping rule the real ViewSets use (`user_has_full_scope`): full access for superusers/managers (`hr.delete` perm), otherwise limited to the user's own Employee record. """ from apps.core.permissions import user_has_full_scope from apps.hr.models import Attendance, Contract, Employee, Payroll HR_TOOL_SCHEMAS = [ { "type": "function", "function": { "name": "hr_list_employees", "description": "List employee directory entries (name, position, department, status). No salary or personal contact info.", "parameters": { "type": "object", "properties": { "department": {"type": "string", "description": "Filter by department name (partial match)."}, "status": { "type": "string", "enum": ["active", "inactive", "on_leave", "terminated"], }, }, }, }, }, { "type": "function", "function": { "name": "hr_my_attendance", "description": "List the current user's own attendance/leave records. Managers may pass an employee name to look up someone else's; non-managers can only see their own regardless of arguments.", "parameters": { "type": "object", "properties": { "employee": {"type": "string", "description": "Employee name to look up (manager-only; ignored for non-managers)."}, }, }, }, }, { "type": "function", "function": { "name": "hr_my_payroll", "description": "List payroll records including salary figures. Non-managers only ever get their own; managers may pass an employee name for anyone's.", "parameters": { "type": "object", "properties": { "employee": {"type": "string", "description": "Employee name to look up (manager-only; ignored for non-managers)."}, }, }, }, }, { "type": "function", "function": { "name": "hr_my_contract", "description": "List contract records (type, dates, status — not the document file or terms text). Non-managers only ever get their own; managers may pass an employee name for anyone's.", "parameters": { "type": "object", "properties": { "employee": {"type": "string", "description": "Employee name to look up (manager-only; ignored for non-managers)."}, }, }, }, }, ] def _own_employee(user, organization): return Employee.objects.filter(user=user, organization=organization).first() def _resolve_target_employee(user, organization, employee_name): """Managers may target another employee by name; everyone else is pinned to their own Employee record regardless of what the model asks for.""" if user_has_full_scope(user, "hr", organization) and employee_name: emp = Employee.objects.select_related("user").filter( organization=organization, user__first_name__icontains=employee_name ).first() or Employee.objects.select_related("user").filter( organization=organization, user__last_name__icontains=employee_name ).first() if emp: return emp, None return None, {"error": f"No employee found matching '{employee_name}'."} return _own_employee(user, organization), None def hr_list_employees(user, organization, department=None, status=None): qs = Employee.objects.select_related("user").filter(organization=organization) if department: qs = qs.filter(department__icontains=department) if status: qs = qs.filter(status=status) return [ { "name": e.user.get_full_name() or e.user.email, "employee_id": e.employee_id, "position": e.position, "department": e.department, "status": e.status, } for e in qs[:50] ] def hr_my_attendance(user, organization, employee=None): emp, error = _resolve_target_employee(user, organization, employee) if error: return error if not emp: return {"error": "No employee record linked to your account."} records = Attendance.objects.filter(employee=emp).order_by("-date")[:30] return [ { "date": str(a.date), "type": a.type, "time": str(a.time) if a.time else None, "location": a.location, } for a in records ] def hr_my_payroll(user, organization, employee=None): emp, error = _resolve_target_employee(user, organization, employee) if error: return error if not emp: return {"error": "No employee record linked to your account."} records = Payroll.objects.filter(employee=emp).order_by("-period_start")[:12] return [ { "period_start": str(p.period_start), "period_end": str(p.period_end), "base_salary": str(p.base_salary), "allowances": str(p.allowances), "deductions": str(p.deductions), "net_salary": str(p.net_salary), "status": p.status, } for p in records ] def hr_my_contract(user, organization, employee=None): emp, error = _resolve_target_employee(user, organization, employee) if error: return error if not emp: return {"error": "No employee record linked to your account."} records = Contract.objects.filter(employee=emp).order_by("-start_date")[:12] return [ { "contract_type": c.contract_type, "start_date": str(c.start_date), "end_date": str(c.end_date) if c.end_date else None, "status": c.status, } for c in records ] HR_DISPATCH = { "hr_list_employees": hr_list_employees, "hr_my_attendance": hr_my_attendance, "hr_my_payroll": hr_my_payroll, "hr_my_contract": hr_my_contract, } # Args allowed per tool, with their enum whitelist (None = free text, capped below). HR_ALLOWED_ARGS = { "hr_list_employees": { "department": None, "status": {"active", "inactive", "on_leave", "terminated"}, }, "hr_my_attendance": {"employee": None}, "hr_my_payroll": {"employee": None}, "hr_my_contract": {"employee": None}, } HR_MAX_ARG_LEN = 200 def call_hr_tool(user, organization, name, arguments: dict): fn = HR_DISPATCH.get(name) if not fn: return {"error": f"Unknown tool: {name}"} allowed = HR_ALLOWED_ARGS[name] if not isinstance(arguments, dict): return {"error": "Invalid tool arguments."} clean = {} for key, value in arguments.items(): if key not in allowed: continue # drop unknown args instead of erroring — model may hallucinate extra keys enum = allowed[key] if not isinstance(value, str): continue value = value[:HR_MAX_ARG_LEN] if enum is not None and value not in enum: continue clean[key] = value return fn(user, organization, **clean)