Add personio-headcount skill (migrated from plugin)

2026-03-05 13:57:46 +00:00
parent 4208187250
commit b1999dce35
1 changed files with 435 additions and 0 deletions
@@ -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 <skill-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("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+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'<font color="{color.hexval()}">{value}</font>',
+                  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 <b>{len(active)}</b> active employees spread across "
+        f"<b>{len(team_hc)}</b> 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 <b>{len(dept_hc)}</b> 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, <b>{total_joiners}</b> employees have joined or are scheduled to join. "
+        f"<b>{future_joiners}</b> 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, <b>{total_leavers}</b> employees have left or are scheduled to leave. "
+        f"<b>{future_leavers}</b> 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()