Szablon danych dla procesu od zatrudnienia do odejścia z pracy: zarządzanie stanowiskami

/*
 This is a logical representation of the Workday Advanced Report configuration.
 It is built using the Workday Report Writer UI, not by writing code.
 Each 'SELECT' block represents a set of conditions within a single 'Evaluate Expression' calculated field for the 'ActivityName' column.
 The report's primary data source is 'Business Process Transactions'.
*/

// Calculated Field for ActivityName: 'cf_ActivityName'
// Type: Evaluate Expression

// Condition 1: Position Request Initiated
WHEN ('Business Process'.Type = 'Create Position' AND 'Business Process Transaction'.Is First Step = true)
THEN 'Position Request Initiated'
// Column Mapping for this condition:
// PositionId: 'Position' From Business Process -> Position ID
// ActivityName: 'cf_ActivityName' (this calculated field)
// EventTime: 'Business Process' -> Initiation Moment
// User: 'Business Process' -> Initiator

// Condition 2: Manager Approval Submitted
WHEN ('Business Process'.Type = 'Create Position' AND 'Business Process Step'.Name CONTAINS 'Manager Approval' AND 'Business Process Transaction'.Status = 'Completed')
THEN 'Manager Approval Submitted'
// Column Mapping for this condition:
// EventTime: 'Business Process Transaction' -> Completed Moment
// User: 'Business Process Transaction' -> Completed By

// Condition 3: HR Approval Submitted
WHEN ('Business Process'.Type = 'Create Position' AND 'Business Process Step'.Name CONTAINS 'HR Approval' AND 'Business Process Transaction'.Status = 'Completed')
THEN 'HR Approval Submitted'
// Column Mapping for this condition:
// EventTime: 'Business Process Transaction' -> Completed Moment
// User: 'Business Process Transaction' -> Completed By

// Condition 4: Budget Approved
WHEN ('Business Process'.Type = 'Create Position' AND 'Business Process Step'.Name CONTAINS 'Budget Approval' AND 'Business Process Transaction'.Status = 'Completed')
THEN 'Budget Approved'
// Column Mapping for this condition:
// EventTime: 'Business Process Transaction' -> Completed Moment
// User: 'Business Process Transaction' -> Completed By

// Condition 5: Position Request Denied
WHEN ('Business Process'.Type = 'Create Position' AND 'Business Process'.Status = 'Denied')
THEN 'Position Request Denied'
// Column Mapping for this condition:
// EventTime: 'Business Process' -> Overall Process Completed Moment
// User: 'Business Process Transaction' -> Completed By (on the denial step)

// Condition 6: Position Created
WHEN ('Business Process'.Type = 'Create Position' AND 'Business Process'.Status = 'Successfully Completed')
THEN 'Position Created'
// Column Mapping for this condition:
// EventTime: 'Business Process' -> Overall Process Completed Moment
// User: 'Business process' -> Initiator

// Condition 7: Position Attributes Modified
WHEN ('Business Process'.Type = 'Edit Position Restrictions' AND 'Business Process'.Status = 'Successfully Completed')
THEN 'Position Attributes Modified'
// Column Mapping for this condition:
// EventTime: 'Business Process' -> Overall Process Completed Moment
// User: 'Business Process' -> Initiator

// Condition 8: Position Reclassified
WHEN ('Business Process'.Type = 'Change Job' AND 'Business Process'.Context.Is Position = true AND 'Business Process'.Status = 'Successfully Completed')
THEN 'Position Reclassified'
// Column Mapping for this condition:
// EventTime: 'Business Process' -> Overall Process Completed Moment
// User: 'Business Process' -> Initiator

// Condition 9: Position Compliance Reviewed
WHEN ('Business Process Step'.Name CONTAINS 'Compliance Review' AND 'Business Process Transaction'.Status = 'Completed')
THEN 'Position Compliance Reviewed'
// Column Mapping for this condition:
// EventTime: 'Business Process Transaction' -> Completed Moment
// User: 'Business Process Transaction' -> Completed By

// Condition 10: Position Frozen
WHEN ('Business Process'.Type = 'Freeze Position' AND 'Business Process'.Status = 'Successfully Completed')
THEN 'Position Frozen'
// Column Mapping for this condition:
// EventTime: 'Business Process' -> Overall Process Completed Moment
// User: 'Business Process' -> Initiator

// Condition 11: Position Unfrozen
// Note: This often relies on a different data source, 'Position History', to track status changes.
// If using 'Business Process Transactions', this assumes an 'Unfreeze' BP exists or is part of another BP.
// The example below assumes an 'Edit Position' BP is used to change the status.
WHEN ('Business Process'.Type = 'Edit Position Restrictions' AND 'Position'.Is Frozen (Prior) = true AND 'Position'.Is Frozen = false AND 'Business Process'.Status = 'Successfully Completed')
THEN 'Position Unfrozen'
// Column Mapping for this condition:
// EventTime: 'Business Process' -> Overall Process Completed Moment
// User: 'Business Process' -> Initiator

// Condition 12: Position Deactivated
WHEN ('Business Process'.Type = 'Inactivate Position' AND 'Business Process'.Status = 'Successfully Completed')
THEN 'Position Deactivated'
// Column Mapping for this condition:
// EventTime: 'Business Process' -> Overall Process Completed Moment
// User: 'Business Process' -> Initiator

// Condition 13: Position Closed
WHEN ('Business Process'.Type = 'Close Position' AND 'Business Process'.Status = 'Successfully Completed')
THEN 'Position Closed'
// Column Mapping for this condition:
// EventTime: 'Business Process' -> Overall Process Completed Moment
// User: 'Business Process' -> Initiator

// Default case for the calculated field
ELSE 'Other Activity'

/*
Final Report Structure:
- Data Source: Business Process Transactions
- Columns:
  - Position ID (from Position)
  - cf_ActivityName
  - Event Time (Calculated field to pick correct timestamp based on activity)
  - User (Calculated field to pick correct user)
  - End Time (same as Event Time for discrete events)
  - Position Status (from Position)
  - Reason For Change (from Business Process -> Reason)
  - Department (Supervisory Organization from Position)
  - Cost Center (from Position)
  - Job Family (from Job Profile)
- Filter:
  - Business Process Type IN ('Create Position', 'Edit Position Restrictions', 'Change Job', 'Freeze Position', 'Inactivate Position', 'Close Position', ...)
  - Completed Date IS IN [Prompted Date Range]
  - cf_ActivityName IS NOT 'Other Activity'
*/

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 Workday RaaS endpoint, username, and password
WORKDAY_RAAS_URL = class="hl-string">'[Your_Workday_RaaS_JSON_Endpoint_URL]'
WORKDAY_USERNAME = class="hl-string">'[Your_ISU_Username]'
WORKDAY_PASSWORD = class="hl-string">'[Your_ISU_Password]'

class=class="hl-string">"hl-comment"># Set the date range for the extraction (YYYY-MM-DD format)
START_DATE = class="hl-string">'2023-01-01'
END_DATE = class="hl-string">'2023-12-31'

class=class="hl-string">"hl-comment"># --- Activity Mapping --- #
class=class="hl-string">"hl-comment"># This dictionary maps Workday processes and statuses to standardized activity names.
class=class="hl-string">"hl-comment"># The keys are tuples: (class="hl-string">'Business Process Name', class="hl-string">'Step Name / Status').
class=class="hl-string">"hl-comment"># You MAY need to adjust the class="hl-string">'Business Process Name' to match your Workday configuration.
activity_map = {
    (class="hl-string">'Create Position', class="hl-string">'Initiation Step'): class="hl-string">'Position Request Initiated',
    (class="hl-string">'Create Position', class="hl-string">'Manager Approval'): class="hl-string">'Manager Approval Submitted',
    (class="hl-string">'Create Position', class="hl-string">'HR Approval'): class="hl-string">'HR Approval Submitted',
    (class="hl-string">'Create Position', class="hl-string">'Budget Approval'): class="hl-string">'Budget Approved',
    (class="hl-string">'Create Position', class="hl-string">'Denied'): class="hl-string">'Position Request Denied',
    (class="hl-string">'Create Position', class="hl-string">'Successfully Completed'): class="hl-string">'Position Created',
    (class="hl-string">'Edit Position Restrictions', class="hl-string">'Successfully Completed'): class="hl-string">'Position Attributes Modified',
    (class="hl-string">'Change Job', class="hl-string">'Successfully Completed'): class="hl-string">'Position Reclassified',
    (class="hl-string">'Freeze Position', class="hl-string">'Successfully Completed'): class="hl-string">'Position Frozen',
    class=class="hl-string">"hl-comment"># Assuming a business process exists for this. If not, this requires a different approach.
    (class="hl-string">'Unfreeze Position', class="hl-string">'Successfully Completed'): class="hl-string">'Position Unfrozen',
    (class="hl-string">'Inactivate Position', class="hl-string">'Successfully Completed'): class="hl-string">'Position Deactivated',
    (class="hl-string">'Close Position', class="hl-string">'Successfully Completed'): class="hl-string">'Position Closed',
    class=class="hl-string">"hl-comment"># Mapping a class="hl-string">'To Do' step within a parent process for compliance review
    (class="hl-string">'Create Position', class="hl-string">'Compliance Review To Do'): class="hl-string">'Position Compliance Reviewed',
    (class="hl-string">'Edit Position Restrictions', class="hl-string">'Compliance Review To Do'): class="hl-string">'Position Compliance Reviewed'
}

def get_activity_name(process, step, status):
    class="hl-string">"""Determines the activity name based on the mapping."""
    class=class="hl-string">"hl-comment"># Denied and successful completion are based on the overall process status
    if status == class="hl-string">'Denied':
        return activity_map.get((process, class="hl-string">'Denied'))
    if status == class="hl-string">'Successfully Completed':
        return activity_map.get((process, class="hl-string">'Successfully Completed'))

    class=class="hl-string">"hl-comment"># Other activities are based on specific steps
    return activity_map.get((process, step))

def fetch_workday_data():
    class="hl-string">"""Fetches data from the Workday RaaS endpoint with pagination."""
    all_entries = []
    page = 1
    while True:
        params = {
            class="hl-string">'format': class="hl-string">'json',
            class="hl-string">'page': page,
            class="hl-string">'StartDate': fclass="hl-string">'{START_DATE}T00:00:00Z',
            class="hl-string">'EndDate': fclass="hl-string">'{END_DATE}T23:59:59Z'
        }
        print(fclass="hl-string">"Fetching page {page}...")
        try:
            response = requests.get(
                WORKDAY_RAAS_URL,
                auth=(WORKDAY_USERNAME, WORKDAY_PASSWORD),
                params=params
            )
            response.raise_for_status() class=class="hl-string">"hl-comment"># Raise an exception for bad status codes
            data = response.json()
            entries = data.get(class="hl-string">'Report_Entry', [])
            if not entries:
                print(class="hl-string">"No more entries found. Concluding fetch.")
                break
            all_entries.extend(entries)
            class=class="hl-string">"hl-comment"># Check if this is the last page
            if len(entries) < data.get(class="hl-string">'Report_Data', {}).get(class="hl-string">'Page_Size', 1000):
                 break
            page += 1
        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"Error fetching data from Workday: {e}")
            break
        except ValueError:
            print(class="hl-string">"Error parsing JSON response. Check the URL and credentials.")
            print(class="hl-string">"Response text:", response.text)
            break

    print(fclass="hl-string">"Total entries fetched: {len(all_entries)}")
    return all_entries

def main():
    class="hl-string">"""Main function to extract, transform, and save data."""
    report_entries = fetch_workday_data()
    if not report_entries:
        print(class="hl-string">"No data extracted. Exiting.")
        return

    event_log = []
    for entry in report_entries:
        class=class="hl-string">"hl-comment"># Assuming your RaaS report column names are aliased for clarity
        process_name = entry.get(class="hl-string">'BusinessProcessType')
        step_name = entry.get(class="hl-string">'BusinessProcessStep')
        status = entry.get(class="hl-string">'TransactionStatus')
        position_id = entry.get(class="hl-string">'PositionID')
        
        activity = get_activity_name(process_name, step_name, status)
        
        if not activity or not position_id:
            continue class=class="hl-string">"hl-comment"># Skip records that don't map to a required activity or have no Position ID

        event_log.append({
            class="hl-string">'PositionId': position_id,
            class="hl-string">'ActivityName': activity,
            class="hl-string">'EventTime': entry.get(class="hl-string">'CompletedDateTime', entry.get(class="hl-string">'EffectiveDateTime')),
            class="hl-string">'EndTime': entry.get(class="hl-string">'CompletedDateTime'),
            class="hl-string">'User': entry.get(class="hl-string">'InitiatorName'),
            class="hl-string">'PositionStatus': entry.get(class="hl-string">'PositionStatus'), class=class="hl-string">"hl-comment"># Assumes this field is in the report
            class="hl-string">'ReasonForChange': entry.get(class="hl-string">'Reason'),
            class="hl-string">'Department': entry.get(class="hl-string">'SupervisoryOrganization'),
            class="hl-string">'CostCenter': entry.get(class="hl-string">'CostCenter'),
            class="hl-string">'JobFamily': entry.get(class="hl-string">'JobFamily')
        })

    if not event_log:
        print(class="hl-string">"No processable events found after mapping. Exiting.")
        return

    df = pd.DataFrame(event_log)

    class=class="hl-string">"hl-comment"># Data Cleaning and Formatting
    df[class="hl-string">'EventTime'] = pd.to_datetime(df[class="hl-string">'EventTime'], errors=class="hl-string">'coerce')
    df.dropna(subset=[class="hl-string">'PositionId', class="hl-string">'ActivityName', class="hl-string">'EventTime'], inplace=True)
    df.sort_values(by=[class="hl-string">'PositionId', class="hl-string">'EventTime'], inplace=True)

    class=class="hl-string">"hl-comment"># Reorder columns for clarity
    final_columns = [
        class="hl-string">'PositionId', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'User', class="hl-string">'EndTime', 
        class="hl-string">'PositionStatus', class="hl-string">'ReasonForChange', class="hl-string">'Department', class="hl-string">'CostCenter', class="hl-string">'JobFamily'
    ]
    df = df.reindex(columns=final_columns)

    output_filename = class="hl-string">'position_management_event_log.csv'
    df.to_csv(output_filename, index=False, encoding=class="hl-string">'utf-8')
    print(fclass="hl-string">"Successfully created event log: {output_filename}")

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

WITH source_events AS (
    SELECT
        [Position ID] AS PositionId,
        [Business Process Name] AS BusinessProcessName,
        [Step Name] AS StepName,
        [Step Status] AS StepStatus,
        [Event Type] AS EventType,
        [Event Timestamp] AS SourceEventTime,
        [Completion Timestamp] AS SourceEndTime,
        [Performed By] AS SourceUser,
        [Position Status] AS SourcePositionStatus,
        [Reason] AS SourceReasonForChange,
        [Department] AS SourceDepartment,
        [Cost Center] AS SourceCostCenter,
        [Job Family] AS SourceJobFamily,
        [Previous Position Status] AS PreviousPositionStatus,
        [Source Event ID] AS SourceEventId
    FROM [Your Workday staged event source]
    WHERE [Event Timestamp] >= CAST('[Your start date]' AS TIMESTAMP)
      AND [Event Timestamp] < CAST('[Your end date]' AS TIMESTAMP)
      AND [Position ID] IS NOT NULL
),
position_attributes AS (
    SELECT
        [Position ID] AS PositionId,
        [Position Status] AS PositionStatus,
        [Reason] AS ReasonForChange,
        [Department] AS Department,
        [Cost Center] AS CostCenter,
        [Job Family] AS JobFamily
    FROM [Your Workday staged position attribute source]
),
normalized_events AS (
    SELECT
        s.PositionId,
        'Position Request Initiated' AS ActivityName,
        s.SourceEventTime AS EventTime,
        s.SourceEndTime AS EndTime,
        s.SourceUser AS User,
        COALESCE(s.SourcePositionStatus, p.PositionStatus) AS PositionStatus,
        s.SourceReasonForChange AS ReasonForChange,
        COALESCE(s.SourceDepartment, p.Department) AS Department,
        COALESCE(s.SourceCostCenter, p.CostCenter) AS CostCenter,
        COALESCE(s.SourceJobFamily, p.JobFamily) AS JobFamily,
        s.SourceEventId AS SourceEventId
    FROM source_events s
    LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Create Position'
      AND (s.EventType = 'Initiated' OR s.StepName = 'Create Position')
    UNION ALL
    SELECT s.PositionId, 'Manager Approval Submitted', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Create Position' AND s.StepName = 'Manager Approval' AND s.StepStatus = 'Completed'
    UNION ALL
    SELECT s.PositionId, 'HR Approval Submitted', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Create Position' AND s.StepName = 'HR Approval' AND s.StepStatus = 'Completed'
    UNION ALL
    SELECT s.PositionId, 'Budget Approved', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Create Position' AND s.StepName = 'Budget Approval' AND s.StepStatus = 'Completed'
    UNION ALL
    SELECT s.PositionId, 'Position Request Denied', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Create Position' AND (s.StepStatus = 'Denied' OR s.EventType = 'Denied')
    UNION ALL
    SELECT s.PositionId, 'Position Created', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Create Position' AND s.StepStatus = 'Completed' AND s.EventType = 'Completed'
    UNION ALL
    SELECT s.PositionId, 'Position Attributes Modified', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Edit Position Restrictions' AND s.StepStatus = 'Completed'
    UNION ALL
    SELECT s.PositionId, 'Position Reclassified', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Change Job' AND s.StepStatus = 'Completed'
    UNION ALL
    SELECT s.PositionId, 'Position Compliance Reviewed', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.StepStatus = 'Completed' AND (s.StepName = 'Compliance Review' OR s.StepName = 'Compliance Checklist' OR s.StepName = 'Compliance To Do')
    UNION ALL
    SELECT s.PositionId, 'Position Frozen', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Freeze Position' AND s.StepStatus = 'Completed'
    UNION ALL
    SELECT s.PositionId, 'Position Unfrozen', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.PreviousPositionStatus = 'Frozen' AND s.SourcePositionStatus = 'Active'
    UNION ALL
    SELECT s.PositionId, 'Position Deactivated', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Inactivate Position' AND s.StepStatus = 'Completed'
    UNION ALL
    SELECT s.PositionId, 'Position Closed', s.SourceEventTime, s.SourceEndTime, s.SourceUser, COALESCE(s.SourcePositionStatus, p.PositionStatus), s.SourceReasonForChange, COALESCE(s.SourceDepartment, p.Department), COALESCE(s.SourceCostCenter, p.CostCenter), COALESCE(s.SourceJobFamily, p.JobFamily), s.SourceEventId
    FROM source_events s LEFT JOIN position_attributes p ON p.PositionId = s.PositionId
    WHERE s.BusinessProcessName = 'Close Position' AND s.StepStatus = 'Completed'
),
deduplicated_events AS (
    SELECT
        PositionId,
        ActivityName,
        EventTime,
        EndTime,
        User,
        PositionStatus,
        ReasonForChange,
        Department,
        CostCenter,
        JobFamily,
        ROW_NUMBER() OVER (
            PARTITION BY PositionId, ActivityName, EventTime, COALESCE(SourceEventId, '')
            ORDER BY EndTime DESC NULLS LAST
        ) AS duplicate_rank
    FROM normalized_events
)
SELECT
    PositionId,
    ActivityName,
    EventTime,
    EndTime,
    User,
    PositionStatus,
    ReasonForChange,
    Department,
    CostCenter,
    JobFamily
FROM deduplicated_events
WHERE duplicate_rank = 1
  AND PositionId IS NOT NULL
  AND ActivityName IS NOT NULL
  AND EventTime IS NOT NULL
ORDER BY PositionId, EventTime, ActivityName;