データテンプレート: 人材採用から退職までの従業員ライフサイクル

import requests
import pandas as pd
import json
import time
import io
from datetime import datetime

class=class="hl-string">"hl-comment"># --- Configuration: Replace with your UKG Pro details ---
UKG_BASE_URL = class="hl-string">"https://service[XX].ultipro.com" class=class="hl-string">"hl-comment"># e.g., https://service5.ultipro.com
USER_NAME = class="hl-string">"[Your_API_Service_Account_Username]"
PASSWORD = class="hl-string">"[Your_API_Service_Account_Password]"
CUSTOMER_API_KEY = class="hl-string">"[Your_Customer_API_Key]"

class=class="hl-string">"hl-comment"># Path to the pre-configured BI report in UKG Pro
REPORT_PATH = class="hl-string">"/content/folder/My Reports/[Your Hire to Retire BI Report Name]"

class=class="hl-string">"hl-comment"># Output file name
OUTPUT_CSV_FILE = class="hl-string">'ukg_hire_to_retire_event_log.csv'

class=class="hl-string">"hl-comment"># --- Activity Mapping: Map your BI Report columns to Activity Names ---
class=class="hl-string">"hl-comment"># This is the most critical part to configure.
class=class="hl-string">"hl-comment"># class="hl-string">'ukg_column': The exact column name from your UKG Pro BI Report.
class=class="hl-string">"hl-comment"># class="hl-string">'activity': The desired ActivityName in the final event log.
ACTIVITY_MAP = [
    {class="hl-string">'ukg_column': class="hl-string">'[RequisitionCreatedDate]', class="hl-string">'activity': class="hl-string">'Job Requisition Created'},
    {class="hl-string">'ukg_column': class="hl-string">'[ApplicationDate]', class="hl-string">'activity': class="hl-string">'Candidate Applied'},
    {class="hl-string">'ukg_column': class="hl-string">'[InterviewScheduledDate]', class="hl-string">'activity': class="hl-string">'Interview Scheduled'},
    {class="hl-string">'ukg_column': class="hl-string">'[OfferExtendedDate]', class="hl-string">'activity': class="hl-string">'Offer Extended'},
    {class="hl-string">'ukg_column': class="hl-string">'[OfferAcceptedDate]', class="hl-string">'activity': class="hl-string">'Offer Accepted'},
    {class="hl-string">'ukg_column': class="hl-string">'[BackgroundCheckInitiatedDate]', class="hl-string">'activity': class="hl-string">'Background Check Initiated'},
    {class="hl-string">'ukg_column': class="hl-string">'[HireDate]', class="hl-string">'activity': class="hl-string">'Employee Hired'},
    {class="hl-string">'ukg_column': class="hl-string">'[OnboardingInitiatedDate]', class="hl-string">'activity': class="hl-string">'Onboarding Initiated'},
    {class="hl-string">'ukg_column': class="hl-string">'[PayrollSetupCompletedDate]', class="hl-string">'activity': class="hl-string">'Payroll Setup Completed'},
    {class="hl-string">'ukg_column': class="hl-string">'[TrainingAssignedDate]', class="hl-string">'activity': class="hl-string">'Training Assigned'},
    {class="hl-string">'ukg_column': class="hl-string">'[PerformanceReviewConductedDate]', class="hl-string">'activity': class="hl-string">'Performance Review Conducted'},
    {class="hl-string">'ukg_column': class="hl-string">'[PromotionApprovedDate]', class="hl-string">'activity': class="hl-string">'Promotion Approved'},
    {class="hl-string">'ukg_column': class="hl-string">'[RoleChangeInitiatedDate]', class="hl-string">'activity': class="hl-string">'Role Change Initiated'},
    {class="hl-string">'ukg_column': class="hl-string">'[TerminationInitiatedDate]', class="hl-string">'activity': class="hl-string">'Termination Initiated'},
    {class="hl-string">'ukg_column': class="hl-string">'[ExitInterviewConductedDate]', class="hl-string">'activity': class="hl-string">'Exit Interview Conducted'},
    {class="hl-string">'ukg_column': class="hl-string">'[TerminationDate]', class="hl-string">'activity': class="hl-string">'Employee Terminated'},
    {class="hl-string">'ukg_column': class="hl-string">'[OffboardingTasksCompletedDate]', class="hl-string">'activity': class="hl-string">'Offboarding Tasks Completed'}
]

def get_service_token():
    class="hl-string">"""Authenticates and retrieves a service token."""
    auth_url = fclass="hl-string">"{UKG_BASE_URL}/personnel/v1/token"
    auth_payload = json.dumps({
        class="hl-string">"username": USER_NAME,
        class="hl-string">"password": PASSWORD,
        class="hl-string">"customerApiKey": CUSTOMER_API_KEY
    })
    headers = {class="hl-string">'Content-Type': class="hl-string">'application/json'}
    print(class="hl-string">"Requesting service token...")
    response = requests.post(auth_url, headers=headers, data=auth_payload)
    response.raise_for_status() class=class="hl-string">"hl-comment"># Raises an exception for bad status codes
    token = response.json().get(class="hl-string">'token')
    print(class="hl-string">"Successfully obtained service token.")
    return token

def execute_report(token):
    class="hl-string">"""Submits a report execution request."""
    exec_url = fclass="hl-string">"{UKG_BASE_URL}/bi/v1/reports/execute"
    headers = {
        class="hl-string">'Content-Type': class="hl-string">'application/json',
        class="hl-string">'Authorization': fclass="hl-string">'Bearer {token}',
        class="hl-string">'US-Customer-Api-Key': CUSTOMER_API_KEY
    }
    exec_payload = json.dumps({
        class="hl-string">"reportPath": REPORT_PATH,
        class="hl-string">"delim": class="hl-string">"," class=class="hl-string">"hl-comment"># Request CSV format
    })
    print(fclass="hl-string">"Executing report: {REPORT_PATH}...")
    response = requests.post(exec_url, headers=headers, data=exec_payload)
    response.raise_for_status()
    report_key = response.json().get(class="hl-string">'reportKey')
    print(fclass="hl-string">"Report execution started. Report Key: {report_key}")
    return report_key

def check_report_status(token, report_key):
    class="hl-string">"""Polls for the report status until completion."""
    status_url = fclass="hl-string">"{UKG_BASE_URL}/bi/v1/reports/status/{report_key}"
    headers = {
        class="hl-string">'Authorization': fclass="hl-string">'Bearer {token}',
        class="hl-string">'US-Customer-Api-Key': CUSTOMER_API_KEY
    }
    while True:
        print(class="hl-string">"Checking report status...")
        response = requests.get(status_url, headers=headers)
        response.raise_for_status()
        status = response.json().get(class="hl-string">'status')
        print(fclass="hl-string">"Current status: {status}")
        if status == class="hl-string">'Complete':
            print(class="hl-string">"Report execution complete.")
            return
        elif status in [class="hl-string">'Failed', class="hl-string">'Cancelled']:
            raise Exception(fclass="hl-string">"Report execution failed with status: {status}")
        time.sleep(10) class=class="hl-string">"hl-comment"># Wait 10 seconds before polling again

def download_report(token, report_key):
    class="hl-string">"""Downloads the completed report data."""
    download_url = fclass="hl-string">"{UKG_BASE_URL}/bi/v1/reports/{report_key}"
    headers = {
        class="hl-string">'Authorization': fclass="hl-string">'Bearer {token}',
        class="hl-string">'US-Customer-Api-Key': CUSTOMER_API_KEY
    }
    print(class="hl-string">"Downloading report data...")
    response = requests.get(download_url, headers=headers)
    response.raise_for_status()
    print(class="hl-string">"Download successful.")
    return response.text

def transform_data_to_event_log(csv_data):
    class="hl-string">"""Transforms the wide report data into a long event log format."""
    print(class="hl-string">"Transforming data into event log format...")
    df_raw = pd.read_csv(io.StringIO(csv_data))
    
    all_events = []
    
    class=class="hl-string">"hl-comment"># Define base attributes that are constant for each employee
    class=class="hl-string">"hl-comment"># Replace with the actual column names from your BI report
    base_cols = {
        class="hl-string">'EmployeeId': class="hl-string">'[EmployeeID]',
        class="hl-string">'JobRole': class="hl-string">'[JobTitle]',
        class="hl-string">'Department': class="hl-string">'[DepartmentName]',
        class="hl-string">'ManagerName': class="hl-string">'[ManagerName]',
        class="hl-string">'TerminationReason': class="hl-string">'[TerminationReason]',
        class="hl-string">'SlaTargetDate': class="hl-string">'[SlaTargetDate]', class=class="hl-string">"hl-comment"># Example, may not exist for all events
        class="hl-string">'JobRequisitionId': class="hl-string">'[JobRequisitionID]',
        class="hl-string">'EventPerformer': class="hl-string">'[LastModifiedBy]' class=class="hl-string">"hl-comment"># Example performer
    }
    
    class=class="hl-string">"hl-comment"># Ensure all configured columns exist in the dataframe
    for activity_info in ACTIVITY_MAP:
        if activity_info[class="hl-string">'ukg_column'] not in df_raw.columns:
            print(fclass="hl-string">"Warning: Column '{activity_info['ukg_column']}' not found in the report. Skipping activity '{activity_info['activity']}'.")
            continue
        
        class=class="hl-string">"hl-comment"># Create a dataframe for the current activity
        df_activity = df_raw.copy()
        df_activity = df_activity.dropna(subset=[activity_info[class="hl-string">'ukg_column']])
        
        if df_activity.empty:
            continue

        df_activity[class="hl-string">'ActivityName'] = activity_info[class="hl-string">'activity']
        df_activity[class="hl-string">'EventTime'] = pd.to_datetime(df_activity[activity_info[class="hl-string">'ukg_column']])
        
        class=class="hl-string">"hl-comment"># Rename columns to standard format
        df_activity = df_activity.rename(columns=base_cols)
        
        class=class="hl-string">"hl-comment"># Select only the required columns for the event log
        final_cols = list(base_cols.keys()) + [class="hl-string">'ActivityName', class="hl-string">'EventTime']
        df_activity = df_activity[final_cols]
        
        all_events.append(df_activity)
    
    if not all_events:
        print(class="hl-string">"No events were generated. Please check your ACTIVITY_MAP and BI report data.")
        return pd.DataFrame()
        
    df_event_log = pd.concat(all_events, ignore_index=True)
    
    class=class="hl-string">"hl-comment"># Add static columns
    df_event_log[class="hl-string">'SourceSystem'] = class="hl-string">'UKG Pro'
    df_event_log[class="hl-string">'LastDataUpdate'] = datetime.utcnow().isoformat() + class="hl-string">'Z'
    
    class=class="hl-string">"hl-comment"># Sort by employee and event time for clarity
    df_event_log = df_event_log.sort_values(by=[class="hl-string">'EmployeeId', class="hl-string">'EventTime'])
    
    print(fclass="hl-string">"Transformation complete. Generated {len(df_event_log)} events.")
    return df_event_log

if __name__ == class="hl-string">"__main__":
    try:
        service_token = get_service_token()
        report_key = execute_report(service_token)
        check_report_status(service_token, report_key)
        csv_content = download_report(service_token, report_key)
        
        if csv_content:
            event_log_df = transform_data_to_event_log(csv_content)
            if not event_log_df.empty:
                event_log_df.to_csv(OUTPUT_CSV_FILE, index=False)
                print(fclass="hl-string">"Event log successfully saved to {OUTPUT_CSV_FILE}")
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"An API error occurred: {e}")
        if e.response is not None:
            print(fclass="hl-string">"Response Body: {e.response.text}")
    except Exception as e:
        print(fclass="hl-string">"An unexpected error occurred: {e}")