계약 관리 데이터 템플릿

import requests
import jwt
import time
import datetime
import csv
import json

class=class="hl-string">"hl-comment"># --- Configuration --- 
class=class="hl-string">"hl-comment"># Replace with your DocuSign account details
INTEGRATION_KEY = class="hl-string">"[Your Integration Key]"
USER_ID = class="hl-string">"[Your User ID GUID]"
ACCOUNT_ID = class="hl-string">"[Your Account ID]"
OAUTH_BASE_PATH = class="hl-string">"account-d.docusign.com" class=class="hl-string">"hl-comment"># Use account.docusign.com for demo, account.docusign.com for production

class=class="hl-string">"hl-comment"># --- Constants ---
SOURCE_SYSTEM = class="hl-string">"DocuSign CLM"
LAST_DATA_UPDATE = datetime.datetime.utcnow().isoformat() + class="hl-string">'Z'

class=class="hl-string">"hl-comment"># --- Functions ---
def get_access_token(private_key_path):
    class="hl-string">"""Generates a JWT and exchanges it for an access token."""
    try:
        with open(private_key_path, class="hl-string">'r') as key_file:
            private_key = key_file.read()

        payload = {
            class="hl-string">"iss": INTEGRATION_KEY,
            class="hl-string">"sub": USER_ID,
            class="hl-string">"iat": int(time.time()),
            class="hl-string">"exp": int(time.time()) + 3600,
            class="hl-string">"aud": OAUTH_BASE_PATH,
            class="hl-string">"scope": class="hl-string">"signature impersonation"
        }

        token = jwt.encode(payload, private_key, algorithm=class="hl-string">'RS256')
        
        headers = {class="hl-string">'Content-Type': class="hl-string">'application/x-www-form-urlencoded'}
        data = {
            class="hl-string">'grant_type': class="hl-string">'urn:ietf:params:oauth:grant-type:jwt-bearer',
            class="hl-string">'assertion': token
        }
        
        response = requests.post(fclass="hl-string">"https протокол://{OAUTH_BASE_PATH}/oauth/token", headers=headers, data=data)
        response.raise_for_status()
        return response.json()[class="hl-string">'access_token']

    except Exception as e:
        print(fclass="hl-string">"Error getting access token: {e}")
        return None

def get_document_attribute(doc_attributes, attribute_name):
    class="hl-string">"""Helper function to safely get a value from the document's attribute list."""
    for attr in doc_attributes:
        if attr.get(class="hl-string">'name') == attribute_name:
            return attr.get(class="hl-string">'value')
    return None

def extract_event_log(access_token, start_date_str, end_date_str):
    class="hl-string">"""Main function to extract, process, and save the event log."""
    
    event_log = []
    api_base_url = fclass="hl-string">"https протокол://{OAUTH_BASE_PATH.replace('account', 'api')}/api/v2/accounts/{ACCOUNT_ID}"
    headers = {class="hl-string">'Authorization': fclass="hl-string">'Bearer {access_token}', class="hl-string">'Content-Type': class="hl-string">'application/json'}
    
    class=class="hl-string">"hl-comment"># Activity mapping: Customize keys to match your workflow step names
    ACTIVITY_MAP = {
        class="hl-string">'Start': class="hl-string">'Contract Request Initiated',
        class="hl-string">'Drafting': class="hl-string">'Contract Drafted',
        class="hl-string">'Internal Review': class="hl-string">'Internal Review Started',
        class="hl-string">'Legal Review': class="hl-string">'Legal Review Started',
        class="hl-string">'Send for Internal Approval': class="hl-string">'Internal Approval Sent',
        class="hl-string">'Internal Approval': class="hl-string">'Internal Approval Received',
        class="hl-string">'Send to Counterparty': class="hl-string">'Sent to Counterparty',
        class="hl-string">'Send for Signature': class="hl-string">'Contract Sent for Signature',
        class="hl-string">'Executed': class="hl-string">'Contract Executed',
        class="hl-string">'Archive': class="hl-string">'Contract Stored in Repository',
        class="hl-string">'Start Obligation Monitoring': class="hl-string">'Obligation Monitoring Started',
        class="hl-string">'Initiate Renewal': class="hl-string">'Contract Renewal Initiated',
        class="hl-string">'Terminate': class="hl-string">'Contract Terminated'
    }
    
    class=class="hl-string">"hl-comment"># 1. Search for documents (contracts)
    search_url = fclass="hl-string">"{api_base_url}/search"
    search_params = {
        class="hl-string">'q': fclass="hl-string">'IsLatestVersion:true AND CreatedDate:[{start_date_str} TO {end_date_str}]',
        class="hl-string">'limit': 100, class=class="hl-string">"hl-comment"># Max limit per page
        class="hl-string">'offset': 0
    }
    
    while True:
        print(fclass="hl-string">"Searching for documents with offset {search_params['offset']}...")
        response = requests.get(search_url, headers=headers, params=search_params)
        if response.status_code != 200:
            print(fclass="hl-string">"Error searching documents: {response.text}")
            break
        
        data = response.json()
        documents = data.get(class="hl-string">'value', [])
        
        if not documents:
            print(class="hl-string">"No more documents found.")
            break

        for doc in documents:
            contract_id = doc.get(class="hl-string">'id')
            doc_attributes = doc.get(class="hl-string">'attributes', [])
            
            base_event_data = {
                class="hl-string">'ContractId': contract_id,
                class="hl-string">'ContractStatus': get_document_attribute(doc_attributes, class="hl-string">'Contract Status'),
                class="hl-string">'ContractOwner': get_document_attribute(doc_attributes, class="hl-string">'Contract Owner'),
                class="hl-string">'ContractType': doc.get(class="hl-string">'documentTypeName'),
                class="hl-string">'ContractValue': get_document_attribute(doc_attributes, class="hl-string">'Contract Value'),
                class="hl-string">'Counterparty': get_document_attribute(doc_attributes, class="hl-string">'Counterparty Name'),
                class="hl-string">'ExpirationDate': get_document_attribute(doc_attributes, class="hl-string">'Expiration Date'),
                class="hl-string">'SourceSystem': SOURCE_SYSTEM,
                class="hl-string">'LastDataUpdate': LAST_DATA_UPDATE,
                class="hl-string">'EndTime': None, class=class="hl-string">"hl-comment"># Can be populated if logic is added
                class="hl-string">'CycleTime': None class=class="hl-string">"hl-comment"># Requires post-processing
            }
            
            class=class="hl-string">"hl-comment"># 2. Get workflow history for each document
            workflow_url = fclass="hl-string">"{api_base_url}/documents/{contract_id}/workflow"
            workflow_response = requests.get(workflow_url, headers=headers)
            
            if workflow_response.status_code == 200:
                workflow_data = workflow_response.json()
                
                class=class="hl-string">"hl-comment"># Process workflow steps
                for step in workflow_data.get(class="hl-string">'workflow', {}).get(class="hl-string">'steps', []):
                    step_name = step.get(class="hl-string">'stepName')
                    if step_name in ACTIVITY_MAP:
                        event = base_event_data.copy()
                        event[class="hl-string">'ActivityName'] = ACTIVITY_MAP[step_name]
                        event[class="hl-string">'EventTime'] = step.get(class="hl-string">'started')
                        event[class="hl-string">'EndTime'] = step.get(class="hl-string">'completed')
                        event_log.append(event)
                        
                        class=class="hl-string">"hl-comment"># Special case for final approval
                        if step_name == class="hl-string">'Internal Approval' and step.get(class="hl-string">'completed'):
                            event_approval = base_event_data.copy()
                            event_approval[class="hl-string">'ActivityName'] = class="hl-string">'Internal Approval Received'
                            event_approval[class="hl-string">'EventTime'] = step.get(class="hl-string">'completed')
                            event_log.append(event_approval)
                            
            class=class="hl-string">"hl-comment"># 3. Get document version history for class="hl-string">'Contract Redlined' events
            versions_url = fclass="hl-string">"{api_base_url}/documents/{contract_id}/versions"
            versions_response = requests.get(versions_url, headers=headers)
            if versions_response.status_code == 200:
                versions_data = versions_response.json()
                for version in versions_data.get(class="hl-string">'value', []):
                    if version.get(class="hl-string">'versionNumber', 0) > 1: class=class="hl-string">"hl-comment"># Version 1 is the initial draft
                        event = base_event_data.copy()
                        event[class="hl-string">'ActivityName'] = class="hl-string">'Contract Redlined'
                        event[class="hl-string">'EventTime'] = version.get(class="hl-string">'createdDate')
                        event_log.append(event)
        
        class=class="hl-string">"hl-comment"># Handle pagination
        search_params[class="hl-string">'offset'] += len(documents)
        if len(documents) < 100:
            break class=class="hl-string">"hl-comment"># Last page

        time.sleep(1) class=class="hl-string">"hl-comment"># Simple rate limit mitigation
        
    class=class="hl-string">"hl-comment"># 4. Save to CSV
    if not event_log:
        print(class="hl-string">"No events were extracted.")
        return

    class=class="hl-string">"hl-comment"># Sort events by contract and time for logical order
    event_log.sort(key=lambda x: (x[class="hl-string">'ContractId'], x[class="hl-string">'EventTime'] or class="hl-string">''))

    fieldnames = [
        class="hl-string">'ContractId', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'EndTime', class="hl-string">'ContractStatus',
        class="hl-string">'ContractOwner', class="hl-string">'ContractType', class="hl-string">'ContractValue', class="hl-string">'Counterparty',
        class="hl-string">'ExpirationDate', class="hl-string">'CycleTime', class="hl-string">'SourceSystem', class="hl-string">'LastDataUpdate'
    ]
    
    with open(class="hl-string">'docusign_clm_event_log.csv', class="hl-string">'w', newline=class="hl-string">'', encoding=class="hl-string">'utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(event_log)
        
    print(fclass="hl-string">"Successfully extracted {len(event_log)} events to docusign_clm_event_log.csv")

if __name__ == class="hl-string">"__main__":
    class=class="hl-string">"hl-comment"># --- Execution --- 
    PRIVATE_KEY_PATH = class="hl-string">"./private.key" class=class="hl-string">"hl-comment"># Path to your RSA private key file
    START_DATE = class="hl-string">"2023-01-01T00:00:00Z"
    END_DATE = class="hl-string">"2023-12-31T23:59:59Z"
    
    access_token = get_access_token(PRIVATE_KEY_PATH)
    if access_token:
        extract_event_log(access_token, START_DATE, END_DATE)