Templat data siklus hidup karyawan dari rekrutmen hingga pensiun

/*
 This is a logical representation of a Workday Advanced Composite Report configuration.
 It is not executable code but a blueprint for a Workday Report Writer.
 The report unions data from different business processes to form a single event log.
*/

REPORT_TYPE: Composite
COMPOSITE_TYPE: Union

// =======================================================
// SUB-REPORT 1: RECRUITMENT AND HIRE EVENTS
// =======================================================

SUB_REPORT_NAME: SR_RecruitHire
DATA_SOURCE: Business Process Transactions

COLUMNS:
  1. Field: Worker > Employee ID | Alias: EmployeeId
  2. Calculated Field: CF_ActivityName_RecruitHire | Alias: ActivityName
  3. Field: Completion Date | Alias: EventTimestamp
  4. Field: Initiator | Alias: EventPerformer
  5. Field: Worker > Supervisory Organization | Alias: Department
  6. Field: Job Application > Job Requisition > Requisition ID | Alias: JobRequisitionId
  7. Calculated Field: CF_LifecycleType_Onboarding | Alias: LifecycleEventType

CALCULATED_FIELDS:
  - Name: CF_ActivityName_RecruitHire
    Type: Evaluate Expression
    Logic:
      IF (Business Process Type = 'Create Job Requisition' AND Status = 'Successfully Completed') THEN 'Job Requisition Created'
      ELSE IF (Business Process Type = 'Job Application' AND Last Step Completed = 'Generate Offer') THEN 'Offer Letter Generated'
      ELSE IF (Business process Type = 'Job Application' AND Last Step Completed = 'Candidate Accepts Offer') THEN 'Offer Accepted'
      ELSE IF (Business Process Type = 'Job Application' AND Last Step Completed = 'Background Check') THEN 'Background Check Initiated'
      ELSE IF (Business Process Type = 'Job Application' AND Last Step Completed = 'Background Check Complete') THEN 'Background Check Completed'
      ELSE IF (Business Process Type = 'Hire' AND Status = 'Successfully Completed') THEN 'Hire Process Completed'
      ELSE ''

  - Name: CF_LifecycleType_Onboarding
    Type: Evaluate Expression
    Logic:
      IF (True) THEN 'Onboarding'

FILTERS:
  - Business Process Type IN ('Create Job Requisition', 'Job Application', 'Hire')
  - Completion Date BETWEEN [Prompt: Start Date] AND [Prompt: End Date]


// =======================================================
// SUB-REPORT 2: ONBOARDING EVENTS
// =======================================================

SUB_REPORT_NAME: SR_Onboarding
DATA_SOURCE: Business Process Transactions

COLUMNS: (Same structure as SUB-REPORT 1)

CALCULATED_FIELDS:
  - Name: CF_ActivityName_Onboarding
    Type: Evaluate Expression
    Logic:
      IF (Business Process Type = 'Onboarding' AND Status = 'In Progress' AND First Step Initiated = 'True') THEN 'Onboarding Initiated'
      ELSE IF (Business Process Type = 'Onboarding' AND Last Step Completed = 'Complete Payroll Forms') THEN 'Payroll Setup Completed'
      ELSE IF (Business Process Type = 'Onboarding' AND Status = 'Successfully Completed') THEN 'Onboarding Tasks Completed'
      ELSE ''

  - Name: CF_LifecycleType_Onboarding
    Type: Evaluate Expression
    Logic:
      IF (True) THEN 'Onboarding'

FILTERS:
  - Business Process Type = 'Onboarding'
  - Completion Date BETWEEN [Prompt: Start Date] AND [Prompt: End Date]


// =======================================================
// SUB-REPORT 3: PERFORMANCE AND CAREER EVENTS
// =======================================================

SUB_REPORT_NAME: SR_PerformanceCareer
DATA_SOURCE: Business Process Transactions

COLUMNS: (Same structure as SUB-REPORT 1, JobRequisitionId may be null)

CALCULATED_FIELDS:
  - Name: CF_ActivityName_PerfCareer
    Type: Evaluate Expression
    Logic:
      IF (Business Process Type = 'Start Performance Review' AND Status = 'Successfully Completed') THEN 'Performance Review Completed'
      ELSE IF (Business Process Type = 'Change Job' AND Status = 'In Progress' AND First Step Initiated = 'True') THEN 'Role Change Initiated'
      ELSE IF (Business Process Type = 'Change Job' AND Last Step Completed = 'Approval' AND (Reason Category = 'Promotion')) THEN 'Promotion Approved'
      ELSE ''

  - Name: CF_LifecycleType_CareerDev
    Type: Evaluate Expression
    Logic:
      IF (True) THEN 'Career Development'

FILTERS:
  - Business Process Type IN ('Start Performance Review', 'Change Job')
  - Completion Date BETWEEN [Prompt: Start Date] AND [Prompt: End Date]


// =======================================================
// SUB-REPORT 4: OFFBOARDING EVENTS
// =======================================================

SUB_REPORT_NAME: SR_Offboarding
DATA_SOURCE: Business Process Transactions

COLUMNS: (Same structure as SUB-REPORT 1, JobRequisitionId will be null)

CALCULATED_FIELDS:
  - Name: CF_ActivityName_Offboarding
    Type: Evaluate Expression
    Logic:
      IF (Business Process Type = 'Terminate Employee' AND Status = 'In Progress' AND First Step Initiated = 'True') THEN 'Termination Initiated'
      ELSE IF (Business Process Type = 'Offboarding' AND Status = 'Successfully Completed') THEN 'Offboarding Tasks Completed'
      ELSE IF (Business Process Type = 'Terminate Employee' AND Status = 'Successfully Completed') THEN 'Employee Terminated'
      ELSE ''

  - Name: CF_LifecycleType_Offboarding
    Type: Evaluate Expression
    Logic:
      IF (True) THEN 'Offboarding'

FILTERS:
  - Business Process Type IN ('Terminate Employee', 'Offboarding')
  - Completion Date BETWEEN [Prompt: Start Date] AND [Prompt: End Date]

import os
import pandas as pd
from zeep import Client, Settings, Transport
from zeep.wsse.username import UsernameToken
from datetime import datetime

class=class="hl-string">"hl-comment"># --- Configuration Settings --- #
class=class="hl-string">"hl-comment"># Replace with your Workday tenant's WSDL URL for Human Resources
WSDL_URL = class="hl-string">'https://[Your_Workday_Tenant_URL]/ccx/service/[Your_Tenant_Name]/Human_Resources/v38.0?wsdl'

class=class="hl-string">"hl-comment"># Replace with your Integration System User (ISU) credentials
TENANT_USERNAME = class="hl-string">'user@tenant'
TENANT_PASSWORD = os.environ.get(class="hl-string">'WORKDAY_PASSWORD') class=class="hl-string">"hl-comment"># Best practice: use environment variables for passwords

class=class="hl-string">"hl-comment"># Define the date range for the extraction
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"># List of Business Processes to extract data from. Names must match your Workday configuration.
BUSINESS_PROCESS_TYPES = [
    class="hl-string">'Create Job Requisition', 
    class="hl-string">'Job Application', 
    class="hl-string">'Hire', 
    class="hl-string">'Onboarding', 
    class="hl-string">'Performance Review', 
    class="hl-string">'Change Job', 
    class="hl-string">'Terminate Employee'
]

class=class="hl-string">"hl-comment"># --- Main Extraction Logic --- #
def get_workday_client(wsdl, user, password):
    class="hl-string">"""Initializes the Zeep SOAP client with WS-Security."""
    session = requests.Session()
    session.verify = True class=class="hl-string">"hl-comment"># Or path to your cert bundle
    transport = Transport(session=session)
    settings = Settings(strict=False, xml_huge_tree=True)
    client = Client(wsdl, wsse=UsernameToken(user, password), transport=transport, settings=settings)
    return client

def map_transaction_to_events(transaction):
    class="hl-string">"""Maps a single Workday business process transaction to one or more event log records."""
    events = []
    process_data = transaction.Business_Process_Data
    process_type = process_data.Business_Process.Business_Process_Definition_Name
    process_status = process_data.Status.value
    case_id = process_data.Business_Process.Initiating_Person_Reference.Employee_ID if process_data.Business_Process.Initiating_Person_Reference else None
    subject_employee_id = None
    if process_data.Business_Process.Business_Process_Target_Reference and hasattr(process_data.Business_Process.Business_Process_Target_Reference, class="hl-string">'Employee_ID'):
        subject_employee_id = process_data.Business_Process.Business_Process_Target_Reference.Employee_ID

    class=class="hl-string">"hl-comment"># Default to subject employee, fallback to initiator for processes like Job Requisition
    employee_id = subject_employee_id or case_id
    if not employee_id:
        return [] class=class="hl-string">"hl-comment"># Skip transactions without a clear employee context

    performer = case_id
    department = None class=class="hl-string">"hl-comment"># Placeholder, would require additional API calls to enrich

    class=class="hl-string">"hl-comment"># General process completion events
    if process_status == class="hl-string">'Successfully Completed':
        completion_time = process_data.Status_Date.isoformat()
        activity_map = {
            class="hl-string">'Create Job Requisition': class="hl-string">'Job Requisition Created',
            class="hl-string">'Hire': class="hl-string">'Hire Process Completed',
            class="hl-string">'Onboarding': class="hl-string">'Onboarding Tasks Completed',
            class="hl-string">'Performance Review': class="hl-string">'Performance Review Completed',
            class="hl-string">'Change Job': class="hl-string">'Promotion Approved', class=class="hl-string">"hl-comment"># Assumption: Change Job completion means approval
            class="hl-string">'Terminate Employee': class="hl-string">'Employee Terminated'
        }
        if process_type in activity_map:
            events.append({
                class="hl-string">'EmployeeId': employee_id,
                class="hl-string">'ActivityName': activity_map[process_type],
                class="hl-string">'EventTimestamp': completion_time,
                class="hl-string">'EventEndTime': completion_time,
                class="hl-string">'EventPerformer': performer,
                class="hl-string">'Department': department,
                class="hl-string">'JobRequisitionId': None, class=class="hl-string">"hl-comment"># Requires enrichment
                class="hl-string">'LifecycleEventType': process_type
            })

    class=class="hl-string">"hl-comment"># Step-specific events from the transaction history
    if transaction.Business_Process_Step_Data:
        for step in transaction.Business_Process_Step_Data:
            step_name = step.Step
            step_completion_time = step.Completed_Date.isoformat() if step.Completed_Date else None
            if not step_completion_time: continue

            activity_name = None
            if process_type == class="hl-string">'Job Application' and class="hl-string">'Offer' in step_name and not class="hl-string">'Accept' in step_name:
                activity_name = class="hl-string">'Offer Letter Generated'
            elif process_type == class="hl-string">'Job Application' and class="hl-string">'Accept' in step_name and class="hl-string">'Offer' in step_name:
                activity_name = class="hl-string">'Offer Accepted'
            elif class="hl-string">'Background Check' in step_name and class="hl-string">'Initiate' in step_name:
                activity_name = class="hl-string">'Background Check Initiated'
            elif class="hl-string">'Background Check' in step_name and class="hl-string">'Complete' in step_name:
                activity_name = class="hl-string">'Background Check Completed'
            elif process_type == class="hl-string">'Hire' and class="hl-string">'Initiation' in step.Status.value:
                 class=class="hl-string">"hl-comment"># Onboarding is typically a sub-process or triggered by Hire completion
                events.append({
                    class="hl-string">'EmployeeId': employee_id,
                    class="hl-string">'ActivityName': class="hl-string">'Onboarding Initiated',
                    class="hl-string">'EventTimestamp': step_completion_time,
                    class="hl-string">'EventEndTime': step_completion_time,
                    class="hl-string">'EventPerformer': performer,
                    class="hl-string">'Department': department, class="hl-string">'JobRequisitionId': None, class="hl-string">'LifecycleEventType': class="hl-string">'Onboarding'
                })
            elif process_type == class="hl-string">'Onboarding' and class="hl-string">'Payroll' in step_name and step.Status.value == class="hl-string">'Successfully Completed':
                activity_name = class="hl-string">'Payroll Setup Completed'
            elif process_type == class="hl-string">'Change Job' and class="hl-string">'Initiation' in step.Status.value:
                activity_name = class="hl-string">'Role Change Initiated'
            elif process_type == class="hl-string">'Terminate Employee' and class="hl-string">'Initiation' in step.Status.value:
                activity_name = class="hl-string">'Termination Initiated'
            elif process_type == class="hl-string">'Terminate Employee' and class="hl-string">'Offboarding' in step_name and step.Status.value == class="hl-string">'Successfully Completed':
                activity_name = class="hl-string">'Offboarding Tasks Completed'

            if activity_name:
                events.append({
                    class="hl-string">'EmployeeId': employee_id,
                    class="hl-string">'ActivityName': activity_name,
                    class="hl-string">'EventTimestamp': step_completion_time,
                    class="hl-string">'EventEndTime': step_completion_time,
                    class="hl-string">'EventPerformer': performer,
                    class="hl-string">'Department': department, class="hl-string">'JobRequisitionId': None, class="hl-string">'LifecycleEventType': process_type
                })
    return events

def main():
    class="hl-string">"""Main function to run the extraction process."""
    if not TENANT_PASSWORD:
        print(class="hl-string">"Error: WORKDAY_PASSWORD environment variable not set.")
        return

    print(class="hl-string">"Initializing Workday SOAP client...")
    client = get_workday_client(WSDL_URL, TENANT_USERNAME, TENANT_PASSWORD)
    factory = client.type_factory(class="hl-string">'ns0')

    all_events = []
    for bp_type_name in BUSINESS_PROCESS_TYPES:
        print(fclass="hl-string">"\nFetching data for business process: {bp_type_name}")
        page = 1
        while True:
            print(fclass="hl-string">"  - Requesting page {page}...")
            request_data = factory.Get_Business_Process_Transactions_RequestType(
                Request_Criteria=factory.Business_Process_Transaction_Request_CriteriaType(
                    Transaction_Date_Range_Data=factory.Transaction_Date_Range_DataType(
                        Effective_From=START_DATE,
                        Effective_To=END_DATE
                    ),
                    Business_Process_Type_Reference=[factory.Business_Process_Type_ReferenceType(
                        Business_Process_Type_Name=bp_type_name
                    )]
                ),
                Response_Filter=factory.Response_FilterType(
                    Page=page,
                    Count=100  class=class="hl-string">"hl-comment"># Page size
                )
            )
            
            try:
                response = client.service.Get_Business_Process_Transactions(request_data)

                if response.Response_Results.Total_Results == 0:
                    print(class="hl-string">"  - No results found for this process type.")
                    break

                if not response.Response_Data or not response.Response_Data.Business_Process_Transactions:
                    print(class="hl-string">"  - Reached end of data.")
                    break
                
                transactions = response.Response_Data.Business_Process_Transactions
                print(fclass="hl-string">"  - Received {len(transactions)} transactions on this page.")

                for trans in transactions:
                    events = map_transaction_to_events(trans)
                    all_events.extend(events)

                if len(transactions) < 100:
                    print(class="hl-string">"  - Last page processed.")
                    break

                page += 1

            except Exception as e:
                print(fclass="hl-string">"An error occurred: {e}")
                break

    if not all_events:
        print(class="hl-string">"\nExtraction complete. No events were generated.")
        return

    print(fclass="hl-string">"\nTotal events extracted: {len(all_events)}")
    df = pd.DataFrame(all_events)
    
    class=class="hl-string">"hl-comment"># Reorder columns for clarity
    cols = [class="hl-string">'EmployeeId', class="hl-string">'ActivityName', class="hl-string">'EventTimestamp', class="hl-string">'EventEndTime', class="hl-string">'EventPerformer', 
            class="hl-string">'Department', class="hl-string">'JobRequisitionId', class="hl-string">'LifecycleEventType']
    df = df[cols]

    class=class="hl-string">"hl-comment"># Sort by employee and timestamp to create a chronological log
    df.sort_values(by=[class="hl-string">'EmployeeId', class="hl-string">'EventTimestamp'], inplace=True)

    output_filename = class="hl-string">'workday_h2r_event_log.csv'
    df.to_csv(output_filename, index=False)
    print(fclass="hl-string">"Successfully exported event log to {output_filename}")

if __name__ == class="hl-string">'__main__':
    import requests class=class="hl-string">"hl-comment"># Add requests for the session object
    main()