您的合同管理数据模板

-- This is a logical representation of the report configuration in the SAP Ariba Analytical Reporting UI.
-- The tool does not use a direct query language; you build this logic by selecting facts, fields, and filters.
-- The final step requires transforming the wide export file into this long format.

-- FACT: Contract Workspace (Projects) as c
-- JOIN: Approval History as a ON c."Project Id" = a."Project Id"
-- JOIN: Task as t ON c."Project Id" = t."Project Id"
-- JOIN: Document as d ON c."Project Id" = d."Project Id"

-- Each of the following SELECT blocks represents a set of rows for a specific activity.

SELECT c."Contract Id" AS ContractId, 'Contract Workspace Created' AS ActivityName, c."Time Created" AS EventTime, NULL AS EventEndTime, c."Project Owner" AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Contract Drafted' AS ActivityName, d."Time Created" AS EventTime, NULL AS EventEndTime, d."Created By" AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c JOIN "Document" AS d ON c."Project Id" = d."Project Id" WHERE d."Is Main Agreement" = TRUE AND d.Version = 1
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Internal Review Started' AS ActivityName, t."Activated Date" AS EventTime, t."Completed Date" AS EventEndTime, t.Assignee AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c JOIN "Task" AS t ON c."Project Id" = t."Project Id" WHERE t."Task Name" LIKE '%Internal Review%'
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Legal Review Started' AS ActivityName, t."Activated Date" AS EventTime, t."Completed Date" AS EventEndTime, t.Assignee AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c JOIN "Task" AS t ON c."Project Id" = t."Project Id" WHERE t."Task Name" LIKE '%Legal Review%'
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Contract Revised' AS ActivityName, d."Time Created" AS EventTime, NULL AS EventEndTime, d."Created By" AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c JOIN "Document" AS d ON c."Project Id" = d."Project Id" WHERE d."Is Main Agreement" = TRUE AND d.Version > 1
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Negotiation With Counterparty' AS ActivityName, t."Activated Date" AS EventTime, t."Completed Date" AS EventEndTime, t.Assignee AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c JOIN "Task" AS t ON c."Project Id" = t."Project Id" WHERE t."Task Name" LIKE '%Negotiation%'
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Internal Approval Completed' AS ActivityName, a."Time Completed" AS EventTime, NULL AS EventEndTime, a."Approver" AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c JOIN "Approval History" AS a ON c."Project Id" = a."Project Id" WHERE a."Approval Task Name" LIKE '%Internal Approval Flow%' AND a.Status = 'Approved' AND a."Is Last Approver" = TRUE
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Counterparty Approval Received' AS ActivityName, t."Completed Date" AS EventTime, NULL AS EventEndTime, t.Assignee AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c JOIN "Task" AS t ON c."Project Id" = t."Project Id" WHERE t."Task Name" LIKE '%Counterparty Signature%'
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Contract Executed' AS ActivityName, c."Executed Date" AS EventTime, NULL AS EventEndTime, c."Project Owner" AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c WHERE c."Executed Date" IS NOT NULL
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Contract Activated' AS ActivityName, c."Activated Date" AS EventTime, NULL AS EventEndTime, c."Project Owner" AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c WHERE c."Activated Date" IS NOT NULL
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Compliance Task Completed' AS ActivityName, t."Completed Date" AS EventTime, NULL AS EventEndTime, t.Assignee AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c JOIN "Task" AS t ON c."Project Id" = t."Project Id" WHERE t."Task Name" LIKE '%Compliance%'
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Amendment Initiated' AS ActivityName, c."Time Created" AS EventTime, NULL AS EventEndTime, c."Project Owner" AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c WHERE c."Is Amendment" = TRUE
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Contract Renewed' AS ActivityName, c."Time Created" AS EventTime, NULL AS EventEndTime, c."Project Owner" AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c WHERE c."Is Renewal" = TRUE
UNION ALL
SELECT c."Contract Id" AS ContractId, 'Contract Closed' AS ActivityName, c."Completed Date" AS EventTime, NULL AS EventEndTime, c."Project Owner" AS UserName, c.Department, c.Amount AS ContractAmount, c."Contract Type Name" AS ContractType, c."Legal Contact" AS LegalCounsel FROM "Contract Workspace (Projects)" AS c WHERE c.Status = 'Completed'

import requests
import pandas as pd
import base64
from datetime import datetime, timedelta

class=class="hl-string">"hl-comment"># --- CONFIGURATION --- #
class=class="hl-string">"hl-comment"># Replace with your SAP Ariba realm and API credentials
REALM = class="hl-string">'[Your Ariba Realm]'
API_KEY = class="hl-string">'[Your Application Key]'
class=class="hl-string">"hl-comment"># The OAUTH_CLIENT_ID is your Application Key and a hyphen, base64 encoded.
class=class="hl-string">"hl-comment"># Example: class="hl-string">'yourappkey:' -> class="hl-string">'eW91cmFwcGtleTo='
OAUTH_CLIENT_ID = class="hl-string">'[Your Base64 Encoded OAuth Client ID]'
OAUTH_CLIENT_SECRET = class="hl-string">'[Your OAuth Client Secret]'

class=class="hl-string">"hl-comment"># --- FILTERS --- #
class=class="hl-string">"hl-comment"># Set the date range for contracts to extract (based on last modified date)
UPDATED_DATE_TO = datetime.now().strftime(class="hl-string">'%Y-%m-%dT%H:%M:%SZ')
UPDATED_DATE_FROM = (datetime.now() - timedelta(days=180)).strftime(class="hl-string">'%Y-%m-%dT%H:%M:%SZ')

class=class="hl-string">"hl-comment"># API Endpoints
AUTH_URL = fclass="hl-string">'https://api.ariba.com/v2/oauth/token'
BASE_API_URL = fclass="hl-string">'https://{REALM}.api.ariba.com/api/contracts/v1/workspaces'

class=class="hl-string">"hl-comment"># --- AUTHENTICATION --- #
def get_access_token():
    class="hl-string">"""Authenticates with Ariba API to get an access token."""
    auth_header = {class="hl-string">'Authorization': fclass="hl-string">'Basic {OAUTH_CLIENT_ID}'}
    payload = {
        class="hl-string">'grant_type': class="hl-string">'client_credentials',
        class="hl-string">'client_secret': OAUTH_CLIENT_SECRET
    }
    try:
        response = requests.post(AUTH_URL, headers=auth_header, data=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

class=class="hl-string">"hl-comment"># --- DATA FETCHING --- #
def fetch_paginated_data(url, headers, params):
    class="hl-string">"""Fetches data from a paginated Ariba API endpoint."""
    all_records = []
    page = 0
    while True:
        params[class="hl-string">'page'] = page
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        data = response.json()
        all_records.extend(data)
        if len(data) == 0: class=class="hl-string">"hl-comment"># Assuming empty list means last page
            break
        page += 1
    return all_records

class=class="hl-string">"hl-comment"># --- EVENT DERIVATION LOGIC --- #
def derive_events_from_workspace(workspace_details):
    class="hl-string">"""Derives multiple event log rows from a single workspace's detailed data."""
    events = []
    case_id = workspace_details.get(class="hl-string">'contractId')
    if not case_id:
        return []

    class=class="hl-string">"hl-comment"># Helper function to create a standard event dictionary
    def create_event(activity, timestamp, user=class="hl-string">'System'):
        return {
            class="hl-string">'ContractId': case_id,
            class="hl-string">'ActivityName': activity,
            class="hl-string">'EventTime': timestamp,
            class="hl-string">'EventEndTime': None, class=class="hl-string">"hl-comment"># Populate if available
            class="hl-string">'UserName': user,
            class="hl-string">'Department': workspace_details.get(class="hl-string">'custom', {}).get(class="hl-string">'Department', None),
            class="hl-string">'ContractAmount': workspace_details.get(class="hl-string">'amount', {}).get(class="hl-string">'amount', None),
            class="hl-string">'ContractType': workspace_details.get(class="hl-string">'contractType', {}).get(class="hl-string">'name', None),
            class="hl-string">'LegalCounsel': next((member.get(class="hl-string">'name') for member in workspace_details.get(class="hl-string">'team', []) if member.get(class="hl-string">'role') == class="hl-string">'Legal Counsel'), None)
        }

    class=class="hl-string">"hl-comment"># 1. Contract Workspace Created
    if workspace_details.get(class="hl-string">'createDate'):
        events.append(create_event(class="hl-string">'Contract Workspace Created', workspace_details[class="hl-string">'createDate']))

    class=class="hl-string">"hl-comment"># 2. Contract Drafted & 5. Contract Revised
    if workspace_details.get(class="hl-string">'documents'):
        main_doc_versions = sorted([v for doc in workspace_details[class="hl-string">'documents'] if doc.get(class="hl-string">'isMainAgreement') for v in doc.get(class="hl-string">'versions', [])], key=lambda x: x[class="hl-string">'createDate'])
        if main_doc_versions:
            events.append(create_event(class="hl-string">'Contract Drafted', main_doc_versions[0][class="hl-string">'createDate'], main_doc_versions[0].get(class="hl-string">'createdBy')))
            for version in main_doc_versions[1:]:
                events.append(create_event(class="hl-string">'Contract Revised', version[class="hl-string">'createDate'], version.get(class="hl-string">'createdBy')))

    class=class="hl-string">"hl-comment"># 3, 4, 7, 11. Task-based activities
    if workspace_details.get(class="hl-string">'tasks'):
        for task in workspace_details[class="hl-string">'tasks']:
            task_name = task.get(class="hl-string">'name', class="hl-string">'').lower()
            if class="hl-string">'internal review' in task_name and task.get(class="hl-string">'startDate'):
                events.append(create_event(class="hl-string">'Internal Review Started', task[class="hl-string">'startDate'], task.get(class="hl-string">'owner')))
            if class="hl-string">'legal review' in task_name and task.get(class="hl-string">'startDate'):
                events.append(create_event(class="hl-string">'Legal Review Started', task[class="hl-string">'startDate'], task.get(class="hl-string">'owner')))
            if class="hl-string">'internal approval' in task_name and task.get(class="hl-string">'completionDate'):
                events.append(create_event(class="hl-string">'Internal Approval Completed', task[class="hl-string">'completionDate'], task.get(class="hl-string">'owner')))
            if class="hl-string">'compliance' in task_name and task.get(class="hl-string">'completionDate'):
                events.append(create_event(class="hl-string">'Compliance Task Completed', task[class="hl-string">'completionDate'], task.get(class="hl-string">'owner')))

    class=class="hl-string">"hl-comment"># 6, 8. Status-based manual activities
    class=class="hl-string">"hl-comment"># This requires interpreting status history, which can be complex. Simplified here.
    if workspace_details.get(class="hl-string">'statusHistory'):
        for status_change in workspace_details.get(class="hl-string">'statusHistory'):
            status = status_change.get(class="hl-string">'status', class="hl-string">'').lower()
            if class="hl-string">'negotiation' in status:
                events.append(create_event(class="hl-string">'Negotiation With Counterparty', status_change[class="hl-string">'date'], status_change.get(class="hl-string">'user')))
            if class="hl-string">'counterparty approval' in status:
                events.append(create_event(class="hl-string">'Counterparty Approval Received', status_change[class="hl-string">'date'], status_change.get(class="hl-string">'user')))

    class=class="hl-string">"hl-comment"># 9. Contract Executed
    if workspace_details.get(class="hl-string">'executionDate'):
        events.append(create_event(class="hl-string">'Contract Executed', workspace_details[class="hl-string">'executionDate']))

    class=class="hl-string">"hl-comment"># 10. Contract Activated
    if workspace_details.get(class="hl-string">'effectiveDate'):
        events.append(create_event(class="hl-string">'Contract Activated', workspace_details[class="hl-string">'effectiveDate']))

    class=class="hl-string">"hl-comment"># 12. Amendment Initiated
    if workspace_details.get(class="hl-string">'isAmendment'):
        events.append(create_event(class="hl-string">'Amendment Initiated', workspace_details[class="hl-string">'createDate']))

    class=class="hl-string">"hl-comment"># 13. Contract Renewed
    if workspace_details.get(class="hl-string">'renewalOfWorkspaceId'):
         events.append(create_event(class="hl-string">'Contract Renewed', workspace_details[class="hl-string">'createDate']))

    class=class="hl-string">"hl-comment"># 14. Contract Closed
    if workspace_details.get(class="hl-string">'closeDate'):
        events.append(create_event(class="hl-string">'Contract Closed', workspace_details[class="hl-string">'closeDate']))

    return events

class=class="hl-string">"hl-comment"># --- MAIN EXECUTION --- #
def main():
    class="hl-string">"""Main function to orchestrate the extraction process."""
    print(class="hl-string">"Attempting to get access token...")
    access_token = get_access_token()
    if not access_token:
        print(class="hl-string">"Failed to get access token. Exiting.")
        return

    print(class="hl-string">"Successfully obtained access token.")
    headers = {
        class="hl-string">'Authorization': fclass="hl-string">'Bearer {access_token}',
        class="hl-string">'apiKey': API_KEY
    }

    print(fclass="hl-string">"Fetching contract workspaces updated from {UPDATED_DATE_FROM} to {UPDATED_DATE_TO}...")
    workspace_params = {
        class="hl-string">'filters': fclass="hl-string">'updatedDateFrom={UPDATED_DATE_FROM}&updatedDateTo={UPDATED_DATE_TO}'
    }
    try:
        workspaces = fetch_paginated_data(BASE_API_URL, headers, workspace_params)
        print(fclass="hl-string">"Found {len(workspaces)} workspaces.")

        all_events = []
        for i, ws_summary in enumerate(workspaces):
            ws_id = ws_summary.get(class="hl-string">'contractId')
            print(fclass="hl-string">"Processing workspace {i+1}/{len(workspaces)}: {ws_id}")
            detail_url = fclass="hl-string">"{BASE_API_URL}/{ws_id}"
            class=class="hl-string">"hl-comment"># Add parameters to get full details
            detail_params = {class="hl-string">'include': class="hl-string">'tasks,documents,team,statusHistory'}
            try:
                ws_details_response = requests.get(detail_url, headers=headers, params=detail_params)
                ws_details_response.raise_for_status()
                ws_details = ws_details_response.json()
                
                events = derive_events_from_workspace(ws_details)
                all_events.extend(events)
            except requests.exceptions.RequestException as e:
                print(fclass="hl-string">"Could not fetch details for workspace {ws_id}: {e}")

        if not all_events:
            print(class="hl-string">"No events were generated. Check API responses and derivation logic.")
            return

        class=class="hl-string">"hl-comment"># Convert to DataFrame and save to CSV
        event_log_df = pd.DataFrame(all_events)
        event_log_df[class="hl-string">'EventTime'] = pd.to_datetime(event_log_df[class="hl-string">'EventTime'])
        event_log_df = event_log_df.sort_values(by=[class="hl-string">'ContractId', class="hl-string">'EventTime'])
        
        output_filename = class="hl-string">'ariba_contract_management_event_log.csv'
        event_log_df.to_csv(output_filename, index=False, date_format=class="hl-string">'%Y-%m-%d %H:%M:%S')
        print(fclass="hl-string">"Successfully created event log: {output_filename}")

    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"An error occurred during API request: {e}")

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