Votre modèle de données Hire to Retire, gestion des postes

/*
Integration Center is a UI-based tool. This script represents the logical configuration
of the entities, fields, and filters for each activity within a single integration job.
Each 'BLOCK' would be configured as a sibling element in the Integration Center UI.
*/

// BLOCK 1: Position Created in HR System
// This is the root element of the integration.
ENTITY: Position
FIELDS:
  PositionId = Position.code
  ActivityName = 'Position Created in HR System' (Fixed Value)
  EventTime = Position.createdDate
  SourceSystemId = '[Your SFTP Server Name]' (Fixed Value)
  LastDataUpdateTime = Current Execution Time (System Variable)
  EventEndTime = Position.createdDate
  UserId = Position.createdBy
  DepartmentName = Position.department.name
  PositionStatus = Position.status
  ApproverId = null (Fixed Value)
FILTERS:
  Position.createdDate >= [Start Date]

-- UNION ALL (Configured by adding a sibling element)

// BLOCK 2: Workflow Events (Initiated, Approved, Rejected)
// This single block uses calculated columns to generate multiple event types.
ENTITY: Position -> positionWorkOrder -> wfRequestNav
FIELDS:
  PositionId = Position.code
  ActivityName = Calculated Field:
    IF wfRequestNav.wfRequestSteps.stepNum = 1 AND wfRequestNav.status = 'PENDING' THEN 'Position Request Initiated'
    IF wfRequestNav.wfRequestSteps.stepNum = [Manager Approval Step Number] AND wfRequestNav.wfRequestSteps.status = 'APPROVED' THEN 'Position Request Approved by Manager'
    IF wfRequestNav.wfRequestSteps.stepNum = [HR Approval Step Number] AND wfRequestNav.wfRequestSteps.status = 'APPROVED' THEN 'Position Request Approved by HR'
    IF wfRequestNav.wfRequestSteps.stepNum = [Budget Approval Step Number] AND wfRequestNav.wfRequestSteps.status = 'APPROVED' THEN 'Position Budget Approved'
    IF wfRequestNav.wfRequestSteps.status = 'REJECTED' THEN 'Position Request Rejected'
    ELSE null
  EventTime = wfRequestNav.wfRequestSteps.completedOn
  SourceSystemId = '[Your SFTP Server Name]' (Fixed Value)
  LastDataUpdateTime = Current Execution Time (System Variable)
  EventEndTime = wfRequestNav.wfRequestSteps.completedOn
  UserId = wfRequestNav.wfRequestSteps.processedBy
  DepartmentName = Position.department.name
  PositionStatus = Position.status
  ApproverId = wfRequestNav.wfRequestSteps.approverId
FILTERS:
  ActivityName IS NOT null
  wfRequestNav.wfRequestSteps.lastModifiedDateTime >= [Start Date]

-- UNION ALL (Configured by adding a sibling element)

// BLOCK 3: Position Activated
ENTITY: Position (Effective Dated History)
FIELDS:
  PositionId = Position.code
  ActivityName = 'Position Activated' (Fixed Value)
  EventTime = Position.effectiveStartDate
  SourceSystemId = '[Your SFTP Server Name]' (Fixed Value)
  LastDataUpdateTime = Current Execution Time (System Variable)
  EventEndTime = Position.effectiveStartDate
  UserId = Position.lastModifiedBy
  DepartmentName = Position.department.name
  PositionStatus = Position.status
  ApproverId = null (Fixed Value)
FILTERS:
  Position.status = 'A' (Active)
  // This requires an advanced filter to check the status of the previous record, which can be complex in Integration Center.
  Position.lastModifiedDateTime >= [Start Date]

-- UNION ALL (Configured by adding a sibling element)

// BLOCK 4: Position Attributes Modified
ENTITY: Position (Effective Dated History)
FIELDS:
  PositionId = Position.code
  ActivityName = 'Position Attributes Modified' (Fixed Value)
  EventTime = Position.lastModifiedDateTime
  SourceSystemId = '[Your SFTP Server Name]' (Fixed Value)
  LastDataUpdateTime = Current Execution Time (System Variable)
  EventEndTime = Position.lastModifiedDateTime
  UserId = Position.lastModifiedBy
  DepartmentName = Position.department.name
  PositionStatus = Position.status
  ApproverId = null (Fixed Value)
FILTERS:
  // This filter should exclude records that are purely status changes captured in other blocks.
  Position.lastModifiedDateTime >= [Start Date]

-- UNION ALL (Configured by adding a sibling element)

// BLOCK 5: Position Reclassified
ENTITY: Position (Effective Dated History)
FIELDS:
  PositionId = Position.code
  ActivityName = 'Position Reclassified' (Fixed Value)
  EventTime = Position.effectiveStartDate
  SourceSystemId = '[Your SFTP Server Name]' (Fixed Value)
  LastDataUpdateTime = Current Execution Time (System Variable)
  EventEndTime = Position.effectiveStartDate
  UserId = Position.lastModifiedBy
  DepartmentName = Position.department.name
  PositionStatus = Position.status
  ApproverId = null (Fixed Value)
FILTERS:
  // Requires advanced filter to check if Position.jobCode has changed from the previous record.
  Position.lastModifiedDateTime >= [Start Date]

-- UNION ALL (Configured by adding a sibling element)

// BLOCK 6: Position Reviewed for Compliance
// This requires a custom MDF object or workflow configured by the customer.
ENTITY: Position -> cust_ComplianceReview.position
FIELDS:
  PositionId = Position.code
  ActivityName = 'Position Reviewed for Compliance' (Fixed Value)
  EventTime = cust_ComplianceReview.lastModifiedDate
  SourceSystemId = '[Your SFTP Server Name]' (Fixed Value)
  LastDataUpdateTime = Current Execution Time (System Variable)
  EventEndTime = cust_ComplianceReview.lastModifiedDate
  UserId = cust_ComplianceReview.lastModifiedBy
  DepartmentName = Position.department.name
  PositionStatus = Position.status
  ApproverId = cust_ComplianceReview.reviewerId
FILTERS:
  cust_ComplianceReview.lastModifiedDate >= [Start Date]

-- UNION ALL (Configured by adding a sibling element)

// BLOCK 7: Position Frozen/Suspended
ENTITY: Position (Effective Dated History)
FIELDS:
  PositionId = Position.code
  ActivityName = 'Position Frozen/Suspended' (Fixed Value)
  EventTime = Position.effectiveStartDate
  SourceSystemId = '[Your SFTP Server Name]' (Fixed Value)
  LastDataUpdateTime = Current Execution Time (System Variable)
  EventEndTime = Position.effectiveStartDate
  UserId = Position.lastModifiedBy
  DepartmentName = Position.department.name
  PositionStatus = Position.status
  ApproverId = null (Fixed Value)
FILTERS:
  Position.status = '[Your Frozen Status Code]'
  Position.lastModifiedDateTime >= [Start Date]

-- UNION ALL (Configured by adding a sibling element)

// BLOCK 8: Position Deactivated
ENTITY: Position (Effective Dated History)
FIELDS:
  PositionId = Position.code
  ActivityName = 'Position Deactivated' (Fixed Value)
  EventTime = Position.effectiveStartDate
  SourceSystemId = '[Your SFTP Server Name]' (Fixed Value)
  LastDataUpdateTime = Current Execution Time (System Variable)
  EventEndTime = Position.effectiveStartDate
  UserId = Position.lastModifiedBy
  DepartmentName = Position.department.name
  PositionStatus = Position.status
  ApproverId = null (Fixed Value)
FILTERS:
  Position.status = 'I' (Inactive)
  Position.lastModifiedDateTime >= [Start Date]

-- UNION ALL (Configured by adding a sibling element)

// BLOCK 9: Position Closed
ENTITY: Position (Effective Dated History)
FIELDS:
  PositionId = Position.code
  ActivityName = 'Position Closed' (Fixed Value)
  EventTime = Position.effectiveStartDate
  SourceSystemId = '[Your SFTP Server Name]' (Fixed Value)
  LastDataUpdateTime = Current Execution Time (System Variable)
  EventEndTime = Position.effectiveStartDate
  UserId = Position.lastModifiedBy
  DepartmentName = Position.department.name
  PositionStatus = Position.status
  ApproverId = null (Fixed Value)
FILTERS:
  Position.status = '[Your Closed/Eliminated Status Code]'
  Position.lastModifiedDateTime >= [Start Date]

SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Request Initiated]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Request Approved by Manager]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Request Approved by HR]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Budget Approved]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Created in HR System]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Activated]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Attributes Modified]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Reclassified]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Reviewed for Compliance]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Frozen or Suspended]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Deactivated]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Request Rejected]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter]
UNION ALL
SELECT
    PositionId,
    ActivityName,
    EventTime,
    SourceSystemId,
    LastDataUpdateTime,
    EventEndTime,
    UserId,
    DepartmentName,
    PositionStatus,
    ApproverId
FROM [Your configured Canvas Report output for Position Closed]
WHERE EventTime >= [Start date parameter]
  AND EventTime < [End date parameter];

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

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># Replace with your SuccessFactors API details
API_ENDPOINT = class="hl-string">"https://apiX.successfactors.com/odata/v2"
COMPANY_ID = class="hl-string">"[Your_Company_ID]"
API_USER = class="hl-string">"[Your_API_User]@[Your_Company_ID]"
API_PASSWORD = class="hl-string">"[Your_API_Password]"

class=class="hl-string">"hl-comment"># Date range for 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"># Pagination size
PAGE_SIZE = 500

class=class="hl-string">"hl-comment"># Mappings (Customize to your SF configuration)
STATUS_ACTIVITY_MAP = {
    class="hl-string">'A': class="hl-string">'Position Activated',
    class="hl-string">'I': class="hl-string">'Position Deactivated',
    class="hl-string">'S': class="hl-string">'Position Frozen/Suspended',
    class="hl-string">'E': class="hl-string">'Position Closed',
    class="hl-string">'F': class="hl-string">'Position Frozen/Suspended'
}

class=class="hl-string">"hl-comment"># Note: Workflow step names can be found in WF_REQUEST_STEP.step_name
WORKFLOW_STEP_MAP = {
    class="hl-string">'Manager Approval': class="hl-string">'Position Request Approved by Manager',
    class="hl-string">'HR Approval': class="hl-string">'Position Request Approved by HR',
    class="hl-string">'Budget Approval': class="hl-string">'Position Budget Approved'
}

class=class="hl-string">"hl-comment"># --- Helper Functions ---
def fetch_sf_data(entity, select_fields, filter_query):
    class="hl-string">"""Fetches paginated data from a SuccessFactors OData entity."""
    all_records = []
    skip = 0
    while True:
        url = fclass="hl-string">"{API_ENDPOINT}/{entity}?$format=json&$top={PAGE_SIZE}&$skip={skip}"
        if select_fields:
            url += fclass="hl-string">"&$select={','.join(select_fields)}"
        if filter_query:
            url += fclass="hl-string">"&$filter={filter_query}"
        
        print(fclass="hl-string">"Fetching {entity} data, page starting at record {skip}...")
        response = requests.get(url, auth=(API_USER, API_PASSWORD))
        if response.status_code != 200:
            print(fclass="hl-string">"Error fetching {entity}: {response.status_code} {response.text}")
            break

        data = response.json()[class="hl-string">'d'][class="hl-string">'results']
        if not data:
            break

        all_records.extend(data)
        skip += PAGE_SIZE

    return all_records

def parse_sf_timestamp(sf_date):
    class="hl-string">"""Parses SuccessFactors OData timestamp format /Date(...)/."""
    if not sf_date:
        return None
    try:
        class=class="hl-string">"hl-comment"># Extract timestamp in milliseconds and convert to datetime
        ts = int(sf_date.split(class="hl-string">'(')[1].split(class="hl-string">')')[0])
        return datetime.fromtimestamp(ts / 1000)
    except (IndexError, ValueError):
        return None

class=class="hl-string">"hl-comment"># --- Main Extraction Logic ---
def main():
    events = []
    source_system_id = class="hl-string">"SAP SuccessFactors"
    last_data_update_time = datetime.now()

    class=class="hl-string">"hl-comment"># 1. Fetch Position history data
    position_filter = fclass="hl-string">"lastModifiedDateTime ge datetime'{START_DATE}T00:00:00Z' and lastModifiedDateTime le datetime'{END_DATE}T23:59:59Z'"
    position_fields = [class="hl-string">"code", class="hl-string">"effectiveStartDate", class="hl-string">"status", class="hl-string">"jobCode", class="hl-string">"lastModifiedBy", class="hl-string">"lastModifiedDateTime", class="hl-string">"createdBy", class="hl-string">"createdDateTime", class="hl-string">"departmentNav/name"]
    positions_raw = fetch_sf_data(class="hl-string">"Position", position_fields, position_filter)

    class=class="hl-string">"hl-comment"># Group by position code to process history
    positions_by_code = {}
    for pos in positions_raw:
        code = pos.get(class="hl-string">'code')
        if code not in positions_by_code:
            positions_by_code[code] = []
        positions_by_code[code].append(pos)

    class=class="hl-string">"hl-comment"># Process Position history for events
    for pos_id, history in positions_by_code.items():
        class=class="hl-string">"hl-comment"># Sort records by effective date and last modified date to establish order
        history.sort(key=lambda x: (parse_sf_timestamp(x[class="hl-string">'createdDateTime']), parse_sf_timestamp(x[class="hl-string">'lastModifiedDateTime'])))

        class=class="hl-string">"hl-comment"># Event: Position Created in HR System
        creation_record = history[0]
        events.append({
            class="hl-string">'PositionId': pos_id,
            class="hl-string">'ActivityName': class="hl-string">'Position Created in HR System',
            class="hl-string">'EventTime': parse_sf_timestamp(creation_record[class="hl-string">'createdDateTime']),
            class="hl-string">'UserId': creation_record.get(class="hl-string">'createdBy'),
            class="hl-string">'DepartmentName': creation_record.get(class="hl-string">'departmentNav', {}).get(class="hl-string">'name'),
            class="hl-string">'PositionStatus': creation_record.get(class="hl-string">'status'),
            class="hl-string">'ApproverId': None
        })
        
        last_status = creation_record.get(class="hl-string">'status')
        last_job_code = creation_record.get(class="hl-string">'jobCode')

        for i in range(len(history)):
            record = history[i]
            event_time = parse_sf_timestamp(record[class="hl-string">'lastModifiedDateTime'])
            department = record.get(class="hl-string">'departmentNav', {}).get(class="hl-string">'name')

            class=class="hl-string">"hl-comment"># Events from status changes (Activated, Deactivated, Frozen, Closed)
            current_status = record.get(class="hl-string">'status')
            if current_status != last_status and current_status in STATUS_ACTIVITY_MAP:
                events.append({
                    class="hl-string">'PositionId': pos_id,
                    class="hl-string">'ActivityName': STATUS_ACTIVITY_MAP[current_status],
                    class="hl-string">'EventTime': event_time,
                    class="hl-string">'UserId': record.get(class="hl-string">'lastModifiedBy'),
                    class="hl-string">'DepartmentName': department,
                    class="hl-string">'PositionStatus': current_status,
                    class="hl-string">'ApproverId': None
                })
            last_status = current_status

            class=class="hl-string">"hl-comment"># Event: Position Reclassified (Job Code changed)
            current_job_code = record.get(class="hl-string">'jobCode')
            if current_job_code != last_job_code and i > 0: class=class="hl-string">"hl-comment"># Avoid flagging on creation
                events.append({
                    class="hl-string">'PositionId': pos_id,
                    class="hl-string">'ActivityName': class="hl-string">'Position Reclassified',
                    class="hl-string">'EventTime': event_time,
                    class="hl-string">'UserId': record.get(class="hl-string">'lastModifiedBy'),
                    class="hl-string">'DepartmentName': department,
                    class="hl-string">'PositionStatus': current_status,
                    class="hl-string">'ApproverId': None
                })
            last_job_code = current_job_code

            class=class="hl-string">"hl-comment"># Event: Position Attributes Modified (generic modification)
            if i > 0: class=class="hl-string">"hl-comment"># This fires for any modification after creation
                 events.append({
                    class="hl-string">'PositionId': pos_id,
                    class="hl-string">'ActivityName': class="hl-string">'Position Attributes Modified',
                    class="hl-string">'EventTime': event_time,
                    class="hl-string">'UserId': record.get(class="hl-string">'lastModifiedBy'),
                    class="hl-string">'DepartmentName': department,
                    class="hl-string">'PositionStatus': current_status,
                    class="hl-string">'ApproverId': None
                })


    class=class="hl-string">"hl-comment"># 2. Fetch Workflow (WfRequest) data for position creations
    wf_filter = fclass="hl-string">"lastModifiedDateTime ge datetime'{START_DATE}T00:00:00Z' and (module eq 'POSITION' or objectType eq 'Position')"
    wf_fields = [class="hl-string">"wfRequestId", class="hl-string">"status", class="hl-string">"effectiveDate", class="hl-string">"createdBy", class="hl-string">"createdDate", class="hl-string">"wfRequestSteps/processedBy", class="hl-string">"wfRequestSteps/status", class="hl-string">"wfRequestSteps/createdDate", class="hl-string">"wfRequestSteps/stepName", class="hl-string">"objectKey"]
    wf_requests_raw = fetch_sf_data(class="hl-string">"WfRequest", wf_fields, wf_filter)

    for wf in wf_requests_raw:
        pos_id = wf.get(class="hl-string">'objectKey')
        if not pos_id: continue

        class=class="hl-string">"hl-comment"># Event: Position Request Initiated
        events.append({
            class="hl-string">'PositionId': pos_id,
            class="hl-string">'ActivityName': class="hl-string">'Position Request Initiated',
            class="hl-string">'EventTime': parse_sf_timestamp(wf[class="hl-string">'createdDate']),
            class="hl-string">'UserId': wf.get(class="hl-string">'createdBy'),
            class="hl-string">'DepartmentName': None, class=class="hl-string">"hl-comment"># Department info is not in WfRequest
            class="hl-string">'PositionStatus': class="hl-string">'Pending',
            class="hl-string">'ApproverId': None
        })

        class=class="hl-string">"hl-comment"># Event: Position Request Rejected
        if wf[class="hl-string">'status'] == class="hl-string">'REJECTED':
            last_step = wf.get(class="hl-string">'wfRequestSteps', {}).get(class="hl-string">'results', [])[-1] if wf.get(class="hl-string">'wfRequestSteps') else {}
            events.append({
                class="hl-string">'PositionId': pos_id,
                class="hl-string">'ActivityName': class="hl-string">'Position Request Rejected',
                class="hl-string">'EventTime': parse_sf_timestamp(last_step.get(class="hl-string">'createdDate')),
                class="hl-string">'UserId': last_step.get(class="hl-string">'processedBy'),
                class="hl-string">'DepartmentName': None,
                class="hl-string">'PositionStatus': class="hl-string">'Rejected',
                class="hl-string">'ApproverId': last_step.get(class="hl-string">'processedBy')
            })

        class=class="hl-string">"hl-comment"># Events for approval steps
        if wf.get(class="hl-string">'wfRequestSteps') and class="hl-string">'results' in wf[class="hl-string">'wfRequestSteps']:
            for step in wf[class="hl-string">'wfRequestSteps'][class="hl-string">'results']:
                step_name = step.get(class="hl-string">'stepName', class="hl-string">'')
                if step[class="hl-string">'status'] == class="hl-string">'APPROVED' and step_name in WORKFLOW_STEP_MAP:
                    events.append({
                        class="hl-string">'PositionId': pos_id,
                        class="hl-string">'ActivityName': WORKFLOW_STEP_MAP[step_name],
                        class="hl-string">'EventTime': parse_sf_timestamp(step[class="hl-string">'createdDate']),
                        class="hl-string">'UserId': step.get(class="hl-string">'processedBy'),
                        class="hl-string">'DepartmentName': None,
                        class="hl-string">'PositionStatus': class="hl-string">'In Approval',
                        class="hl-string">'ApproverId': step.get(class="hl-string">'processedBy')
                    })

    class=class="hl-string">"hl-comment"># 3. Placeholder for class="hl-string">'Position Reviewed for Compliance'
    class=class="hl-string">"hl-comment"># This event is highly customer-specific. You need to implement the logic here.
    class=class="hl-string">"hl-comment"># Example: Query a custom MDF object or a specific workflow type.
    class=class="hl-string">"hl-comment"># compliance_events = get_compliance_review_events(START_DATE, END_DATE)
    class=class="hl-string">"hl-comment"># events.extend(compliance_events)

    class=class="hl-string">"hl-comment"># --- Final Dataframe Creation ---
    if not events:
        print(class="hl-string">"No events found for the specified period.")
        return

    df = pd.DataFrame(events)

    class=class="hl-string">"hl-comment"># Add standard columns
    df[class="hl-string">'SourceSystemId'] = source_system_id
    df[class="hl-string">'LastDataUpdateTime'] = last_data_update_time
    df[class="hl-string">'EventEndTime'] = df[class="hl-string">'EventTime'] class=class="hl-string">"hl-comment"># For atomic events, end time is the same as start time

    class=class="hl-string">"hl-comment"># Standardize column names and order
    final_columns = [
        class="hl-string">'PositionId', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'EventEndTime', class="hl-string">'UserId',
        class="hl-string">'DepartmentName', class="hl-string">'PositionStatus', class="hl-string">'ApproverId', class="hl-string">'SourceSystemId', class="hl-string">'LastDataUpdateTime'
    ]
    df = df.reindex(columns=final_columns)

    class=class="hl-string">"hl-comment"># Clean up and save
    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)

    output_filename = class="hl-string">'position_management_event_log.csv'
    df.to_csv(output_filename, index=False)
    print(fclass="hl-string">"Event log successfully created: {output_filename}")

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