Mẫu dữ liệu tiếp nhận nhân viên mới

/*
This is a conceptual representation of the Workday Custom Report configuration.
It is built using the Workday Report Writer UI, not a scriptable language.

**Report Name:** ProcessMind_Onboarding_Event_Log
**Report Type:** Advanced
**Data Source:** Business Process Transactions
*/

// 1. **Data Source Filter Configuration**
// Apply these filters in the 'Filter' tab of the report definition.
FILTER ON BusinessProcess.ProcessType IN ('Hire', 'Onboarding', 'Rescind Business Process')
AND BusinessProcess.TransactionStatus IN ('Successfully Completed', 'Canceled')
AND BusinessProcess.CompletedDateTime BETWEEN [Start Date Prompt] AND [End Date Prompt]

// 2. **Column and Calculated Field Configuration**
// These are configured in the 'Columns' tab. Many require creating 'Calculated Fields'.

{
  // **Required Attributes**
  "OnboardingCase": "Calculated Field (Lookup Related Value): Lookup Employee ID from Business Process -> Worker. Fallback to Candidate ID.",
  "ActivityName": "Calculated Field (Evaluate Expression): See logic below.",
  "EventTime": "Field: Business Process Transaction -> Completed Date/Time",
  "SourceSystem": "Calculated Field (Text Constant): 'Workday Onboarding'",
  "LastDataUpdate": "Calculated Field (Current Moment)",

  // **Recommended Attributes**
  "EmployeeId": "Calculated Field (Lookup Related Value): Lookup Employee ID from Business Process -> Worker",
  "HiringManager": "Calculated Field (Lookup Related Value): Lookup Hiring Manager from Business Process -> Worker",
  "Department": "Calculated Field (Lookup Related Value): Lookup Supervisory Organization from Business Process -> Worker",
  "OnboardingStatus": "Calculated Field (Evaluate Expression): Based on the overall status of the 'Onboarding' business process for the worker.",
  "OfficialStartDate": "Calculated Field (Lookup Related Value): Lookup Hire Date from Business Process -> Worker",
  "ComplianceDueDate": "Calculated Field (Lookup Related Value): Lookup Due Date from specific 'To Do' steps like compliance documents.",
  "SystemInitiatingActivity": "Field: Business Process Transaction -> Initiating Person"
}

// 3. **Logic for 'ActivityName' Calculated Field (Evaluate Expression)**
// This logic maps specific business process events to standardized activity names.
CASE
  WHEN BusinessProcess.ProcessType = 'Hire' AND BusinessProcess.Step CONTAINS 'Offer' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Offer Accepted'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Background Check' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Background Check Initiated'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Background Check Results Received' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Background Check Completed'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Complete Onboarding Paperwork' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Pre-Employment Paperwork Sent'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Review Documents' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Pre-Employment Paperwork Completed'
  WHEN BusinessProcess.ProcessType = 'Hire' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'HR Profile Created'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Request Account Provisioning' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'IT Account Provisioning Initiated'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Request Equipment' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Equipment Ordered'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Confirm Account Provisioning' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'IT Account Provisioning Completed'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Confirm Equipment Delivery' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Equipment Delivered'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'First Day' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Employee Started'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Assign Learning' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Mandatory Training Assigned'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS 'Complete Mandatory Training' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Mandatory Training Completed'
  WHEN BusinessProcess.ProcessType = 'Onboarding' AND BusinessProcess.Step CONTAINS '30 Day Check-in' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN '30-Day Check-in Performed'
  WHEN BusinessProcess.ProcessType = 'Rescind Business Process' AND BusinessProcess.TransactionStatus = 'Successfully Completed' THEN 'Onboarding Canceled'
  ELSE NULL // Exclude rows that don't match a required activity
END

import requests
import pandas as pd
from datetime import datetime, timezone
import xml.etree.ElementTree as ET
import json

class=class="hl-string">"hl-comment"># --- Configuration Settings --- #
class=class="hl-string">"hl-comment"># Replace with your Workday tenant, credentials, and API endpoint
WORKDAY_TENANT = class="hl-string">'[Your_Workday_Tenant]'
WORKDAY_USERNAME = class="hl-string">'[Your_Integration_System_User]'
WORKDAY_PASSWORD = class="hl-string">'[Your_Integration_System_Password]'
WORKDAY_API_ENDPOINT = fclass="hl-string">'https://wd2-impl-services1.workday.com/ccx/api/v1/{WORKDAY_TENANT}'

class=class="hl-string">"hl-comment"># Define the date range for hires to extract
START_DATE = class="hl-string">'2023-01-01T00:00:00Z'
END_DATE = class="hl-string">'2023-12-31T23:59:59Z'

class=class="hl-string">"hl-comment"># Mapping of Workday business process steps to standardized activity names
class=class="hl-string">"hl-comment"># CRITICAL: These keys may need to be adjusted to match your specific Workday configuration
STEP_TO_ACTIVITY_MAPPING = {
    class="hl-string">'Offer': class="hl-string">'Offer Accepted',
    class="hl-string">'Background Check': class="hl-string">'Background Check Initiated',
    class="hl-string">'Review Background Check': class="hl-string">'Background Check Completed',
    class="hl-string">'Onboarding Setup': class="hl-string">'Pre-Employment Paperwork Sent',
    class="hl-string">'Complete Form I-9': class="hl-string">'Pre-Employment Paperwork Completed',
    class="hl-string">'Complete Federal and State Withholding': class="hl-string">'Pre-Employment Paperwork Completed',
    class="hl-string">'Propose Compensation Hire': class="hl-string">'HR Profile Created', class=class="hl-string">"hl-comment"># Often the final step
    class="hl-string">'Create User Account': class="hl-string">'IT Account Provisioning Initiated',
    class="hl-string">'Order Equipment': class="hl-string">'Equipment Ordered',
    class="hl-string">'Review New Hire Equipment': class="hl-string">'Equipment Delivered',
    class="hl-string">'Assign Learning': class="hl-string">'Mandatory Training Assigned',
    class="hl-string">'Complete All Onboarding Questionnaires': class="hl-string">'Mandatory Training Completed',
    class="hl-string">'Manager Check-in: 30 Day': class="hl-string">'30-Day Check-in Performed'
}

def fetch_workday_data(url, params={}):
    class="hl-string">"""Generic function to fetch data from Workday REST API."""
    try:
        response = requests.get(url, auth=(WORKDAY_USERNAME, WORKDAY_PASSWORD), params=params, headers={class="hl-string">'Accept': class="hl-string">'application/json'})
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(fclass="hl-string">"HTTP Error fetching {url}: {e}")
        print(fclass="hl-string">"Response body: {e.response.text}")
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"Request Error fetching {url}: {e}")
    return None

def main():
    class="hl-string">"""Main function to orchestrate the extraction process."""
    print(class="hl-string">"Starting Workday onboarding data extraction...")
    
    class=class="hl-string">"hl-comment"># 1. Fetch workers hired within the date range
    workers_url = fclass="hl-string">'{WORKDAY_API_ENDPOINT}/workers'
    workers_params = {class="hl-string">'hireDate': fclass="hl-string">'gt:{START_DATE},lt:{END_DATE}', class="hl-string">'limit': 100} class=class="hl-string">"hl-comment"># Use paging for >100
    workers_response = fetch_workday_data(workers_url, workers_params)
    
    if not workers_response or class="hl-string">'data' not in workers_response:
        print(class="hl-string">"Failed to fetch workers or no workers found in date range.")
        return

    all_events = []
    last_data_update = datetime.now(timezone.utc).isoformat()

    class=class="hl-string">"hl-comment"># 2. For each worker, fetch their hire business process history
    for worker in workers_response[class="hl-string">'data']:
        employee_id = worker.get(class="hl-string">'id', class="hl-string">'')
        print(fclass="hl-string">"Processing worker: {employee_id}")
        
        class=class="hl-string">"hl-comment"># Find the Hire business process instance
        processes_url = fclass="hl-string">'{WORKDAY_API_ENDPOINT}/workers/{employee_id}/businessProcesses'
        processes_response = fetch_workday_data(processes_url)
        
        if not processes_response or class="hl-string">'data' not in processes_response:
            continue

        hire_process_id = None
        for process in processes_response[class="hl-string">'data']:
            if process.get(class="hl-string">'businessProcess', {}).get(class="hl-string">'descriptor', class="hl-string">'').startswith(class="hl-string">'Hire'):
                hire_process_id = process.get(class="hl-string">'id')
                break

        if not hire_process_id:
            continue

        class=class="hl-string">"hl-comment"># 3. Fetch details and events for the Hire process
        process_details_url = fclass="hl-string">'{WORKDAY_API_ENDPOINT}/values/businessProcessInstances/{hire_process_id}'
        process_details = fetch_workday_data(process_details_url)

        if not process_details:
            continue

        class=class="hl-string">"hl-comment"># Extract common attributes
        onboarding_case_id = hire_process_id
        hire_date = worker.get(class="hl-string">'hireDate', class="hl-string">'')
        hiring_manager = worker.get(class="hl-string">'manager', {}).get(class="hl-string">'descriptor', class="hl-string">'[Not Available]')
        department = worker.get(class="hl-string">'supervisoryOrganization', {}).get(class="hl-string">'descriptor', class="hl-string">'[Not Available]')
        process_status = process_details.get(class="hl-string">'status', {}).get(class="hl-string">'descriptor', class="hl-string">'')

        base_event = {
            class="hl-string">'OnboardingCase': onboarding_case_id,
            class="hl-string">'EmployeeId': employee_id,
            class="hl-string">'HiringManager': hiring_manager,
            class="hl-string">'Department': department,
            class="hl-string">'OfficialStartDate': hire_date,
            class="hl-string">'SourceSystem': class="hl-string">'Workday Onboarding',
            class="hl-string">'LastDataUpdate': last_data_update,
            class="hl-string">'SystemInitiatingActivity': class="hl-string">'Workday',
            class="hl-string">'ComplianceDueDate': None class=class="hl-string">"hl-comment"># Placeholder, needs specific logic if available
        }

        class=class="hl-string">"hl-comment"># Handle canceled onboarding
        if process_status in [class="hl-string">'Canceled', class="hl-string">'Rescinded']:
            event = base_event.copy()
            event[class="hl-string">'ActivityName'] = class="hl-string">'Onboarding Canceled'
            event[class="hl-string">'EventTime'] = process_details.get(class="hl-string">'lastUpdated', class="hl-string">'')
            event[class="hl-string">'OnboardingStatus'] = class="hl-string">'Canceled'
            all_events.append(event)
            continue class=class="hl-string">"hl-comment"># Stop processing other events for this canceled case
        
        class=class="hl-string">"hl-comment"># Add class="hl-string">'Employee Started' event based on hire date
        if hire_date:
            event = base_event.copy()
            event[class="hl-string">'ActivityName'] = class="hl-string">'Employee Started'
            event[class="hl-string">'EventTime'] = fclass="hl-string">"{hire_date}T09:00:00Z" class=class="hl-string">"hl-comment"># Assume start of day
            event[class="hl-string">'OnboardingStatus'] = class="hl-string">'In Progress'
            all_events.append(event)

        class=class="hl-string">"hl-comment"># Process all steps in the business process
        steps = process_details.get(class="hl-string">'steps', [])
        for step in steps:
            step_name = step.get(class="hl-string">'name', class="hl-string">'')
            completion_time = step.get(class="hl-string">'completed', class="hl-string">'')
            
            if step_name in STEP_TO_ACTIVITY_MAPPING and completion_time:
                activity_name = STEP_TO_ACTIVITY_MAPPING[step_name]
                
                class=class="hl-string">"hl-comment"># Handle activities that can occur multiple times, like paperwork
                if activity_name == class="hl-string">'Pre-Employment Paperwork Completed':
                    class=class="hl-string">"hl-comment"># Check if a similar event already exists to avoid duplicates from multiple forms
                    if any(e[class="hl-string">'ActivityName'] == activity_name and e[class="hl-string">'OnboardingCase'] == onboarding_case_id for e in all_events):
                        continue

                event = base_event.copy()
                event[class="hl-string">'ActivityName'] = activity_name
                event[class="hl-string">'EventTime'] = completion_time
                event[class="hl-string">'OnboardingStatus'] = class="hl-string">'In Progress'
                all_events.append(event)
                
                class=class="hl-string">"hl-comment"># Special case for IT Provisioning Completed
                if activity_name == class="hl-string">'IT Account Provisioning Initiated':
                    class=class="hl-string">"hl-comment"># Assume completion happens after a delay, or find a specific step if it exists
                    event_completed = event.copy()
                    event_completed[class="hl-string">'ActivityName'] = class="hl-string">'IT Account Provisioning Completed'
                    class=class="hl-string">"hl-comment"># This is a placeholder; real completion time should be from a specific step
                    event_completed[class="hl-string">'EventTime'] = completion_time 
                    all_events.append(event_completed)

    class=class="hl-string">"hl-comment"># 4. Create DataFrame and export to CSV
    if not all_events:
        print(class="hl-string">"No events were generated.")
        return

    df = pd.DataFrame(all_events)
    
    class=class="hl-string">"hl-comment"># Reorder columns for clarity
    column_order = [
        class="hl-string">'OnboardingCase', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'SourceSystem', class="hl-string">'LastDataUpdate',
        class="hl-string">'EmployeeId', class="hl-string">'HiringManager', class="hl-string">'Department', class="hl-string">'OnboardingStatus', 
        class="hl-string">'OfficialStartDate', class="hl-string">'ComplianceDueDate', class="hl-string">'SystemInitiatingActivity'
    ]
    df = df[column_order]

    df.to_csv(class="hl-string">'workday_onboarding_event_log.csv', index=False)
    print(fclass="hl-string">"Successfully extracted {len(df)} events to workday_onboarding_event_log.csv")

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