您的员工入职数据模板

import pandas as pd
import numpy as np
from datetime import datetime

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># The input file exported from BambooHR.
BAMBOOHR_EXPORT_FILE = class="hl-string">'bamboohr_onboarding_export.csv'

class=class="hl-string">"hl-comment"># The final output file ready for ProcessMind.
EVENT_LOG_FILE = class="hl-string">'event_log.csv'

class=class="hl-string">"hl-comment"># This dictionary maps the column names from your BambooHR export to the desired Activity Name.
class=class="hl-string">"hl-comment"># You MUST update the keys (the column names) to exactly match the headers in your exported CSV file.
class=class="hl-string">"hl-comment"># For example, if your class="hl-string">'Background Check Completed' task completion date column is named class="hl-string">'Onboarding: Background Check - Completed On', update it below.
ACTIVITY_MAPPING = {
    class="hl-string">'[Offer Accepted Date Custom Field]': class="hl-string">'Offer Accepted',
    class="hl-string">'[Background Check Initiated Task Completion Date]': class="hl-string">'Background Check Initiated',
    class="hl-string">'[Background Check Completed Task Completion Date]': class="hl-string">'Background Check Completed',
    class="hl-string">'[Paperwork Packet Sent Task Completion Date]': class="hl-string">'Paperwork Packet Sent',
    class="hl-string">'[Paperwork Packet Completed Task Completion Date]': class="hl-string">'Paperwork Packet Completed',
    class="hl-string">'Hire Date': class="hl-string">'Employee Profile Created',
    class="hl-string">'[IT Provisioning Requested Task Completion Date]': class="hl-string">'IT Provisioning Requested',
    class="hl-string">'[Equipment Ordered Task Completion Date]': class="hl-string">'Equipment Ordered',
    class="hl-string">'[Equipment Delivered Task Completion Date]': class="hl-string">'Equipment Delivered',
    class="hl-string">'[First Day Orientation Completed Task Completion Date]': class="hl-string">'First Day Orientation Completed',
    class="hl-string">'[Mandatory Training Assigned Date]': class="hl-string">'Mandatory Training Assigned',
    class="hl-string">'[Team Introduction Completed Task Completion Date]': class="hl-string">'Team Introduction Completed',
    class="hl-string">'[Mandatory Training Completed Date]': class="hl-string">'Mandatory Training Completed',
    class="hl-string">'[30-Day Check-in Completed Task Completion Date]': class="hl-string">'30-Day Check-in Completed',
    class="hl-string">'[Onboarding Tasks Completed Date]': class="hl-string">'Onboarding Tasks Completed'
}

class=class="hl-string">"hl-comment"># --- Main Script ---
def transform_bamboohr_export():
    class="hl-string">"""Loads the wide BambooHR report, transforms it into a long event log format, and saves it to a new CSV."""
    print(fclass="hl-string">"Reading BambooHR export from '{BAMBOOHR_EXPORT_FILE}'...")
    try:
        df = pd.read_csv(BAMBOOHR_EXPORT_FILE)
    except FileNotFoundError:
        print(fclass="hl-string">"Error: The file '{BAMBOOHR_EXPORT_FILE}' was not found. Please ensure it is in the same directory as the script.")
        return

    print(class="hl-string">"Identifying attribute and activity columns...")
    class=class="hl-string">"hl-comment"># Use Employee # as the unique case identifier.
    id_vars = [class="hl-string">'Employee class="hl-comment">#']
    
    class=class="hl-string">"hl-comment"># Define recommended attributes to keep for each event.
    class=class="hl-string">"hl-comment"># Update these column names if they are different in your export.
    attribute_vars = [
        class="hl-string">'Department', 
        class="hl-string">'Job Title', 
        class="hl-string">'Location', 
        class="hl-string">'Hiring Manager'
    ]

    class=class="hl-string">"hl-comment"># All columns that exist in the dataframe and are in our mapping are treated as value_vars for melting.
    value_vars = [col for col in ACTIVITY_MAPPING.keys() if col in df.columns]
    
    if not value_vars:
        print(class="hl-string">"Error: None of the specified activity columns in ACTIVITY_MAPPING were found in the CSV.")
        print(class="hl-string">"Please check your column names.")
        return

    class=class="hl-string">"hl-comment"># Ensure all identifier and attribute columns are present.
    all_id_vars = id_vars + attribute_vars
    missing_cols = [col for col in all_id_vars if col not in df.columns]
    if missing_cols:
        print(fclass="hl-string">"Warning: The following columns were not found and will be skipped: {', '.join(missing_cols)}")
        all_id_vars = [col for col in all_id_vars if col in df.columns]

    print(class="hl-string">"Unpivoting data to create event log...")
    class=class="hl-string">"hl-comment"># Unpivot the dataframe from wide to long format.
    event_log_df = df.melt(
        id_vars=all_id_vars,
        value_vars=value_vars,
        var_name=class="hl-string">'ActivityName',
        value_name=class="hl-string">'EventTimestamp'
    )

    class=class="hl-string">"hl-comment"># Drop rows where the event did not occur (timestamp is null).
    event_log_df.dropna(subset=[class="hl-string">'EventTimestamp'], inplace=True)
    
    print(class="hl-string">"Formatting columns for ProcessMind...")
    class=class="hl-string">"hl-comment"># Replace the source column names with the clean activity names.
    event_log_df[class="hl-string">'ActivityName'] = event_log_df[class="hl-string">'ActivityName'].map(ACTIVITY_MAPPING)

    class=class="hl-string">"hl-comment"># Rename columns to match the required schema.
    event_log_df.rename(columns={class="hl-string">'Employee class="hl-comment">#': class="hl-string">'OnboardingCase'}, inplace=True)
    event_log_df[class="hl-string">'EmployeeId'] = event_log_df[class="hl-string">'OnboardingCase']

    class=class="hl-string">"hl-comment"># Add static columns required by the schema.
    event_log_df[class="hl-string">'SourceSystem'] = class="hl-string">'BambooHR'
    event_log_df[class="hl-string">'LastDataUpdate'] = datetime.now().strftime(class="hl-string">'%Y-%m-%d %H:%M:%S')
    event_log_df[class="hl-string">'EndTimestamp'] = event_log_df[class="hl-string">'EventTimestamp'] class=class="hl-string">"hl-comment"># Assuming point-in-time events
    
    class=class="hl-string">"hl-comment"># Ensure timestamp format is consistent.
    event_log_df[class="hl-string">'EventTimestamp'] = pd.to_datetime(event_log_df[class="hl-string">'EventTimestamp']).dt.strftime(class="hl-string">'%Y-%m-%d %H:%M:%S')
    event_log_df[class="hl-string">'EndTimestamp'] = pd.to_datetime(event_log_df[class="hl-string">'EndTimestamp']).dt.strftime(class="hl-string">'%Y-%m-%d %H:%M:%S')

    class=class="hl-string">"hl-comment"># Reorder columns for clarity.
    final_columns = [
        class="hl-string">'OnboardingCase', class="hl-string">'ActivityName', class="hl-string">'EventTimestamp', class="hl-string">'SourceSystem', class="hl-string">'LastDataUpdate',
        class="hl-string">'EmployeeId', class="hl-string">'Department', class="hl-string">'Job Title', class="hl-string">'Location', class="hl-string">'HiringManager', class="hl-string">'EndTimestamp'
    ]
    class=class="hl-string">"hl-comment"># Filter to only include columns that exist.
    final_columns = [col for col in final_columns if col in event_log_df.columns]
    event_log_df = event_log_df[final_columns]

    class=class="hl-string">"hl-comment"># Sort the data.
    event_log_df.sort_values(by=[class="hl-string">'OnboardingCase', class="hl-string">'EventTimestamp'], inplace=True)

    print(fclass="hl-string">"Saving event log to '{EVENT_LOG_FILE}'...")
    event_log_df.to_csv(EVENT_LOG_FILE, index=False)
    print(class="hl-string">"Transformation complete.")

if __name__ == class="hl-string">'__main__':
    transform_bamboohr_export()

import requests
import pandas as pd
from datetime import datetime

class=class="hl-string">"hl-comment"># --- Configuration Settings --- #
class=class="hl-string">"hl-comment"># Replace with your companyclass="hl-string">'s subdomain (e.g., 'yourcompany' from yourcompany.bamboohr.com)
COMPANY_SUBDOMAIN = class="hl-string">'[Your Company Subdomain]'
class=class="hl-string">"hl-comment"># Replace with your BambooHR API Key
API_KEY = class="hl-string">'[Your BambooHR API Key]'

class=class="hl-string">"hl-comment"># Define the date range for new hires to extract (based on hire date)
START_DATE = class="hl-string">'2023-01-01'
END_DATE = class="hl-string">'2023-12-31'

SOURCE_SYSTEM = class="hl-string">'BambooHR'

class=class="hl-string">"hl-comment"># --- Critical: Customize this mapping to match your BambooHR task names --- #
TASK_TO_ACTIVITY_MAP = {
    class="hl-string">'Background Check': class="hl-string">'Background Check Completed',
    class="hl-string">'Send New Hire Packet': class="hl-string">'Paperwork Packet Sent',
    class="hl-string">'Complete New Hire Packet': class="hl-string">'Paperwork Packet Completed',
    class="hl-string">'IT Provisioning Request': class="hl-string">'IT Provisioning Requested',
    class="hl-string">'Order Equipment': class="hl-string">'Equipment Ordered',
    class="hl-string">'Equipment Delivered': class="hl-string">'Equipment Delivered',
    class="hl-string">'First Day Orientation': class="hl-string">'First Day Orientation Completed',
    class="hl-string">'Introduce to Team': class="hl-string">'Team Introduction Completed',
    class="hl-string">'30-Day Check-in': class="hl-string">'30-Day Check-in Completed',
    class="hl-string">'Complete Onboarding': class="hl-string">'Onboarding Tasks Completed'
}

class=class="hl-string">"hl-comment"># --- Main Script Logic --- #

def get_new_hires(start_date, end_date):
    class="hl-string">"""Fetches employees hired within the specified date range."""
    url = fclass="hl-string">'https://api.bamboohr.com/api/gateway.php/{COMPANY_SUBDOMAIN}/v1/reports/custom'
    headers = {class="hl-string">'Accept': class="hl-string">'application/json'}
    auth = (API_KEY, class="hl-string">'x')
    
    payload = {
        class="hl-string">'title': class="hl-string">'New Hire Report',
        class="hl-string">'fields': [class="hl-string">'employeeId', class="hl-string">'created', class="hl-string">'hireDate', class="hl-string">'department', class="hl-string">'jobTitle', class="hl-string">'location', class="hl-string">'supervisor'],
        class="hl-string">'filters': {
            class="hl-string">'lastChanged': {
                class="hl-string">'includeNull': class="hl-string">'false'
            },
            class="hl-string">'hireDate': {
                class="hl-string">'operator': class="hl-string">'between',
                class="hl-string">'value': start_date,
                class="hl-string">'value2': end_date
            }
        }
    }
    
    print(fclass="hl-string">"Fetching new hires from {start_date} to {end_date}...")
    response = requests.post(url, auth=auth, headers=headers, json=payload)
    response.raise_for_status()
    data = response.json()
    print(fclass="hl-string">"Found {len(data.get('employees', []))} new hires.")
    return data.get(class="hl-string">'employees', [])

def get_onboarding_tasks(employee_id):
    class="hl-string">"""Fetches onboarding tasks for a specific employee."""
    url = fclass="hl-string">'https://api.bamboohr.com/api/gateway.php/{COMPANY_SUBDOMAIN}/v1/employees/{employee_id}/onboarding/tasks'
    headers = {class="hl-string">'Accept': class="hl-string">'application/json'}
    auth = (API_KEY, class="hl-string">'x')
    response = requests.get(url, auth=auth, headers=headers)
    if response.status_code == 200:
        return response.json().get(class="hl-string">'tasks', [])
    return []

def get_training_records(employee_id):
    class="hl-string">"""Fetches training records for a specific employee."""
    url = fclass="hl-string">'https://api.bamboohr.com/api/gateway.php/{COMPANY_SUBDOMAIN}/v1/employees/{employee_id}/training/'
    headers = {class="hl-string">'Accept': class="hl-string">'application/json'}
    auth = (API_KEY, class="hl-string">'x')
    response = requests.get(url, auth=auth, headers=headers)
    if response.status_code == 200:
        return response.json()
    return []

def main():
    class="hl-string">"""Main function to orchestrate the extraction and transformation."""
    try:
        new_hires = get_new_hires(START_DATE, END_DATE)
    except requests.exceptions.HTTPError as e:
        print(fclass="hl-string">"Error fetching new hires: {e}")
        print(fclass="hl-string">"Response content: {e.response.text}")
        return

    event_log = []
    last_data_update = datetime.utcnow().isoformat() + class="hl-string">'Z'

    for employee in new_hires:
        emp_id = employee[class="hl-string">'employeeId']
        print(fclass="hl-string">"Processing Employee ID: {emp_id}...")

        base_event = {
            class="hl-string">'OnboardingCase': emp_id,
            class="hl-string">'EmployeeId': emp_id,
            class="hl-string">'Department': employee.get(class="hl-string">'department'),
            class="hl-string">'JobTitle': employee.get(class="hl-string">'jobTitle'),
            class="hl-string">'Location': employee.get(class="hl-string">'location'),
            class="hl-string">'HiringManager': employee.get(class="hl-string">'supervisor'),
            class="hl-string">'SourceSystem': SOURCE_SYSTEM,
            class="hl-string">'LastDataUpdate': last_data_update
        }

        class=class="hl-string">"hl-comment"># 1. Offer Accepted (inferred from hireDate)
        if employee.get(class="hl-string">'hireDate'):
            event_log.append({
                **base_event,
                class="hl-string">'ActivityName': class="hl-string">'Offer Accepted',
                class="hl-string">'EventTimestamp': fclass="hl-string">"{employee['hireDate']}T09:00:00Z",
                class="hl-string">'EndTimestamp': fclass="hl-string">"{employee['hireDate']}T09:00:00Z"
            })

        class=class="hl-string">"hl-comment"># 2. Employee Profile Created
        if employee.get(class="hl-string">'created'):
            event_log.append({
                **base_event,
                class="hl-string">'ActivityName': class="hl-string">'Employee Profile Created',
                class="hl-string">'EventTimestamp': employee[class="hl-string">'created'].replace(class="hl-string">' ', class="hl-string">'T'),
                class="hl-string">'EndTimestamp': employee[class="hl-string">'created'].replace(class="hl-string">' ', class="hl-string">'T')
            })
        
        class=class="hl-string">"hl-comment"># 3. Onboarding Task Events
        tasks = get_onboarding_tasks(emp_id)
        task_completion_dates = []
        for task in tasks:
            activity_name = TASK_TO_ACTIVITY_MAP.get(task.get(class="hl-string">'title'))
            if activity_name and task.get(class="hl-string">'completedDate'):
                event_log.append({
                    **base_event,
                    class="hl-string">'ActivityName': activity_name,
                    class="hl-string">'EventTimestamp': fclass="hl-string">"{task['completedDate']}T17:00:00Z",
                    class="hl-string">'EndTimestamp': fclass="hl-string">"{task['completedDate']}T17:00:00Z"
                })
                task_completion_dates.append(task.get(class="hl-string">'completedDate'))

            class=class="hl-string">"hl-comment"># Handle Background Check Initiated (assumed from task creation)
            if task.get(class="hl-string">'title') == class="hl-string">'Background Check' and task.get(class="hl-string">'createdDate'):
                 event_log.append({
                    **base_event,
                    class="hl-string">'ActivityName': class="hl-string">'Background Check Initiated',
                    class="hl-string">'EventTimestamp': fclass="hl-string">"{task['createdDate']}T17:00:00Z",
                    class="hl-string">'EndTimestamp': fclass="hl-string">"{task['createdDate']}T17:00:00Z"
                })

        class=class="hl-string">"hl-comment"># 4. Training Events
        trainings = get_training_records(emp_id)
        for training in trainings:
            if training.get(class="hl-string">'type') in [class="hl-string">'Compliance Training', class="hl-string">'Role-Specific Training', class="hl-string">'Mandatory Training']:
                if training.get(class="hl-string">'assignedDate'):
                    event_log.append({
                        **base_event,
                        class="hl-string">'ActivityName': class="hl-string">'Mandatory Training Assigned',
                        class="hl-string">'EventTimestamp': fclass="hl-string">"{training['assignedDate']}T17:00:00Z",
                        class="hl-string">'EndTimestamp': fclass="hl-string">"{training['assignedDate']}T17:00:00Z"
                    })
                if training.get(class="hl-string">'completed'):
                    event_log.append({
                        **base_event,
                        class="hl-string">'ActivityName': class="hl-string">'Mandatory Training Completed',
                        class="hl-string">'EventTimestamp': fclass="hl-string">"{training['completed']}T17:00:00Z",
                        class="hl-string">'EndTimestamp': fclass="hl-string">"{training['completed']}T17:00:00Z"
                    })

    if not event_log:
        print(class="hl-string">"No events were generated. Please check your date range and API permissions.")
        return

    df = pd.DataFrame(event_log)
    df[class="hl-string">'EventTimestamp'] = pd.to_datetime(df[class="hl-string">'EventTimestamp'])
    df = df.sort_values(by=[class="hl-string">'OnboardingCase', class="hl-string">'EventTimestamp'])
    
    output_filename = class="hl-string">'bamboohr_onboarding_event_log.csv'
    df.to_csv(output_filename, index=False, date_format=class="hl-string">'%Y-%m-%d %H:%M:%S')
    print(fclass="hl-string">"\nExtraction complete. Event log saved to '{output_filename}'.")

if __name__ == class="hl-string">'__main__':
    main()