決算報告 - 仕訳入力データテンプレート

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

class=class="hl-string">"hl-comment"># --- Configuration Settings ---
class=class="hl-string">"hl-comment"># Replace with your Workiva instance details and credentials
WORKIVA_DOMAIN = class="hl-string">"https://[YourWorkivaDomain].workiva.com"
CLIENT_ID = class="hl-string">"[Your-Client-ID]"
CLIENT_SECRET = class="hl-string">"[Your-Client-Secret]"

class=class="hl-string">"hl-comment"># --- Extraction Parameters ---
class=class="hl-string">"hl-comment"># Define the date range for the journal entry extraction
START_DATE = class="hl-string">"2023-01-01T00:00:00Z"
END_DATE = class="hl-string">"2023-12-31T23:59:59Z"

class=class="hl-string">"hl-comment"># --- API Endpoint Definitions ---
class=class="hl-string">"hl-comment"># Use the correct API version and paths for your instance
TOKEN_URL = fclass="hl-string">"{WORKIVA_DOMAIN}/oauth2/v2/token"
JOURNALS_API_ENDPOINT = fclass="hl-string">"{WORKIVA_DOMAIN}/api/v1/journals"

class=class="hl-string">"hl-comment"># Mapping from Workiva status changes to standardized activity names
class=class="hl-string">"hl-comment"># This may need adjustment based on your specific Workiva workflow configuration.
STATUS_TO_ACTIVITY_MAP = {
    (class="hl-string">'Draft', class="hl-string">'In Review'): class="hl-string">'Journal Entry Submitted for Review',
    (class="hl-string">'In Review', class="hl-string">'Pending Approval'): class="hl-string">'Journal Entry Sent For Approval',
    (class="hl-string">'Pending Approval', class="hl-string">'Approved'): class="hl-string">'Journal Entry Approved',
    (class="hl-string">'Approved', class="hl-string">'Posted'): class="hl-string">'Journal Entry Posted to GL',
    (class="hl-string">'Posted', class="hl-string">'Reconciled'): class="hl-string">'Journal Entry Reconciled',
    (class="hl-string">'In Review', class="hl-string">'Rejected'): class="hl-string">'Journal Entry Rejected',
    (class="hl-string">'Pending Approval', class="hl-string">'Rejected'): class="hl-string">'Journal Entry Rejected'
}

def get_access_token():
    class="hl-string">"""Authenticates with Workiva and retrieves an OAuth2 access 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
    }
    response = requests.post(TOKEN_URL, data=payload)
    response.raise_for_status() class=class="hl-string">"hl-comment"># Raise an exception for bad status codes
    return response.json()[class="hl-string">'access_token']

def fetch_paginated_data(url, headers, params):
    class="hl-string">"""Fetches data from a paginated API endpoint."""
    all_data = []
    while url:
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        data = response.json()
        all_data.extend(data.get(class="hl-string">'data', []))
        url = data.get(class="hl-string">'metadata', {}).get(class="hl-string">'nextPageUrl') class=class="hl-string">"hl-comment"># Adjust based on actual API response structure
        params = {} class=class="hl-string">"hl-comment"># Subsequent requests use the full URL from nextPageUrl
        time.sleep(0.5) class=class="hl-string">"hl-comment"># Respect rate limits
    return all_data

def map_history_to_activity(event, previous_event, journal_details):
    class="hl-string">"""Maps a single history event to a structured activity record."""
    activity_record = {
        class="hl-string">'JournalEntryId': journal_details.get(class="hl-string">'id'),
        class="hl-string">'EventTime': event.get(class="hl-string">'timestamp'),
        class="hl-string">'User': event.get(class="hl-string">'user', {}).get(class="hl-string">'name', class="hl-string">'System'),
        class="hl-string">'JournalEntryType': journal_details.get(class="hl-string">'type'),
        class="hl-string">'JournalEntryAmount': journal_details.get(class="hl-string">'amount'),
        class="hl-string">'JournalEntryStatus': event.get(class="hl-string">'details', {}).get(class="hl-string">'newValue'),
        class="hl-string">'Department': journal_details.get(class="hl-string">'department')
    }

    event_type = event.get(class="hl-string">'eventType')

    if event_type == class="hl-string">'CREATE':
        activity_record[class="hl-string">'ActivityName'] = class="hl-string">'Journal Entry Created'
        activity_record[class="hl-string">'JournalEntryStatus'] = class="hl-string">'Draft'
        return activity_record

    if event_type == class="hl-string">'ATTACHMENT_ADD':
        activity_record[class="hl-string">'ActivityName'] = class="hl-string">'Supporting Documentation Attached'
        return activity_record

    if event_type == class="hl-string">'REVERSAL_PROCESSED':
        activity_record[class="hl-string">'ActivityName'] = class="hl-string">'Journal Entry Reversal Processed'
        return activity_record

    if event_type == class="hl-string">'VALIDATION_RUN':
        status = event.get(class="hl-string">'details', {}).get(class="hl-string">'status')
        if status == class="hl-string">'SUCCESS':
            activity_record[class="hl-string">'ActivityName'] = class="hl-string">'Automated Validation Run'
        elif status == class="hl-string">'FAILURE':
            activity_record[class="hl-string">'ActivityName'] = class="hl-string">'Automated Validation Failed'
        return activity_record

    if event_type == class="hl-string">'UPDATE' and event.get(class="hl-string">'details', {}).get(class="hl-string">'field') == class="hl-string">'status':
        old_status = event.get(class="hl-string">'details', {}).get(class="hl-string">'oldValue')
        new_status = event.get(class="hl-string">'details', {}).get(class="hl-string">'newValue')
        activity_tuple = (old_status, new_status)
        if activity_tuple in STATUS_TO_ACTIVITY_MAP:
            activity_record[class="hl-string">'ActivityName'] = STATUS_TO_ACTIVITY_MAP[activity_tuple]
            return activity_record
        class=class="hl-string">"hl-comment"># Infer other activities based on status
        if new_status == class="hl-string">'In Review' and previous_event and previous_event.get(class="hl-string">'ActivityName') != class="hl-string">'Journal Entry Submitted for Review':
            activity_record[class="hl-string">'ActivityName'] = class="hl-string">'Journal Entry Review Started' class=class="hl-string">"hl-comment"># Inferred event
            return activity_record

    if event_type == class="hl-string">'UPDATE' and previous_event and previous_event.get(class="hl-string">'JournalEntryStatus') == class="hl-string">'Rejected':
        activity_record[class="hl-string">'ActivityName'] = class="hl-string">'Journal Entry Corrected' class=class="hl-string">"hl-comment"># Inferred event
        return activity_record
    
    class=class="hl-string">"hl-comment"># Fallback for unmapped status changes that might represent a review completion
    if event_type == class="hl-string">'UPDATE' and event.get(class="hl-string">'details', {}).get(class="hl-string">'field') == class="hl-string">'status':
        if event.get(class="hl-string">'details', {}).get(class="hl-string">'oldValue') == class="hl-string">'In Review':
             activity_record[class="hl-string">'ActivityName'] = class="hl-string">'Journal Entry Reviewed' class=class="hl-string">"hl-comment"># Inferred event
             return activity_record

    return None

def main():
    class="hl-string">"""Main function to orchestrate the extraction process."""
    print(class="hl-string">"Starting extraction from Workiva API...")
    try:
        access_token = get_access_token()
        headers = {class="hl-string">'Authorization': fclass="hl-string">'Bearer {access_token}'}
        
        class=class="hl-string">"hl-comment"># Fetch all journal entries within the specified date range
        print(fclass="hl-string">"Fetching journal entries from {START_DATE} to {END_DATE}...")
        journal_params = {class="hl-string">'createdAfter': START_DATE, class="hl-string">'createdBefore': END_DATE}
        journals = fetch_paginated_data(JOURNALS_API_ENDPOINT, headers, journal_params)
        print(fclass="hl-string">"Found {len(journals)} journal entries.")

        event_log = []
        last_data_update = datetime.utcnow().isoformat() + class="hl-string">'Z'

        for i, journal in enumerate(journals):
            journal_id = journal.get(class="hl-string">'id')
            print(fclass="hl-string">"Processing journal {i+1}/{len(journals)}: {journal_id}")

            class=class="hl-string">"hl-comment"># Fetch the detailed history for each journal entry
            history_url = fclass="hl-string">"{JOURNALS_API_ENDPOINT}/{journal_id}/history"
            try:
                history_events = fetch_paginated_data(history_url, headers, {})
            except requests.exceptions.HTTPError as e:
                print(fclass="hl-string">"Could not fetch history for {journal_id}: {e}")
                continue

            processed_activities = []
            for idx, event in enumerate(history_events):
                previous_event = processed_activities[-1] if processed_activities else None
                activity_record = map_history_to_activity(event, previous_event, journal)
                if activity_record:
                    activity_record[class="hl-string">'SourceSystem'] = class="hl-string">'Workiva'
                    activity_record[class="hl-string">'LastDataUpdate'] = last_data_update
                    processed_activities.append(activity_record)
            
            event_log.extend(processed_activities)

        if not event_log:
            print(class="hl-string">"No events were generated. The event log is empty.")
            return

        class=class="hl-string">"hl-comment"># Create DataFrame and save to CSV
        df = pd.DataFrame(event_log)
        df[class="hl-string">'EventTime'] = pd.to_datetime(df[class="hl-string">'EventTime'])
        df = df.sort_values(by=[class="hl-string">'JournalEntryId', class="hl-string">'EventTime'])
        
        output_filename = class="hl-string">'workiva_journal_entry_event_log.csv'
        df.to_csv(output_filename, index=False, encoding=class="hl-string">'utf-8')
        print(fclass="hl-string">"Extraction complete. Event log saved to {output_filename}")

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

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