from collections import defaultdict from datetime import date, timedelta from decimal import Decimal from django.db.models import OuterRef, Prefetch, Subquery from django.utils import timezone from attendance.models import AttendanceRecord from employees.models import Employee, EmployeeAction from employees.services.personnel_state import ( calculate_employee_tenure_days, resolve_employee_personnel_state, ) from finance.analytics.models import ReportingEntry from finance.enums import PayrollRunStatus from finance.models import PayrollOffCyclePayment, PayrollRun, PayrollStatement from .sales import ZERO, months def mean(values): return sum(values, ZERO) / len(values) if values else None def absence_days(employee, actions, start, end): vacation = sick = 0 day = start while day <= end: status = ( resolve_employee_personnel_state(employee, day, actions=actions).status if actions else "normal" ) vacation += int(status == "on_leave") sick += int(status == "on_sick_leave") day += timedelta(days=1) return vacation, sick def people_report(start, end, currency): employees = list( Employee.objects.prefetch_related( Prefetch( "actions", queryset=EmployeeAction.objects.select_related( "source_request" ).order_by("date", "pk"), ) ).order_by("last_name", "first_name", "pk") ) latest_run = ( PayrollRun.objects.filter( period_id=OuterRef("run__period_id"), status=PayrollRunStatus.PUBLISHED ) .order_by("-revision") .values("pk")[:1] ) statements = PayrollStatement.objects.filter( run_id=Subquery(latest_run), currency=currency, run__period__date_from__gte=start, run__period__date_to__lte=end, ).select_related("run__period") pay = defaultdict(lambda: defaultdict(lambda: ZERO)) for statement in statements: month = statement.run.period.date_to.strftime("%Y-%m") pay[statement.employee_id][month] += statement.gross_total attendance = defaultdict(dict) for record in AttendanceRecord.objects.filter( date__range=(start, end), analysis_run__status="success" ): attendance[record.employee_id][record.date] = record taxes = defaultdict(lambda: ZERO) tax_counts = defaultdict(int) for entry in ReportingEntry.objects.filter( kind="tax", date__range=(start, end), currency=currency, voided_at__isnull=True ): taxes[entry.employee_id] += entry.amount tax_counts[entry.employee_id] += 1 off_cycle = list( PayrollOffCyclePayment.objects.filter( paid_on__range=(start, end), currency=currency, voided_at__isnull=True, ).select_related("source_period") ) cash_by_employee = defaultdict(lambda: ZERO) cash_by_month = defaultdict(lambda: ZERO) for payment in off_cycle: cash_by_employee[payment.employee_id] += payment.amount cash_by_month[payment.paid_on.strftime("%Y-%m")] += payment.amount rows = [] month_keys = list(months(start, end)) comparison_months = sorted( {month for employee_pay in pay.values() for month in employee_pay} ) demographic = {month: [] for month in month_keys} for employee in employees: actions = list(employee.actions.all()) tenure = calculate_employee_tenure_days(employee, end, actions=actions) had_future_hire = any( a.action in ["hired", "rehired"] and timezone.localtime(a.date).date() > end for a in actions ) state = resolve_employee_personnel_state(employee, end, actions=actions) employed = (tenure is not None and state.status != "dismissed") or ( tenure is None and employee.is_active and not had_future_hire ) salary = pay[employee.pk] records = attendance[employee.pk] employed_at_start = ( calculate_employee_tenure_days(employee, start, actions=actions) is not None and resolve_employee_personnel_state( employee, start, actions=actions ).status != "dismissed" ) worked_during_period = (tenure or 0) > ( calculate_employee_tenure_days(employee, start, actions=actions) or 0 ) if ( not employed and not employed_at_start and not worked_during_period and not salary and not records and not tax_counts[employee.pk] and not cash_by_employee[employee.pk] ): continue vacation, sick = absence_days(employee, actions, start, end) age = None if employee.birth_date and employee.birth_date <= end: born = employee.birth_date age = end.year - born.year - ((end.month, end.day) < (born.month, born.day)) first = salary.get(comparison_months[0]) if comparison_months else None last = salary.get(comparison_months[-1]) if comparison_months else None growth = ( (last / first - 1) * 100 if len(comparison_months) > 1 and first and last is not None else None ) row = { "id": employee.pk, "name": " ".join( filter( None, [employee.last_name, employee.first_name, employee.patronymic] ) ), "employed": employed, "gender": employee.gender, "age": age, "tenure_days": tenure, "salary_total": sum(salary.values(), ZERO), "salary_months": len(salary), "average_salary": mean(list(salary.values())), "salary_growth_percent": growth, "worked_days": sum( 1 for r in records.values() if r.work_hours is not None and r.work_hours > 0 and not r.is_absent ), "attendance_days": len(records), "vacation_days": vacation, "sick_days": sick, "tax_paid": taxes[employee.pk] if tax_counts[employee.pk] else None, "tax_payments": tax_counts[employee.pk], "off_cycle_paid": cash_by_employee[employee.pk], } rows.append(row) for month in month_keys: month_start = date.fromisoformat(month + "-01") as_of = min( end, (month_start.replace(day=28) + timedelta(days=4)).replace(day=1) - timedelta(days=1), ) service = calculate_employee_tenure_days(employee, as_of, actions=actions) if ( service is not None and resolve_employee_personnel_state( employee, as_of, actions=actions ).status != "dismissed" ): born = employee.birth_date historical_age = ( as_of.year - born.year - ((as_of.month, as_of.day) < (born.month, born.day)) if born and born <= as_of else None ) demographic[month].append((service, historical_age, employee.gender)) current = [r for r in rows if r["employed"]] comparable = [r for r in rows if r["salary_growth_percent"] is not None] ordered = sorted(comparable, key=lambda r: (r["salary_growth_percent"], r["id"])) monthly = [] for month in month_keys: salaries = [p[month] for p in pay.values() if month in p] population = demographic[month] monthly.append( { "id": month, "average_salary": mean(salaries), "salary_total": sum(salaries, ZERO), "salary_employees": len(salaries), "off_cycle_paid": cash_by_month[month], "headcount_with_history": len(population), "average_tenure_days": mean([Decimal(p[0]) for p in population]), "average_age": mean( [Decimal(p[1]) for p in population if p[1] is not None] ), "men": sum(p[2] == 1 for p in population), "women": sum(p[2] == 2 for p in population), } ) return { "employees": rows, "monthly": monthly, "off_cycle_payments": [ { "id": payment.pk, "name": payment.payee_name, "paid_on": payment.paid_on, "amount": payment.amount, "source_period": ( payment.source_period.name or payment.source_period.code ) if payment.source_period else "—", "note": payment.note, } for payment in off_cycle ], "summary": { "off_cycle_paid": sum((payment.amount for payment in off_cycle), ZERO), "off_cycle_payment_count": len(off_cycle), "growth_from": comparison_months[0] if comparison_months else None, "growth_to": comparison_months[-1] if comparison_months else None, "headcount": len(current), "men": sum(r["gender"] == 1 for r in current), "women": sum(r["gender"] == 2 for r in current), "unspecified_gender": sum(r["gender"] == 0 for r in current), "average_age": mean( [Decimal(r["age"]) for r in current if r["age"] is not None] ), "average_tenure_days": mean( [ Decimal(r["tenure_days"]) for r in current if r["tenure_days"] is not None ] ), "missing_hire_dates": sum(r["tenure_days"] is None for r in current), "missing_birth_dates": sum(r["age"] is None for r in current), "average_salary": mean( [value for p in pay.values() for value in p.values()] ), "salary_total": sum((sum(p.values(), ZERO) for p in pay.values()), ZERO), "tax_paid": sum(taxes.values(), ZERO) if any(tax_counts.values()) else None, "fastest_growth": ordered[-1]["id"] if ordered else None, "slowest_growth": ordered[0]["id"] if ordered else None, }, }