Ihre Vorlage für GehaltsabrechnungsDaten

/* 
   UKG PRO COGNOS ANALYTICS - REPORT SPECIFICATION LOGIC 
   Language: Cognos Expression / Pseudo-SQL Configuration
   NOTE: This logic represents the 15 segments required in the Union Query.
   Each block represents a separate Query Subject in Cognos that must be Unioned.
*/

/* 1. Time Sheet Submitted */
SELECT
  ([TimeMgmt].[EmployeeID] + '-' + [TimeMgmt].[PayPeriodEnd]) AS PayrollRecordId,
  'Time Sheet Submitted' AS ActivityName,
  [TimeMgmt].[Timecard_Submission_Date] AS EventTimestamp,
  'UKG Time' AS SourceSystem,
  [TimeMgmt].[Last_Updated] AS LastUpdateDate,
  [TimeMgmt].[Approver_ID] AS PayrollSpecialist,
  [TimeMgmt].[Gross_Pay_Est] AS GrossPayAmount,
  [TimeMgmt].[Department] AS DepartmentCode,
  [TimeMgmt].[PayGroup] AS PayGroup
FROM [Time Management Package]
WHERE [TimeMgmt].[Status] = 'Submitted'

UNION ALL

/* 2. Time Sheet Approved */
SELECT
  ([TimeMgmt].[EmployeeID] + '-' + [TimeMgmt].[PayPeriodEnd]) AS PayrollRecordId,
  'Time Sheet Approved' AS ActivityName,
  [TimeMgmt].[Timecard_Approval_Date] AS EventTimestamp,
  'UKG Time' AS SourceSystem,
  [TimeMgmt].[Last_Updated] AS LastUpdateDate,
  [TimeMgmt].[Manager_ID] AS PayrollSpecialist,
  [TimeMgmt].[Gross_Pay_Est] AS GrossPayAmount,
  [TimeMgmt].[Department] AS DepartmentCode,
  [TimeMgmt].[PayGroup] AS PayGroup
FROM [Time Management Package]
WHERE [TimeMgmt].[Status] = 'Approved'

UNION ALL

/* 3. Payroll Record Initialized */
SELECT
  ([PayRegister].[EmployeeID] + '-' + [PayRegister].[PeriodEndDate]) AS PayrollRecordId,
  'Payroll Record Initialized' AS ActivityName,
  [PayRegister].[Record_Creation_Date] AS EventTimestamp,
  'UKG Payroll' AS SourceSystem,
  [PayRegister].[Last_Mod_Date] AS LastUpdateDate,
  'System' AS PayrollSpecialist,
  0 AS GrossPayAmount,
  [PayRegister].[Home_Dept] AS DepartmentCode,
  [PayRegister].[PayGroup] AS PayGroup
FROM [Payroll Data Package]

UNION ALL

/* 4. Incentive Data Imported */
SELECT
  ([PayDetail].[EmployeeID] + '-' + [PayDetail].[PeriodEndDate]) AS PayrollRecordId,
  'Incentive Data Imported' AS ActivityName,
  [PayDetail].[Import_Date] AS EventTimestamp,
  'UKG Payroll' AS SourceSystem,
  [PayDetail].[Import_Date] AS LastUpdateDate,
  [PayDetail].[User_ID] AS PayrollSpecialist,
  [PayDetail].[Amount] AS GrossPayAmount,
  [PayDetail].[Department] AS DepartmentCode,
  [PayDetail].[PayGroup] AS PayGroup
FROM [Payroll Detail Package]
WHERE [PayDetail].[Source_Type] = 'Import' AND [PayDetail].[Earn_Ded_Type] = 'Earning'

UNION ALL

/* 5. Gross Pay Calculated */
SELECT
  ([PayHistory].[EmployeeID] + '-' + [PayHistory].[PeriodEndDate]) AS PayrollRecordId,
  'Gross Pay Calculated' AS ActivityName,
  [PayHistory].[Calculation_Time] AS EventTimestamp,
  'UKG Calc Engine' AS SourceSystem,
  [PayHistory].[Calculation_Time] AS LastUpdateDate,
  'System' AS PayrollSpecialist,
  [PayHistory].[Total_Gross] AS GrossPayAmount,
  [PayHistory].[Department] AS DepartmentCode,
  [PayHistory].[PayGroup] AS PayGroup
FROM [Payroll History Package]

UNION ALL

/* 6. Benefits Deductions Applied */
SELECT
  ([PayHistory].[EmployeeID] + '-' + [PayHistory].[PeriodEndDate]) AS PayrollRecordId,
  'Benefits Deductions Applied' AS ActivityName,
  _add_seconds([PayHistory].[Calculation_Time], 5) AS EventTimestamp,
  'UKG Calc Engine' AS SourceSystem,
  [PayHistory].[Calculation_Time] AS LastUpdateDate,
  'System' AS PayrollSpecialist,
  [PayHistory].[Total_Gross] AS GrossPayAmount,
  [PayHistory].[Department] AS DepartmentCode,
  [PayHistory].[PayGroup] AS PayGroup
FROM [Payroll History Package]
WHERE [PayHistory].[Total_Deductions] > 0

UNION ALL

/* 7. Taxes Calculated */
SELECT
  ([PayHistory].[EmployeeID] + '-' + [PayHistory].[PeriodEndDate]) AS PayrollRecordId,
  'Taxes Calculated' AS ActivityName,
  _add_seconds([PayHistory].[Calculation_Time], 10) AS EventTimestamp,
  'UKG Tax Engine' AS SourceSystem,
  [PayHistory].[Calculation_Time] AS LastUpdateDate,
  'System' AS PayrollSpecialist,
  [PayHistory].[Total_Gross] AS GrossPayAmount,
  [PayHistory].[Department] AS DepartmentCode,
  [PayHistory].[PayGroup] AS PayGroup
FROM [Payroll History Package]
WHERE [PayHistory].[Total_Taxes] > 0

UNION ALL

/* 8. Payroll Result Previewed */
SELECT
  ([AuditLog].[Target_EmployeeID] + '-' + [AuditLog].[Context_Date]) AS PayrollRecordId,
  'Payroll Result Previewed' AS ActivityName,
  [AuditLog].[Access_Time] AS EventTimestamp,
  'UKG Audit' AS SourceSystem,
  [AuditLog].[Access_Time] AS LastUpdateDate,
  [AuditLog].[User_Name] AS PayrollSpecialist,
  0 AS GrossPayAmount,
  [AuditLog].[Department] AS DepartmentCode,
  [AuditLog].[PayGroup] AS PayGroup
FROM [System Audit Package]
WHERE [AuditLog].[Action] = 'View Report' AND [AuditLog].[Report_Name] LIKE '%Pay Register%'

UNION ALL

/* 9. Audit Exception Flagged */
SELECT
  ([PayMessages].[EmployeeID] + '-' + [PayMessages].[PeriodEndDate]) AS PayrollRecordId,
  'Audit Exception Flagged' AS ActivityName,
  [PayMessages].[Message_Time] AS EventTimestamp,
  'UKG Payroll' AS SourceSystem,
  [PayMessages].[Message_Time] AS LastUpdateDate,
  'System' AS PayrollSpecialist,
  0 AS GrossPayAmount,
  [PayMessages].[Department] AS DepartmentCode,
  [PayMessages].[PayGroup] AS PayGroup
FROM [Payroll Messages Package]
WHERE [PayMessages].[Severity] IN ('Warning', 'Error')

UNION ALL

/* 10. Data Correction Performed */
SELECT
  ([AuditData].[EmployeeID] + '-' + [AuditData].[Target_Period_End]) AS PayrollRecordId,
  'Data Correction Performed' AS ActivityName,
  [AuditData].[Change_Time] AS EventTimestamp,
  'UKG Audit' AS SourceSystem,
  [AuditData].[Change_Time] AS LastUpdateDate,
  [AuditData].[User_Changed_By] AS PayrollSpecialist,
  0 AS GrossPayAmount,
  [AuditData].[Department] AS DepartmentCode,
  [AuditData].[PayGroup] AS PayGroup
FROM [Data Audit Package]
WHERE [AuditData].[Table_Name] = 'PayData' AND [AuditData].[Operation] = 'Update'

UNION ALL

/* 11. Payroll Record Approved */
SELECT
  ([PayControl].[EmployeeID] + '-' + [PayControl].[PeriodEndDate]) AS PayrollRecordId,
  'Payroll Record Approved' AS ActivityName,
  [PayControl].[Approval_Time] AS EventTimestamp,
  'UKG Payroll' AS SourceSystem,
  [PayControl].[Approval_Time] AS LastUpdateDate,
  [PayControl].[Approver_ID] AS PayrollSpecialist,
  [PayControl].[Final_Gross] AS GrossPayAmount,
  [PayControl].[Department] AS DepartmentCode,
  [PayControl].[PayGroup] AS PayGroup
FROM [Payroll Control Package]
WHERE [PayControl].[Status] = 'Finalized'

UNION ALL

/* 12. Bank Transfer File Generated */
SELECT
  ([AchFile].[EmployeeID] + '-' + [AchFile].[PeriodEndDate]) AS PayrollRecordId,
  'Bank Transfer File Generated' AS ActivityName,
  [AchFile].[File_Creation_Timestamp] AS EventTimestamp,
  'UKG Banking' AS SourceSystem,
  [AchFile].[File_Creation_Timestamp] AS LastUpdateDate,
  [AchFile].[Generated_By] AS PayrollSpecialist,
  [AchFile].[Net_Pay] AS GrossPayAmount,
  [AchFile].[Department] AS DepartmentCode,
  [AchFile].[PayGroup] AS PayGroup
FROM [Banking / ACH Package]

UNION ALL

/* 13. Payment Executed */
SELECT
  ([CheckRegister].[EmployeeID] + '-' + [CheckRegister].[PeriodEndDate]) AS PayrollRecordId,
  'Payment Executed' AS ActivityName,
  [CheckRegister].[Check_Date] AS EventTimestamp,
  'UKG Payment' AS SourceSystem,
  [CheckRegister].[Check_Date] AS LastUpdateDate,
  'System' AS PayrollSpecialist,
  [CheckRegister].[Gross_Amount] AS GrossPayAmount,
  [CheckRegister].[Department] AS DepartmentCode,
  [CheckRegister].[PayGroup] AS PayGroup
FROM [Check Register Package]

UNION ALL

/* 14. Pay Slip Published */
SELECT
  ([CheckRegister].[EmployeeID] + '-' + [CheckRegister].[PeriodEndDate]) AS PayrollRecordId,
  'Pay Slip Published' AS ActivityName,
  [CheckRegister].[Advice_Available_Date] AS EventTimestamp,
  'UKG Self Service' AS SourceSystem,
  [CheckRegister].[Advice_Available_Date] AS LastUpdateDate,
  'System' AS PayrollSpecialist,
  0 AS GrossPayAmount,
  [CheckRegister].[Department] AS DepartmentCode,
  [CheckRegister].[PayGroup] AS PayGroup
FROM [Check Register Package]
WHERE [CheckRegister].[Is_Void] = 'False'

UNION ALL

/* 15. Tax Filing Completed */
SELECT
  ([TaxHistory].[EmployeeID] + '-' + [TaxHistory].[PeriodEndDate]) AS PayrollRecordId,
  'Tax Filing Completed' AS ActivityName,
  [TaxHistory].[Filing_Submission_Date] AS EventTimestamp,
  'UKG Tax' AS SourceSystem,
  [TaxHistory].[Last_Update] AS LastUpdateDate,
  'System' AS PayrollSpecialist,
  0 AS GrossPayAmount,
  [TaxHistory].[Department] AS DepartmentCode,
  [TaxHistory].[PayGroup] AS PayGroup
FROM [Tax Filing Package]
WHERE [TaxHistory].[Status] = 'Filed'

import requests
import pandas as pd
import os
import json
from datetime import datetime, timedelta

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># Load credentials from environment variables for security
CLIENT_ID = os.environ.get(class="hl-string">'UKG_CLIENT_ID', class="hl-string">'YOUR_CLIENT_ID')
CLIENT_SECRET = os.environ.get(class="hl-string">'UKG_CLIENT_SECRET', class="hl-string">'YOUR_CLIENT_SECRET')
API_KEY = os.environ.get(class="hl-string">'UKG_API_KEY', class="hl-string">'YOUR_API_KEY')
API_BASE_URL = os.environ.get(class="hl-string">'UKG_BASE_URL', class="hl-string">'https://api.ukg.com') class=class="hl-string">"hl-comment"># Adjust if you have a different base URL

class=class="hl-string">"hl-comment"># Set the desired date range for the extraction
START_DATE = class="hl-string">'2023-01-01T00:00:00Z'
END_DATE = class="hl-string">'2023-06-30T23:59:59Z'

def get_access_token():
    class="hl-string">"""Authenticates with UKG Pro and retrieves a bearer token."""
    auth_url = fclass="hl-string">"{API_BASE_URL}/auth/token"
    headers = {class="hl-string">'Content-Type': class="hl-string">'application/json', class="hl-string">'x-api-key': API_KEY}
    payload = {
        class="hl-string">'grant_type': class="hl-string">'client_credentials',
        class="hl-string">'client_id': CLIENT_ID,
        class="hl-string">'client_secret': CLIENT_SECRET,
        class="hl-string">'scope': class="hl-string">'payroll'
    }
    try:
        response = requests.post(auth_url, headers=headers, data=json.dumps(payload))
        response.raise_for_status()
        return response.json().get(class="hl-string">'access_token')
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"Error getting access token: {e}")
        return None

def fetch_paginated_data(url, headers):
    class="hl-string">"""Fetches all pages of data from a UKG Pro API endpoint."""
    all_records = []
    while url:
        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()
            data = response.json()
            all_records.extend(data)

            class=class="hl-string">"hl-comment"># UKG Pro uses a class="hl-string">'Link' header for pagination
            next_link = response.headers.get(class="hl-string">'Link')
            url = None
            if next_link:
                links = next_link.split(class="hl-string">',')
                for link in links:
                    if class="hl-string">'rel="next"' in link:
                        url = link[link.find(class="hl-string">'<')+1:link.find(class="hl-string">'>')]
                        break
        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"Error fetching data from {url}: {e}")
            break
    return all_records

def main():
    class="hl-string">"""Main function to extract, transform, and save payroll data."""
    token = get_access_token()
    if not token:
        print(class="hl-string">"Failed to authenticate. Exiting.")
        return

    headers = {
        class="hl-string">'Authorization': fclass="hl-string">'Bearer {token}',
        class="hl-string">'x-api-key': API_KEY
    }

    print(class="hl-string">"Fetching payroll history summaries...")
    class=class="hl-string">"hl-comment"># This endpoint provides core pay calculation and payment data.
    class=class="hl-string">"hl-comment"># We filter by payPeriodEndDate for a relevant time slice.
    pay_history_url = fclass="hl-string">"{API_BASE_URL}/v1/payroll/pay-history/pay-summaries?payPeriodEndDate_from={START_DATE}&payPeriodEndDate_to={END_DATE}&per_page=100"
    pay_records = fetch_paginated_data(pay_history_url, headers)

    class=class="hl-string">"hl-comment"># Note: A comprehensive audit log endpoint is assumed for workflow steps.
    class=class="hl-string">"hl-comment"># The exact endpoint and filter parameters may vary. This is a representative example.
    print(class="hl-string">"Fetching audit log data for workflow events...")
    audit_log_url = fclass="hl-string">"{API_BASE_URL}/platform/v1/audit/logs?startTime={START_DATE}&endTime={END_DATE}&per_page=100"
    audit_logs = fetch_paginated_data(audit_log_url, headers)

    class=class="hl-string">"hl-comment"># Process audit logs into a dictionary for easy lookup
    audit_events_by_record = {}
    for log in audit_logs:
        class=class="hl-string">"hl-comment"># Assumption: The audit log contains a reference to the payroll record ID
        record_id = log.get(class="hl-string">'entityId')
        if record_id:
            if record_id not in audit_events_by_record:
                audit_events_by_record[record_id] = []
            audit_events_by_record[record_id].append(log)

    events = []
    print(fclass="hl-string">"Processing {len(pay_records)} payroll records...")
    for record in pay_records:
        payroll_record_id = fclass="hl-string">"{record.get('employeeId')}-{record.get('payPeriodEndDate')}"
        pay_date = datetime.fromisoformat(record.get(class="hl-string">'checkDate').replace(class="hl-string">'Z', class="hl-string">'+00:00'))
        pay_period_end_date = datetime.fromisoformat(record.get(class="hl-string">'payPeriodEndDate').replace(class="hl-string">'Z', class="hl-string">'+00:00'))
        class=class="hl-string">"hl-comment"># Heuristic: Assume initialization happens 1 day after period end
        initialized_ts = pay_period_end_date + timedelta(days=1)
        class=class="hl-string">"hl-comment"># Heuristic: Assume calculation happens 3 days after period end
        calculated_ts = pay_period_end_date + timedelta(days=3)
        
        class=class="hl-string">"hl-comment"># Use a placeholder for SLA deadline, e.g., 1 day before pay date
        sla_deadline = pay_date - timedelta(days=1)

        common_attributes = {
            class="hl-string">'PayrollRecordId': payroll_record_id,
            class="hl-string">'SourceSystem': class="hl-string">'UKG Pro',
            class="hl-string">'PayrollSpecialist': record.get(class="hl-string">'processingUser', class="hl-string">'System'),
            class="hl-string">'DepartmentCode': record.get(class="hl-string">'department'),
            class="hl-string">'CostCenter': record.get(class="hl-string">'costCenter'),
            class="hl-string">'PayGroup': record.get(class="hl-string">'payGroup'),
            class="hl-string">'GrossPayAmount': record.get(class="hl-string">'grossEarnings'),
            class="hl-string">'TaxJurisdiction': record.get(class="hl-string">'primaryTaxState'),
            class="hl-string">'SlaProcessingDeadline': sla_deadline.isoformat(),
            class="hl-string">'IsOffCycle': record.get(class="hl-string">'isOffCycle', False),
            class="hl-string">'PayPeriodEndDate': pay_period_end_date.isoformat(),
            class="hl-string">'HasAuditException': False, class=class="hl-string">"hl-comment"># To be updated by audit events
        }

        class=class="hl-string">"hl-comment"># 1. Time Sheet Submitted (from audit)
        class=class="hl-string">"hl-comment"># 2. Time Sheet Approved (from audit)
        class=class="hl-string">"hl-comment"># These events are highly dependent on audit log content. Using placeholders.
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Time Sheet Submitted', class="hl-string">'EventTimestamp': (initialized_ts - timedelta(hours=12)).isoformat()})
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Time Sheet Approved', class="hl-string">'EventTimestamp': (initialized_ts - timedelta(hours=6)).isoformat()})

        class=class="hl-string">"hl-comment"># 3. Payroll Record Initialized
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Payroll Record Initialized', class="hl-string">'EventTimestamp': initialized_ts.isoformat()})

        class=class="hl-string">"hl-comment"># 4. Incentive Data Imported (from audit)
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Incentive Data Imported', class="hl-string">'EventTimestamp': (initialized_ts + timedelta(days=1)).isoformat()})

        class=class="hl-string">"hl-comment"># 5-7. Calculation Steps
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Gross Pay Calculated', class="hl-string">'EventTimestamp': calculated_ts.isoformat()})
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Benefits Deductions Applied', class="hl-string">'EventTimestamp': (calculated_ts + timedelta(seconds=10)).isoformat()})
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Taxes Calculated', class="hl-string">'EventTimestamp': (calculated_ts + timedelta(seconds=20)).isoformat()})

        class=class="hl-string">"hl-comment"># 8. Payroll Result Previewed (from audit)
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Payroll Result Previewed', class="hl-string">'EventTimestamp': (calculated_ts + timedelta(hours=1)).isoformat()})

        class=class="hl-string">"hl-comment"># 9-10. Audit/Correction Events (from audit log)
        record_audits = audit_events_by_record.get(payroll_record_id, [])
        for audit_event in record_audits:
            event_ts = audit_event.get(class="hl-string">'eventDateTime')
            if class="hl-string">'Exception' in audit_event.get(class="hl-string">'eventName', class="hl-string">''):
                events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Audit Exception Flagged', class="hl-string">'EventTimestamp': event_ts, class="hl-string">'HasAuditException': True})
            if class="hl-string">'Correction' in audit_event.get(class="hl-string">'eventName', class="hl-string">''):
                events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Data Correction Performed', class="hl-string">'EventTimestamp': event_ts})

        class=class="hl-string">"hl-comment"># 11. Payroll Record Approved
        class=class="hl-string">"hl-comment"># Heuristic: Approval happens 2 days before pay date
        approved_ts = pay_date - timedelta(days=2)
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Payroll Record Approved', class="hl-string">'EventTimestamp': approved_ts.isoformat()})

        class=class="hl-string">"hl-comment"># 12. Bank Transfer File Generated
        class=class="hl-string">"hl-comment"># Heuristic: File is generated the day before pay date
        bank_file_ts = pay_date - timedelta(days=1, hours=4)
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Bank Transfer File Generated', class="hl-string">'EventTimestamp': bank_file_ts.isoformat()})
        
        class=class="hl-string">"hl-comment"># 13. Payment Executed
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Payment Executed', class="hl-string">'EventTimestamp': pay_date.isoformat()})

        class=class="hl-string">"hl-comment"># 14. Pay Slip Published
        class=class="hl-string">"hl-comment"># Assumption: Published on the pay date
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Pay Slip Published', class="hl-string">'EventTimestamp': (pay_date + timedelta(hours=8)).isoformat()})

        class=class="hl-string">"hl-comment"># 15. Tax Filing Completed
        class=class="hl-string">"hl-comment"># Heuristic: Filing completed 2 days after pay date
        tax_filed_ts = pay_date + timedelta(days=2)
        events.append({**common_attributes, class="hl-string">'ActivityName': class="hl-string">'Tax Filing Completed', class="hl-string">'EventTimestamp': tax_filed_ts.isoformat()})

    if not events:
        print(class="hl-string">"No events generated. Check API response or date range.")
        return

    class=class="hl-string">"hl-comment"># Create DataFrame and add final attributes
    df = pd.DataFrame(events)
    df[class="hl-string">'EventTimestamp'] = pd.to_datetime(df[class="hl-string">'EventTimestamp'])
    df = df.sort_values(by=[class="hl-string">'PayrollRecordId', class="hl-string">'EventTimestamp']).reset_index(drop=True)

    df[class="hl-string">'LastUpdateDate'] = datetime.now().isoformat()

    class=class="hl-string">"hl-comment"># Calculate cycle time and SLA breach
    start_times = df[df[class="hl-string">'ActivityName'] == class="hl-string">'Payroll Record Initialized'][[class="hl-string">'PayrollRecordId', class="hl-string">'EventTimestamp']].rename(columns={class="hl-string">'EventTimestamp': class="hl-string">'StartTime'})
    end_times = df[df[class="hl-string">'ActivityName'] == class="hl-string">'Payment Executed'][[class="hl-string">'PayrollRecordId', class="hl-string">'EventTimestamp']].rename(columns={class="hl-string">'EventTimestamp': class="hl-string">'EndTime'})
    sla_deadlines = df[[class="hl-string">'PayrollRecordId', class="hl-string">'SlaProcessingDeadline']].drop_duplicates()

    df = df.merge(start_times, on=class="hl-string">'PayrollRecordId', how=class="hl-string">'left')
    df = df.merge(end_times, on=class="hl-string">'PayrollRecordId', how=class="hl-string">'left')
    df = df.merge(sla_deadlines, on=class="hl-string">'PayrollRecordId', how=class="hl-string">'left')

    df[class="hl-string">'CycleTimeDays'] = (df[class="hl-string">'EndTime'] - df[class="hl-string">'StartTime']).dt.total_seconds() / (24 * 3600)
    df[class="hl-string">'IsSlaBreached'] = df[class="hl-string">'EndTime'] > pd.to_datetime(df[class="hl-string">'SlaProcessingDeadline'])
    
    df.drop(columns=[class="hl-string">'StartTime', class="hl-string">'EndTime'], inplace=True)

    class=class="hl-string">"hl-comment"># Reorder columns to match requirements
    final_columns = [
        class="hl-string">'PayrollRecordId', class="hl-string">'ActivityName', class="hl-string">'EventTimestamp', class="hl-string">'SourceSystem', class="hl-string">'LastUpdateDate',
        class="hl-string">'PayrollSpecialist', class="hl-string">'DepartmentCode', class="hl-string">'CostCenter', class="hl-string">'PayGroup', class="hl-string">'GrossPayAmount',
        class="hl-string">'TaxJurisdiction', class="hl-string">'SlaProcessingDeadline', class="hl-string">'IsOffCycle', class="hl-string">'HasAuditException',
        class="hl-string">'PayPeriodEndDate', class="hl-string">'IsSlaBreached', class="hl-string">'CycleTimeDays'
    ]
    df = df[final_columns]

    class=class="hl-string">"hl-comment"># Save to CSV
    output_filename = class="hl-string">'ukg_payroll_event_log.csv'
    df.to_csv(output_filename, index=False, date_format=class="hl-string">'%Y-%m-%dT%H:%M:%S.%fZ')
    print(fclass="hl-string">"Extraction complete. Event log saved to {output_filename}")

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