貴社のRecord to Report - 月次締め処理および照合データテンプレート

import requests
import pandas as pd
from datetime import datetime
import os

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># Replace with your Oracle Fusion instance details
BASE_URL = class="hl-string">"https://[your-fusion-instance].oraclecloud.com"
USERNAME = os.environ.get(class="hl-string">"FUSION_USER", class="hl-string">"your_username")
PASSWORD = os.environ.get(class="hl-string">"FUSION_PASSWORD", class="hl-string">"your_password")

class=class="hl-string">"hl-comment"># Set the date range for extraction (YYYY-MM-DD)
START_DATE = class="hl-string">"2023-01-01"
END_DATE = class="hl-string">"2023-12-31"

class=class="hl-string">"hl-comment"># --- Generic API Fetch Function ---
def fetch_data(endpoint, q_filter, fields):
    all_items = []
    offset = 0
    limit = 500
    url = fclass="hl-string">"{BASE_URL}{endpoint}"
    
    headers = {
        class="hl-string">'Content-Type': class="hl-string">'application/vnd.oracle.adf.resourceitem+json'
    }
    
    while True:
        params = {
            class="hl-string">'q': q_filter,
            class="hl-string">'fields': fields,
            class="hl-string">'limit': limit,
            class="hl-string">'offset': offset,
            class="hl-string">'onlyData': class="hl-string">'true'
        }
        try:
            response = requests.get(url, auth=(USERNAME, PASSWORD), headers=headers, params=params)
            response.raise_for_status()
            data = response.json()
            items = data.get(class="hl-string">'items', [])
            all_items.extend(items)
            
            if not data.get(class="hl-string">'hasMore', False):
                break
            offset += limit
        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"Error fetching data from {endpoint}: {e}")
            break
            
    return all_items

class=class="hl-string">"hl-comment"># --- Main Extraction Logic ---
def main():
    event_log = []
    now_utc = datetime.utcnow().isoformat() + class="hl-string">'Z'
    date_filter = fclass="hl-string">"lastUpdateDate between '{START_DATE}T00:00:00.000Z' and '{END_DATE}T23:59:59.999Z'"

    class=class="hl-string">"hl-comment"># 1. Period Opened, Closed, Permanently Closed
    print(class="hl-string">"Fetching Period Status changes...")
    period_data = fetch_data(class="hl-string">'/fscmRestApi/resources/11.13.18.05/glPeriods', date_filter, class="hl-string">"periodName,periodStatus,lastUpdateDate,lastUpdatedBy,ledgerName,ledgerId")
    for p in period_data:
        activity_map = {
            class="hl-string">'OPEN': class="hl-string">'Period Opened',
            class="hl-string">'CLOSED': class="hl-string">'Period Set to \'Closed\class="hl-string">'',
            class="hl-string">'PERMANENTLY_CLOSED': class="hl-string">'Period Permanently Closed'
        }
        activity_name = activity_map.get(p.get(class="hl-string">'periodStatus'))
        if activity_name:
            event_log.append({
                class="hl-string">'FinancialPeriod': p.get(class="hl-string">'periodName'),
                class="hl-string">'ActivityName': activity_name,
                class="hl-string">'EventTime': p.get(class="hl-string">'lastUpdateDate'),
                class="hl-string">'EndTime': p.get(class="hl-string">'lastUpdateDate'),
                class="hl-string">'ResponsibleUser': p.get(class="hl-string">'lastUpdatedBy'),
                class="hl-string">'CompanyCode': p.get(class="hl-string">'ledgerId'),
                class="hl-string">'ActivityStatus': class="hl-string">'Completed',
                class="hl-string">'JournalEntryType': None,
                class="hl-string">'ReconciliationType': None,
                class="hl-string">'Department': None,
                class="hl-string">'TargetCompletionDate': None
            })
            
    class=class="hl-string">"hl-comment"># 2. Subledger Period Closed
    print(class="hl-string">"Fetching Subledger Period Close status...")
    class=class="hl-string">"hl-comment"># This is inferred from the GL period's closingStatus property.
    class=class="hl-string">"hl-comment"># A status of class="hl-string">'C' (Closed) or class="hl-string">'W' (Close pending) in sub-fields often indicates subledger closure.
    class=class="hl-string">"hl-comment"># For a more direct approach, specific subledger period APIs would be needed.
    subledger_period_data = fetch_data(class="hl-string">'/fscmRestApi/resources/11.13.18.05/glPeriods', date_filter, class="hl-string">"periodName,closingStatus,lastUpdateDate,lastUpdatedBy,ledgerName,ledgerId")
    for p in subledger_period_data:
        if p.get(class="hl-string">'closingStatus') in [class="hl-string">'C', class="hl-string">'W']:
            event_log.append({
                class="hl-string">'FinancialPeriod': p.get(class="hl-string">'periodName'),
                class="hl-string">'ActivityName': class="hl-string">'Subledger Period Closed',
                class="hl-string">'EventTime': p.get(class="hl-string">'lastUpdateDate'),
                class="hl-string">'EndTime': p.get(class="hl-string">'lastUpdateDate'),
                class="hl-string">'ResponsibleUser': p.get(class="hl-string">'lastUpdatedBy'),
                class="hl-string">'CompanyCode': p.get(class="hl-string">'ledgerId'),
                class="hl-string">'ActivityStatus': class="hl-string">'Completed',
                class="hl-string">'JournalEntryType': None,
                class="hl-string">'ReconciliationType': None,
                class="hl-string">'Department': None,
                class="hl-string">'TargetCompletionDate': None
            })

    class=class="hl-string">"hl-comment"># 3. Journal Entries Imported & 8. Adjusting Journal Entry Posted
    print(class="hl-string">"Fetching Journal Entries...")
    journal_filter = fclass="hl-string">"actualFlag='A' and postedDate between '{START_DATE}' and '{END_DATE}'"
    journal_data = fetch_data(class="hl-string">'/fscmRestApi/resources/11.13.18.05/journalBatches', journal_filter, class="hl-string">"journalBatchName,postedDate,createdBy,jeSource,jeCategory,periodName,ledgerId")
    for j in journal_data:
        class=class="hl-string">"hl-comment"># Imported journals often come from specific sources like class="hl-string">'Payables', class="hl-string">'Receivables', etc.
        if j.get(class="hl-string">'jeSource') not in [class="hl-string">'Manual']:
            activity_name = class="hl-string">'Journal Entries Imported'
            entry_type = class="hl-string">'Imported'
        else:
            activity_name = class="hl-string">'Adjusting Journal Entry Posted'
            entry_type = class="hl-string">'Adjusting'
        event_log.append({
            class="hl-string">'FinancialPeriod': j.get(class="hl-string">'periodName'),
            class="hl-string">'ActivityName': activity_name,
            class="hl-string">'EventTime': j.get(class="hl-string">'postedDate'),
            class="hl-string">'EndTime': j.get(class="hl-string">'postedDate'),
            class="hl-string">'ResponsibleUser': j.get(class="hl-string">'createdBy'),
            class="hl-string">'CompanyCode': j.get(class="hl-string">'ledgerId'),
            class="hl-string">'ActivityStatus': class="hl-string">'Completed',
            class="hl-string">'JournalEntryType': entry_type,
            class="hl-string">'ReconciliationType': None,
            class="hl-string">'Department': None,
            class="hl-string">'TargetCompletionDate': None
        })

    class=class="hl-string">"hl-comment"># 4, 5, 6, 7. Reconciliation Activities (Placeholder - requires ARCS or equivalent module)
    print(class="hl-string">"Fetching Reconciliation activities (Placeholder)...")
    class=class="hl-string">"hl-comment"># These activities are typically in Oracle ARCS. The endpoint below is a professional placeholder.
    class=class="hl-string">"hl-comment"># You would need to use the ARCS REST API if you have that module.
    class=class="hl-string">"hl-comment"># Example endpoint: /armRestApi/v1/reconciliations

    class=class="hl-string">"hl-comment"># 9, 10, 11. Report Generation and Consolidation (via ESS Jobs)
    print(class="hl-string">"Fetching ESS Job history for reports and consolidation...")
    ess_filter = fclass="hl-string">"(jobDefinitionName like '%TrialBalance%' or jobDefinitionName like '%Consolidation%' or jobDefinitionName like '%FinancialStatement%') and lastUpdateDate between '{START_DATE}T00:00:00.000Z' and '{END_DATE}T23:59:59.999Z'"
    class=class="hl-string">"hl-comment"># Note: The ESS job history endpoint is often accessed via SOAP APIs or specific REST resources.
    class=class="hl-string">"hl-comment"># The below is a representation of what such a query might look like.
    class=class="hl-string">"hl-comment"># A common method is a BI Publisher report on ESS tables, exposed as a REST service.
    
    class=class="hl-string">"hl-comment"># 12. Financial Statements Reviewed
    class=class="hl-string">"hl-comment"># This is often a manual step and may not be logged in a standard way. It could be a specific ESS job
    class=class="hl-string">"hl-comment"># status change, a custom object, or inferred from a subsequent action.

    class=class="hl-string">"hl-comment"># --- Final Data Processing ---
    if not event_log:
        print(class="hl-string">"No events found for the specified criteria.")
        return

    df = pd.DataFrame(event_log)
    df[class="hl-string">'SourceSystem'] = class="hl-string">'Oracle Fusion Financials'
    df[class="hl-string">'LastDataUpdate'] = now_utc

    class=class="hl-string">"hl-comment"># Reorder and fill missing columns
    final_cols = [class="hl-string">'FinancialPeriod', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'EndTime', class="hl-string">'ResponsibleUser', class="hl-string">'Department', class="hl-string">'CompanyCode', class="hl-string">'ReconciliationType', class="hl-string">'ActivityStatus', class="hl-string">'JournalEntryType', class="hl-string">'TargetCompletionDate', class="hl-string">'SourceSystem', class="hl-string">'LastDataUpdate']
    for col in final_cols:
        if col not in df.columns:
            df[col] = None
    df = df[final_cols]
    
    class=class="hl-string">"hl-comment"># Sort by time
    df[class="hl-string">'EventTime'] = pd.to_datetime(df[class="hl-string">'EventTime'])
    df = df.sort_values(by=[class="hl-string">'FinancialPeriod', class="hl-string">'EventTime']).reset_index(drop=True)

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

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