您的 KYC 客户准入数据模板

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

class=class="hl-string">"hl-comment"># --- Configuration Section --- #
class=class="hl-string">"hl-comment"># Replace with your actual LexisNexis API credentials and endpoints
API_KEY = class="hl-string">'[Your_API_Key]'
API_SECRET = class="hl-string">'[Your_API_Secret]'
BASE_URL = class="hl-string">'[https://api.example.lexisnexis.com/v1]' class=class="hl-string">"hl-comment"># e.g., your specific API gateway URL

class=class="hl-string">"hl-comment"># Set the date range for the extraction
START_DATE = class="hl-string">'2023-01-01'
END_DATE = class="hl-string">'2023-06-30'

class=class="hl-string">"hl-comment"># Delay between API calls to avoid rate limiting
API_CALL_DELAY_SECONDS = 0.5

class=class="hl-string">"hl-comment"># Mapping from LexisNexis event types to standard activity names
class=class="hl-string">"hl-comment"># CRITICAL: The keys of this dictionary must be adjusted to match the event names
class=class="hl-string">"hl-comment"># returned by your specific LexisNexis API endpoint.
ACTIVITY_MAP = {
    class="hl-string">'application.submitted': class="hl-string">'Application Submitted',
    class="hl-string">'screening.initial.performed': class="hl-string">'Initial Screening Performed',
    class="hl-string">'verification.identity.initiated': class="hl-string">'Identity Verification Initiated',
    class="hl-string">'documents.requested': class="hl-string">'Documents Requested',
    class="hl-string">'documents.received': class="hl-string">'Documents Received',
    class="hl-string">'review.document.performed': class="hl-string">'Document Review Performed',
    class="hl-string">'assessment.risk.performed': class="hl-string">'Risk Assessment Performed',
    class="hl-string">'review.compliance.initiated': class="hl-string">'Compliance Review Initiated',
    class="hl-string">'information.additional.requested': class="hl-string">'Additional Information Requested',
    class="hl-string">'review.compliance.completed': class="hl-string">'Compliance Review Completed',
    class="hl-string">'application.approved': class="hl-string">'Application Approved',
    class="hl-string">'account.activated': class="hl-string">'Account Activated',
    class="hl-string">'onboarding.completed': class="hl-string">'Customer Onboarding Completed',
    class="hl-string">'application.rejected': class="hl-string">'Application Rejected'
}

def get_auth_token():
    class="hl-string">"""Handles authentication and returns an access token."""
    class=class="hl-string">"hl-comment"># This is a placeholder for the actual authentication mechanism.
    class=class="hl-string">"hl-comment"># Adjust according to LexisNexis API documentation (e.g., OAuth2).
    class=class="hl-string">"hl-comment"># For simplicity, we assume the API Key can be used directly in a header.
    class=class="hl-string">"hl-comment"># In a real scenario, you might POST to a token endpoint.
    print(class="hl-string">"Authentication successful (simulated).")
    return fclass="hl-string">'Bearer {API_KEY}' class=class="hl-string">"hl-comment"># Simplified for example purposes

def main():
    class="hl-string">"""Main function to orchestrate the data extraction process."""
    auth_token = get_auth_token()
    if not auth_token:
        print(class="hl-string">"Failed to authenticate. Exiting.")
        return

    headers = {
        class="hl-string">'Authorization': auth_token,
        class="hl-string">'Content-Type': class="hl-string">'application/json'
    }

    print(fclass="hl-string">"Fetching applications from {START_DATE} to {END_DATE}...")
    
    class=class="hl-string">"hl-comment"># 1. Fetch all application IDs within the date range
    try:
        applications_endpoint = fclass="hl-string">'{BASE_URL}/applications'
        params = {class="hl-string">'startDate': START_DATE, class="hl-string">'endDate': END_DATE}
        response = requests.get(applications_endpoint, headers=headers, params=params)
        response.raise_for_status()
        applications = response.json().get(class="hl-string">'applications', [])
        print(fclass="hl-string">"Found {len(applications)} applications to process.")
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"Error fetching applications: {e}")
        return

    all_events = []

    class=class="hl-string">"hl-comment"># 2. For each application, fetch its detailed event history
    for i, app in enumerate(applications):
        app_id = app.get(class="hl-string">'customerApplicationId')
        if not app_id:
            continue

        print(fclass="hl-string">"Processing application {i+1}/{len(applications)}: {app_id}")
        try:
            events_endpoint = fclass="hl-string">'{BASE_URL}/applications/{app_id}/events'
            event_response = requests.get(events_endpoint, headers=headers)
            event_response.raise_for_status()
            events_data = event_response.json().get(class="hl-string">'events', [])
            
            class=class="hl-string">"hl-comment"># 3. Parse events and map them to the event log format
            for event in events_data:
                event_type = event.get(class="hl-string">'eventType')
                if event_type in ACTIVITY_MAP:
                    activity_name = ACTIVITY_MAP[event_type]
                    
                    class=class="hl-string">"hl-comment"># Extract attributes, using .get() to handle missing data gracefully
                    record = {
                        class="hl-string">'CustomerApplication': app_id,
                        class="hl-string">'ActivityName': activity_name,
                        class="hl-string">'EventTimestamp': event.get(class="hl-string">'eventTimestamp'),
                        class="hl-string">'EndTime': event.get(class="hl-string">'eventCompletionTimestamp', None), class=class="hl-string">"hl-comment"># Optional
                        class="hl-string">'AssignedUser': event.get(class="hl-string">'userId', class="hl-string">'System'),
                        class="hl-string">'Department': event.get(class="hl-string">'userDepartment', None),
                        class="hl-string">'ApplicationStatus': event.get(class="hl-string">'applicationStatusAfterEvent'),
                        class="hl-string">'RiskLevel': event.get(class="hl-string">'riskAssessment', {}).get(class="hl-string">'level', None),
                        class="hl-string">'SlaTargetDate': app.get(class="hl-string">'slaTargetDate', None)
                    }
                    all_events.append(record)

        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"Could not fetch events for application {app_id}: {e}")
        
        time.sleep(API_CALL_DELAY_SECONDS)

    if not all_events:
        print(class="hl-string">"No events were extracted. The output file will be empty.")
        return

    class=class="hl-string">"hl-comment"># 4. Create a DataFrame and save to CSV
    df = pd.DataFrame(all_events)
    df[class="hl-string">'EventTimestamp'] = pd.to_datetime(df[class="hl-string">'EventTimestamp'])
    df.sort_values(by=[class="hl-string">'CustomerApplication', class="hl-string">'EventTimestamp'], inplace=True)
    
    output_filename = class="hl-string">'lexisnexis_event_log.csv'
    df.to_csv(output_filename, index=False, date_format=class="hl-string">'%Y-%m-%d %H:%M:%S')
    print(fclass="hl-string">"Extraction complete. Event log saved to {output_filename}")

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