您的 KYC 客户开户数据模板

/*
 Part 1: BIX Rule-Admin-Extract Configuration (XML Representation)
 This XML defines the BIX rule. You would configure this using the Pega UI.
 The key is extracting from the History-Work class and including the raw event fields.
*/

<?xml version="1.0"?>
<pagedata>
    <pxObjClass>Rule-Admin-Extract</pxObjClass>
    <pyClassName>[Your-FW-KYC-History-Work]</pyClassName>
    <pyExtractName>ExtractCustomerOnboardingHistory</pyExtractName>
    <pyOutputFormat>CSV</pyOutputFormat>
    <pyDelimiter>,</pyDelimiter>
    <pyTargetFile>kyc_event_log.csv</pyTargetFile>
    <!-- Filter Criteria: Filter for cases created in the last 180 days -->
    <pyFilterCriteria>
        <rowdata REPEATINGTYPE="PageList">
            <row RIX="1">
                <pyFilterProperty>pxHistoryForReference.pxCreateDateTime</pyFilterProperty>
                <pyCondition>IS GREATER THAN</pyCondition>
                <pyFilterValue>CurrentDate-180</pyFilterValue>
            </row>
        </rowdata>
    </pyFilterCriteria>
    <!-- Properties to Extract -->
    <pyExtractProperties>
        <rowdata REPEATINGTYPE="PageList">
            <row RIX="1"><pyTargetProperty>CustomerApplication</pyTargetProperty><pySourceProperty>.pxHistoryForReference.pyID</pySourceProperty></row>
            <row RIX="2"><pyTargetProperty>EventTime</pyTargetProperty><pySourceProperty>.pxTimeCreated</pySourceProperty></row>
            <row RIX="3"><pyTargetProperty>RawActivity</pyTargetProperty><pySourceProperty>.pyMessageKey</pySourceProperty></row>
            <row RIX="4"><pyTargetProperty>HistoryType</pyTargetProperty><pySourceProperty>.pyHistoryType</pySourceProperty></row>
            <row RIX="5"><pyTargetProperty>OperatorId</pyTargetProperty><pySourceProperty>.pxPerformer</pySourceProperty></row>
            <row RIX="6"><pyTargetProperty>WorkGroup</pyTargetProperty><pySourceProperty>.pyAssignedToWorkgroup</pySourceProperty></row>
            <row RIX="7"><pyTargetProperty>ApplicationStatus</pyTargetProperty><pySourceProperty>.pxHistoryForReference.pyStatusWork</pySourceProperty></row>
            <row RIX="8"><pyTargetProperty>SlaTargetDate</pyTargetProperty><pySourceProperty>.pxHistoryForReference.pxGoalTime</pySourceProperty></row>
            <row RIX="9"><pyTargetProperty>RiskLevel</pyTargetProperty><pySourceProperty>.pxHistoryForReference.[YourRiskLevelProperty]</pySourceProperty></row>
            <row RIX="10"><pyTargetProperty>RejectionReason</pyTargetProperty><pySourceProperty>.pxHistoryForReference.[YourRejectionReasonProperty]</pySourceProperty></row>
        </rowdata>
    </pyExtractProperties>
</pagedata>

/*
 Part 2: SQL Transformation Query
 This script is run after BIX exports the raw data. It translates the technical
 Pega history events into the required business-friendly activity names.
 Assumes the BIX output has been loaded into a staging table named 'bix_raw_history'.
*/

SELECT
    b.CustomerApplication,
    CASE
        WHEN b.HistoryType = 'New' THEN 'Application Submitted'
        WHEN b.RawActivity = 'workitem new' THEN 'Application Submitted'
        WHEN b.RawActivity LIKE 'Status changed from New to Pending-Documents' THEN 'Initial Screening Performed'
        WHEN b.RawActivity LIKE 'Correspondence generated for document request' THEN 'Documents Requested'
        WHEN b.RawActivity LIKE 'Attachment added' THEN 'Documents Received'
        WHEN b.RawActivity = 'Flow Action: CompleteDocumentReview' THEN 'Document Review Completed'
        WHEN b.RawActivity LIKE '%BackgroundCheck%' AND b.HistoryType = 'Status Change' THEN 'Background Check Initiated'
        WHEN b.RawActivity = 'StageCompleted: RiskAssessment' THEN 'Risk Assessment Performed'
        WHEN b.RawActivity = 'AssignmentCreated: ComplianceQueue' THEN 'Compliance Review Initiated'
        WHEN b.RawActivity LIKE 'Correspondence sent to customer for info' THEN 'Additional Information Requested'
        WHEN b.RawActivity = 'StageCompleted: ComplianceReview' THEN 'Compliance Review Completed'
        WHEN b.RawActivity = 'Resolved-Completed' AND b.ApplicationStatus LIKE 'Approved%' THEN 'Application Approved'
        WHEN b.RawActivity = 'Status changed from Approved to Active' THEN 'Account Activated'
        WHEN b.RawActivity = 'Resolved-Completed' THEN 'Onboarding Completed'
        WHEN b.RawActivity = 'Resolved-Rejected' THEN 'Application Rejected'
        ELSE NULL -- Exclude rows that do not map to a required activity
    END AS ActivityName,
    b.EventTime,
    NULL AS EndTime, -- EndTime is not directly available from Pega history
    b.OperatorId,
    b.WorkGroup,
    b.ApplicationStatus,
    b.RejectionReason,
    b.SlaTargetDate,
    b.RiskLevel
FROM
    bix_raw_history AS b
WHERE
    CASE
        WHEN b.HistoryType = 'New' THEN 'Application Submitted'
        WHEN b.RawActivity = 'workitem new' THEN 'Application Submitted'
        WHEN b.RawActivity LIKE 'Status changed from New to Pending-Documents' THEN 'Initial Screening Performed'
        WHEN b.RawActivity LIKE 'Correspondence generated for document request' THEN 'Documents Requested'
        WHEN b.RawActivity LIKE 'Attachment added' THEN 'Documents Received'
        WHEN b.RawActivity = 'Flow Action: CompleteDocumentReview' THEN 'Document Review Completed'
        WHEN b.RawActivity LIKE '%BackgroundCheck%' AND b.HistoryType = 'Status Change' THEN 'Background Check Initiated'
        WHEN b.RawActivity = 'StageCompleted: RiskAssessment' THEN 'Risk Assessment Performed'
        WHEN b.RawActivity = 'AssignmentCreated: ComplianceQueue' THEN 'Compliance Review Initiated'
        WHEN b.RawActivity LIKE 'Correspondence sent to customer for info' THEN 'Additional Information Requested'
        WHEN b.RawActivity = 'StageCompleted: ComplianceReview' THEN 'Compliance Review Completed'
        WHEN b.RawActivity = 'Resolved-Completed' AND b.ApplicationStatus LIKE 'Approved%' THEN 'Application Approved'
        WHEN b.RawActivity = 'Status changed from Approved to Active' THEN 'Account Activated'
        WHEN b.RawActivity = 'Resolved-Completed' THEN 'Onboarding Completed'
        WHEN b.RawActivity = 'Resolved-Rejected' THEN 'Application Rejected'
        ELSE NULL
    END IS NOT NULL;

import requests
import pandas as pd
from datetime import datetime

class=class="hl-string">"hl-comment"># --- Configuration Section --- #
class=class="hl-string">"hl-comment"># Replace with your Pega environment details
PEGA_BASE_URL = class="hl-string">"https://[your-pega-server]/prweb/api/v1"
CLIENT_ID = class="hl-string">"[your_client_id]"
CLIENT_SECRET = class="hl-string">"[your_client_secret]"

class=class="hl-string">"hl-comment"># Pega Case Type for KYC Customer Onboarding
CASE_TYPE = class="hl-string">"[Your-KYC-Case-Type]"

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"># Mapping from Pega statuses/audit notes to standard Activity Names
class=class="hl-string">"hl-comment"># This section requires customization based on your specific Pega application configuration.
ACTIVITY_MAPPING = {
    class="hl-string">'Pending-InitialScreening': class="hl-string">'Initial Screening Performed',
    class="hl-string">'Pending-Customer-Docs': class="hl-string">'Documents Requested',
    class="hl-string">'Attachment-Added': class="hl-string">'Documents Received',
    class="hl-string">'Pending-DocReview': class="hl-string">'Document Review Completed',
    class="hl-string">'Pending-BackgroundCheck': class="hl-string">'Background Check Initiated',
    class="hl-string">'Pending-RiskAssessment': class="hl-string">'Risk Assessment Performed',
    class="hl-string">'Pending-ComplianceReview': class="hl-string">'Compliance Review Initiated',
    class="hl-string">'Correspondence-Sent-InfoRequest': class="hl-string">'Additional Information Requested',
    class="hl-string">'Pending-FinalDecision': class="hl-string">'Compliance Review Completed',
    class="hl-string">'Resolved-Completed': class="hl-string">'Application Approved',
    class="hl-string">'Status-AccountActivated': class="hl-string">'Account Activated',
    class="hl-string">'Resolved-Onboarded': class="hl-string">'Onboarding Completed',
    class="hl-string">'Resolved-Rejected': class="hl-string">'Application Rejected'
}

class=class="hl-string">"hl-comment"># --- Script Logic --- #

def get_access_token():
    class="hl-string">"""Authenticates with Pega and returns an access token."""
    token_url = fclass="hl-string">"{PEGA_BASE_URL.replace('/api/v1', '')}/PRRestService/oauth2/v1/token"
    auth_payload = {
        class="hl-string">'grant_type': class="hl-string">'client_credentials'
    }
    try:
        response = requests.post(token_url, auth=(CLIENT_ID, CLIENT_SECRET), data=auth_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 get_case_ids(token):
    class="hl-string">"""Retrieves a list of case IDs for the specified date range and case type."""
    case_ids = []
    page = 1
    headers = {class="hl-string">'Authorization': fclass="hl-string">'Bearer {token}'}
    
    while True:
        query_params = {
            class="hl-string">'caseTypeID': CASE_TYPE,
            class="hl-string">'query': fclass="hl-string">'.pxCreateDateTime > "{START_DATE}T00:00:00.000Z" AND .pxCreateDateTime < "{END_DATE}T23:59:59.000Z"',
            class="hl-string">'page': page,
            class="hl-string">'pageSize': 1000  class=class="hl-string">"hl-comment"># Adjust page size as needed
        }
        try:
            response = requests.get(fclass="hl-string">"{PEGA_BASE_URL}/cases", headers=headers, params=query_params)
            response.raise_for_status()
            data = response.json()
            cases_on_page = data.get(class="hl-string">'cases', [])
            if not cases_on_page:
                break class=class="hl-string">"hl-comment"># Exit loop if no more cases are returned
            
            for case in cases_on_page:
                case_ids.append(case.get(class="hl-string">'ID'))
            
            page += 1
        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"Error fetching case IDs on page {page}: {e}")
            break
    return case_ids

def main():
    class="hl-string">"""Main function to extract, process, and save event data."""
    print(class="hl-string">"Starting Pega KYC data extraction...")
    access_token = get_access_token()
    if not access_token:
        return

    print(fclass="hl-string">"Successfully authenticated. Fetching case IDs between {START_DATE} and {END_DATE}...")
    case_ids = get_case_ids(access_token)
    print(fclass="hl-string">"Found {len(case_ids)} cases to process.")

    event_log = []
    headers = {class="hl-string">'Authorization': fclass="hl-string">'Bearer {access_token}'}

    for i, case_id in enumerate(case_ids):
        print(fclass="hl-string">"Processing case {i+1}/{len(case_ids)}: {case_id}")
        try:
            case_response = requests.get(fclass="hl-string">"{PEGA_BASE_URL}/cases/{case_id}", headers=headers)
            case_response.raise_for_status()
            case_data = case_response.json()

            class=class="hl-string">"hl-comment"># Extract case-level attributes
            case_info = case_data.get(class="hl-string">'content', {})
            rejection_reason = case_info.get(class="hl-string">'RejectionReason', None)
            sla_date = case_info.get(class="hl-string">'pySLADeadline', None)
            risk_level = case_info.get(class="hl-string">'RiskLevel', None)

            class=class="hl-string">"hl-comment"># 1. Application Submitted Event (from case creation time)
            event_log.append({
                class="hl-string">'CustomerApplication': case_id,
                class="hl-string">'ActivityName': class="hl-string">'Application Submitted',
                class="hl-string">'EventTime': case_info.get(class="hl-string">'pxCreateDateTime'),
                class="hl-string">'EndTime': None, class=class="hl-string">"hl-comment"># EndTime can be derived in post-processing if needed
                class="hl-string">'OperatorId': case_info.get(class="hl-string">'pxCreateOperator'),
                class="hl-string">'WorkGroup': case_info.get(class="hl-string">'pxCreateOpWorkgroup', None),
                class="hl-string">'ApplicationStatus': case_info.get(class="hl-string">'pyStatusWork'),
                class="hl-string">'RejectionReason': rejection_reason,
                class="hl-string">'SlaTargetDate': sla_date,
                class="hl-string">'RiskLevel': risk_level
            })

            class=class="hl-string">"hl-comment"># 2. Process Audit History for other events
            audit_history = case_data.get(class="hl-string">'audit', [])
            for audit_entry in audit_history:
                timestamp = audit_entry.get(class="hl-string">'timeStamp')
                operator = audit_entry.get(class="hl-string">'performedBy', {}).get(class="hl-string">'userID')
                message = audit_entry.get(class="hl-string">'message')
                
                class=class="hl-string">"hl-comment"># Map status changes and specific messages to activities
                activity_name = None
                if class="hl-string">'changed status to' in message:
                    new_status = message.split(class="hl-string">' to ')[-1].strip()
                    activity_name = ACTIVITY_MAPPING.get(new_status)
                elif class="hl-string">'Created attachment' in message:
                    class=class="hl-string">"hl-comment"># This is a generic mapping, may need refinement
                    activity_name = ACTIVITY_MAPPING.get(class="hl-string">'Attachment-Added')
                elif class="hl-string">'Sent correspondence' in message and class="hl-string">'Request for Information' in message:
                     activity_name = ACTIVITY_MAPPING.get(class="hl-string">'Correspondence-Sent-InfoRequest')
                
                if activity_name:
                    event_log.append({
                        class="hl-string">'CustomerApplication': case_id,
                        class="hl-string">'ActivityName': activity_name,
                        class="hl-string">'EventTime': timestamp,
                        class="hl-string">'EndTime': None,
                        class="hl-string">'OperatorId': operator,
                        class="hl-string">'WorkGroup': None, class=class="hl-string">"hl-comment"># Workgroup is not always available in audit trail
                        class="hl-string">'ApplicationStatus': case_info.get(class="hl-string">'pyStatusWork'),
                        class="hl-string">'RejectionReason': rejection_reason,
                        class="hl-string">'SlaTargetDate': sla_date,
                        class="hl-string">'RiskLevel': risk_level
                    })

        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"Could not process case {case_id}: {e}")
        except Exception as ex:
            print(fclass="hl-string">"An unexpected error occurred processing case {case_id}: {ex}")

    print(class="hl-string">"Processing complete. Generating CSV file...")
    df = pd.DataFrame(event_log)
    df = df.sort_values(by=[class="hl-string">'CustomerApplication', class="hl-string">'EventTime'], ascending=[True, True])
    df.to_csv(class="hl-string">'pega_kyc_event_log.csv', index=False)
    print(class="hl-string">"Event log saved to pega_kyc_event_log.csv")

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