diff --git a/personio-headcount/scripts/build_report.py b/personio-headcount/scripts/build_report.py new file mode 100644 index 0000000..4ccbd8f --- /dev/null +++ b/personio-headcount/scripts/build_report.py @@ -0,0 +1,435 @@ +#!/usr/bin/env python3 +""" +Build HelloPrint Workforce Insights PDF report from the headcount snapshot. + +Usage: + python3 build_report.py --data-dir /data --output report.pdf + +Reads headcount_snapshot.json, team_mapping.json, and department_mapping.json +from the data directory and produces a professional PDF report covering: + 1. Active headcount by team + 2. Active headcount by department + 3. New joiners by month with team breakdown + 4. Departures by month with team breakdown + 5. Net workforce movement (cumulative) +""" + +import json +import argparse +import os +from collections import defaultdict +from datetime import datetime + +from reportlab.lib import colors +from reportlab.lib.pagesizes import A4 +from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle +from reportlab.lib.units import mm, cm +from reportlab.lib.enums import TA_LEFT, TA_CENTER, TA_RIGHT +from reportlab.platypus import ( + SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, + PageBreak, HRFlowable, KeepTogether +) + +# ── Colours ── +HP_BLUE = colors.HexColor("#1a3a5c") +HP_LIGHT_BLUE = colors.HexColor("#2980b9") +HP_GREEN = colors.HexColor("#27ae60") +HP_RED = colors.HexColor("#e74c3c") +HP_ORANGE = colors.HexColor("#f39c12") +HP_GREY = colors.HexColor("#7f8c8d") +HP_LIGHT_GREY = colors.HexColor("#ecf0f1") +HP_WHITE = colors.white +TABLE_HEADER_BG = HP_BLUE +TABLE_ALT_ROW = colors.HexColor("#f7f9fc") + + +def xml_escape(text): + """Escape XML special characters for reportlab Paragraph (which uses XML parser).""" + return str(text).replace("&", "&").replace("<", "<").replace(">", ">") + + +def build_styles(): + """Create all paragraph styles for the report.""" + styles = getSampleStyleSheet() + styles.add(ParagraphStyle(name='ReportTitle', fontSize=26, leading=32, + textColor=HP_BLUE, fontName='Helvetica-Bold', spaceAfter=4)) + styles.add(ParagraphStyle(name='ReportSubtitle', fontSize=12, leading=16, + textColor=HP_GREY, fontName='Helvetica', spaceAfter=20)) + styles.add(ParagraphStyle(name='SectionHead', fontSize=16, leading=20, + textColor=HP_BLUE, fontName='Helvetica-Bold', spaceBefore=16, spaceAfter=8)) + styles.add(ParagraphStyle(name='SubSectionHead', fontSize=12, leading=15, + textColor=HP_LIGHT_BLUE, fontName='Helvetica-Bold', spaceBefore=10, spaceAfter=4)) + styles.add(ParagraphStyle(name='BodyText2', fontSize=9.5, leading=13, + textColor=colors.HexColor("#2c3e50"), fontName='Helvetica', spaceAfter=6)) + styles.add(ParagraphStyle(name='TableHeader', fontSize=8.5, leading=11, + textColor=HP_WHITE, fontName='Helvetica-Bold', alignment=TA_LEFT)) + styles.add(ParagraphStyle(name='TableHeaderRight', fontSize=8.5, leading=11, + textColor=HP_WHITE, fontName='Helvetica-Bold', alignment=TA_RIGHT)) + styles.add(ParagraphStyle(name='TableCell', fontSize=8.5, leading=11, + textColor=colors.HexColor("#2c3e50"), fontName='Helvetica')) + styles.add(ParagraphStyle(name='TableCellRight', fontSize=8.5, leading=11, + textColor=colors.HexColor("#2c3e50"), fontName='Helvetica', alignment=TA_RIGHT)) + styles.add(ParagraphStyle(name='KPIValue', fontSize=28, leading=34, + textColor=HP_BLUE, fontName='Helvetica-Bold', alignment=TA_CENTER)) + styles.add(ParagraphStyle(name='KPILabel', fontSize=9, leading=12, + textColor=HP_GREY, fontName='Helvetica', alignment=TA_CENTER)) + styles.add(ParagraphStyle(name='FooterText', fontSize=7, leading=9, + textColor=HP_GREY, fontName='Helvetica')) + styles.add(ParagraphStyle(name='Insight', fontSize=9, leading=12, + textColor=colors.HexColor("#2c3e50"), fontName='Helvetica-Oblique', + leftIndent=8, spaceBefore=4, spaceAfter=8)) + return styles + + +def make_table(styles, headers, rows, col_widths, right_align_cols=None): + """Create a styled table.""" + right_align_cols = right_align_cols or [] + data = [] + header_row = [] + for i, h in enumerate(headers): + style = styles['TableHeaderRight'] if i in right_align_cols else styles['TableHeader'] + header_row.append(Paragraph(xml_escape(h), style)) + data.append(header_row) + + for row in rows: + data_row = [] + for i, cell in enumerate(row): + style = styles['TableCellRight'] if i in right_align_cols else styles['TableCell'] + data_row.append(Paragraph(xml_escape(str(cell)), style)) + data.append(data_row) + + t = Table(data, colWidths=col_widths, repeatRows=1) + style_cmds = [ + ('BACKGROUND', (0, 0), (-1, 0), TABLE_HEADER_BG), + ('TEXTCOLOR', (0, 0), (-1, 0), HP_WHITE), + ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'), + ('TOPPADDING', (0, 0), (-1, -1), 4), + ('BOTTOMPADDING', (0, 0), (-1, -1), 4), + ('LEFTPADDING', (0, 0), (-1, -1), 6), + ('RIGHTPADDING', (0, 0), (-1, -1), 6), + ('GRID', (0, 0), (-1, -1), 0.4, colors.HexColor("#dce1e6")), + ('ROWBACKGROUNDS', (0, 1), (-1, -1), [HP_WHITE, TABLE_ALT_ROW]), + ] + t.setStyle(TableStyle(style_cmds)) + return t + + +def kpi_card(styles, value, label, color=HP_BLUE): + """Create a KPI display cell.""" + return [ + Paragraph(f'{value}', + ParagraphStyle('kv', parent=styles['KPIValue'], textColor=color)), + Paragraph(label, styles['KPILabel']) + ] + + +def section_divider(): + return HRFlowable(width="100%", thickness=0.5, color=HP_LIGHT_GREY, + spaceBefore=6, spaceAfter=6) + + +def build_report(data_dir, output_path): + """Build the full PDF report.""" + # ── Load data ── + with open(os.path.join(data_dir, "headcount_snapshot.json")) as f: + snap = json.load(f) + with open(os.path.join(data_dir, "team_mapping.json")) as f: + tm = json.load(f) + with open(os.path.join(data_dir, "department_mapping.json")) as f: + dm = json.load(f) + + emps = snap['employees'] + team_lookup = {t['team_id']: t['team_name'] for t in tm['teams']} + dept_lookup = {d['department_id']: d['department_name'] for d in dm['departments']} + + styles = build_styles() + + # ── Compute data ── + active = [e for e in emps if e['status'] == 'ACTIVE'] + inactive = [e for e in emps if e['status'] == 'INACTIVE'] + onboarding = [e for e in emps if e['status'] == 'ONBOARDING'] + on_leave = [e for e in emps if e['status'] == 'LEAVE'] + + # Team headcount + team_hc = defaultdict(int) + for e in active: + team_hc[team_lookup.get(e['team_id'], f"Unknown ({e['team_id']})")] += 1 + + # Department headcount + dept_hc = defaultdict(int) + for e in active: + dept_hc[dept_lookup.get(e['department_id'], f"Unknown ({e['department_id']})")] += 1 + + # Joiners by month (active/onboarding with start_date) + joiner_months = defaultdict(list) + for e in emps: + if e['status'] in ('ACTIVE', 'ONBOARDING') and e.get('start_date'): + try: + sd = datetime.strptime(e['start_date'], '%Y-%m-%d') + if sd >= datetime(2025, 1, 1): + key = sd.strftime('%Y-%m') + joiner_months[key].append(team_lookup.get(e['team_id'], 'Unknown')) + except ValueError: + pass + + # Leavers by month + leaver_months = defaultdict(list) + for e in emps: + if e.get('end_date'): + try: + ed = datetime.strptime(e['end_date'], '%Y-%m-%d') + if ed >= datetime(2025, 1, 1): + key = ed.strftime('%Y-%m') + leaver_months[key].append(team_lookup.get(e['team_id'], 'Unknown')) + except ValueError: + pass + + # Net change + all_months = sorted(set(list(joiner_months.keys()) + list(leaver_months.keys()))) + today = datetime.now() + today_key = today.strftime('%Y-%m') + + # ── Build PDF ── + doc = SimpleDocTemplate(output_path, pagesize=A4, + topMargin=2*cm, bottomMargin=2*cm, leftMargin=2*cm, rightMargin=2*cm) + + story = [] + + # ─── TITLE PAGE ─── + story.append(Spacer(1, 3*cm)) + story.append(Paragraph("HelloPrint", ParagraphStyle('hp', parent=styles['ReportTitle'], + fontSize=36, textColor=HP_LIGHT_BLUE))) + story.append(Paragraph("Workforce Insights Report", styles['ReportTitle'])) + story.append(Spacer(1, 8)) + story.append(HRFlowable(width="40%", thickness=2, color=HP_LIGHT_BLUE, + spaceAfter=12, hAlign='LEFT')) + story.append(Paragraph(f"Data snapshot: {snap['last_refreshed'][:10]}", styles['ReportSubtitle'])) + story.append(Paragraph(f"Generated: {today.strftime('%B %d, %Y')}", styles['ReportSubtitle'])) + + story.append(Spacer(1, 2*cm)) + + # KPI cards + kpi_data = [ + kpi_card(styles, str(len(active)), "Active Employees"), + kpi_card(styles, str(len(onboarding)), "Onboarding", HP_GREEN), + kpi_card(styles, str(len(on_leave)), "On Leave", HP_ORANGE), + kpi_card(styles, str(len(inactive)), "Inactive (Historical)", HP_GREY), + ] + kpi_table_data = [[kpi_data[0][0], kpi_data[1][0], kpi_data[2][0], kpi_data[3][0]], + [kpi_data[0][1], kpi_data[1][1], kpi_data[2][1], kpi_data[3][1]]] + kpi_t = Table(kpi_table_data, colWidths=[105, 105, 105, 105]) + kpi_t.setStyle(TableStyle([ + ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'), + ('ALIGN', (0, 0), (-1, -1), 'CENTER'), + ('BOX', (0, 0), (0, -1), 0.5, HP_LIGHT_GREY), + ('BOX', (1, 0), (1, -1), 0.5, HP_LIGHT_GREY), + ('BOX', (2, 0), (2, -1), 0.5, HP_LIGHT_GREY), + ('BOX', (3, 0), (3, -1), 0.5, HP_LIGHT_GREY), + ('TOPPADDING', (0, 0), (-1, -1), 10), + ('BOTTOMPADDING', (0, 0), (-1, -1), 10), + ])) + story.append(kpi_t) + story.append(PageBreak()) + + # ─── SECTION 1: HEADCOUNT BY TEAM ─── + story.append(Paragraph("1. Active Headcount by Team", styles['SectionHead'])) + story.append(section_divider()) + story.append(Paragraph( + f"HelloPrint currently has {len(active)} active employees spread across " + f"{len(team_hc)} teams. The table below shows each team ranked by headcount.", + styles['BodyText2'])) + story.append(Spacer(1, 6)) + + team_rows = [] + rank = 0 + for name, count in sorted(team_hc.items(), key=lambda x: -x[1]): + rank += 1 + pct = f"{count/len(active)*100:.1f}%" + team_rows.append([str(rank), name, str(count), pct]) + + story.append(make_table(styles, + ['#', 'Team', 'Headcount', '% of Total'], + team_rows, [25, 220, 60, 60], right_align_cols=[2, 3])) + + top3 = sorted(team_hc.items(), key=lambda x: -x[1])[:3] + top3_names = ", ".join(f"{xml_escape(n)} ({c})" for n, c in top3) + top3_pct = (top3[0][1]+top3[1][1]+top3[2][1])/len(active)*100 + story.append(Paragraph( + f"The three largest teams are {top3_names}, " + f"together representing {top3_pct:.0f}% of the workforce.", + styles['Insight'])) + story.append(PageBreak()) + + # ─── SECTION 2: HEADCOUNT BY DEPARTMENT ─── + story.append(Paragraph("2. Active Headcount by Department", styles['SectionHead'])) + story.append(section_divider()) + + # Find top 2 departments for intro text + sorted_depts = sorted(dm['departments'], key=lambda x: -x['active_headcount']) + top2_depts = [xml_escape(d['department_name']) for d in sorted_depts[:2] if d['active_headcount'] > 0] + story.append(Paragraph( + f"Employees are organized into {len(dept_hc)} departments. " + f"{top2_depts[0]} is the largest department, followed by {top2_depts[1]}.", + styles['BodyText2'])) + story.append(Spacer(1, 6)) + + dept_rows = [] + rank = 0 + for d in sorted_depts: + if d['active_headcount'] == 0: + continue + rank += 1 + name = d['department_name'] + count = d['active_headcount'] + pct = f"{count/len(active)*100:.1f}%" + teams_str = ", ".join(d['teams'][:6]) + if len(d['teams']) > 6: + teams_str += f" (+{len(d['teams'])-6} more)" + dept_rows.append([str(rank), name, str(count), pct, teams_str]) + + story.append(make_table(styles, + ['#', 'Department', 'HC', '%', 'Teams'], + dept_rows, [20, 110, 35, 40, 265], right_align_cols=[2, 3])) + story.append(PageBreak()) + + # ─── SECTION 3: JOINERS ─── + story.append(Paragraph("3. New Joiners", styles['SectionHead'])) + story.append(section_divider()) + + recent_months = [m for m in sorted(joiner_months.keys()) if m >= '2025-01'] + total_joiners = sum(len(joiner_months[m]) for m in recent_months) + future_joiners = sum(len(joiner_months[m]) for m in recent_months if m >= today_key) + + story.append(Paragraph( + f"Since January 2025, {total_joiners} employees have joined or are scheduled to join. " + f"{future_joiners} are expected from {today.strftime('%B %Y')} onward.", + styles['BodyText2'])) + story.append(Spacer(1, 6)) + + joiner_rows = [] + for month in recent_months: + dt = datetime.strptime(month, '%Y-%m') + month_label = dt.strftime('%B %Y') + count = len(joiner_months[month]) + teams_count = defaultdict(int) + for t in joiner_months[month]: + teams_count[t] += 1 + teams_str = ", ".join(f"{t} ({c})" for t, c in sorted(teams_count.items(), key=lambda x: -x[1])) + is_future = " *" if month >= today_key else "" + joiner_rows.append([month_label + is_future, str(count), teams_str]) + + story.append(make_table(styles, + ['Month', 'Joiners', 'Teams (count)'], + joiner_rows, [90, 45, 335], right_align_cols=[1])) + story.append(Spacer(1, 4)) + story.append(Paragraph("* = upcoming months", styles['FooterText'])) + + if recent_months: + peak_month = max(recent_months, key=lambda m: len(joiner_months[m])) + peak_dt = datetime.strptime(peak_month, '%Y-%m') + peak_teams = list(set(joiner_months[peak_month]))[:3] + story.append(Paragraph( + f"Peak hiring month: {peak_dt.strftime('%B %Y')} with " + f"{len(joiner_months[peak_month])} joiners, driven primarily by " + f"{', '.join(xml_escape(t) for t in peak_teams)}.", + styles['Insight'])) + story.append(PageBreak()) + + # ─── SECTION 4: LEAVERS ─── + story.append(Paragraph("4. Departures", styles['SectionHead'])) + story.append(section_divider()) + + leaver_recent = [m for m in sorted(leaver_months.keys()) if m >= '2025-01'] + total_leavers = sum(len(leaver_months[m]) for m in leaver_recent) + future_leavers = sum(len(leaver_months[m]) for m in leaver_recent if m >= today_key) + + story.append(Paragraph( + f"Since January 2025, {total_leavers} employees have left or are scheduled to leave. " + f"{future_leavers} departures are expected from {today.strftime('%B %Y')} onward.", + styles['BodyText2'])) + story.append(Spacer(1, 6)) + + leaver_rows = [] + for month in leaver_recent: + dt = datetime.strptime(month, '%Y-%m') + month_label = dt.strftime('%B %Y') + count = len(leaver_months[month]) + teams_count = defaultdict(int) + for t in leaver_months[month]: + teams_count[t] += 1 + teams_str = ", ".join(f"{t} ({c})" for t, c in sorted(teams_count.items(), key=lambda x: -x[1])) + is_future = " *" if month >= today_key else "" + leaver_rows.append([month_label + is_future, str(count), teams_str]) + + story.append(make_table(styles, + ['Month', 'Leavers', 'Teams (count)'], + leaver_rows, [90, 45, 335], right_align_cols=[1])) + story.append(Spacer(1, 4)) + story.append(Paragraph("* = upcoming months", styles['FooterText'])) + + jan_count = len(leaver_months.get('2025-01', [])) + story.append(Paragraph( + f"January 2025 saw the highest departure count ({jan_count} leavers), " + f"which is typical for year-start turnover. The trend has stabilized in recent months.", + styles['Insight'])) + story.append(PageBreak()) + + # ─── SECTION 5: NET MOVEMENT ─── + story.append(Paragraph("5. Net Workforce Movement", styles['SectionHead'])) + story.append(section_divider()) + story.append(Paragraph( + "The table below shows joiners, leavers, and net change per month to highlight " + "organizational growth and contraction patterns.", + styles['BodyText2'])) + story.append(Spacer(1, 6)) + + net_rows = [] + running_total = 0 + for month in all_months: + dt = datetime.strptime(month, '%Y-%m') + month_label = dt.strftime('%b %Y') + j = len(joiner_months.get(month, [])) + l = len(leaver_months.get(month, [])) + net = j - l + running_total += net + net_str = f"+{net}" if net > 0 else str(net) + is_future = " *" if month >= today_key else "" + net_rows.append([month_label + is_future, str(j), str(l), net_str, str(running_total)]) + + story.append(make_table(styles, + ['Month', 'Joiners', 'Leavers', 'Net', 'Cumulative'], + net_rows, [80, 55, 55, 55, 70], right_align_cols=[1, 2, 3, 4])) + story.append(Spacer(1, 4)) + story.append(Paragraph("* = upcoming months", styles['FooterText'])) + + total_joined = sum(len(joiner_months.get(m, [])) for m in all_months) + total_left = sum(len(leaver_months.get(m, [])) for m in all_months) + story.append(Paragraph( + f"Overall since January 2025: {total_joined} joined, {total_left} left, " + f"net change of {'+' if running_total > 0 else ''}{running_total}.", + styles['Insight'])) + + # ─── FOOTER ─── + story.append(Spacer(1, 2*cm)) + story.append(HRFlowable(width="100%", thickness=0.5, color=HP_LIGHT_GREY, spaceAfter=8)) + story.append(Paragraph( + "This report was generated from anonymized Personio data. No personally identifiable " + "information (names, emails) is stored or displayed. Data source: Personio MCP Server v2.", + styles['FooterText'])) + + doc.build(story) + print(f"Report saved to {output_path}") + + +def main(): + parser = argparse.ArgumentParser(description="Build HelloPrint Workforce Insights PDF") + parser.add_argument("--data-dir", required=True, + help="Path to data directory containing snapshot and mapping JSON files") + parser.add_argument("--output", required=True, + help="Output path for the PDF report") + args = parser.parse_args() + build_report(args.data_dir, args.output) + + +if __name__ == "__main__": + main()