계약 관리 데이터 템플릿

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 Agiloft instance details
BASE_URL = class="hl-string">"https://[your_domain].agiloft.com/gui2/rest"
USERNAME = class="hl-string">"[your_api_user]"
PASSWORD = class="hl-string">"[your_api_password]"
API_KEY = class="hl-string">"[your_api_key]"
KB_NAME = class="hl-string">"contract"  class=class="hl-string">"hl-comment"># The API name of your contracts knowledgebase
PAGE_SIZE = 200

class=class="hl-string">"hl-comment"># Filter for contracts created in a specific range (adjust as needed)
class=class="hl-string">"hl-comment"># Format for Agiloft API dates is class="hl-string">'YYYY-MM-DD HH:MM:SS'
SEARCH_FILTER = {
    class="hl-string">"search_stype": class="hl-string">"field_based",
    class="hl-string">"searches": [
        {class="hl-string">"field": class="hl-string">"date_created", class="hl-string">"op": class="hl-string">"ge", class="hl-string">"val": class="hl-string">"2023-01-01 00:00:00"}
    ]
}

class=class="hl-string">"hl-comment"># --- SCRIPT --- #

def get_agiloft_session(base_url, username, password, api_key):
    class="hl-string">"""Authenticates with Agiloft and returns a session token."""
    login_url = fclass="hl-string">"{base_url}/login"
    headers = {class="hl-string">"Content-Type": class="hl-string">"application/json", class="hl-string">"API-Key": api_key}
    payload = {class="hl-string">"username": username, class="hl-string">"password": password}
    try:
        response = requests.post(login_url, headers=headers, json=payload, timeout=60)
        response.raise_for_status()
        return response.cookies
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"Error: Authentication failed. {e}")
        return None

def fetch_all_records(session, base_url, kb_name, page_size, search_filter):
    class="hl-string">"""Fetches all records from a knowledgebase with pagination."""
    records = []
    page_num = 0
    while True:
        url = fclass="hl-string">"{base_url}/kb/{kb_name}"
        params = {
            class="hl-string">"page_size": page_size,
            class="hl-string">"page_num": page_num,
            class="hl-string">"view_name": class="hl-string">"Default",
            class="hl-string">"search": json.dumps(search_filter),
            class="hl-string">"fields": class="hl-string">"id,history,contract_status,contract_type,contract_value,created_by,owner_group,counterparty,expiration_date"
        }
        try:
            response = requests.get(url, cookies=session, params=params, timeout=120)
            response.raise_for_status()
            data = response.json()
            records.extend(data.get(class="hl-string">'records', []))
            if len(data.get(class="hl-string">'records', [])) < page_size:
                break
            page_num += 1
        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"Error fetching page {page_num}: {e}")
            break
    return records

def map_history_to_activity(entry_text):
    class="hl-string">"""Maps a history log entry to a defined activity name."""
    entry_lower = entry_text.lower()
    if class="hl-string">"record created by" in entry_lower:
        return class="hl-string">"Contract Request Initiated"
    if class="hl-string">"status changed from" in entry_lower:
        if class="hl-string">"draft" in entry_lower and class="hl-string">"internal review" in entry_lower:
            return class="hl-string">"Internal Review Submitted"
        if class="hl-string">"internal review" in entry_lower and class="hl-string">"legal approved" in entry_lower:
            return class="hl-string">"Legal Review Conducted"
        if class="hl-string">"drafting" in entry_lower and class="hl-string">"in negotiation" in entry_lower:
            return class="hl-string">"Counterparty Negotiation Started"
        if class="hl-string">"negotiation" in entry_lower and class="hl-string">"fully approved" in entry_lower:
            return class="hl-string">"Internal Approvals Obtained"
        if class="hl-string">"approved" in entry_lower and class="hl-string">"sent for signature" in entry_lower:
            return class="hl-string">"Sent For Signature"
        if class="hl-string">"signature" in entry_lower and class="hl-string">"executed" in entry_lower:
            return class="hl-string">"Contract Executed/Signed"
        if class="hl-string">"executed" in entry_lower and class="hl-string">"active" in entry_lower:
            return class="hl-string">"Contract Activated"
        if class="hl-string">"active" in entry_lower and class="hl-string">"renewed" in entry_lower:
            return class="hl-string">"Contract Renewed"
        if class="hl-string">"active" in entry_lower and class="hl-string">"terminated" in entry_lower:
            return class="hl-string">"Contract Terminated"
        if class="hl-string">"canceled" in entry_lower or class="hl-string">"withdrawn" in entry_lower:
            return class="hl-string">"Contract Canceled"
        if class="hl-string">"active" in entry_lower and class="hl-string">"amending" in entry_lower:
            return class="hl-string">"Amendment Initiated"
        if class="hl-string">"active" in entry_lower and class="hl-string">"compliance review" in entry_lower:
            return class="hl-string">"Compliance Review Performed"
    if class="hl-string">"a new file was attached" in entry_lower:
        return class="hl-string">"Contract Redlined/Revised"
    if class="hl-string">"status changed to 'drafting complete'" in entry_lower:
        return class="hl-string">"Contract Drafted"
    return None


def main():
    print(class="hl-string">"Starting Agiloft data extraction...")
    session = get_agiloft_session(BASE_URL, USERNAME, PASSWORD, API_KEY)
    if not session:
        return

    print(fclass="hl-string">"Fetching records from '{KB_NAME}' knowledgebase...")
    contracts = fetch_all_records(session, BASE_URL, KB_NAME, PAGE_SIZE, SEARCH_FILTER)
    print(fclass="hl-string">"Found {len(contracts)} contract records.")

    if not contracts:
        print(class="hl-string">"No contracts found for the specified criteria.")
        return

    event_log_rows = []
    last_data_update = datetime.now().strftime(class="hl-string">'%Y-%m-%d %H:%M:%S')

    for contract in contracts:
        contract_id = contract.get(class="hl-string">'id')
        history_raw = contract.get(class="hl-string">'history', class="hl-string">'')

        class=class="hl-string">"hl-comment"># Process history entries
        if history_raw:
            history_entries = history_raw.split(class="hl-string">'\n')
            for entry in history_entries:
                if not entry.strip():
                    continue
                
                activity_name = map_history_to_activity(entry)
                if activity_name:
                    try:
                        class=class="hl-string">"hl-comment"># Agiloft history format: class="hl-string">'YYYY-MM-DD HH:MM:SS [User Name]: Event Description'
                        timestamp_str = entry.split(class="hl-string">' [')[0]
                        event_time = datetime.strptime(timestamp_str, class="hl-string">'%Y-%m-%d %H:%M:%S').isoformat() + class="hl-string">'Z'
                        user_name = entry.split(class="hl-string">'[')[1].split(class="hl-string">']')[0]
                    except (IndexError, ValueError):
                        event_time = None
                        user_name = class="hl-string">'System'

                    event_log_rows.append({
                        class="hl-string">"ContractId": contract_id,
                        class="hl-string">"ActivityName": activity_name,
                        class="hl-string">"EventTime": event_time,
                        class="hl-string">"UserName": user_name,
                        class="hl-string">"UserDepartment": contract.get(class="hl-string">'owner_group'),
                        class="hl-string">"ContractType": contract.get(class="hl-string">'contract_type'),
                        class="hl-string">"ContractValue": contract.get(class="hl-string">'contract_value'),
                        class="hl-string">"ContractStatus": contract.get(class="hl-string">'contract_status'),
                        class="hl-string">"CounterpartyName": contract.get(class="hl-string">'counterparty'),
                        class="hl-string">"ExpirationDate": contract.get(class="hl-string">'expiration_date'),
                        class="hl-string">"EventEndTime": None,  class=class="hl-string">"hl-comment"># Populate if available
                        class="hl-string">"SourceSystem": class="hl-string">"Agiloft",
                        class="hl-string">"LastDataUpdate": last_data_update
                    })

        class=class="hl-string">"hl-comment"># Handle calculated class="hl-string">'Contract Expired' event
        exp_date_str = contract.get(class="hl-string">'expiration_date')
        current_status = contract.get(class="hl-string">'contract_status', class="hl-string">'').lower()
        if exp_date_str and current_status not in [class="hl-string">'terminated', class="hl-string">'canceled', class="hl-string">'expired']:
            try:
                exp_date = datetime.strptime(exp_date_str, class="hl-string">'%Y-%m-%d %H:%M:%S')
                if exp_date < datetime.now():
                    event_log_rows.append({
                        class="hl-string">"ContractId": contract_id,
                        class="hl-string">"ActivityName": class="hl-string">"Contract Expired",
                        class="hl-string">"EventTime": exp_date.isoformat() + class="hl-string">'Z',
                        class="hl-string">"UserName": class="hl-string">"System",
                        class="hl-string">"UserDepartment": contract.get(class="hl-string">'owner_group'),
                        class="hl-string">"ContractType": contract.get(class="hl-string">'contract_type'),
                        class="hl-string">"ContractValue": contract.get(class="hl-string">'contract_value'),
                        class="hl-string">"ContractStatus": class="hl-string">"Expired",
                        class="hl-string">"CounterpartyName": contract.get(class="hl-string">'counterparty'),
                        class="hl-string">"ExpirationDate": contract.get(class="hl-string">'expiration_date'),
                        class="hl-string">"EventEndTime": None,
                        class="hl-string">"SourceSystem": class="hl-string">"Agiloft",
                        class="hl-string">"LastDataUpdate": last_data_update
                    })
            except ValueError:
                pass class=class="hl-string">"hl-comment"># Ignore invalid date formats

    if not event_log_rows:
        print(class="hl-string">"No process events could be generated from the contract histories.")
        return

    df = pd.DataFrame(event_log_rows)
    df = df.sort_values(by=[class="hl-string">'ContractId', class="hl-string">'EventTime']).reset_index(drop=True)
    
    class=class="hl-string">"hl-comment"># Reorder columns for clarity
    column_order = [
        class="hl-string">'ContractId', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'EventEndTime', class="hl-string">'UserName',
        class="hl-string">'UserDepartment', class="hl-string">'ContractType', class="hl-string">'ContractValue', class="hl-string">'ContractStatus',
        class="hl-string">'CounterpartyName', class="hl-string">'ExpirationDate', class="hl-string">'SourceSystem', class="hl-string">'LastDataUpdate'
    ]
    df = df[column_order]

    output_filename = class="hl-string">'agiloft_contract_event_log.csv'
    df.to_csv(output_filename, index=False)
    print(fclass="hl-string">"Extraction complete. Event log saved to {output_filename}")

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