お客様の支払い処理データテンプレート

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

class=class="hl-string">"hl-comment"># === CONFIGURATION ===
API_BASE_URL = class="hl-string">"https://[Your FIS API Endpoint]/v1"
CLIENT_ID = class="hl-string">"[Your Client ID]"
CLIENT_SECRET = class="hl-string">"[Your Client Secret]"
START_DATE = class="hl-string">"2023-01-01"
END_DATE = class="hl-string">"2023-01-31"
OUTPUT_FILE = class="hl-string">"fis_payment_event_log.csv"

class=class="hl-string">"hl-comment"># === MAPPINGS ===
class=class="hl-string">"hl-comment"># Maps FIS internal status codes to ProcessMind Activity Names
STATUS_MAP = {
    class="hl-string">"INITIATED": class="hl-string">"Payment Request Created",
    class="hl-string">"VALIDATED": class="hl-string">"Payment Details Validated",
    class="hl-string">"PENDING_AUTH": class="hl-string">"Payment Sent For Approval",
    class="hl-string">"APPROVED": class="hl-string">"Payment Approved",
    class="hl-string">"REJECTED": class="hl-string">"Payment Rejected",
    class="hl-string">"AUTHORIZED": class="hl-string">"Payment Authorized",
    class="hl-string">"SENT_TO_NETWORK": class="hl-string">"Payment Instruction Sent",
    class="hl-string">"ACKNOWLEDGED": class="hl-string">"Payment Confirmed",
    class="hl-string">"SETTLED": class="hl-string">"Payment Settled",
    class="hl-string">"EXCEPTION": class="hl-string">"Payment Error Identified",
    class="hl-string">"EXCEPTION_CLEARED": class="hl-string">"Payment Error Resolved",
    class="hl-string">"RECONCILED": class="hl-string">"Payment Reconciled"
}

def get_access_token():
    url = fclass="hl-string">"{API_BASE_URL}/oauth/token"
    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
    }
    try:
        response = requests.post(url, data=payload)
        response.raise_for_status()
        return response.json()[class="hl-string">"access_token"]
    except Exception as e:
        print(fclass="hl-string">"Authentication failed: {e}")
        exit(1)

def get_payments(token, start, end):
    headers = {class="hl-string">"Authorization": fclass="hl-string">"Bearer {token}"}
    payments = []
    page = 1
    has_more = True
    
    while has_more:
        class=class="hl-string">"hl-comment"># Placeholder endpoint for searching payments
        url = fclass="hl-string">"{API_BASE_URL}/payments?startDate={start}&endDate={end}&page={page}&pageSize=100"
        try:
            resp = requests.get(url, headers=headers)
            resp.raise_for_status()
            data = resp.json()
            
            batch = data.get(class="hl-string">"items", [])
            payments.extend(batch)
            
            if len(batch) < 100:
                has_more = False
            else:
                page += 1
        except Exception as e:
            print(fclass="hl-string">"Error fetching payments page {page}: {e}")
            has_more = False
            
    return payments

def get_payment_history(token, payment_id):
    headers = {class="hl-string">"Authorization": fclass="hl-string">"Bearer {token}"}
    try:
        class=class="hl-string">"hl-comment"># Placeholder endpoint for audit history
        url = fclass="hl-string">"{API_BASE_URL}/payments/{payment_id}/audit"
        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
        return resp.json().get(class="hl-string">"events", [])
    except Exception as e:
        print(fclass="hl-string">"Error fetching history for {payment_id}: {e}")
        return []

def process_data():
    print(class="hl-string">"Authenticating...")
    token = get_access_token()
    
    print(fclass="hl-string">"Fetching payments from {START_DATE} to {END_DATE}...")
    raw_payments = get_payments(token, START_DATE, END_DATE)
    
    event_log = []
    
    print(fclass="hl-string">"Processing {len(raw_payments)} transactions...")
    
    for p in raw_payments:
        p_id = p.get(class="hl-string">"paymentId")
        amount = p.get(class="hl-string">"amount")
        currency = p.get(class="hl-string">"currency")
        p_type = p.get(class="hl-string">"paymentType")
        due_date_str = p.get(class="hl-string">"dueDate")
        unit = p.get(class="hl-string">"businessUnit")
        
        class=class="hl-string">"hl-comment"># Fetch history for this specific payment
        history = get_payment_history(token, p_id)
        
        class=class="hl-string">"hl-comment"># Variables for derivative logic
        settlement_date = None
        due_date = None
        
        if due_date_str:
            try:
                due_date = datetime.fromisoformat(due_date_str.replace(class="hl-string">"Z", class="hl-string">"+00:00"))
            except ValueError:
                pass

        class=class="hl-string">"hl-comment"># Process explicit API events
        for h in history:
            raw_status = h.get(class="hl-string">"status")
            timestamp_str = h.get(class="hl-string">"timestamp")
            user = h.get(class="hl-string">"userId")
            
            class=class="hl-string">"hl-comment"># Map to standard Activity Name
            activity_name = STATUS_MAP.get(raw_status, raw_status)
            
            class=class="hl-string">"hl-comment"># Only include relevant activities
            if activity_name:
                event_entry = {
                    class="hl-string">"PaymentTransactionId": p_id,
                    class="hl-string">"ActivityName": activity_name,
                    class="hl-string">"EventTimestamp": timestamp_str,
                    class="hl-string">"ProcessingUser": user,
                    class="hl-string">"PaymentAmount": amount,
                    class="hl-string">"CurrencyCode": currency,
                    class="hl-string">"PaymentType": p_type,
                    class="hl-string">"PaymentDueDate": due_date_str,
                    class="hl-string">"BusinessUnit": unit,
                    class="hl-string">"ActualSettlementDate": None class=class="hl-string">"hl-comment"># To be filled if known
                }
                
                class=class="hl-string">"hl-comment"># Capture settlement date for logic
                if activity_name == class="hl-string">"Payment Settled":
                    event_entry[class="hl-string">"ActualSettlementDate"] = timestamp_str
                    try:
                        settlement_date = datetime.fromisoformat(timestamp_str.replace(class="hl-string">"Z", class="hl-string">"+00:00"))
                    except ValueError:
                        pass
                        
                event_log.append(event_entry)

        class=class="hl-string">"hl-comment"># Logic for Derivative Activity: Late Payment Identified
        class=class="hl-string">"hl-comment"># Rules: If settled and due date exists, and settled > due date
        if settlement_date and due_date:
            if settlement_date > due_date:
                event_log.append({
                    class="hl-string">"PaymentTransactionId": p_id,
                    class="hl-string">"ActivityName": class="hl-string">"Late Payment Identified",
                    class="hl-string">"EventTimestamp": settlement_date.isoformat(),
                    class="hl-string">"ProcessingUser": class="hl-string">"System Calculation",
                    class="hl-string">"PaymentAmount": amount,
                    class="hl-string">"CurrencyCode": currency,
                    class="hl-string">"PaymentType": p_type,
                    class="hl-string">"PaymentDueDate": due_date_str,
                    class="hl-string">"ActualSettlementDate": settlement_date.isoformat(),
                    class="hl-string">"BusinessUnit": unit
                })

    class=class="hl-string">"hl-comment"># Export to CSV
    df = pd.DataFrame(event_log)
    class=class="hl-string">"hl-comment"># Ensure timestamps are consistent
    if not df.empty:
        df[class="hl-string">'EventTimestamp'] = pd.to_datetime(df[class="hl-string">'EventTimestamp']).dt.strftime(class="hl-string">'%Y-%m-%d %H:%M:%S')
        
    df.to_csv(OUTPUT_FILE, index=False)
    print(fclass="hl-string">"Extraction complete. Data saved to {OUTPUT_FILE}")

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