MF会計APIで仕訳を自動登録するときの二重登録対策と冪等設計

マネーフォワード クラウド会計の未仕訳明細を、スクリプトで取得して仕訳として登録する。この自動化を組んでいる、または組もうとしている人向けの記事です。

最初の1回は動きます。トラブルは、2回目以降の実行、通信が途中で切れたとき、cronが重なったときに出ます。この記事では、仕訳登録の自動化で起きやすい失敗を先に挙げ、それを冪等設計で潰していく手順を書きます。

結論

• 冪等性はAPI側に期待せず、自分の側の台帳で担保する
• 明細から決定的に作れる識別キーを決め、台帳の主キーにする
• 送信する前に台帳へ「送信中」を書いておく
• タイムアウトは失敗ではなく「不明」として扱い、再送せずに照合する
• 同日・同額・同摘要の正当な2件を重複扱いしない仕組みを入れる

コードはPythonで、台帳にはSQLiteを使います。APIのエンドポイント名やフィールド名は、以下では抽象化したクライアントとして書きます。実際の仕様は公式ドキュメントで確認してください。

問題の深掘り：なぜ二重登録は起きるのか

仕訳登録の自動化で二重登録が起きる経路は、大きく4つあります。

1つ目は、単純な再実行です。途中で落ちたバッチを頭からやり直すと、登録済みの明細をもう一度登録します。

2つ目は、タイムアウトです。組み込みの通信でいえば、コマンドを送ったのにACKが返ってこない状態です。相手に届いていないのか、処理されたのに返事だけ来ていないのか、送信側からは区別できません。ここで「失敗」と判断して再送すると、処理済みだった場合に二重になります。

3つ目は、明細の再取得でキーがずれることです。取得し直したときに摘要の表記が変わったり、取得範囲が変わって並び順が動いたりすると、同じ取引が別物に見えます。

4つ目は逆向きで、重複と誤判定して正当な明細を落とすパターンです。同じ日に同じ店で同じ金額を2回使うことは普通にあります。日付・金額・摘要が一致するだけで重複とみなすと、2件目が消えます。

二重登録は修正が面倒です。登録後の仕訳を探して消す作業が必要で、月次を締めたあとなら影響範囲も広がります。一方、登録漏れは未仕訳として画面に残るので気づけます。重複より漏れのほうが安全に倒せるため、迷ったら登録しない側に倒す設計にします。

この手の問題を前提にして自分で作り、運用しているキットが JournalPilot です。以下の冪等設計は、そこで採っている考え方の骨格にあたります。

解決手順

ステップ1：識別キーを決める

台帳の主キーになる source_key を、明細から決定的に作ります。取得元が安定した取引IDを返すなら、それを使うのが一番です。IDがない、または信用できない場合は、口座・日付・金額・正規化した摘要から作ります。

同日・同額・同摘要の取引が複数ある場合は、取得バッチ内での出現順を連番として足します。

import hashlib
import unicodedata
from collections import Counter

def normalize(text: str) -> str:
    return unicodedata.normalize("NFKC", text).strip()

def build_keys(transactions: list[dict]) -> list[str]:
    seen = Counter()
    keys = []
    for tx in transactions:
        base = "|".join([
            tx["account_id"],
            tx["date"],
            str(tx["amount"]),
            normalize(tx["description"]),
        ])
        seen[base] += 1
        raw = f"{base}|{seen[base]}"
        keys.append(hashlib.sha256(raw.encode("utf-8")).hexdigest()[:32])
    return keys

連番方式には弱点があります。取得範囲を変えると連番がずれる可能性があるため、取得範囲は固定します。取得元に取引IDがあるなら、連番方式より取引IDを優先してください。

ステップ2：台帳を作る

状態は4つに絞ります。送信前に書く submitting、成功した done、結果が分からない unknown、4xxなどで確定的に失敗した failed です。

import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS ledger (
  source_key TEXT PRIMARY KEY,
  state      TEXT NOT NULL,
  remote_id  TEXT,
  updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
"""

def open_ledger(path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute(SCHEMA)
    conn.commit()
    return conn

def mark(conn, key, state, remote_id=None):
    conn.execute(
        "UPDATE ledger SET state=?, remote_id=?, updated_at=CURRENT_TIMESTAMP "
        "WHERE source_key=?",
        (state, remote_id, key),
    )
    conn.commit()

PRIMARY KEY があるので、同じキーのINSERTはDBが拒否します。アプリ側の判定が漏れても、最後の防波堤として効きます。

ステップ3：送信前に書き、結果で状態を分ける

処理の核は、API呼び出しの前に submitting をコミットしておく点です。送信直後にプロセスが落ちても、台帳には「送ったかもしれない」という痕跡が残ります。

def process(conn, client, key, journal):
    row = conn.execute(
        "SELECT state FROM ledger WHERE source_key=?", (key,)
    ).fetchone()

    if row and row[0] in ("done", "failed"):
        return row[0]

    if row:  # submitting / unknown：再送の前に照合する
        remote = client.find_by_key(key)
        if remote:
            mark(conn, key, "done", remote["id"])
            return "done"
    else:
        conn.execute(
            "INSERT INTO ledger(source_key, state) VALUES(?, 'submitting')",
            (key,),
        )
        conn.commit()

    try:
        res = client.post_journal(journal)  # 備考相当の欄に key を入れておく
    except (TimeoutError, ConnectionError):
        mark(conn, key, "unknown")
        return "unknown"
    except client.ClientError:
        mark(conn, key, "failed")
        return "failed"

    mark(conn, key, "done", res["id"])
    return "done"

ポイントは3つです。

• 4xx系は確定失敗なので failed にして人間に回す
• タイムアウトと接続断は unknown にして、再送しない
• 再実行時に submitting または unknown が残っていたら、再送の前にリモート側へ照合する

ステップ4：照合のためにキーを仕訳側にも埋める

find_by_key が成り立つには、登録する仕訳の側にもキーが残っていなければなりません。仕訳のメモや備考に相当するフィールドに source_key を入れておけば、「台帳では不明だが、リモートには存在する」ケースを後から拾えます。使えるフィールドと検索条件は、公式ドキュメントで確認してください。

登録直後の検索結果に反映されるまで時間差がある可能性も考えて、unknown は一定時間置いてから照合するのが安全です。照合しても見つからなければ、そこで初めて再送します。

ステップ5：実行前に dry-run を出す

登録の前に、新規件数・照合待ち件数・失敗件数を出力して、人間が見られるようにします。初回や取得範囲を変えた直後は、dry-run の件数が想定と違わないかを確認してから本実行に進みます。

落とし穴と対処

実際にこの種の自動化を組むとき、冪等設計で先に潰しておきたい穴を挙げます。

タイムアウトを失敗扱いにしてリトライする。 もっとも起きやすい経路です。ネットワークエラー全般を一括りに except して再試行するコードは、書きがちです。対処はステップ3のとおり、確定失敗と結果不明を分けることです。

キーに摘要を入れたせいで、取得し直しで別物になる。 摘要は表記ゆれの影響を受けます。NFKC正規化で全角半角の揺れは吸収できますが、サービス側が摘要を書き換える場合は吸収できません。取得元に取引IDがあるなら、摘要をキーから外すほうが堅いです。

cronの多重起動。 前回の実行が終わる前に次が始まると、同じ明細を2つのプロセスが処理します。台帳の主キー制約が最後の防波堤になりますが、それに頼らず、実行の入口でファイルロックを取るようにします。

台帳が真実の情報源になるので、失うと冪等性も失う。 台帳ファイルを消した、別のマシンで動かした、という状況では、全件が新規に見えます。台帳はバックアップ対象にし、起動時に台帳が空なら警告を出す程度のガードは入れておく価値があります。

AI判定を再実行のたびにやり直す。 勘定科目の判定にLLMを使う場合、同じ明細でも判定が揺れることがあります。判定結果と根拠は台帳に保存し、登録済みや照合待ちの明細は再判定しないようにします。再実行で科目が変わると、冪等でなくなります。

関連ツール

ここまでの設計（明細取得、勘定科目のAI判定、冪等な仕訳登録、売掛金の発生・回収チェック）をまとめた自己ホスト型のキットが JournalPilot です。台帳の設計を一から組む時間を省きたい人向けに、note で配布しています。

JournalPilot：マネーフォワード会計の記帳を自動化する自己ホストキット（明細取込、勘定科目AI判定、仕訳登録）

まとめ

仕訳登録の自動化で厄介なのは、1回目の成功ではなく、2回目以降と異常系です。冪等性は、APIに期待せず自前の台帳で作ります。

• 明細から決定的に作れるキーで一意にする
• 送信前に状態を書く
• タイムアウトは「不明」として、再送より照合を先にする
• 迷ったら登録しない側に倒す

この4点を押さえておけば、二重登録の大半は防げるはずです。

---

著者：ぽん（@pon_freelance）
C言語実務23年、組み込み／制御系。
副業で技術記事販売と自作ツール販売をやっている。

書いているもの：

• JournalPilot - マネーフォワード会計の記帳を自動化する自己ホストキット（明細取込→勘定科目AI判定→仕訳登録）
  （その他：（なし））

同じ手で作った道具を、インストールなしで試せます — 自分が実際に使っている自作ツールを、ブラウザからそのまま動かせる形で並べています。登録は不要です。
→ RunStack（無料で試す）