OIC 演習1：スケジュール統合で FBDI 取込を自動化する

統合は 2 本 です。スケジュール統合（ファイル取得 → 入力チェック → 変換 → zip → Import Bulk Data）と、コールバック統合（取込の完了を受けて結果を通知）に分けます
FBDI の CSV は 見出し行なし・各行の最後が END・列順固定 です。Write File → Zip File で作り、zip の中のファイル名はテンプレートのままにします
取込前に 入力チェック で止め、取込後は コールバック でログとリジェクトまで確認します。コールバックの「成功」だけでは、取込の成功とは言えません
確認する場所は 4 つです。OIC の Instances（アクティビティ・ストリーム）、Fusion の スケジュール済プロセス、請求書、通知メール

この演習で身につくこと

要件から 統合設計書（統合一覧・スケジュール・ファイル処理・マッピング・エラー処理）を表で作れる
スケジュール統合、Stage File、ERP Cloud アダプタ（Import Bulk Data とコールバック）を組み合わせて実装できる
入力チェック、フォルト処理、通知、リトライを設計に入れられる
Run（Ad hoc request）でテストし、Instances と Fusion のジョブで結果を確認できる
環境がなくても、設計書の机上演習とモック API で、取込から結果確認までの流れを再現できる

背景の知識はスケジュール統合とファイル処理、接続とアダプタ、障害処理・テスト・監視・移送を参照してください。

前提・準備（Fusion 環境あり／なし）

シナリオ（架空）

項目	内容
会社	サンプル商事株式会社（架空）
取込元	経費精算システム（架空）。毎晩 21:30 までに SFTP へ 2 つの CSV を置く
ファイル	`exp_inv_header.csv`（請求書ヘッダー）、`exp_inv_lines.csv`（明細）
取込先	Fusion の AP（買掛管理）。JP01 営業単位（架空。自社のテスト用 BU に読み替える）
起動	毎晩 22:00（日本時間）

要件

ファイルが届いていない日は、運用担当に通知する
入力チェックで 1 件でも不合格なら、ファイル全体を取り込まず 通知する
取込の結果（リジェクト件数を含む）を運用担当に通知する
同じファイルを 二重に取り込まない
失敗したとき、運用担当が原因を調べて 再実行 できる

取込元の CSV（サンプル。架空のデータ）です。ファイルとして保存して使います。

ExtInvoiceId,InvoiceNumber,SupplierCode,SiteCode,InvoiceDate,Currency,Amount,Memo
100001,SMP-2026-1001,10001,TOKYO,2026-10-15,JPY,110000,サーバー保守・監視 10月分
100002,SMP-2026-1002,10002,SAPPORO,2026-10-15,JPY,55000,消耗品 10月分

ExtInvoiceId,LineNo,LineAmount,AccountCode,LineMemo
100001,1,80000,5120,サーバー保守 10月分
100001,2,30000,5120,監視サービス 10月分
100002,1,50000,5120,消耗品 10月分
100002,2,5000,5130,配送料

Fusion／OIC 環境ありルート

OIC：インスタンスと、統合を作れるロール（ServiceDeveloper など）。テスト用のプロジェクトを使います
Fusion のテスト環境と統合ユーザー。ロールは Integration Specialist、取込先 BU のデータ・アクセス、AttachmentsUser、SOAOperator（コールバックはイベントで届くため）が目安です。正確な要件は公式の「Assign Required Roles to an Integration User」で確認します
SFTP：OIC の File Server か、テスト用の SFTP サーバー
テスト用の仕入先とサイト：サンプル CSV の仕入先名・サイトは、自分の環境に存在する値に置き換えます
本番環境では行いません

環境なしルート

OIC や Fusion がなくても、次の 2 つで同じ考え方を学べます。

机上演習：要件から統合設計書（表）を作り、解答例と照合します
モック API でのフロー再現：Python（標準ライブラリのみ）で、CSV の検証 → zip 作成 → importBulkData → 状況照会までを再現します（練習用 API プレイグラウンドと同じモック API を使います）

手順

ステップ1 接続を作る

プロジェクト（例：SMP_AP_FBDI）を作成します
ERP Cloud 接続 SMP_ERP_CLOUD：アダプタは Oracle ERP Cloud、ロールは トリガーと呼出し（取込統合では呼出し、コールバック統合ではトリガーに使うため）。ホスト名を入力し、セキュリティ・ポリシー（例：Username Password Token）と統合ユーザーを設定して Test します
SFTP 接続 SMP_SFTP：アダプタは FTP、ロールは呼出し。ホスト・ポート・SFTP の設定と認証を入力して Test します

ステップ2 ルックアップを作る

システムごとのコードを Fusion の値に変換する表を作ります（マッパー・ルックアップ・JavaScript ライブラリ）。

ルックアップ	SYSTEM_A	FUSION
`SMP_SUPPLIER`	10001	サンプル商事株式会社
`SMP_SUPPLIER`	10002	仮想オフィス用品株式会社
`SMP_LINE_TYPE`	5120	Item
`SMP_LINE_TYPE`	5130	Freight

仕入先名は、自分の環境の仕入先に置き換えます。

ステップ3 スケジュール統合を作る

統合を作成し、パターンに Schedule を選びます。名前と識別子は AP_INV_FBDI_LOAD にします
キャンバスの Schedule の Actions > Edit > Configure schedule parameters で、BusinessUnit（既定値：JP01 営業単位）と SourceName（既定値：External）を追加します
ビジネス識別子 に BusinessUnit を設定します。スケジュール統合にはトリガーのペイロードがないため、スケジュール・パラメータを使います

ステップ4 ファイルを取得して読む

SFTP 接続の呼出しを追加し、Download File でヘッダーと明細の 2 ファイルを取得します。取得したファイルは ファイル参照 として得られます
Stage File の Read Entire File を追加します（ヘッダー用）。Configure File Reference は Yes、スキーマは Sample delimited document で exp_inv_header.csv をサンプルにし、Use First Row as Column Headers を有効にします
明細用にも同じ設定で Read Entire File を追加します（exp_inv_lines.csv をサンプルにします）

ファイルは 10MB 未満の想定なので Read Entire File を使います。件数が増える見込みなら、発展課題 2 で Read File in Segments に置き換えます。

ステップ5 入力チェックを作る

Assign で、ヘッダーの Amount の合計と、明細の LineAmount の合計を、式ビルダーの XPath の sum() で変数に入れます。件数（count()）も取ります
Switch で「件数が 0」「合計が一致しない」を判定し、該当すれば Throw New Fault（コード INPUT_CHECK_NG、理由、詳細）で業務エラーにします。これで取込に進みません

ステップ6 変換して CSV と zip を作る

マップ → Write File で ApInvoicesInterface.csv を作ります。スキーマは FBDI テンプレートから生成した CSV をサンプルにして、列順をテンプレートのままにします。対応付けは、次の机上演習のマッピング設計に従います
同じ要領で ApInvoiceLinesInterface.csv を作ります。出力先ディレクトリは両方とも専用の fbdi_out にし、追記は使いません（1 回の Write でファイル全体を作ります）
Zip File で fbdi_out を ApInvoices_<日付>_01.zip にまとめます。出力先は別のディレクトリ（fbdi_zip）にします

ステップ7 取込を依頼する

ERP Cloud 接続の呼出しを追加し、Import Bulk Data into Oracle ERP Cloud を選びます
Select Bulk Data Import Process で、AP 請求書のインポート（Import Payables Invoices）を選びます
Response ページで Enable Callback を選びます
マップで、zip のファイル参照と、ジョブのパラメータ（BU、Source、Import Set など）を対応付けます。Import Set は実行ごとに一意 にします（日付時刻や OIC のフロー ID から作ります）。並びと項目はマッパーの画面に従います
Logger で、zip 名・Import Set・BU を記録します
依頼が受理されたら、SFTP 接続で入力ファイルを 依頼済みフォルダーへ移します（翌日以降に同じファイルを読み直さないため）。取込の成否は、次のステップのコールバックで通知します

ステップ8 コールバック統合を作る

新しい統合（パターンは Application）を AP_INV_FBDI_CALLBACK として作ります
トリガーに SMP_ERP_CLOUD を使い、Receive Callback Message upon completion of FBDI bulk Import Job submitted via another Integration を選びます。対象のジョブは Import Payables Invoices、Download Options は Always にします
Notification で、運用担当に結果（ジョブの状態）を通知します。ログ・出力ファイルは Stage File（Unzip File → Read）で読めます（発展課題 3）
有効化します。有効化したときに購読が自動で登録されます。取込統合より先に有効化します

ステップ9 フォルト処理を作る

AP_INV_FBDI_LOAD の グローバル・フォルトハンドラ に、Switch（フォルトの種類）→ Notification（統合名・フロー ID・BusinessUnit・エラー内容）→ Re-throw Fault を作ります。Stop は使いません
ファイルの取得（Download File）を Scope で囲み、Default Handler で「ファイル未着」を通知して Re-throw します
Fusion の呼出しを Scope で囲み、一時的なエラーは 3 回までリトライします（障害処理・テスト・監視・移送のリトライのフロー）

ステップ10 テストする

2 本の統合を 有効化 します（コールバック統合が先）
AP_INV_FBDI_LOAD の Actions > Run で Configure and run を開き、Ad hoc request で Run します
Observability > Instances で、LOAD とコールバックのインスタンスを開き、アクティビティ・ストリームで流れを確認します
Fusion の ナビゲータ > ツール > スケジュール済プロセス で、Load Interface File for Import と Import Payables Invoices が Succeeded か確認します
請求書の画面で、請求書番号（SMP-2026-1001、SMP-2026-1002）を検索し、2 件できていることを確認します

異常系もテストします。

ケース	操作	期待する結果
入力チェック不合格	明細の金額を 1 つ変えて、合計を合わなくする	取込前に止まり、通知が届く。Fusion にジョブが出ない
ファイル未着	SFTP のファイルを退避して Run	「ファイル未着」の通知。Errors に出る
Fusion 側のリジェクト	存在しない仕入先名にする	取込は完了するがリジェクトされ、コールバックの通知で気付ける
再実行	修正したファイルを置き直して Run	新しい Import Set で取り込まれ、二重にならない

ステップ11 スケジュールを設定して監視する

入力する iCal 式です（毎日 22:00。実際の時刻はタイム・ゾーンで決まります）。

FREQ=DAILY;BYHOUR=22;BYMINUTE=0;

Schedule の iCal（直接入力） に上の式を入力して Validate します。タイム・ゾーンは日本時間を選びます。開始（From）は「スケジュール開始時」、終了（Until）は「なし」にします
Start Schedule で開始し、翌日 22:00 の予定が Future Instances に表示されることを確認します
翌朝、Observability > Instances で夜間の実行を確認します。予定のインスタンスが、実行後に結果へ変わります

確認ポイント

机上演習（環境なしルート①）

課題

上のシナリオと要件から、統合設計書 を表で作ってください。

表1 統合一覧（統合名・パターン・起動・役割）
表2 スケジュール設計（式・タイム・ゾーン・パラメータ・実行ユーザー）
表3 Stage File の設計（操作・対象・設定）
表4 マッピング設計（ヘッダーと明細。FBDI の列・取込元・変換ルール）
表5 エラー設計（事象・検知・処理・通知・再実行）

FBDI の列の表記は、AP のテンプレート（PayablesStandardInvoiceImportTemplate.xlsm）の列見出しを参考にしてください。列順・必須・桁数・形式はテンプレートで確認します。

解答例を表示

表1 統合一覧

統合名	パターン	起動	役割
AP_INV_FBDI_LOAD	スケジュール	毎晩 22:00（日本時間）	ファイル取得 → 入力チェック → 変換 → zip → Import Bulk Data
AP_INV_FBDI_CALLBACK	アプリケーション	ERP Cloud アダプタ（コールバック）	取込の完了通知を受け、ログを取得して結果を通知

表2 スケジュール設計

項目	設計
式	`FREQ=DAILY;BYHOUR=22;BYMINUTE=0;`（毎日 22:00 に固定するため iCal。BYMINUTE を付ける）
タイム・ゾーン	日本時間を明示的に選ぶ（既定値は利用者の設定に依存）
開始・終了	開始はスケジュール開始時、終了は Never
パラメータ	`BusinessUnit`（既定 JP01 営業単位）、`SourceName`（既定 External）。環境ごとの値は Update schedule parameters で設定
実行ユーザー	専用の技術ユーザー。作成者がロックされると実行されないため、個人で開始しない
時間	外部システムの出力完了（21:30）から 30 分あけて 22:00。処理は 6 時間以内に終わる見込み

表3 Stage File の設計

順	操作	対象	設定
1	Read Entire File	exp_inv_header.csv	ファイル参照 Yes、CSV サンプル、1 行目を列名
2	Read Entire File	exp_inv_lines.csv	同上
3	Write File	ApInvoicesInterface.csv	出力先 fbdi_out、見出しなし、追記しない、全列＋END
4	Write File	ApInvoiceLinesInterface.csv	同上
5	Zip File	fbdi_out	ApInvoices_日付_連番.zip を fbdi_zip へ

1 日 5,000 件程度（1MB 前後）なので Read Entire File（10MB 未満）で足ります。10MB を超える見込みなら、Read File in Segments に変え、取込を複数の zip に分けます。

表4 マッピング設計（ヘッダー：ApInvoicesInterface）

FBDI の列（例）	取込元	変換ルール
Invoice ID	ExtInvoiceId	そのまま（ファイル内で一意な数値）
Business Unit	スケジュール・パラメータ	`BusinessUnit`
Source	スケジュール・パラメータ	`SourceName`
Invoice Number	InvoiceNumber	そのまま
Invoice Amount	Amount	数値（カンマなし）
Invoice Date	InvoiceDate	`YYYY-MM-DD` → `YYYY/MM/DD`
Supplier Name	SupplierCode	ルックアップ `SMP_SUPPLIER`（該当なしは `#NOT_FOUND`）
Supplier Site	SiteCode	そのまま
Invoice Currency	Currency	そのまま
Description	Memo	240 文字に切り詰め
Import Set	（実行時に生成）	実行ごとに一意（例：`IMP` ＋日付時刻）
END	—	固定値 `END`

表4 マッピング設計（明細：ApInvoiceLinesInterface）

FBDI の列（例）	取込元	変換ルール
Invoice ID	ExtInvoiceId	そのまま（ヘッダーと同じ値）
Line Number	LineNo	そのまま
Line Type	AccountCode	ルックアップ `SMP_LINE_TYPE`（5120 → Item、5130 → Freight）
Amount	LineAmount	数値
Description	LineMemo	240 文字に切り詰め
Distribution Combination	AccountCode	例：`101-1000-{勘定}-000`（会社-部門-勘定-補助。実環境の勘定体系に合わせる）
END	—	固定値 `END`

表5 エラー設計

事象	検知	処理	通知	再実行
ファイル未着	Download File のフォルト（Scope）	取込せず Re-throw	運用担当へ「ファイル未着」	配置後に Run（Ad hoc request）
入力チェック不合格	Switch	Throw New Fault で取込しない	運用担当と依頼元へ（内容つき）	依頼元が修正して置き直し、Run
Fusion 呼出しの一時エラー	invoke のフォルト	3 回リトライ、だめなら Re-throw	3 回失敗で通知	Run で再実行（Import Set は新しく作る）
取込ジョブの ERROR	コールバック（Download Options：Always）	ログを取得	運用担当へ（ログつき）	Fusion 側の原因を直して再実行
一部リジェクト	コールバックのログ・出力	リジェクト件数を確認	運用担当へ（件数）	Correct Import Errors で修正（請求書）
二重取込	依頼済みフォルダー、Import Set の一意化	同じファイルを読み直さない	—	依頼済みフォルダーのファイルを戻したときだけ再処理

解説

スケジュール統合には呼出し元がないため、通知を設計に含めない と失敗に気付けません。グローバル・フォルトハンドラは Stop にせず、通知してから Re-throw します
取込前の入力チェックは、ファイル全体を止める（部分取込にしない）設計にすると、再実行が単純になります
コールバックは「ジョブが終わった」ことを知らせます。取込が正常終了でも一部がリジェクトされることがあるため、ログと出力の内容まで確認します
取込元の件数・サイズが増えたときの方針（Read File in Segments、複数 zip、Import Set の分割）を、設計書に書いておきます

モック API でフローを再現する（環境なしルート②）

練習用モック API の importBulkData（zip を受け取り、要求 ID を返す）と、状況照会（ESSJobStatusRF）を使い、OIC のスケジュール統合の流れを Python で再現します。モック API は 架空の応答を返すだけ で、保存はしません（仕様は練習用 API プレイグラウンドを参照）。

モック API では、zip のファイル名に error を含める とジョブが ERROR に、warning を含める と WARNING になります
状態は、投入から約 5 秒が WAIT、約 15 秒までが RUNNING、その後に最終状態になります
OIC ではコールバックで受ける完了通知を、ここでは状況照会（ポーリング）で代用します
モック API の BU 名は JP01 Business Unit です

上の 2 つの CSV を、同じフォルダーに保存します
次のスクリプトを fbdi_rehearsal.py として保存します
環境変数を設定して実行します

export MOCK_BASE="https://<このサイトのURL>/api/mock/fscmRestApi/resources/11.13.18.05"
export MOCK_USER=learner MOCK_PASS=welcome1

python3 fbdi_rehearsal.py                          # 正常（SUCCEEDED）
python3 fbdi_rehearsal.py ApInvoices_error_01.zip   # ファイル名に error → ERROR
python3 fbdi_rehearsal.py ApInvoices_warning_01.zip # ファイル名に warning → WARNING

fbdi_rehearsal.py を表示

"""fbdi_rehearsal.py：OIC のスケジュール統合（FBDI 取込）の流れを、練習用モック API で再現する。

使い方:
  export MOCK_BASE="https://<このサイトのURL>/api/mock/fscmRestApi/resources/11.13.18.05"
  export MOCK_USER=learner MOCK_PASS=welcome1
  python3 fbdi_rehearsal.py [zip ファイル名]
"""
import base64
import csv
import io
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from datetime import datetime

BASE = os.environ["MOCK_BASE"].rstrip("/")
AUTH = "Basic " + base64.b64encode(f'{os.environ["MOCK_USER"]}:{os.environ["MOCK_PASS"]}'.encode()).decode()

BUSINESS_UNIT = "JP01 Business Unit"  # モック API の BU 名。実機は自社のテスト用 BU に読み替える
SOURCE = "External"
SUPPLIER_LOOKUP = {"10001": "サンプル商事株式会社", "10002": "仮想オフィス用品株式会社"}  # OIC ではルックアップ
LINE_TYPE_LOOKUP = {"5120": "Item", "5130": "Freight"}  # 勘定コード → 明細タイプ（OIC ではルックアップ）

# 実機の FBDI は、テンプレートの列をすべて、この順序で並べる。ここでは学習用に主要な列だけにした
HEADER_COLUMNS = ["InvoiceId", "BusinessUnit", "Source", "InvoiceNumber", "InvoiceAmount", "InvoiceDate",
                  "SupplierName", "SupplierSite", "InvoiceCurrency", "Description", "ImportSet", "END"]
LINE_COLUMNS = ["InvoiceId", "LineNumber", "LineType", "LineAmount", "Description",
                "DistributionCombination", "END"]


def call(method, path, body=None):
    data = None if body is None else json.dumps(body).encode("utf-8")
    req = urllib.request.Request(BASE + path, method=method, data=data,
                                 headers={"Authorization": AUTH, "Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=30) as res:
            return json.load(res)
    except urllib.error.HTTPError as e:
        sys.exit(f"HTTP {e.code}: {e.read().decode('utf-8', 'replace')}")


def read_csv(path):
    with open(path, newline="", encoding="utf-8-sig") as f:
        return list(csv.DictReader(f))


def validate(headers, lines):
    """取込前の入力チェック。1 件でも不合格ならファイルごと止める（部分取込にしない）"""
    errors, ids = [], {h["ExtInvoiceId"] for h in headers}
    totals = {}
    for ln in lines:
        if ln["ExtInvoiceId"] not in ids:
            errors.append(f"明細の請求書 ID {ln['ExtInvoiceId']} がヘッダーにありません")
        totals[ln["ExtInvoiceId"]] = totals.get(ln["ExtInvoiceId"], 0) + int(ln["LineAmount"])
        if ln["AccountCode"] not in LINE_TYPE_LOOKUP:
            errors.append(f"勘定コード {ln['AccountCode']} が変換表にありません")
    for h in headers:
        if h["SupplierCode"] not in SUPPLIER_LOOKUP:
            errors.append(f"仕入先コード {h['SupplierCode']} が変換表にありません")
        if totals.get(h["ExtInvoiceId"], 0) != int(h["Amount"]):
            errors.append(f"{h['InvoiceNumber']}: 明細の合計 {totals.get(h['ExtInvoiceId'], 0)} とヘッダー金額 {h['Amount']} が一致しません")
    return errors


def build_zip(headers, lines, import_set):
    """FBDI の zip を作る。CSV に見出し行は付けず、各行の最後に END を付ける"""
    def to_csv(rows):
        buf = io.StringIO()
        csv.writer(buf, lineterminator="\n").writerows(rows)
        return buf.getvalue().encode("utf-8")

    header_rows = [[h["ExtInvoiceId"], BUSINESS_UNIT, SOURCE, h["InvoiceNumber"], h["Amount"],
                    h["InvoiceDate"].replace("-", "/"), SUPPLIER_LOOKUP[h["SupplierCode"]], h["SiteCode"],
                    h["Currency"], h["Memo"], import_set, "END"] for h in headers]
    line_rows = [[ln["ExtInvoiceId"], ln["LineNo"], LINE_TYPE_LOOKUP[ln["AccountCode"]], ln["LineAmount"],
                  ln["LineMemo"], f"101-1000-{ln['AccountCode']}-000", "END"] for ln in lines]
    # 列数のずれは FBDI のロード失敗の典型的な原因。zip にする前に検出する
    assert all(len(r) == len(HEADER_COLUMNS) for r in header_rows), "ヘッダーの列数が不正です"
    assert all(len(r) == len(LINE_COLUMNS) for r in line_rows), "明細の列数が不正です"
    out = io.BytesIO()
    with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as z:
        z.writestr("ApInvoicesInterface.csv", to_csv(header_rows))      # zip 内のファイル名はテンプレートのまま
        z.writestr("ApInvoiceLinesInterface.csv", to_csv(line_rows))
    return out.getvalue()


def import_bulk_data(zip_bytes, zip_name, import_set):
    body = {
        "OperationName": "importBulkData",
        "DocumentContent": base64.b64encode(zip_bytes).decode("ascii"),
        "FileName": zip_name,
        "ContentType": "zip",
        "DocumentAccount": "fin$/payables$/import$",
        "JobName": "oracle/apps/ess/financials/payables/invoices/transactions,APXIIMPT",
        "ParameterList": f"{BUSINESS_UNIT},{SOURCE},{import_set}",  # 並びは例。実機は標準ジョブの定義に合わせる
    }
    return call("POST", "/erpintegrations", body)["ReqstId"]


def wait_for_job(request_id, timeout=90, interval=3):
    """OIC ではコールバックで受ける完了通知を、ここでは状況照会（ポーリング）で代用する"""
    path = "/erpintegrations?" + urllib.parse.urlencode(
        {"finder": f"ESSJobStatusRF;requestId={request_id}", "onlyData": "true"})
    started = time.time()
    while time.time() - started < timeout:
        items = call("GET", path)["items"]
        status = items[0]["RequestStatus"] if items else "UNKNOWN"
        print(f"  状態: {status}")
        if status in ("SUCCEEDED", "WARNING", "ERROR"):
            return status
        time.sleep(interval)
    return "TIMEOUT"


def fetch_output(request_id):
    # fileType=LOG は「ログと出力」の両方、OUT は出力のみ（公式の REST ガイド）。ALL は公式の値ではない
    path = "/erpintegrations?" + urllib.parse.urlencode(
        {"finder": f"ESSJobExecutionDetailsRF;requestId={request_id},fileType=LOG", "onlyData": "true"})
    items = call("GET", path)["items"]
    data = base64.b64decode(items[0]["DocumentContent"])
    with zipfile.ZipFile(io.BytesIO(data)) as z:
        return {name: z.read(name).decode("utf-8", "replace") for name in z.namelist()}


def main():
    now = datetime.now()
    zip_name = sys.argv[1] if len(sys.argv) > 1 else f"ApInvoices_{now:%Y%m%d}_01.zip"
    import_set = f"IMP{now:%Y%m%d%H%M%S}"  # 要求ごとに一意にする

    headers, lines = read_csv("exp_inv_header.csv"), read_csv("exp_inv_lines.csv")
    print(f"[1] 入力を読み込みました：ヘッダー {len(headers)} 件、明細 {len(lines)} 件")
    errors = validate(headers, lines)
    if errors:
        print("[2] 入力チェック NG。取込を中止します（ファイルは処理済みにしない）")
        for e in errors:
            print("   -", e)
        sys.exit(1)
    print("[2] 入力チェック OK")

    zip_bytes = build_zip(headers, lines, import_set)
    print(f"[3] zip を作成しました：{zip_name}（{len(zip_bytes)} バイト、Import Set = {import_set}）")

    request_id = import_bulk_data(zip_bytes, zip_name, import_set)
    print(f"[4] importBulkData を依頼しました：ReqstId = {request_id}")

    status = wait_for_job(request_id)
    print(f"[5] ジョブの最終状態：{status}")
    if status == "TIMEOUT":
        sys.exit("NOTIFY: 時間内に終わりませんでした。後で状況を確認してください")

    files = fetch_output(request_id)
    for name, text in files.items():
        print(f"--- {name} ---\n{text.strip()}")
    out = next((t for n, t in files.items() if n.endswith(".out")), "")
    rows = list(csv.DictReader(io.StringIO(out)))
    rejected = int(rows[0]["RECORDS_REJECTED"]) if rows else 0  # モックの出力形式。実機は出力レポートの内容を確認する

    if status == "SUCCEEDED" and rejected == 0:
        print("[6] 完了：入力ファイルを処理済みフォルダーへ移す（この演習では省略）")
    else:
        print(f"[6] NOTIFY: 状態 {status}、リジェクト {rejected} 件。運用担当へ通知し、ファイルは処理済みにしない")
        sys.exit(2)


if __name__ == "__main__":
    main()

次のことを確認してください。

正常のとき、入力チェック OK → zip 作成 → 要求 ID → 状態の遷移（WAIT → RUNNING → SUCCEEDED）→ 出力ファイルの表示 → 完了、の順に出る
error を含めたとき、状態が ERROR になり、NOTIFY の行が出て終了コードが 2 になる（OIC の通知とエラー処理に相当）
exp_inv_lines.csv の金額を 1 つ変えて実行すると、入力チェックで止まり、モック API を呼ばずに終了コード 1 になる
HEADER_COLUMNS の列を 1 つ減らすと、列数のチェックで止まる（FBDI の列数のずれが、ロード失敗の典型的な原因であることの再現）

OIC の実装との違いも整理してください。OIC では次のように役割が分かれます。

CSV の作成は Stage File の Write File、zip は Zip File が行います（スクリプトでは build_zip()）
完了通知は、ポーリングではなく コールバック統合 が受けます（スクリプトでは wait_for_job()）
スケジュールの管理、フォルト処理、実行履歴は OIC が受け持ちます
スクリプトは完了を待ってから入力ファイルの扱いを決めますが、OIC の設計では 依頼の直後に依頼済みフォルダーへ移し、結果はコールバックの通知で受けます（結果で振り分ける方法は発展課題 4）

よくあるエラーと対処

症状	原因と対処
SFTP 接続の Test が失敗する	ホスト・ポート・認証方式、接続元 IP の許可を確認する。閉域網のサーバーなら接続エージェントを使う
CSV の列がずれて読まれる	区切り文字、Optionally Enclosed By（引用符）、文字コードを確認する。サンプル CSV を実データに近づける
Read Entire File が失敗する	ファイルが 10MB 以上でないか確認する。大きいなら Read File in Segments にする
Load Interface File for Import が失敗する	列数・列順・zip 内のファイル名を確認する。見出し行が入っていないか、列を削除していないか
Import Payables Invoices が 0 件になる	Source や BU のジョブ・パラメータが CSV の値と一致しているか、Import Set が一致しているかを確認する
コールバック統合が起動しない	コールバック統合が有効か、取込の呼出しで Enable Callback を選んだか、統合ユーザーのロールが足りているかを確認する。プライベート・エンドポイント経由のイベント受信には制約がある
毎晩のスケジュールが動かない	Start Schedule で開始したか、統合が有効か、タイム・ゾーン、スケジュールを作成したユーザーがロックされていないかを確認する
zip に余計なファイルが入る	Zip File はディレクトリ全体が対象。出力先を取込ごとに専用にする
同じ請求書が二重に取り込まれそうになる	入力ファイルを依頼済みフォルダーへ移しているか、Import Set を実行ごとに一意にしているかを確認する。請求書番号の重複は Fusion 側でもエラーになる

発展課題

請求書ごとの突合：ヘッダーの Amount と、その請求書の明細の合計を、請求書単位で突合する（For-Each の入れ子、または明細を ExtInvoiceId で集計）。不一致の請求書だけをエラーにする
大容量対応：Read Entire File を Read File in Segments に置き換え、セグメントごとの変換を設計する。Write File の追記の制限（スケジュール統合とファイル処理）を踏まえ、複数の zip に分けて取り込む案を比較する
結果通知の充実：コールバックで取得したログ・出力の zip を Unzip File で展開して Read Entire File で読み、リジェクト件数と内容を通知本文に入れる
退避の高度化：コールバックの結果に応じて、入力ファイルを「成功」「失敗」のフォルダーへ移す仕組みを考える（コールバックから元のファイルを特定する方法を含む）
再処理の指定：スケジュール・パラメータで処理対象日を指定できるようにし、Run の Ad hoc request で過去日を再処理できるようにする
暗号化ファイルの受信：取込元が PGP で暗号化したファイルを送ってくる想定で、Decrypt File を入れる（PGP 鍵を Certificates に登録）
突合：翌朝に、REST で取り込んだ請求書の件数・金額を取得し、取込元の件数・金額と突き合わせる統合を作る（ESS 演習のジョブ監視の考え方）