第166回 SQLiteバックアップを隔離環境へ戻す:Pythonで作る復元訓練と業務スキーマ確認

第165回では、稼働中のSQLiteから整合したバックアップを作り、SHA-256とSQLiteの検査結果をマニフェストへ残しました。ただし、バックアップを作成できたことと、必要な状態で復元できることは同じではありません。

今回は、本番データベースを上書きせず、新しい隔離用ディレクトリで復元訓練を行います。コピー後のファイルを再読して計算したSHA-256、SQLiteの整合性、外部キー、PRAGMA user_version、必須テーブル、列、行数範囲を確認します。

実装にはPython 3.10以上の標準ライブラリだけを使います。本番データベースとの切り替えや、アプリケーションからの書き込み試験は対象外です。

今回の復元訓練で確認すること

  1. 別に保管したSHA-256と、コピー後のファイルを再読して計算した値が一致する
  2. 復元先を新しく作成し、既存のファイルやディレクトリを上書きしない
  3. integrity_check(1)がokを返す
  4. foreign_key_checkで違反が見つからない
  5. PRAGMA user_versionが復元ポリシーと一致する
  6. 必要なテーブルと列が存在し、行数が許容範囲内にある
  7. すべての検査を通過した後でreport.jsonを確定する
項目 今回の契約
入力 SQLiteのバックアップファイル、期待するSHA-256、復元ポリシーJSON
復元先 実行前に存在しないディレクトリ
生成物 recovered.sqliteとreport.json
完了条件 report.jsonをJSONとして読み取れ、statusがpassedであること
上書き 既存の復元先は拒否する
対象外 本番への切り替え、業務データの意味上の正しさ、添付データベース、外部ファイル、アプリケーション固有の検索結果

入力には、SQLiteのバックアップAPIなどで作成した、単独で開けるバックアップファイルを使います。稼働中のメインデータベースだけをファイルコピーすると、WALファイルに残っている変更を含められないことがあります。

復元ポリシーを用意する

次の内容をrecovery-policy.jsonとして保存します。この例は、後ほど作成する動作確認用データベースに対応しています。

{
  "version": 1,
  "sqlite_user_version": 1,
  "tables": {
    "review_items": {
      "required_columns": [
        "batch_id",
        "record_id",
        "status"
      ],
      "min_rows": 2,
      "max_rows": 2
    }
  }
}

min_rowsとmax_rowsは、動作確認用データが2行であるため同じ値にしています。実際のバックアップでは、復旧時点として許容できる範囲を設定します。上限を設けない場合はmax_rowsをnullにします。

このポリシーは必須列を検査しますが、追加の列やテーブルは拒否しません。検査対象には通常列だけでなく、生成列と仮想テーブルの隠し列も含めます。そのため、列一覧の取得には、それらを省略するPRAGMA table_infoではなくPRAGMA table_xinfoを使います。列名は大文字と小文字を含めてポリシーと一致させます。PRAGMA user_versionはSQLiteが自動更新する値ではありません。スキーマ変更時に、データベースとポリシーを同じ値へ更新します。

復元訓練CLIを実装する

次のコードをrestore_drill.pyとして保存します。復元先ディレクトリを作成し、コピー後の候補ファイルを再読してSHA-256を計算してから、読み取り専用のSQLite接続で検査します。

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import sqlite3
import stat
import sys
from pathlib import Path
from typing import Any

CHUNK_SIZE = 1024 * 1024


class DrillError(Exception):
    def __init__(self, exit_code: int, kind: str, message: str) -> None:
        super().__init__(message)
        self.exit_code = exit_code
        self.kind = kind
        self.message = message


def write_json_line(payload: dict[str, Any], stream: Any) -> None:
    print(
        json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
        file=stream,
    )


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        write_json_line(
            {"status": "error", "kind": "arguments", "message": message},
            sys.stderr,
        )
        raise SystemExit(2)


def reject_duplicate_keys(
    pairs: list[tuple[str, Any]],
) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = value
    return result


def reject_constant(value: str) -> None:
    raise ValueError(f"non-standard JSON number: {value}")


def validate_name(value: Any, label: str) -> str:
    if type(value) is not str:
        raise DrillError(2, "policy", f"{label} must be a string")
    if (
        not value
        or value != value.strip()
        or any(ord(char) < 32 or ord(char) == 127 for char in value)
    ):
        raise DrillError(
            2,
            "policy",
            f"{label} contains an invalid character",
        )
    return value


def load_policy(path: Path) -> dict[str, Any]:
    try:
        with path.open("r", encoding="utf-8") as policy_file:
            raw = json.load(
                policy_file,
                object_pairs_hook=reject_duplicate_keys,
                parse_constant=reject_constant,
            )
    except (OSError, UnicodeError, ValueError) as exc:
        raise DrillError(
            2,
            "policy",
            f"cannot load policy: {exc}",
        ) from exc

    required_top = {"version", "sqlite_user_version", "tables"}
    if type(raw) is not dict or set(raw) != required_top:
        raise DrillError(
            2,
            "policy",
            "policy keys must be version, sqlite_user_version, and tables",
        )
    if type(raw["version"]) is not int or raw["version"] != 1:
        raise DrillError(
            2,
            "policy",
            "policy version must be integer 1",
        )

    user_version = raw["sqlite_user_version"]
    if (
        type(user_version) is not int
        or not 0 <= user_version <= 2_147_483_647
    ):
        raise DrillError(
            2,
            "policy",
            "sqlite_user_version must be an integer from 0 to 2147483647",
        )

    tables = raw["tables"]
    if type(tables) is not dict or not tables:
        raise DrillError(
            2,
            "policy",
            "tables must be a non-empty object",
        )

    required_spec = {"required_columns", "min_rows", "max_rows"}
    for raw_table_name, spec in tables.items():
        table_name = validate_name(raw_table_name, "table name")
        if table_name.lower().startswith("sqlite_"):
            raise DrillError(
                2,
                "policy",
                "internal SQLite tables are not allowed",
            )
        if type(spec) is not dict or set(spec) != required_spec:
            raise DrillError(
                2,
                "policy",
                f"{table_name} must define required_columns, min_rows, and max_rows",
            )

        columns = spec["required_columns"]
        if type(columns) is not list or not columns:
            raise DrillError(
                2,
                "policy",
                f"{table_name}.required_columns must be a non-empty array",
            )
        checked_columns = [
            validate_name(column, f"{table_name} column")
            for column in columns
        ]
        if len(checked_columns) != len(set(checked_columns)):
            raise DrillError(
                2,
                "policy",
                f"{table_name}.required_columns contains a duplicate",
            )

        min_rows = spec["min_rows"]
        max_rows = spec["max_rows"]
        if type(min_rows) is not int or min_rows < 0:
            raise DrillError(
                2,
                "policy",
                f"{table_name}.min_rows must be a non-negative integer",
            )
        if max_rows is not None and (
            type(max_rows) is not int or max_rows < min_rows
        ):
            raise DrillError(
                2,
                "policy",
                f"{table_name}.max_rows must be null or at least min_rows",
            )

    return raw


def parse_expected_sha256(value: str) -> str:
    if re.fullmatch(r"[0-9a-fA-F]{64}", value) is None:
        raise DrillError(
            2,
            "arguments",
            "expected SHA-256 must contain exactly 64 hexadecimal characters",
        )
    return value.lower()


def copy_and_hash_destination(
    source: Path,
    destination: Path,
) -> tuple[str, int]:
    try:
        with source.open("rb") as source_file:
            source_mode = os.fstat(source_file.fileno()).st_mode
            if not stat.S_ISREG(source_mode):
                raise DrillError(
                    2,
                    "input",
                    "backup must be a regular file",
                )
            with destination.open("xb") as destination_file:
                while True:
                    chunk = source_file.read(CHUNK_SIZE)
                    if not chunk:
                        break
                    written = destination_file.write(chunk)
                    if written != len(chunk):
                        raise OSError("short write while copying backup")
                destination_file.flush()
                os.fsync(destination_file.fileno())
    except DrillError:
        raise
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"backup copy failed: {exc}",
        ) from exc

    digest = hashlib.sha256()
    total = 0
    try:
        with destination.open("rb") as copied_file:
            while True:
                chunk = copied_file.read(CHUNK_SIZE)
                if not chunk:
                    break
                digest.update(chunk)
                total += len(chunk)
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"cannot read copied backup: {exc}",
        ) from exc

    if total == 0:
        raise DrillError(4, "sqlite_check", "backup file is empty")
    return digest.hexdigest(), total


def quote_identifier(name: str) -> str:
    return '"' + name.replace('"', '""') + '"'


def validate_database(
    path: Path,
    policy: dict[str, Any],
) -> dict[str, Any]:
    uri = f"{path.resolve().as_uri()}?mode=ro&immutable=1"
    try:
        connection = sqlite3.connect(uri, uri=True)
        try:
            connection.execute("PRAGMA foreign_keys = ON")
            connection.execute("PRAGMA query_only = ON")

            integrity_rows = connection.execute(
                "PRAGMA integrity_check(1)"
            ).fetchall()
            if integrity_rows != [("ok",)]:
                raise DrillError(
                    4,
                    "sqlite_check",
                    "integrity_check did not return ok",
                )

            foreign_key_row = connection.execute(
                "PRAGMA foreign_key_check"
            ).fetchone()
            if foreign_key_row is not None:
                raise DrillError(
                    4,
                    "sqlite_check",
                    "foreign_key_check found a violation",
                )

            actual_user_version = connection.execute(
                "PRAGMA user_version"
            ).fetchone()[0]
            if actual_user_version != policy["sqlite_user_version"]:
                raise DrillError(
                    4,
                    "sqlite_check",
                    "SQLite user_version does not match the policy",
                )

            actual_tables = {
                row[0]
                for row in connection.execute(
                    "SELECT name FROM sqlite_schema WHERE type = 'table'"
                )
            }
            table_results: dict[str, Any] = {}

            for table_name, spec in policy["tables"].items():
                if table_name not in actual_tables:
                    raise DrillError(
                        4,
                        "sqlite_check",
                        f"required table is missing: {table_name}",
                    )

                table_sql = quote_identifier(table_name)
                columns = [
                    row[1]
                    for row in connection.execute(
                        f"PRAGMA table_xinfo({table_sql})"
                    )
                ]
                missing_columns = [
                    column
                    for column in spec["required_columns"]
                    if column not in columns
                ]
                if missing_columns:
                    raise DrillError(
                        4,
                        "sqlite_check",
                        f"required column is missing from {table_name}",
                    )

                row_count = connection.execute(
                    f"SELECT COUNT(*) FROM {table_sql}"
                ).fetchone()[0]
                if row_count < spec["min_rows"]:
                    raise DrillError(
                        4,
                        "sqlite_check",
                        f"row count is below the minimum for {table_name}",
                    )
                max_rows = spec["max_rows"]
                if max_rows is not None and row_count > max_rows:
                    raise DrillError(
                        4,
                        "sqlite_check",
                        f"row count is above the maximum for {table_name}",
                    )

                table_results[table_name] = {
                    "row_count": row_count,
                    "required_columns_checked": len(
                        spec["required_columns"]
                    ),
                }
        finally:
            connection.close()
    except DrillError:
        raise
    except sqlite3.Error as exc:
        raise DrillError(
            4,
            "sqlite_check",
            f"SQLite validation failed: {exc}",
        ) from exc

    return {
        "integrity_check": "ok",
        "foreign_key_violations": 0,
        "sqlite_user_version": actual_user_version,
        "tables": table_results,
    }


def write_report_candidate(path: Path, payload: dict[str, Any]) -> None:
    try:
        with path.open("x", encoding="utf-8") as report_file:
            json.dump(payload, report_file, ensure_ascii=False, indent=2)
            print(file=report_file)
            report_file.flush()
            os.fsync(report_file.fileno())
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"cannot write report: {exc}",
        ) from exc


def rename_new(source: Path, target: Path) -> None:
    if os.path.lexists(target):
        raise DrillError(
            5,
            "output",
            f"internal destination already exists: {target}",
        )
    try:
        source.rename(target)
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"cannot finalize {target}: {exc}",
        ) from exc


def cleanup_created_output(output_dir: Path, paths: list[Path]) -> None:
    for path in paths:
        try:
            path.unlink()
        except FileNotFoundError:
            pass
        except OSError:
            pass
    try:
        output_dir.rmdir()
    except OSError:
        pass


def build_parser() -> argparse.ArgumentParser:
    parser = JsonArgumentParser(
        description="Restore a SQLite backup into an isolated directory."
    )
    parser.add_argument("--backup", required=True, type=Path)
    parser.add_argument("--expected-sha256", required=True)
    parser.add_argument("--policy", required=True, type=Path)
    parser.add_argument("--output-dir", required=True, type=Path)
    return parser


def run(args: argparse.Namespace) -> dict[str, Any]:
    expected_sha256 = parse_expected_sha256(args.expected_sha256)
    policy = load_policy(args.policy)

    if not args.backup.is_file():
        raise DrillError(
            2,
            "input",
            "backup does not exist or is not a regular file",
        )
    if args.output_dir.exists():
        raise DrillError(2, "output", "output path already exists")
    if not args.output_dir.parent.is_dir():
        raise DrillError(
            2,
            "output",
            "output parent does not exist or is not a directory",
        )

    try:
        args.output_dir.mkdir(mode=0o700)
    except FileExistsError as exc:
        raise DrillError(
            2,
            "output",
            "output path already exists",
        ) from exc
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"cannot create output directory: {exc}",
        ) from exc

    candidate = args.output_dir / ".recovered.sqlite.part"
    recovered = args.output_dir / "recovered.sqlite"
    report_candidate = args.output_dir / ".report.json.part"
    report_path = args.output_dir / "report.json"
    known_paths = [report_path, report_candidate, recovered, candidate]
    success = False

    try:
        actual_sha256, size_bytes = copy_and_hash_destination(
            args.backup,
            candidate,
        )
        if actual_sha256 != expected_sha256:
            raise DrillError(
                3,
                "hash_mismatch",
                "copied file does not match the expected SHA-256",
            )

        checks = validate_database(candidate, policy)
        report = {
            "status": "passed",
            "backup_name": args.backup.name,
            "output_dir": str(args.output_dir),
            "database": str(recovered),
            "report": str(report_path),
            "sha256": actual_sha256,
            "size_bytes": size_bytes,
            "checks": checks,
        }

        rename_new(candidate, recovered)
        write_report_candidate(report_candidate, report)
        rename_new(report_candidate, report_path)
        success = True
        return report
    finally:
        if not success:
            cleanup_created_output(args.output_dir, known_paths)


def main() -> int:
    args = build_parser().parse_args()
    try:
        report = run(args)
    except DrillError as exc:
        write_json_line(
            {
                "status": "error",
                "kind": exc.kind,
                "message": exc.message,
            },
            sys.stderr,
        )
        return exc.exit_code
    except KeyboardInterrupt:
        write_json_line(
            {
                "status": "error",
                "kind": "interrupted",
                "message": "interrupted by user",
            },
            sys.stderr,
        )
        return 130
    except Exception:
        write_json_line(
            {
                "status": "error",
                "kind": "unexpected",
                "message": "unexpected execution error",
            },
            sys.stderr,
        )
        return 5

    write_json_line(report, sys.stdout)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

immutable=1は、コピー後の候補ファイルを検査中に変更しないことを前提に使います。SQLiteはこの指定があるファイルのロックと変更検出を省略します。書き込みが続いている本番データベースには使えません。

report.jsonは最後に候補名から確定名へ変更します。途中で停止してrecovered.sqliteだけが残っても、完了レポートがなければ成功として扱いません。

動作確認用のSQLiteを作る

次のコードをmake_demo_backup.pyとして保存します。動作確認専用のSQLiteファイルを新規作成し、サンプルデータを2行登録します。

from __future__ import annotations

import hashlib
import json
import sqlite3
from pathlib import Path

DB_PATH = Path("queue-backup.sqlite")


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as source:
        while True:
            chunk = source.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
    return digest.hexdigest()


def main() -> None:
    try:
        DB_PATH.touch(exist_ok=False)
    except FileExistsError as exc:
        raise SystemExit("queue-backup.sqlite already exists") from exc

    try:
        with sqlite3.connect(DB_PATH) as connection:
            connection.execute("PRAGMA foreign_keys = ON")
            connection.execute("PRAGMA user_version = 1")
            connection.execute(
                """
                CREATE TABLE review_items (
                    batch_id TEXT NOT NULL,
                    record_id TEXT NOT NULL,
                    status TEXT NOT NULL CHECK (
                        status IN (
                            'pending',
                            'accepted',
                            'corrected',
                            'rejected'
                        )
                    ),
                    PRIMARY KEY (batch_id, record_id)
                )
                """
            )
            connection.executemany(
                """
                INSERT INTO review_items (
                    batch_id,
                    record_id,
                    status
                ) VALUES (?, ?, ?)
                """,
                [
                    ("demo-batch", "R-001", "pending"),
                    ("demo-batch", "R-002", "accepted"),
                ],
            )
    except BaseException:
        DB_PATH.unlink(missing_ok=True)
        raise

    print(
        json.dumps(
            {
                "backup": str(DB_PATH),
                "sha256": file_sha256(DB_PATH),
            },
            ensure_ascii=False,
            separators=(",", ":"),
        )
    )


if __name__ == "__main__":
    main()

順番に実行する

以下はBashでの実行例です。最初に動作確認用SQLiteを作り、そのSHA-256をJSONへ保存します。

python make_demo_backup.py > demo-result.json
python -m json.tool demo-result.json

続いてJSONからSHA-256を取り出し、復元訓練を実行します。

EXPECTED_SHA256=$(python -c 'import json,pathlib; data=json.loads(pathlib.Path("demo-result.json").read_text(encoding="utf-8")); print(data["sha256"])')
python restore_drill.py --backup queue-backup.sqlite --expected-sha256 "$EXPECTED_SHA256" --policy recovery-policy.json --output-dir drill-20261007

成功時の終了コードは0です。標準出力には、実際のハッシュ、サイズ、検査結果を含む1行のJSONが出ます。復元先には次の2ファイルが残ります。

drill-20261007/
├── recovered.sqlite
└── report.json

レポートを読み直します。

python -m json.tool drill-20261007/report.json

statusがpassedであることに加え、sqlite_user_version、テーブルごとのrow_count、検査した必須列数を確認します。

失敗条件と終了コードを確認する

終了コード 条件 後続処理
0 すべての検査に合格 隔離環境での読み取り試験へ進める
2 引数、ポリシー、入力パス、復元先の問題 設定を修正する
3 コピー後のファイルのSHA-256が期待値と不一致 バックアップとハッシュの取得元を確認する
4 SQLite、外部キー、スキーマ、行数の検査に不合格 復元物を使用しない
5 ファイル操作または予期しない実行エラー 空き容量、権限、パス、実行環境を確認する
130 Ctrl+Cによる中断 残存ディレクトリを確認してから再実行する

動作確認では、次の条件を個別に試せます。試すたびに新しい復元先を指定します。

  • 表示されたSHA-256を1文字変更すると終了コード3になる
  • ポリシーのmin_rowsとmax_rowsをともに3へ変更すると終了コード4になる
  • 成功済みのdrill-20261007を再指定すると終了コード2になる
  • 存在しない列をrequired_columnsへ追加すると終了コード4になる

スクリプトが捕捉した失敗では、自身が作成した既知のファイルと、空になった復元先ディレクトリの削除を試みます。強制終了、停電、ファイルシステム障害、削除権限の問題がある場合は途中生成物が残り得ます。既存の復元先は再利用せず、内容を確認してから別の名前でやり直します。

実データへ適用するときの注意点

  • 期待するSHA-256は、バックアップ本体とは別に保護したマニフェストから取得する
  • バックアップと期待値を同時に書き換えられる環境では、SHA-256だけを改ざん対策にしない
  • 行数範囲は現在の本番件数ではなく、対象バックアップの復旧時点に対して決める
  • COUNT(*)を使う行数検査は、大きなテーブルでの実行時間を事前に測る
  • 復元先にはバックアップと同程度以上の空き容量を確保する
  • mode=0o700だけに依存せず、OSとファイルシステムの権限やACLも確認する
  • 同じOS利用者による競合操作や、突然の電源断に対するファイルシステム上の完全性までは保証しない

integrity_checkはSQLiteの構造を検査しますが、外部キー違反は検査しません。そのため、foreign_key_checkを別に実行しています。どちらも、レビュー内容の正しさや、必要な時点までデータが含まれていることまでは保証しません。

次の実務作業は、合格したrecovered.sqliteを隔離したアプリケーションへ読み取り専用で接続し、担当者が使う一覧表示や件数集計を確認することです。本番ファイルとの切り替えは、サービス停止、権限、切り戻し条件を含む別の手順として管理します。

一次資料

第165回 AIレビューキューを一貫した状態で退避する:PythonとSQLiteで作る期限・検証付きバックアップ

第164回では、AIレビュー項目を期限付きで担当者へ割り当てる仕組みを作りました。レビューキューが運用データになると、次に必要なのは障害や操作ミスに備えたバックアップです。

SQLiteのデータベースを、稼働中に通常のファイルコピーだけで退避するのは避けます。WALモードでは確定済みの変更が別のファイルに残っている場合があり、コピー中に書き込みが発生する可能性もあるためです。

今回はPython標準ライブラリのsqlite3.Connection.backup()を使い、SQLiteが管理する整合したスナップショットを新しいファイルへ保存します。保存後にはデータベース構造、外部キー、SHA-256を検査し、結果をマニフェストへ記録します。特定のレビューキューの列構成には依存しません。

Connection.backup()は、ロックや並行更新が続くと再試行を繰り返し、完了しない可能性があります。接続時のtimeout=5.0はバックアップ処理全体の期限ではありません。この実装では、必須引数--deadline-secondsとprogressコールバック、単調時計を使って経過時間を確認し、期限を超えた実行を例外で終了させます。

今回固定するバックアップ契約

項目 仕様
入力 ローカルディスク上に存在するSQLiteのメインデータベース
入力の扱い mode=roで開き、元データベースへ書き込まない
出力 新しいSQLiteファイル。既存ファイルは上書きしない
完了期限 --deadline-secondsで正の有限秒数を必ず指定し、超過をprogressコールバックで検出する
マニフェスト 出力ファイル名、SHA-256、サイズ、ページ情報、検査結果を隣接するJSONへ保存する
検査 integrity_check(1)とforeign_key_checkを実行する
検証 verifyコマンドでハッシュ、サイズ、ページ情報、SQLiteの検査をやり直す
対象外 添付データベース、外部ファイル、未確定トランザクション、業務上の値の正しさ

integrity_check(1)の引数1は、構造上の問題を最初の1件まで報告する指定です。外部キー違反はこの検査だけでは確認できないため、foreign_key_checkも別に実行します。

SHA-256は転送ミスや意図しない変更の検出には使えますが、マニフェスト自体も同時に書き換えられる環境では改ざんの証明になりません。攻撃者による変更まで検出する必要がある場合は、ハッシュをバックアップとは別のアクセス管理下に保管する設計が必要です。

バックアップCLIを実装する

Python 3.10以上を前提とします。次のコードをsqlite_snapshot.pyとして保存してください。バックアップ先とマニフェストは、作成できる権限を持つ既存ディレクトリに指定します。

from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import sqlite3
import sys
import time
from contextlib import closing
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


FORMAT = 'sqlite-snapshot-v1'
MANIFEST_LIMIT = 16 * 1024


class ToolError(Exception):
    def __init__(
        self, message: str, exit_code: int, error: str
    ) -> None:
        super().__init__(message)
        self.exit_code = exit_code
        self.error = error


def emit(payload: dict[str, Any], stream: Any) -> None:
    print(
        json.dumps(
            payload, ensure_ascii=False, separators=(',', ':')
        ),
        file=stream,
    )


def positive_finite_seconds(value: str) -> float:
    try:
        seconds = float(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            '正の有限秒数を指定してください'
        ) from exc
    if not math.isfinite(seconds) or seconds <= 0:
        raise argparse.ArgumentTypeError(
            '正の有限秒数を指定してください'
        )
    return seconds


def resolve_existing_file(path: Path) -> Path:
    try:
        resolved = path.expanduser().resolve(strict=True)
    except FileNotFoundError as exc:
        raise ToolError(
            f'ファイルが存在しません: {path}', 2, 'input_error'
        ) from exc
    except OSError as exc:
        raise ToolError(
            f'パスを確認できません: {path}', 3, 'io_error'
        ) from exc

    if not resolved.is_file():
        raise ToolError(
            f'通常ファイルではありません: {resolved}',
            2,
            'input_error',
        )
    return resolved


def resolve_new_file(path: Path) -> Path:
    candidate = path.expanduser()
    try:
        if candidate.exists() or candidate.is_symlink():
            raise ToolError(
                f'既存ファイルです: {candidate}', 2, 'input_error'
            )
        resolved = candidate.resolve(strict=False)
    except ToolError:
        raise
    except OSError as exc:
        raise ToolError(
            f'出力パスを確認できません: {candidate}',
            3,
            'io_error',
        ) from exc

    if not resolved.parent.is_dir():
        raise ToolError(
            f'出力ディレクトリが存在しません: {resolved.parent}',
            2,
            'input_error',
        )
    if resolved.exists():
        raise ToolError(
            f'既存ファイルです: {resolved}', 2, 'input_error'
        )
    return resolved


def read_only_connection(path: Path) -> sqlite3.Connection:
    uri = path.as_uri() + '?mode=ro'
    return sqlite3.connect(uri, uri=True, timeout=5.0)


def inspect_database(path: Path) -> dict[str, int]:
    with closing(read_only_connection(path)) as database:
        database.execute('PRAGMA query_only = ON')

        integrity = database.execute(
            'PRAGMA integrity_check(1)'
        ).fetchone()
        if integrity != ('ok',):
            raise ToolError(
                'SQLiteの整合性検査に失敗しました',
                4,
                'verification_failed',
            )

        foreign_key_problem = database.execute(
            'PRAGMA foreign_key_check'
        ).fetchone()
        if foreign_key_problem is not None:
            raise ToolError(
                '外部キー違反を検出しました',
                4,
                'verification_failed',
            )

        return {
            'page_count': int(
                database.execute('PRAGMA page_count').fetchone()[0]
            ),
            'page_size': int(
                database.execute('PRAGMA page_size').fetchone()[0]
            ),
            'user_version': int(
                database.execute('PRAGMA user_version').fetchone()[0]
            ),
        }


def file_signature(
    value: os.stat_result,
) -> tuple[int, int, int, int]:
    return (
        value.st_dev,
        value.st_ino,
        value.st_size,
        value.st_mtime_ns,
    )


def hash_file(
    path: Path,
) -> tuple[str, int, tuple[int, int, int, int]]:
    digest = hashlib.sha256()
    with path.open('rb') as handle:
        before = file_signature(os.fstat(handle.fileno()))
        while True:
            chunk = handle.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
        after = file_signature(os.fstat(handle.fileno()))

    if before != after:
        raise ToolError(
            'ハッシュ計算中にファイルが変更されました',
            4,
            'verification_failed',
        )
    return digest.hexdigest(), after[2], after


def create_exclusive_file(path: Path) -> int:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    try:
        return os.open(path, flags, 0o600)
    except FileExistsError as exc:
        raise ToolError(
            f'既存ファイルです: {path}', 2, 'input_error'
        ) from exc
    except OSError as exc:
        raise ToolError(
            f'ファイルを作成できません: {path}', 3, 'io_error'
        ) from exc


def cleanup_created(paths: list[Path]) -> None:
    for path in reversed(paths):
        try:
            path.unlink()
        except FileNotFoundError:
            pass
        except OSError:
            pass


def copy_snapshot(
    source: Path,
    destination: Path,
    deadline_seconds: float,
) -> None:
    started = time.monotonic()

    def check_deadline(
        status: int, remaining: int, total: int
    ) -> None:
        del status, remaining, total
        if time.monotonic() - started >= deadline_seconds:
            raise ToolError(
                'バックアップが完了期限を超えました',
                5,
                'backup_deadline_exceeded',
            )

    with closing(read_only_connection(source)) as source_database:
        source_database.execute('PRAGMA query_only = ON')
        with closing(
            sqlite3.connect(destination, timeout=5.0)
        ) as destination_database:
            destination_database.execute('PRAGMA synchronous = FULL')
            source_database.backup(
                destination_database,
                pages=256,
                progress=check_deadline,
                sleep=0.1,
            )
            destination_database.commit()

            journal_mode = destination_database.execute(
                'PRAGMA journal_mode = DELETE'
            ).fetchone()
            if (
                journal_mode is None
                or str(journal_mode[0]).lower() != 'delete'
            ):
                raise ToolError(
                    '出力をDELETEジャーナルへ変更できませんでした',
                    3,
                    'sqlite_error',
                )
            destination_database.commit()


def create_backup(
    raw_source: Path,
    raw_destination: Path,
    deadline_seconds: float,
) -> dict[str, Any]:
    source = resolve_existing_file(raw_source)
    destination = resolve_new_file(raw_destination)
    manifest = resolve_new_file(
        Path(str(destination) + '.manifest.json')
    )

    if source == destination or source == manifest:
        raise ToolError(
            '入力と出力には別のパスを指定してください',
            2,
            'input_error',
        )

    created_paths: list[Path] = []
    try:
        destination_fd = create_exclusive_file(destination)
        created_paths.append(destination)
        os.close(destination_fd)

        copy_snapshot(source, destination, deadline_seconds)
        database_info = inspect_database(destination)
        digest, size_bytes, _ = hash_file(destination)

        document = {
            'format': FORMAT,
            'created_at_utc': datetime.now(
                timezone.utc
            ).strftime('%Y-%m-%dT%H:%M:%SZ'),
            'backup_name': destination.name,
            'sha256': digest,
            'size_bytes': size_bytes,
            **database_info,
            'checks': {
                'integrity_check': 'ok',
                'foreign_key_check': 'ok',
            },
        }

        manifest_fd = create_exclusive_file(manifest)
        created_paths.append(manifest)
        with os.fdopen(
            manifest_fd, 'w', encoding='utf-8'
        ) as handle:
            json.dump(
                document,
                handle,
                ensure_ascii=False,
                sort_keys=True,
                separators=(',', ':'),
            )
            handle.write(chr(10))
            handle.flush()
            os.fsync(handle.fileno())

        return {
            'ok': True,
            'action': 'backup',
            'backup': str(destination),
            'manifest': str(manifest),
            'sha256': digest,
        }
    except BaseException:
        cleanup_created(created_paths)
        raise


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f'重複キーです: {key}')
        result[key] = value
    return result


def reject_constant(value: str) -> Any:
    raise ValueError(f'許可しない数値定数です: {value}')


def read_manifest(path: Path) -> dict[str, Any]:
    if path.stat().st_size > MANIFEST_LIMIT:
        raise ToolError(
            'マニフェストが上限サイズを超えています',
            4,
            'verification_failed',
        )

    try:
        with path.open('r', encoding='utf-8') as handle:
            document = json.load(
                handle,
                object_pairs_hook=unique_object,
                parse_constant=reject_constant,
            )
    except (UnicodeError, json.JSONDecodeError, ValueError) as exc:
        raise ToolError(
            'マニフェストを解釈できません',
            4,
            'verification_failed',
        ) from exc

    if type(document) is not dict:
        raise ToolError(
            'マニフェストの最上位はオブジェクトではありません',
            4,
            'verification_failed',
        )

    expected_fields = {
        'format',
        'created_at_utc',
        'backup_name',
        'sha256',
        'size_bytes',
        'page_count',
        'page_size',
        'user_version',
        'checks',
    }
    if set(document) != expected_fields:
        raise ToolError(
            'マニフェストの項目が契約と一致しません',
            4,
            'verification_failed',
        )
    if document['format'] != FORMAT:
        raise ToolError(
            '未対応のマニフェスト形式です',
            4,
            'verification_failed',
        )
    if type(document['backup_name']) is not str:
        raise ToolError(
            'backup_nameの型が不正です', 4, 'verification_failed'
        )
    if (
        type(document['sha256']) is not str
        or re.fullmatch('[0-9a-f]{64}', document['sha256']) is None
    ):
        raise ToolError(
            'sha256の形式が不正です', 4, 'verification_failed'
        )

    for key in ('size_bytes', 'page_count', 'page_size', 'user_version'):
        if type(document[key]) is not int:
            raise ToolError(
                f'{key}の型が不正です', 4, 'verification_failed'
            )
    if (
        document['size_bytes'] < 0
        or document['page_count'] < 0
        or document['page_size'] <= 0
    ):
        raise ToolError(
            'マニフェストの数値範囲が不正です',
            4,
            'verification_failed',
        )
    if document['checks'] != {
        'integrity_check': 'ok',
        'foreign_key_check': 'ok',
    }:
        raise ToolError(
            'マニフェストの検査結果が不正です',
            4,
            'verification_failed',
        )
    if type(document['created_at_utc']) is not str:
        raise ToolError(
            'created_at_utcの型が不正です',
            4,
            'verification_failed',
        )
    try:
        datetime.strptime(
            document['created_at_utc'], '%Y-%m-%dT%H:%M:%SZ'
        )
    except ValueError as exc:
        raise ToolError(
            'created_at_utcの形式が不正です',
            4,
            'verification_failed',
        ) from exc

    return document


def verify_backup(raw_backup: Path) -> dict[str, Any]:
    backup = resolve_existing_file(raw_backup)
    manifest = resolve_existing_file(
        Path(str(backup) + '.manifest.json')
    )
    document = read_manifest(manifest)

    if document['backup_name'] != backup.name:
        raise ToolError(
            'バックアップ名がマニフェストと一致しません',
            4,
            'verification_failed',
        )

    digest, size_bytes, signature = hash_file(backup)
    if digest != document['sha256'] or size_bytes != document['size_bytes']:
        raise ToolError(
            'ファイルのハッシュまたはサイズが一致しません',
            4,
            'verification_failed',
        )

    database_info = inspect_database(backup)
    for key in ('page_count', 'page_size', 'user_version'):
        if database_info[key] != document[key]:
            raise ToolError(
                f'{key}がマニフェストと一致しません',
                4,
                'verification_failed',
            )

    if file_signature(backup.stat()) != signature:
        raise ToolError(
            '検証中にバックアップが変更されました',
            4,
            'verification_failed',
        )

    return {
        'ok': True,
        'action': 'verify',
        'backup': str(backup),
        'manifest': str(manifest),
        'sha256': digest,
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description='SQLiteを新規ファイルへバックアップして検証する'
    )
    subparsers = parser.add_subparsers(dest='command', required=True)

    backup_parser = subparsers.add_parser(
        'backup', help='バックアップとマニフェストを作成する'
    )
    backup_parser.add_argument(
        '--deadline-seconds',
        type=positive_finite_seconds,
        required=True,
        help='バックアップ処理の完了期限を秒数で指定する',
    )
    backup_parser.add_argument('source', type=Path)
    backup_parser.add_argument('destination', type=Path)
    backup_parser.set_defaults(
        handler=lambda args: create_backup(
            args.source,
            args.destination,
            args.deadline_seconds,
        )
    )

    verify_parser = subparsers.add_parser(
        'verify', help='バックアップを再検証する'
    )
    verify_parser.add_argument('backup', type=Path)
    verify_parser.set_defaults(
        handler=lambda args: verify_backup(args.backup)
    )
    return parser


def main() -> int:
    args = build_parser().parse_args()
    try:
        result = args.handler(args)
    except ToolError as exc:
        emit(
            {
                'ok': False,
                'error': exc.error,
                'message': str(exc),
            },
            sys.stderr,
        )
        return exc.exit_code
    except KeyboardInterrupt:
        emit(
            {
                'ok': False,
                'error': 'interrupted',
                'message': '処理を中断しました',
            },
            sys.stderr,
        )
        return 130
    except sqlite3.Error as exc:
        emit(
            {
                'ok': False,
                'error': 'sqlite_error',
                'message': str(exc),
            },
            sys.stderr,
        )
        return 3
    except OSError as exc:
        emit(
            {
                'ok': False,
                'error': 'io_error',
                'message': str(exc),
            },
            sys.stderr,
        )
        return 3
    except Exception as exc:
        emit(
            {
                'ok': False,
                'error': 'internal_error',
                'message': type(exc).__name__,
            },
            sys.stderr,
        )
        return 3

    emit(result, sys.stdout)
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

出力先は先に排他的に作成するため、同名ファイルが存在すれば停止します。作成途中で通常の例外、期限超過、Ctrl+Cが発生した場合は、この実行で作った出力だけを削除します。ただし、電源断やプロセスの強制終了では途中ファイルが残る可能性があります。成功時のJSONとマニフェストがそろうまでは、完成したバックアップとして扱わないでください。

完了期限の扱い

timeout=5.0は、SQLite接続のロック待ちに関する設定であり、backup()全体を5秒で打ち切る設定ではありません。バックアップAPIはSQLITE_BUSYまたはSQLITE_LOCKEDを受けると、指定したsleepの後で再試行します。並行更新による再開始が続く場合も、処理が完了しない可能性があります。

このコードはtime.monotonic()で経過時間を測り、バックアップAPIからprogressコールバックが呼ばれた時点で期限超過を検出します。検出するとToolErrorを送出し、終了コード5を返して、この実行で作成した出力を削除します。

期限の確認はprogressコールバックが呼ばれるタイミングで行うため、指定秒数ちょうどにプロセスを強制停止する仕組みではありません。プロセスを厳密な時刻で終了させる必要がある運用では、この内部期限に加えて、実行環境側でも監視時間を定めます。

再現用の小さなデータベースを作る

動作確認用として、個人情報を含まない小さなレビューキューを作ります。固定した時刻はテストデータであり、現在時刻を表すものではありません。次をmake_demo_db.pyとして保存します。

import json
import os
import sqlite3
from contextlib import closing
from pathlib import Path


path = Path('review-demo.sqlite3')
try:
    file_descriptor = os.open(
        path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
    )
except FileExistsError:
    raise SystemExit('review-demo.sqlite3 はすでに存在します') from None

os.close(file_descriptor)

try:
    with closing(sqlite3.connect(path)) as database:
        database.execute('PRAGMA foreign_keys = ON')
        database.executescript(
            '''
            CREATE TABLE review_items (
                item_id INTEGER PRIMARY KEY,
                record_id TEXT NOT NULL UNIQUE,
                status TEXT NOT NULL CHECK (
                    status IN (
                        'pending', 'accepted', 'corrected', 'rejected'
                    )
                )
            );

            CREATE TABLE review_events (
                event_id INTEGER PRIMARY KEY,
                item_id INTEGER NOT NULL
                    REFERENCES review_items(item_id),
                event_type TEXT NOT NULL CHECK (
                    event_type IN ('enqueued', 'decided')
                ),
                occurred_at INTEGER NOT NULL
            );

            PRAGMA user_version = 1;
            '''
        )
        database.executemany(
            'INSERT INTO review_items VALUES (?, ?, ?)',
            [
                (1, 'R-001', 'pending'),
                (2, 'R-002', 'accepted'),
            ],
        )
        database.executemany(
            'INSERT INTO review_events VALUES (?, ?, ?, ?)',
            [
                (1, 1, 'enqueued', 1700000000),
                (2, 2, 'enqueued', 1700000000),
                (3, 2, 'decided', 1700000060),
            ],
        )
        database.commit()
except BaseException:
    path.unlink(missing_ok=True)
    raise

print(
    json.dumps(
        {'ok': True, 'database': str(path)},
        ensure_ascii=False,
        separators=(',', ':'),
    )
)

実行順を確認する

  1. Pythonのバージョンを確認します。
python --version
  1. デモ用データベースを作ります。
python make_demo_db.py
  1. 空の出力ディレクトリを作り、バックアップします。次の30秒は動作確認用の指定です。実運用ではデータベースの大きさ、書き込み状況、失敗を検知するまでの許容時間に合わせて明示的に決めます。
mkdir backup-out
python sqlite_snapshot.py backup --deadline-seconds 30 review-demo.sqlite3 backup-out/review-demo.sqlite3

成功時は標準出力へ1行のJSONを返し、終了コードは0になります。backup、manifest、sha256には、その実行で生成した値が入ります。SQLiteのバージョンやファイル内容によってハッシュは変わるため、記事内では固定値を示しません。

  1. 作成した成果物を別の読み取りとして再検証します。
python sqlite_snapshot.py verify backup-out/review-demo.sqlite3

verifyは、隣接するreview-demo.sqlite3.manifest.jsonを自動的に読みます。バックアップとマニフェストの名前を個別に変更すると対応関係を確認できなくなるため、2ファイルを同じ名前の組として扱います。

  1. 同じ出力先でもう一度バックアップし、上書きされないことを確認します。
python sqlite_snapshot.py backup --deadline-seconds 30 review-demo.sqlite3 backup-out/review-demo.sqlite3

この実行は終了コード2になり、既存のバックアップとマニフェストを変更しません。新しいバックアップには、日時や連番を含む別名を運用側で指定します。

終了コードと後続処理

終了コード 意味 対応
0 バックアップまたは検証が完了 成果物を保管工程へ渡す
2 入力パス、出力先、期限、既存ファイルなどの指定不備 引数と保存先を確認する
3 ファイル操作またはSQLite処理の失敗 権限、空き容量、ロック状況、元DBを確認する
4 ハッシュ、構造、外部キー、マニフェストの検証失敗 成果物を使用せず、原因を調査する
5 バックアップの完了期限を超過 成果物を使用せず、書き込み状況と期限設定を確認する
130 Ctrl+Cによる中断 途中ファイルの有無を確認して新しい名前で再実行する

argparseが検出する引数不足、正でない期限、有限でない期限、未知のオプションは、標準の使用方法を標準エラー出力へ表示して終了コード2を返します。それ以外の実行結果は、標準出力または標準エラー出力へ1行のJSONとして返します。

この実装で保証しないこと

  • バックアップ中に確定した書き込みはSQLiteが整合する状態で処理しますが、開始時刻と完全に同一の瞬間を指定する仕組みではありません。
  • ロックや並行更新が続く場合、バックアップAPIは再試行を続ける可能性があります。この実装はprogressコールバックで期限超過を検出しますが、指定秒数ちょうどにプロセスを強制終了するものではありません。
  • 新しい接続のmainデータベースだけを対象とします。別ファイルとして管理する添付DB、CSV、AIの元レスポンス、添付資料は個別に退避します。
  • 構造検査と外部キー検査は、レビュー判断や業務ルールの正しさを証明しません。必要な件数や状態別件数は、復旧訓練でアプリケーション固有の条件として確認します。
  • 暗号化は行いません。出力時に可能な環境では所有者だけが読み書きできるモードで作成しますが、OSや保存先のアクセス制御も別に設定します。
  • バックアップとマニフェストの作成は、2ファイルをまたぐ単一の原子的操作ではありません。処理中の出力先を別プロセスから読み書きしない運用が前提です。

実運用へ入れる前の確認

  • 元データベースと異なる障害範囲の保存先へ、バックアップとマニフェストを組で保管する
  • バックアップの内部期限と、必要であれば実行環境側の監視時間を決める
  • 保存期間、実行頻度、担当者、失敗時の連絡先を決める
  • verifyの成功だけで終わらせず、新しいテスト用パスで定期的に復旧確認を行う
  • 復旧確認ではレビュー総数、未確定件数、確定済み件数など、業務側の期待値も照合する
  • 稼働中の正本をこのCLIから自動的に置き換えない

期限付きロックやレビュー結果を安全に運用するには、処理コードだけでなく、終了しない処理を検知する仕組みと、壊れたときに戻せる手順が必要です。まずはデモDBでバックアップと検証を通し、その後に完了期限、権限、保存先、復旧時の業務確認を含めた手順へ落とし込んでください。

第164回 AIレビューを期限付きで割り当てる:PythonとSQLiteで作る担当ロック

第163回では、AIの要確認結果をSQLiteへ取り込み、担当者が採用・修正・却下を記録するレビューキューを作りました。担当者が複数になると、同じ項目を重複して処理しない仕組みが必要です。

今回は、レビュー項目を一定時間だけ担当者へ割り当てる期限付きロックを実装します。有効期限内は別の担当者による取得を拒否し、失効後は新しい担当者が再取得できます。古いトークンからの確定操作も拒否します。

Python標準ライブラリだけを使い、AIサービスへの通信や元データの更新は行いません。第163回のデータベースへ直接適用する移行スクリプトではないため、動作確認には新しいデータベースを使ってください。

今回実装する契約

項目 仕様
取得順 未確定かつ未取得、またはロックが失効した項目を、登録時刻、バッチID、レコードIDの順で1件取得する
ロック時間 30秒以上3600秒以下の整数。現在時刻を整数秒へ切り捨てるため、実際の有効時間は指定値より最大1秒未満短くなり得る
確定条件 有効期限内のロックトークンを持つこと
確定結果 accepted、corrected、rejectedのいずれか
時刻 time.time()から得たUnix秒を整数で保存する
出力 ヘルプを除き、標準出力または標準エラー出力へ1行のJSONを返す
対象環境 同じ端末上の少数プロセスが、ローカルディスク上のSQLiteを共有する構成

reviewerは担当者名の記録であり、本人確認ではありません。トークンも認証やアクセス制御の代わりにはなりません。OSのファイル権限や利用者認証は別に用意します。

期限付きロックの状態変化

現在の状態 操作 結果
未取得 claim 担当者、トークン、有効期限を記録する
有効な取得中 renew 現在時刻を基準に有効期限を設定し直す
有効な取得中 release 確定せずロックを解放する
有効な取得中 decide レビュー結果を確定し、ロック情報を消去する
失効済み 別の担当者がclaim 新しいトークンで再取得する
再取得後 古いトークンでdecide 確定を拒否する

レビュー用CLIを作る

次のコードをreview_leases.pyとして保存します。Python 3.10以上を前提とします。

from __future__ import annotations

import argparse
import json
import secrets
import sqlite3
import sys
import time
from contextlib import closing
from pathlib import Path
from typing import Any


SCHEMA_VERSION = 1
MIN_LEASE_SECONDS = 30
MAX_LEASE_SECONDS = 3600
MAX_CLAIM_FILE_BYTES = 16_384


class AppError(Exception):
    def __init__(self, code: str, message: str, exit_code: int = 2):
        super().__init__(message)
        self.code = code
        self.message = message
        self.exit_code = exit_code


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str):
        raise AppError('argument_error', message)


def emit(payload: dict[str, Any], *, error: bool = False) -> None:
    stream = sys.stderr if error else sys.stdout
    print(
        json.dumps(payload, ensure_ascii=False, separators=(',', ':')),
        file=stream,
    )


def lease_seconds(value: str) -> int:
    try:
        seconds = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError('整数を指定してください') from exc

    if not MIN_LEASE_SECONDS <= seconds <= MAX_LEASE_SECONDS:
        raise argparse.ArgumentTypeError(
            f'{MIN_LEASE_SECONDS}以上{MAX_LEASE_SECONDS}以下で指定してください'
        )
    return seconds


def validate_text(value: str, field: str, max_length: int) -> str:
    if not value or value != value.strip():
        raise AppError('invalid_text', f'{field}は前後空白のない文字列にしてください')
    if len(value) > max_length:
        raise AppError('invalid_text', f'{field}は{max_length}文字以内にしてください')
    if any(ord(char) < 32 or ord(char) == 127 for char in value):
        raise AppError('invalid_text', f'{field}にASCII制御文字は使用できません')
    return value


def validate_token(value: str) -> str:
    valid_chars = all(
        char.isascii() and (char.isalnum() or char in '-_')
        for char in value
    )
    if not 20 <= len(value) <= 200 or not valid_chars:
        raise AppError('invalid_token', 'ロックトークンの形式が正しくありません')
    return value


def configure(connection: sqlite3.Connection) -> None:
    connection.row_factory = sqlite3.Row
    connection.execute('PRAGMA foreign_keys = ON')
    connection.execute('PRAGMA busy_timeout = 5000')


def connect_new(path: Path) -> sqlite3.Connection:
    if path.exists():
        raise AppError('database_exists', '初期化先がすでに存在します')
    connection = sqlite3.connect(
        str(path), timeout=5.0, isolation_level=None
    )
    configure(connection)
    return connection


def connect_existing(path: Path) -> sqlite3.Connection:
    if not path.is_file():
        raise AppError('database_not_found', 'データベースが見つかりません')
    uri = path.resolve().as_uri() + '?mode=rw'
    connection = sqlite3.connect(
        uri, uri=True, timeout=5.0, isolation_level=None
    )
    configure(connection)
    return connection


def verify_schema(connection: sqlite3.Connection) -> None:
    version = connection.execute('PRAGMA user_version').fetchone()[0]
    if version != SCHEMA_VERSION:
        raise AppError(
            'unsupported_schema',
            f'対応していないスキーマです: {version}',
        )


def load_claim_token(args: argparse.Namespace) -> str:
    if args.token is not None:
        return validate_token(args.token)

    try:
        with args.claim_json.open('rb') as source:
            data = source.read(MAX_CLAIM_FILE_BYTES + 1)
    except OSError as exc:
        raise AppError(
            'claim_file_error',
            f'取得結果を読めません: {exc}',
        ) from exc

    if len(data) > MAX_CLAIM_FILE_BYTES:
        raise AppError('claim_file_error', '取得結果のファイルが大きすぎます')

    try:
        payload = json.loads(data.decode('utf-8'))
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise AppError(
            'claim_file_error',
            f'取得結果がUTF-8のJSONではありません: {exc}',
        ) from exc

    if not isinstance(payload, dict) or not isinstance(
        payload.get('claim_token'), str
    ):
        raise AppError(
            'claim_file_error',
            '取得結果にclaim_tokenがありません',
        )
    return validate_token(payload['claim_token'])


CREATE_CATEGORY_SQL = '''
CREATE TABLE allowed_category (
    category TEXT PRIMARY KEY,
    CHECK(length(trim(category)) > 0)
)
'''

CREATE_REVIEW_SQL = '''
CREATE TABLE review_item (
    batch_id TEXT NOT NULL,
    record_id TEXT NOT NULL,
    proposed_category TEXT NOT NULL
        REFERENCES allowed_category(category),
    reason TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending'
        CHECK(status IN ('pending', 'accepted', 'corrected', 'rejected')),
    final_category TEXT REFERENCES allowed_category(category),
    reviewed_by TEXT,
    decided_at INTEGER,
    created_at INTEGER NOT NULL,
    claim_token TEXT UNIQUE,
    claimed_by TEXT,
    claimed_at INTEGER,
    claim_expires_at INTEGER,
    PRIMARY KEY(batch_id, record_id),
    CHECK(length(batch_id) BETWEEN 1 AND 100),
    CHECK(length(record_id) BETWEEN 1 AND 100),
    CHECK(length(reason) BETWEEN 1 AND 1000),
    CHECK(
        (status = 'pending'
            AND final_category IS NULL
            AND reviewed_by IS NULL
            AND decided_at IS NULL)
        OR
        (status IN ('accepted', 'corrected')
            AND final_category IS NOT NULL
            AND reviewed_by IS NOT NULL
            AND decided_at IS NOT NULL)
        OR
        (status = 'rejected'
            AND final_category IS NULL
            AND reviewed_by IS NOT NULL
            AND decided_at IS NOT NULL)
    ),
    CHECK(status <> 'accepted' OR final_category = proposed_category),
    CHECK(status <> 'corrected' OR final_category <> proposed_category),
    CHECK(
        (claim_token IS NULL
            AND claimed_by IS NULL
            AND claimed_at IS NULL
            AND claim_expires_at IS NULL)
        OR
        (status = 'pending'
            AND claim_token IS NOT NULL
            AND claimed_by IS NOT NULL
            AND claimed_at IS NOT NULL
            AND claim_expires_at > claimed_at)
    )
)
'''

CREATE_INDEX_SQL = '''
CREATE INDEX review_item_claim_order
ON review_item(status, claim_expires_at, created_at, batch_id, record_id)
'''


def cmd_init(args: argparse.Namespace) -> dict[str, Any]:
    categories = [
        validate_text(item, 'category', 50) for item in args.category
    ]
    if len(categories) != len(set(categories)):
        raise AppError('duplicate_category', 'カテゴリが重複しています')

    with closing(connect_new(args.db)) as connection:
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            connection.execute(CREATE_CATEGORY_SQL)
            connection.execute(CREATE_REVIEW_SQL)
            connection.execute(CREATE_INDEX_SQL)
            connection.executemany(
                'INSERT INTO allowed_category(category) VALUES (?)',
                [(category,) for category in categories],
            )
            connection.execute(f'PRAGMA user_version = {SCHEMA_VERSION}')

    return {
        'database': str(args.db),
        'category_count': len(categories),
    }


def cmd_add(args: argparse.Namespace) -> dict[str, Any]:
    batch_id = validate_text(args.batch_id, 'batch_id', 100)
    record_id = validate_text(args.record_id, 'record_id', 100)
    category = validate_text(args.proposed_category, 'proposed_category', 50)
    reason = validate_text(args.reason, 'reason', 1000)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            if connection.execute(
                'SELECT 1 FROM allowed_category WHERE category = ?',
                (category,),
            ).fetchone() is None:
                raise AppError('unknown_category', '許可されていないカテゴリです')

            if connection.execute(
                '''
                SELECT 1 FROM review_item
                WHERE batch_id = ? AND record_id = ?
                ''',
                (batch_id, record_id),
            ).fetchone() is not None:
                raise AppError('duplicate_item', 'レビュー項目がすでに存在します')

            connection.execute(
                '''
                INSERT INTO review_item(
                    batch_id, record_id, proposed_category, reason, created_at
                ) VALUES (?, ?, ?, ?, ?)
                ''',
                (batch_id, record_id, category, reason, int(time.time())),
            )

    return {'added': True, 'batch_id': batch_id, 'record_id': record_id}


def cmd_claim(args: argparse.Namespace) -> dict[str, Any]:
    reviewer = validate_text(args.reviewer, 'reviewer', 100)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            now = int(time.time())
            row = connection.execute(
                '''
                SELECT batch_id, record_id, proposed_category, reason
                FROM review_item
                WHERE status = 'pending'
                  AND (claim_token IS NULL OR claim_expires_at <= ?)
                ORDER BY created_at, batch_id, record_id
                LIMIT 1
                ''',
                (now,),
            ).fetchone()

            if row is None:
                result = {'claimed': False}
            else:
                token = secrets.token_urlsafe(24)
                expires_at = now + args.lease_seconds
                cursor = connection.execute(
                    '''
                    UPDATE review_item
                    SET claim_token = ?, claimed_by = ?,
                        claimed_at = ?, claim_expires_at = ?
                    WHERE batch_id = ? AND record_id = ?
                      AND status = 'pending'
                      AND (claim_token IS NULL OR claim_expires_at <= ?)
                    ''',
                    (
                        token,
                        reviewer,
                        now,
                        expires_at,
                        row['batch_id'],
                        row['record_id'],
                        now,
                    ),
                )
                if cursor.rowcount != 1:
                    raise AppError('claim_conflict', '取得状態が変わりました')

                result = {
                    'claimed': True,
                    'batch_id': row['batch_id'],
                    'record_id': row['record_id'],
                    'proposed_category': row['proposed_category'],
                    'reason': row['reason'],
                    'claimed_by': reviewer,
                    'claimed_at': now,
                    'claim_expires_at': expires_at,
                    'claim_token': token,
                }

    return result


def cmd_renew(args: argparse.Namespace) -> dict[str, Any]:
    token = load_claim_token(args)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            now = int(time.time())
            row = connection.execute(
                '''
                SELECT batch_id, record_id, claim_expires_at
                FROM review_item
                WHERE status = 'pending' AND claim_token = ?
                ''',
                (token,),
            ).fetchone()
            if row is None:
                raise AppError('claim_not_found', '有効な取得情報がありません')
            if row['claim_expires_at'] <= now:
                raise AppError('claim_expired', 'ロックはすでに失効しています')

            expires_at = now + args.lease_seconds
            connection.execute(
                'UPDATE review_item SET claim_expires_at = ? WHERE claim_token = ?',
                (expires_at, token),
            )

    return {
        'renewed': True,
        'batch_id': row['batch_id'],
        'record_id': row['record_id'],
        'claim_expires_at': expires_at,
    }


def cmd_release(args: argparse.Namespace) -> dict[str, Any]:
    token = load_claim_token(args)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            now = int(time.time())
            row = connection.execute(
                '''
                SELECT batch_id, record_id, claim_expires_at
                FROM review_item
                WHERE status = 'pending' AND claim_token = ?
                ''',
                (token,),
            ).fetchone()
            if row is None:
                raise AppError('claim_not_found', '有効な取得情報がありません')
            if row['claim_expires_at'] <= now:
                raise AppError('claim_expired', 'ロックはすでに失効しています')

            connection.execute(
                '''
                UPDATE review_item
                SET claim_token = NULL, claimed_by = NULL,
                    claimed_at = NULL, claim_expires_at = NULL
                WHERE claim_token = ?
                ''',
                (token,),
            )

    return {
        'released': True,
        'batch_id': row['batch_id'],
        'record_id': row['record_id'],
    }


def cmd_decide(args: argparse.Namespace) -> dict[str, Any]:
    token = load_claim_token(args)

    if args.decision == 'corrected':
        if args.category is None:
            raise AppError('category_required', 'correctedには--categoryが必要です')
        category = validate_text(args.category, 'category', 50)
    elif args.category is not None:
        raise AppError(
            'unexpected_category',
            'acceptedとrejectedでは--categoryを指定しません',
        )
    else:
        category = None

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            now = int(time.time())
            row = connection.execute(
                '''
                SELECT batch_id, record_id, proposed_category,
                       claimed_by, claim_expires_at
                FROM review_item
                WHERE status = 'pending' AND claim_token = ?
                ''',
                (token,),
            ).fetchone()
            if row is None:
                raise AppError('claim_not_found', '有効な取得情報がありません')
            if row['claim_expires_at'] <= now:
                raise AppError('claim_expired', 'ロックはすでに失効しています')

            if args.decision == 'corrected':
                if connection.execute(
                    'SELECT 1 FROM allowed_category WHERE category = ?',
                    (category,),
                ).fetchone() is None:
                    raise AppError('unknown_category', '許可されていないカテゴリです')
                if category == row['proposed_category']:
                    raise AppError(
                        'category_unchanged',
                        'correctedでは提案と異なるカテゴリを指定してください',
                    )
                final_category = category
            elif args.decision == 'accepted':
                final_category = row['proposed_category']
            else:
                final_category = None

            cursor = connection.execute(
                '''
                UPDATE review_item
                SET status = ?, final_category = ?, reviewed_by = ?,
                    decided_at = ?, claim_token = NULL, claimed_by = NULL,
                    claimed_at = NULL, claim_expires_at = NULL
                WHERE batch_id = ? AND record_id = ?
                  AND status = 'pending' AND claim_token = ?
                ''',
                (
                    args.decision,
                    final_category,
                    row['claimed_by'],
                    now,
                    row['batch_id'],
                    row['record_id'],
                    token,
                ),
            )
            if cursor.rowcount != 1:
                raise AppError('decision_conflict', '確定状態が変わりました')

    return {
        'batch_id': row['batch_id'],
        'record_id': row['record_id'],
        'status': args.decision,
        'final_category': final_category,
        'reviewed_by': row['claimed_by'],
    }


def cmd_show(args: argparse.Namespace) -> dict[str, Any]:
    batch_id = validate_text(args.batch_id, 'batch_id', 100)
    record_id = validate_text(args.record_id, 'record_id', 100)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        row = connection.execute(
            '''
            SELECT batch_id, record_id, status, proposed_category,
                   final_category, reviewed_by, claim_token,
                   claimed_by, claim_expires_at
            FROM review_item
            WHERE batch_id = ? AND record_id = ?
            ''',
            (batch_id, record_id),
        ).fetchone()

    if row is None:
        raise AppError('item_not_found', 'レビュー項目が見つかりません')

    now = int(time.time())
    if row['status'] != 'pending':
        claim_state = 'not_applicable'
    elif row['claim_token'] is None:
        claim_state = 'unclaimed'
    elif row['claim_expires_at'] <= now:
        claim_state = 'expired'
    else:
        claim_state = 'active'

    return {
        'batch_id': row['batch_id'],
        'record_id': row['record_id'],
        'status': row['status'],
        'proposed_category': row['proposed_category'],
        'final_category': row['final_category'],
        'reviewed_by': row['reviewed_by'],
        'claimed_by': row['claimed_by'],
        'claim_state': claim_state,
    }


def add_claim_source(parser: argparse.ArgumentParser) -> None:
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument('--token')
    group.add_argument('--claim-json', type=Path)


def build_parser() -> argparse.ArgumentParser:
    parser = JsonArgumentParser(description='期限付きレビューキュー')
    subparsers = parser.add_subparsers(dest='command', required=True)

    init_parser = subparsers.add_parser('init')
    init_parser.add_argument('db', type=Path)
    init_parser.add_argument('--category', action='append', required=True)
    init_parser.set_defaults(handler=cmd_init)

    add_parser = subparsers.add_parser('add')
    add_parser.add_argument('db', type=Path)
    add_parser.add_argument('--batch-id', required=True)
    add_parser.add_argument('--record-id', required=True)
    add_parser.add_argument('--proposed-category', required=True)
    add_parser.add_argument('--reason', required=True)
    add_parser.set_defaults(handler=cmd_add)

    claim_parser = subparsers.add_parser('claim')
    claim_parser.add_argument('db', type=Path)
    claim_parser.add_argument('--reviewer', required=True)
    claim_parser.add_argument(
        '--lease-seconds', type=lease_seconds, default=300
    )
    claim_parser.set_defaults(handler=cmd_claim)

    renew_parser = subparsers.add_parser('renew')
    renew_parser.add_argument('db', type=Path)
    renew_parser.add_argument(
        '--lease-seconds', type=lease_seconds, default=300
    )
    add_claim_source(renew_parser)
    renew_parser.set_defaults(handler=cmd_renew)

    release_parser = subparsers.add_parser('release')
    release_parser.add_argument('db', type=Path)
    add_claim_source(release_parser)
    release_parser.set_defaults(handler=cmd_release)

    decide_parser = subparsers.add_parser('decide')
    decide_parser.add_argument('db', type=Path)
    decide_parser.add_argument(
        '--decision',
        choices=['accepted', 'corrected', 'rejected'],
        required=True,
    )
    decide_parser.add_argument('--category')
    add_claim_source(decide_parser)
    decide_parser.set_defaults(handler=cmd_decide)

    show_parser = subparsers.add_parser('show')
    show_parser.add_argument('db', type=Path)
    show_parser.add_argument('--batch-id', required=True)
    show_parser.add_argument('--record-id', required=True)
    show_parser.set_defaults(handler=cmd_show)

    return parser


def main() -> int:
    try:
        args = build_parser().parse_args()
        result = args.handler(args)
    except AppError as exc:
        emit(
            {'ok': False, 'error': exc.code, 'message': exc.message},
            error=True,
        )
        return exc.exit_code
    except sqlite3.Error as exc:
        emit(
            {
                'ok': False,
                'error': 'database_error',
                'message': f'SQLite処理に失敗しました: {exc}',
            },
            error=True,
        )
        return 3
    except OSError as exc:
        emit(
            {
                'ok': False,
                'error': 'io_error',
                'message': f'ファイル操作に失敗しました: {exc}',
            },
            error=True,
        )
        return 3
    except KeyboardInterrupt:
        emit(
            {'ok': False, 'error': 'interrupted', 'message': '中断しました'},
            error=True,
        )
        return 130

    emit({'ok': True, **result})
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

1. データベースと確認用データを作る

既存の第163回用データベースではなく、新しいファイルを指定します。

python3 review_leases.py init review-leases.db --category 配送 --category 請求 --category その他
python3 review_leases.py add review-leases.db --batch-id batch-20261005 --record-id C-001 --proposed-category 配送 --reason '提案カテゴリを元データと照合する'

成功時の出力は次のとおりです。

{"ok":true,"database":"review-leases.db","category_count":3}
{"ok":true,"added":true,"batch_id":"batch-20261005","record_id":"C-001"}

initは既存ファイルを拒否します。カテゴリの追加や変更を行う管理コマンドは、この最小実装には含めていません。

2. レビュー項目を取得する

取得結果にはレビュー理由とロックトークンが含まれます。次の例では、未使用のファイルへ保存します。シェルのリダイレクトは既存ファイルを上書きすることがあるため、既存ファイルを出力先に指定しないでください。

python3 review_leases.py claim review-leases.db --reviewer tanaka --lease-seconds 300 > claim-tanaka.json
python3 -m json.tool claim-tanaka.json

取得に成功した場合、claim-tanaka.jsonには対象ID、提案カテゴリ、理由、担当者、取得時刻、有効期限、トークンが入ります。時刻とトークンは実行ごとに変わります。最初のコマンドの終了コードが0であり、JSONのclaimedがtrueであることを確認してから確定操作へ進みます。

この例では未確定項目が1件だけなので、有効期限内に別の担当者が取得すると、次の結果になります。

python3 review_leases.py claim review-leases.db --reviewer suzuki --lease-seconds 300
{"ok":true,"claimed":false}

claimed=falseは異常終了ではなく、現在取得できる項目がないことを表します。この結果にはトークンがないため、decideには使えません。

3. 必要なら期限を更新または解放する

レビューを続ける場合は、有効期限が切れる前にrenewを実行します。

python3 review_leases.py renew review-leases.db --claim-json claim-tanaka.json --lease-seconds 300

新しい期限は、以前の期限ではなくrenewの実行時刻を基準に計算されます。指定した秒数が現在の残り時間より短い場合は期限が前倒しされるため、通常は取得時と同じ値を指定し、失効間際になる前に更新します。更新結果のJSONに新しいclaim_expires_atが返りますが、保存済みのclaim-tanaka.json内の時刻は書き換わりません。トークンはそのまま利用できます。

確定せず別の担当者へ戻す場合はreleaseを使います。次のコマンドは、手順4とは別の分岐例です。手順4で同じ取得結果を使って確定する場合は実行しません。

python3 review_leases.py release review-leases.db --claim-json claim-tanaka.json

失効後のトークンでは更新も解放もできません。失効した項目は、次のclaimによって新しい担当者とトークンへ置き換えられます。

4. レビュー結果を確定する

提案された配送を請求へ修正する場合は、許可済みカテゴリを指定します。

python3 review_leases.py decide review-leases.db --claim-json claim-tanaka.json --decision corrected --category 請求
{"ok":true,"batch_id":"batch-20261005","record_id":"C-001","status":"corrected","final_category":"請求","reviewed_by":"tanaka"}

判定ごとの引数は次のとおりです。

  • accepted:提案カテゴリを採用する。--categoryは指定しない
  • corrected:提案と異なる許可カテゴリを--categoryで指定する
  • rejected:最終カテゴリを持たせない。--categoryは指定しない

確定後はトークンがデータベースから消えるため、同じ取得結果を再利用したdecideはclaim_not_foundで失敗します。失効後、まだ再取得されていないトークンを使った場合はclaim_expiredになります。

5. 保存状態を確認する

python3 review_leases.py show review-leases.db --batch-id batch-20261005 --record-id C-001
{"ok":true,"batch_id":"batch-20261005","record_id":"C-001","status":"corrected","proposed_category":"配送","final_category":"請求","reviewed_by":"tanaka","claimed_by":null,"claim_state":"not_applicable"}

showはトークンを表示しません。未確定項目のclaim_stateは、unclaimed、active、expiredのいずれかです。

同時取得を防ぐ仕組み

claimは、対象の検索前にBEGIN IMMEDIATEを実行します。SQLiteの書き込みトランザクションを確保してから、取得可能な行の検索と更新を同じトランザクション内で行います。このCLIだけがデータベースを更新する構成では、別プロセスが同じ行を同時に取得することを防げます。

ロック待機時間は5秒です。競合が長引いた場合はSQLiteエラーとなり、終了コード3を返すことがあります。高い並列処理性能を目的とした設計ではありません。

終了コード 意味
0 操作成功。取得対象がないclaimed=falseも含む
2 引数、カテゴリ、トークン、取得結果ファイル、有効期限、状態遷移など、入力または業務条件による拒否
3 SQLite処理、または想定外のファイル操作の失敗
130 Ctrl+Cによる中断

実運用へ入れる前の制約

  • 取得結果とデータベースにはトークンが含まれるため、アクセスできる利用者をOSの権限で限定する
  • 端末の時刻変更は有効期限の判定へ影響するため、各プロセスが同じ時刻基準を使える環境で運用する
  • プロセスが異常終了した場合、明示的な解放は行われず、有効期限までロックが残る
  • 期限切れ後に元担当者の作業画面を停止する仕組みはない。別の担当者が再取得した場合、元担当者からの確定は拒否されるが、レビュー作業自体は重複する可能性がある
  • ネットワーク共有上のSQLiteは、ファイルロックの挙動を検証できない限り使用しない
  • 多数の同時利用者、利用者認証、権限分離、通知が必要な場合は、SQLiteだけで要件を満たそうとしない
  • この実装が保存するのは現在状態だけであり、変更不能な監査ログは作らない。監査要件がある場合は、決定履歴を別の仕組みで記録する

まずはサンプル1件で、取得中の再取得拒否、期限内の確定、解放後の再取得を確認します。既存のレビューキューへ統合する場合は、データベースを退避したうえで、実際のテーブル名と制約に合わせた移行処理を別に用意してください。

第163回 AIの要確認結果を人へ回す:PythonとSQLiteで作るレビューキュー

第162回では、AIが返したJSONLの項目、型、許可カテゴリ、重複IDなどを検査しました。ただし、構造が正しいことと、判定内容が業務上正しいことは別です。needs_reviewがtrueの結果は、元データと照合してから確定します。

今回は、検証済みJSONLから要確認行だけをSQLiteへ取り込み、pending、accepted、corrected、rejectedの状態を管理します。Python標準ライブラリだけを使い、AIサービスへの通信や元データの更新は行いません。

今回作るレビューキューの契約

項目 仕様
入力 第162回と同じ4項目を持つ、BOMなしUTF-8の検証済みJSONL
登録対象 needs_reviewがtrueの行だけ
識別子 batch_idとrecord_idの組み合わせ。同じレコードを別バッチで再確認できる
カテゴリ データベース作成時に登録した許可カテゴリだけ
判定 提案を採用、許可カテゴリへ修正、または却下
再判定 このCLIでは確定済みの行を変更しない
書き込み バッチの取り込みと1件の判定を、それぞれSQLiteトランザクションで処理する
結果出力 ヘルプ表示を除き、出力に成功した場合は標準出力または標準エラー出力へ1行のJSONを返す

needs_review=falseの行は件数だけを数え、データベースには保存しません。検証済みJSONLは変更せず、別途保管します。

レビュー理由はAIの説明であり、正しさの証明ではありません。担当者はrecord_idを使い、権限のある元データと照合します。

SQLiteを使う範囲と制約

この実装は、1台の端末または少人数で扱う小規模なレビューキューを対象とします。複数人が同時に操作するWeb画面、認証、権限管理、改ざん検知は備えていません。--reviewerも利用者が入力する文字列であり、本人確認にはなりません。

  • Python 3.10以上を使用する
  • データベースはローカルディスクに置き、アクセス権とバックアップを管理する
  • 理由やレビュー注記に機密情報を入れる場合は、データベースとバックアップも同じ基準で保護する
  • カテゴリ方針を変えるときは、既存データとの整合性を確認して移行する。このスクリプトにはカテゴリ変更機能を設けない
  • 大人数での同時編集、ネットワーク共有上での利用、レビュー画面が必要な業務では、専用のデータベースとアプリケーションを検討する

1.動作確認用JSONLを用意する

次の架空データを、BOMなしUTF-8のvalid.jsonlとして保存します。1行に1つのJSONオブジェクトを書きます。

{"record_id":"C-001","category":"配送","needs_review":true,"reason":"配送先変更と到着日の質問が同時に含まれる"}
{"record_id":"C-002","category":"請求","needs_review":false,"reason":"請求書再発行の依頼が明確"}
{"record_id":"C-003","category":"配送","needs_review":true,"reason":"配送状況と請求額の両方に言及している"}

2.レビューキューを実装する

次のコードをreview_queue.pyとして保存します。

import argparse
import json
import os
import re
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path


SCHEMA_VERSION = 1
EXPECTED_KEYS = frozenset(
    {"record_id", "category", "needs_review", "reason"}
)
BATCH_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}")

CREATE_STATEMENTS = (
    """
    CREATE TABLE categories (
        category TEXT NOT NULL PRIMARY KEY,
        CHECK (length(category) > 0)
    )
    """,
    """
    CREATE TABLE review_batches (
        batch_id TEXT NOT NULL PRIMARY KEY,
        imported_at TEXT NOT NULL,
        source_name TEXT NOT NULL,
        total_rows INTEGER NOT NULL CHECK (total_rows >= 0),
        queued_rows INTEGER NOT NULL CHECK (queued_rows >= 0),
        skipped_rows INTEGER NOT NULL CHECK (skipped_rows >= 0),
        CHECK (length(batch_id) > 0),
        CHECK (length(source_name) > 0),
        CHECK (total_rows = queued_rows + skipped_rows)
    )
    """,
    """
    CREATE TABLE review_items (
        batch_id TEXT NOT NULL,
        record_id TEXT NOT NULL,
        proposed_category TEXT NOT NULL,
        reason TEXT NOT NULL,
        source_line INTEGER NOT NULL CHECK (source_line >= 1),
        status TEXT NOT NULL
            CHECK (status IN ('pending', 'accepted', 'corrected', 'rejected')),
        final_category TEXT,
        reviewer TEXT,
        review_note TEXT,
        decided_at TEXT,
        PRIMARY KEY (batch_id, record_id),
        UNIQUE (batch_id, source_line),
        CHECK (length(record_id) > 0),
        CHECK (length(reason) > 0),
        CHECK (
            (
                status = 'pending'
                AND final_category IS NULL
                AND reviewer IS NULL
                AND review_note IS NULL
                AND decided_at IS NULL
            )
            OR
            (
                status = 'accepted'
                AND final_category IS NOT NULL
                AND final_category = proposed_category
                AND reviewer IS NOT NULL
                AND length(reviewer) > 0
                AND review_note IS NOT NULL
                AND length(review_note) > 0
                AND decided_at IS NOT NULL
            )
            OR
            (
                status = 'corrected'
                AND final_category IS NOT NULL
                AND final_category <> proposed_category
                AND reviewer IS NOT NULL
                AND length(reviewer) > 0
                AND review_note IS NOT NULL
                AND length(review_note) > 0
                AND decided_at IS NOT NULL
            )
            OR
            (
                status = 'rejected'
                AND final_category IS NULL
                AND reviewer IS NOT NULL
                AND length(reviewer) > 0
                AND review_note IS NOT NULL
                AND length(review_note) > 0
                AND decided_at IS NOT NULL
            )
        ),
        FOREIGN KEY (batch_id)
            REFERENCES review_batches(batch_id) ON DELETE RESTRICT,
        FOREIGN KEY (proposed_category)
            REFERENCES categories(category) ON DELETE RESTRICT,
        FOREIGN KEY (final_category)
            REFERENCES categories(category) ON DELETE RESTRICT
    )
    """,
)


class AppError(Exception):
    def __init__(self, error, details=None):
        super().__init__(error)
        self.error = error
        self.details = details


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message):
        raise AppError("invalid_arguments", {"message": message})


def emit(stream, payload):
    line = json.dumps(
        payload,
        ensure_ascii=False,
        separators=(",", ":"),
    ) + "\n"
    stream.write(line)
    stream.flush()


def has_control_character(value):
    return any(
        ord(character) < 32 or ord(character) == 127
        for character in value
    )


def validate_plain_text(value, field):
    if (
        value == ""
        or value != value.strip()
        or has_control_character(value)
    ):
        raise AppError("invalid_text_argument", {"field": field})
    return value


def validate_batch_id(value):
    if BATCH_PATTERN.fullmatch(value) is None:
        raise AppError(
            "invalid_batch_id",
            {
                "rule": (
                    "1-64 ASCII characters; first is a letter or digit; "
                    "the rest may also include dot, underscore, or hyphen"
                )
            },
        )
    return value


def validate_categories(values):
    categories = []
    seen = set()

    for value in values:
        category = validate_plain_text(value, "category")
        if category in seen:
            raise AppError("duplicate_category")
        seen.add(category)
        categories.append(category)

    return categories


def limit_value(raw_value):
    try:
        value = int(raw_value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("limit must be an integer") from exc

    if value < 1 or value > 500:
        raise argparse.ArgumentTypeError("limit must be between 1 and 500")
    return value


def utc_now():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def connect_database(path):
    return sqlite3.connect(
        path,
        timeout=5.0,
        isolation_level=None,
    )


def configure_connection(connection):
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys = ON")
    connection.execute("PRAGMA busy_timeout = 5000")


def safe_rollback(connection):
    try:
        connection.rollback()
    except sqlite3.Error:
        pass


def open_database(path):
    if not path.is_file():
        raise AppError("database_not_found", {"db": str(path)})

    connection = connect_database(path)
    try:
        configure_connection(connection)
        version = connection.execute("PRAGMA user_version").fetchone()[0]
        if version != SCHEMA_VERSION:
            raise AppError(
                "unsupported_schema_version",
                {"expected": SCHEMA_VERSION, "actual": version},
            )
    except BaseException:
        connection.close()
        raise

    return connection


def reject_constant(_value):
    raise ValueError("non-standard JSON number")


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate JSON key")
        result[key] = value
    return result


def parse_result(raw_line, line_number):
    try:
        item = json.loads(
            raw_line,
            parse_constant=reject_constant,
            object_pairs_hook=unique_object,
        )
    except (json.JSONDecodeError, ValueError) as exc:
        raise AppError("invalid_json", {"line": line_number}) from exc

    if type(item) is not dict:
        raise AppError("json_object_required", {"line": line_number})

    keys = set(item)
    if keys != EXPECTED_KEYS:
        raise AppError(
            "invalid_fields",
            {
                "line": line_number,
                "missing": sorted(EXPECTED_KEYS - keys),
                "unknown": sorted(keys - EXPECTED_KEYS),
            },
        )

    for field in ("record_id", "category", "reason"):
        value = item[field]
        if type(value) is not str or value.strip() == "":
            raise AppError(
                "invalid_field_value",
                {"line": line_number, "field": field},
            )

    record_id = item["record_id"]
    if (
        record_id != record_id.strip()
        or has_control_character(record_id)
    ):
        raise AppError(
            "invalid_field_value",
            {"line": line_number, "field": "record_id"},
        )

    if type(item["needs_review"]) is not bool:
        raise AppError(
            "invalid_field_value",
            {"line": line_number, "field": "needs_review"},
        )

    return item


def cmd_init(args):
    categories = validate_categories(args.category)
    if args.db.exists():
        raise AppError("database_already_exists", {"db": str(args.db)})

    connection = None
    created = False

    try:
        descriptor = os.open(
            args.db,
            os.O_CREAT | os.O_EXCL | os.O_WRONLY,
            0o600,
        )
        created = True
        os.close(descriptor)

        connection = connect_database(args.db)
        configure_connection(connection)
        connection.execute("BEGIN IMMEDIATE")

        for statement in CREATE_STATEMENTS:
            connection.execute(statement)

        connection.executemany(
            "INSERT INTO categories(category) VALUES (?)",
            ((category,) for category in categories),
        )
        connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
        connection.commit()
    except BaseException:
        if connection is not None:
            safe_rollback(connection)
            connection.close()
            connection = None
        if created:
            try:
                args.db.unlink()
            except OSError:
                pass
        raise
    finally:
        if connection is not None:
            connection.close()

    return {
        "ok": True,
        "command": "init",
        "db": str(args.db),
        "schema_version": SCHEMA_VERSION,
        "category_count": len(categories),
    }


def cmd_import_results(args):
    batch_id = validate_batch_id(args.batch_id)
    if not args.input.is_file():
        raise AppError("input_not_found", {"input": str(args.input)})

    connection = open_database(args.db)
    total_rows = 0
    queued_rows = 0
    skipped_rows = 0
    seen_record_ids = {}
    imported_at = utc_now()

    try:
        connection.execute("BEGIN IMMEDIATE")

        existing_batch = connection.execute(
            "SELECT 1 FROM review_batches WHERE batch_id = ?",
            (batch_id,),
        ).fetchone()
        if existing_batch is not None:
            raise AppError("batch_already_exists", {"batch_id": batch_id})

        connection.execute(
            """
            INSERT INTO review_batches(
                batch_id, imported_at, source_name,
                total_rows, queued_rows, skipped_rows
            ) VALUES (?, ?, ?, 0, 0, 0)
            """,
            (batch_id, imported_at, args.input.name),
        )

        allowed_categories = {
            row["category"]
            for row in connection.execute("SELECT category FROM categories")
        }

        try:
            with args.input.open("r", encoding="utf-8", newline="") as source:
                for line_number, raw_line in enumerate(source, start=1):
                    total_rows += 1
                    if raw_line.strip() == "":
                        raise AppError("blank_line", {"line": line_number})

                    item = parse_result(raw_line, line_number)
                    record_id = item["record_id"]

                    if item["category"] not in allowed_categories:
                        raise AppError(
                            "category_not_allowed",
                            {"line": line_number},
                        )

                    if record_id in seen_record_ids:
                        raise AppError(
                            "duplicate_record_id",
                            {
                                "line": line_number,
                                "first_line": seen_record_ids[record_id],
                            },
                        )
                    seen_record_ids[record_id] = line_number

                    if item["needs_review"] is False:
                        skipped_rows += 1
                        continue

                    connection.execute(
                        """
                        INSERT INTO review_items(
                            batch_id, record_id, proposed_category,
                            reason, source_line, status
                        ) VALUES (?, ?, ?, ?, ?, 'pending')
                        """,
                        (
                            batch_id,
                            record_id,
                            item["category"],
                            item["reason"],
                            line_number,
                        ),
                    )
                    queued_rows += 1
        except UnicodeDecodeError as exc:
            raise AppError("input_not_utf8") from exc

        if total_rows == 0:
            raise AppError("empty_input")

        connection.execute(
            """
            UPDATE review_batches
            SET total_rows = ?, queued_rows = ?, skipped_rows = ?
            WHERE batch_id = ?
            """,
            (total_rows, queued_rows, skipped_rows, batch_id),
        )
        connection.commit()
    except BaseException:
        safe_rollback(connection)
        raise
    finally:
        connection.close()

    return {
        "ok": True,
        "command": "import-results",
        "batch_id": batch_id,
        "total_rows": total_rows,
        "queued_rows": queued_rows,
        "skipped_rows": skipped_rows,
    }


def cmd_pending(args):
    batch_id = validate_batch_id(args.batch_id)
    connection = open_database(args.db)

    try:
        connection.execute("BEGIN")
        batch = connection.execute(
            "SELECT 1 FROM review_batches WHERE batch_id = ?",
            (batch_id,),
        ).fetchone()
        if batch is None:
            raise AppError("batch_not_found", {"batch_id": batch_id})

        pending_count = connection.execute(
            """
            SELECT COUNT(*)
            FROM review_items
            WHERE batch_id = ? AND status = 'pending'
            """,
            (batch_id,),
        ).fetchone()[0]

        rows = connection.execute(
            """
            SELECT record_id, proposed_category, reason, source_line
            FROM review_items
            WHERE batch_id = ? AND status = 'pending'
            ORDER BY source_line
            LIMIT ?
            """,
            (batch_id, args.limit),
        ).fetchall()
        connection.commit()
    except BaseException:
        safe_rollback(connection)
        raise
    finally:
        connection.close()

    items = [
        {
            "record_id": row["record_id"],
            "proposed_category": row["proposed_category"],
            "reason": row["reason"],
            "source_line": row["source_line"],
        }
        for row in rows
    ]

    return {
        "ok": True,
        "command": "pending",
        "batch_id": batch_id,
        "pending_count": pending_count,
        "returned_count": len(items),
        "items": items,
    }


def cmd_decide(args):
    batch_id = validate_batch_id(args.batch_id)
    record_id = validate_plain_text(args.record_id, "record_id")
    reviewer = validate_plain_text(args.reviewer, "reviewer")
    note = validate_plain_text(args.note, "note")

    requested_category = None
    if args.decision == "correct":
        if args.final_category is None:
            raise AppError("final_category_required")
        requested_category = validate_plain_text(
            args.final_category,
            "final_category",
        )
    elif args.final_category is not None:
        raise AppError(
            "final_category_not_allowed_for_decision",
            {"decision": args.decision},
        )

    connection = open_database(args.db)

    try:
        connection.execute("BEGIN IMMEDIATE")
        row = connection.execute(
            """
            SELECT proposed_category, status
            FROM review_items
            WHERE batch_id = ? AND record_id = ?
            """,
            (batch_id, record_id),
        ).fetchone()

        if row is None:
            raise AppError("review_item_not_found")
        if row["status"] != "pending":
            raise AppError(
                "review_item_already_decided",
                {"status": row["status"]},
            )

        if args.decision == "accept":
            status = "accepted"
            final_category = row["proposed_category"]
        elif args.decision == "correct":
            status = "corrected"
            final_category = requested_category
            if final_category == row["proposed_category"]:
                raise AppError("correction_must_change_category")

            allowed = connection.execute(
                "SELECT 1 FROM categories WHERE category = ?",
                (final_category,),
            ).fetchone()
            if allowed is None:
                raise AppError("final_category_not_allowed")
        else:
            status = "rejected"
            final_category = None

        cursor = connection.execute(
            """
            UPDATE review_items
            SET status = ?, final_category = ?, reviewer = ?,
                review_note = ?, decided_at = ?
            WHERE batch_id = ? AND record_id = ? AND status = 'pending'
            """,
            (
                status,
                final_category,
                reviewer,
                note,
                utc_now(),
                batch_id,
                record_id,
            ),
        )
        if cursor.rowcount != 1:
            raise AppError("concurrent_update_detected")

        connection.commit()
    except BaseException:
        safe_rollback(connection)
        raise
    finally:
        connection.close()

    return {
        "ok": True,
        "command": "decide",
        "batch_id": batch_id,
        "record_id": record_id,
        "status": status,
        "final_category": final_category,
    }


def build_parser():
    parser = JsonArgumentParser(description="SQLite review queue")
    subparsers = parser.add_subparsers(dest="command", required=True)

    init_parser = subparsers.add_parser("init")
    init_parser.add_argument("--db", type=Path, required=True)
    init_parser.add_argument(
        "--category",
        action="append",
        required=True,
        help="Repeat for every allowed category",
    )
    init_parser.set_defaults(handler=cmd_init)

    import_parser = subparsers.add_parser("import-results")
    import_parser.add_argument("--db", type=Path, required=True)
    import_parser.add_argument("--input", type=Path, required=True)
    import_parser.add_argument("--batch-id", required=True)
    import_parser.set_defaults(handler=cmd_import_results)

    pending_parser = subparsers.add_parser("pending")
    pending_parser.add_argument("--db", type=Path, required=True)
    pending_parser.add_argument("--batch-id", required=True)
    pending_parser.add_argument("--limit", type=limit_value, default=50)
    pending_parser.set_defaults(handler=cmd_pending)

    decide_parser = subparsers.add_parser("decide")
    decide_parser.add_argument("--db", type=Path, required=True)
    decide_parser.add_argument("--batch-id", required=True)
    decide_parser.add_argument("--record-id", required=True)
    decide_parser.add_argument(
        "--decision",
        choices=("accept", "correct", "reject"),
        required=True,
    )
    decide_parser.add_argument("--final-category")
    decide_parser.add_argument("--reviewer", required=True)
    decide_parser.add_argument("--note", required=True)
    decide_parser.set_defaults(handler=cmd_decide)

    return parser


def main():
    exit_code = 0
    stream = sys.stdout

    try:
        args = build_parser().parse_args()
        payload = args.handler(args)
    except AppError as exc:
        payload = {"ok": False, "error": exc.error}
        if exc.details is not None:
            payload["details"] = exc.details
        stream = sys.stderr
        exit_code = 2
    except KeyboardInterrupt:
        payload = {"ok": False, "error": "interrupted"}
        stream = sys.stderr
        exit_code = 130
    except sqlite3.Error as exc:
        payload = {
            "ok": False,
            "error": "database_error",
            "details": {"type": type(exc).__name__},
        }
        stream = sys.stderr
        exit_code = 3
    except (OSError, UnicodeError) as exc:
        payload = {
            "ok": False,
            "error": "io_error",
            "details": {"type": type(exc).__name__},
        }
        stream = sys.stderr
        exit_code = 3
    except Exception as exc:
        payload = {
            "ok": False,
            "error": "unexpected_error",
            "details": {"type": type(exc).__name__},
        }
        stream = sys.stderr
        exit_code = 3

    try:
        emit(stream, payload)
    except KeyboardInterrupt:
        return 130
    except (OSError, UnicodeError):
        return 3
    except Exception:
        return 3

    return exit_code


if __name__ == "__main__":
    sys.exit(main())

接続ではisolation_level=Noneを指定し、自動的なトランザクション開始に依存しません。読み取りはBEGIN、取り込みと判定は書き込みを先に確保するBEGIN IMMEDIATEで開始し、成功時だけ明示的にコミットします。

emit()はJSONと改行を一度に書き、続けてflush()します。成功結果を含む最終出力も例外処理の対象にするため、出力時のOSErrorやUnicodeErrorは終了コード3、出力中の割り込みは終了コード130になります。ただし、出力先が閉じている場合などは、JSONが出ないか途中までしか書かれないことがあります。

取り込み時にも項目、型、重複キーなどを確認します。ただし、第162回で設定した行サイズや文字数の上限を、このスクリプトが代替するわけではありません。第162回の検証を通過したvalid.jsonlだけを入力にします。

3.許可カテゴリを固定してデータベースを作る

同名のreview.dbが存在しないディレクトリで実行します。--categoryは必要な数だけ繰り返せます。

python review_queue.py init \
  --db review.db \
  --category 配送 \
  --category 請求 \
  --category その他

成功すると、標準出力へ次の1行が出ます。

{"ok":true,"command":"init","db":"review.db","schema_version":1,"category_count":3}

既存ファイルは上書きしません。作成時には排他的なファイル作成を使います。POSIX環境ではモード0600を指定し、実際の権限にはOSのumaskも適用されます。保存先のアクセス制御も確認してください。

4.検証済み結果をバッチとして取り込む

python review_queue.py import-results \
  --db review.db \
  --input valid.jsonl \
  --batch-id 2026-10-02-a

サンプルでは3行を読み、2行をレビュー対象として登録します。

{"ok":true,"command":"import-results","batch_id":"2026-10-02-a","total_rows":3,"queued_rows":2,"skipped_rows":1}

バッチIDの重複、空行、不正なJSON、重複したrecord_id、未知のカテゴリが1件でもあれば、バッチ全体をロールバックします。一部の行だけが登録された状態にはしません。

5.未確認の項目を取得する

python review_queue.py pending \
  --db review.db \
  --batch-id 2026-10-02-a \
  --limit 50

出力は1つのJSONオブジェクトです。理由に改行などが含まれていても、JSON文字列としてエスケープされます。

{"ok":true,"command":"pending","batch_id":"2026-10-02-a","pending_count":2,"returned_count":2,"items":[{"record_id":"C-001","proposed_category":"配送","reason":"配送先変更と到着日の質問が同時に含まれる","source_line":1},{"record_id":"C-003","proposed_category":"配送","reason":"配送状況と請求額の両方に言及している","source_line":3}]}

pending_countは未確認の総数、returned_countは今回返した件数です。総数が上限を超える場合は、表示された項目を判定してから同じコマンドを再実行します。

6.採用、修正、却下を記録する

1つの項目に実行できる判定は1回だけです。ここでは採用例をC-001に実行した後、C-003について「修正」または「却下」のどちらか一方を選びます。C-003の2つの例は相互に排他的であり、同じデータベースへ続けて実行する例ではありません。

AIの提案カテゴリを採用する

python review_queue.py decide \
  --db review.db \
  --batch-id 2026-10-02-a \
  --record-id C-001 \
  --decision accept \
  --reviewer sato \
  --note 原文照合済み
{"ok":true,"command":"decide","batch_id":"2026-10-02-a","record_id":"C-001","status":"accepted","final_category":"配送"}

C-003を別の許可カテゴリへ修正する場合

python review_queue.py decide \
  --db review.db \
  --batch-id 2026-10-02-a \
  --record-id C-003 \
  --decision correct \
  --final-category 請求 \
  --reviewer sato \
  --note 請求内容を主題として確定
{"ok":true,"command":"decide","batch_id":"2026-10-02-a","record_id":"C-003","status":"corrected","final_category":"請求"}

correctでは、登録済みの許可カテゴリだけを指定できます。提案と同じカテゴリは、修正にならないため拒否します。

C-003を却下する場合

次の例を試す場合は、C-003の修正例を実行しないでください。すでに修正例を実行した場合は、別名の新しいデータベースを作成し、手順3と4をそのデータベース名に置き換えて実行してから、次のコマンドでも同じ名前を指定します。

却下では--decision rejectを指定し、--final-categoryは付けません。保存される最終カテゴリはNULLになり、JSON出力ではnullになります。

python review_queue.py decide \
  --db review.db \
  --batch-id 2026-10-02-a \
  --record-id C-003 \
  --decision reject \
  --reviewer sato \
  --note 元データだけでは分類不能
{"ok":true,"command":"decide","batch_id":"2026-10-02-a","record_id":"C-003","status":"rejected","final_category":null}

いずれの判定も、対象がpendingである場合に一度だけ実行できます。確定済みレコードへの再実行は終了コード2で停止し、保存済みの判定を変更しません。

終了コードと失敗時の扱い

終了コード 意味 確認事項
0 コマンド処理と結果出力が成功 成功JSONの内容を確認する
2 引数、入力、カテゴリ、状態遷移の契約違反 入力内容や現在の状態を直す
3 ファイル操作、データベース、結果出力、または予期しない実行エラー 成功JSONがなければ、再実行前にデータベースの状態を確認する
130 処理中または結果出力中の割り込み 再実行前にデータベースの状態を確認する

処理中の例外では未確定のトランザクションをロールバックします。ただし、コミット直後や結果出力中に入出力エラーや割り込みが起きた場合、データベースへの反映が完了していても、呼び出し側が完全な成功JSONを受け取れないことがあります。出力先そのものが利用できない場合は、エラーJSONも出力できません。終了コードだけで未反映と決めず、同じbatch_idや対象レコードを照会してから再実行します。

SQLiteの待機時間は5秒です。別プロセスが書き込みを保持し続けた場合は、データベースエラーとして終了します。実行中の別作業を確認してから再試行します。

運用へ入れる前の確認

  1. 第162回の検証ゲートを通過したJSONLだけを入力にする
  2. カテゴリ一覧を業務担当者と確認してからデータベースを作る
  3. 処理単位ごとに重複しないbatch_idを付ける
  4. AIの理由だけで判断せず、権限のある元データを照合する
  5. pending_countが0になったことを確認する
  6. データベースを元データと同じ機密区分で保護し、必要な期間だけ保存する

pending_count=0が示すのは、そのバッチの要確認行に判定が付いたことだけです。自動判定された行の正しさや、後続システムへの反映完了までは保証しません。

まずはサンプル用の新しいデータベースで、取り込み、一覧取得、判定、再判定の拒否までを確認します。その後、業務データへ適用してください。

仕様を確認する一次資料

第162回 AIのJSONL結果をそのまま使わない:Pythonで作る行単位の検証ゲート

第161回では、AIなどの後続処理へ渡すCSVを許可列だけに絞りました。入力を整えても、AIが返した結果を無検査でCSV更新や集計へ渡すことはできません。真偽値が文字列になっている、許可していない分類名が混ざる、同じIDが二度返る、といった問題があるためです。

今回は、AIの分類結果を1行1オブジェクトのJSONLとして受け取り、Python標準ライブラリだけで検査します。特定のAIサービスへ通信するコードではありません。実際のレスポンスをJSONLへ変換する部分は、利用するサービスの仕様に合わせて分離します。

今回固定する結果契約

各行のJSONオブジェクトは、次の4項目だけを持つものとします。

項目 条件
record_id 空でない文字列。前後空白、U+0000~U+001F、U+007Fを認めない。項目、型、値の検査を通過した行同士で重複させない
category ポリシーで許可した文字列のいずれか
needs_review JSONの真偽値であるtrueまたはfalse。文字列は不可
reason 空白だけではない文字列。ポリシーで決めた長さ以内

不足項目だけでなく、未知の項目が増えた場合も拒否します。PythonのJSONデコーダが既定で受け入れるNaNやInfinityも拒否します。同じオブジェクト内の重複キーも、後に書かれた値で上書きせずエラーにします。

長さの判定にはPythonのlen()を使います。結合文字や絵文字の並びを、画面上の見た目どおりの1文字として数える処理ではありません。上限は後続システムと同じ数え方で決める必要があります。

各行について記録するのは、最初に検出したエラー1件です。検査結果は新しいディレクトリへ出力し、元のJSONLは変更しません。

生成物 用途
valid.jsonl 契約を通過した行を、項目順をそろえて再出力
errors.jsonl 拒否した物理行番号とエラーコードを記録。元の値は複製しない
summary.json 件数、判定、入力とポリシーのSHA-256を記録

入力が0バイトの場合は、行ごとの判定を作れません。契約違反として終了コード2を返し、作成途中の出力ディレクトリを削除します。

1.検査ポリシーを用意する

次をUTF-8のresult_policy.jsonとして保存します。UTF-8 BOMが付いていても読み込めます。

{
  "schema_version": 1,
  "allowed_categories": ["配送", "請求", "その他"],
  "max_record_id_chars": 64,
  "max_reason_chars": 200,
  "max_line_bytes": 8192
}

64、200、8192は、この例の設定値です。AIサービス共通の上限ではありません。実データのID仕様、確認画面、後続システムの制約から決めます。max_line_bytesは1物理行のバイト数で、改行がある場合は改行コードも含みます。

2.拒否例を含むJSONLを用意する

次をUTF-8のai_results.jsonlとして保存します。JSONLなので、配列を囲む角括弧や行末のカンマは付けません。

{"record_id":"R-001","category":"配送","needs_review":false,"reason":"到着日の確認依頼"}
{"record_id":"R-002","category":"返金","needs_review":false,"reason":"許可されていない分類"}
{"record_id":"R-003","category":"請求","needs_review":"false","reason":"真偽値が文字列"}

1行目は契約どおりです。2行目の「返金」は許可カテゴリにありません。3行目のneeds_reviewは真偽値ではなく文字列です。

3.行単位の検証ゲートを実装する

次をvalidate_ai_jsonl.pyとして保存します。Python 3.10以上と標準ライブラリだけを使います。

from __future__ import annotations

import argparse
import hashlib
import json
import shutil
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, BinaryIO, Iterator


FIELDS = {"record_id", "category", "needs_review", "reason"}


class ContractError(Exception):
    pass


class RowError(Exception):
    def __init__(self, code: str):
        super().__init__(code)
        self.code = code


@dataclass(frozen=True)
class Policy:
    allowed_categories: frozenset[str]
    max_record_id_chars: int
    max_reason_chars: int
    max_line_bytes: int


def has_surrogate(text: str) -> bool:
    return any(0xD800 <= ord(char) <= 0xDFFF for char in text)


def strict_object(pairs: list[tuple[str, object]]) -> dict[str, object]:
    result: dict[str, object] = {}
    for key, value in pairs:
        if key in result:
            raise RowError("DUPLICATE_KEY")
        result[key] = value
    return result


def reject_constant(_: str) -> object:
    raise RowError("NON_STANDARD_NUMBER")


def load_policy(path: Path) -> tuple[Policy, str]:
    raw = path.read_bytes()
    digest = hashlib.sha256(raw).hexdigest()

    try:
        text = raw.decode("utf-8-sig")
    except UnicodeDecodeError as exc:
        raise ContractError("policy must be UTF-8") from exc

    try:
        data = json.loads(
            text,
            object_pairs_hook=strict_object,
            parse_constant=reject_constant,
        )
    except (ValueError, RecursionError, RowError) as exc:
        raise ContractError("policy is not valid strict JSON") from exc

    expected = {
        "schema_version",
        "allowed_categories",
        "max_record_id_chars",
        "max_reason_chars",
        "max_line_bytes",
    }
    if type(data) is not dict or set(data) != expected:
        raise ContractError("policy fields do not match the required schema")
    if type(data["schema_version"]) is not int or data["schema_version"] != 1:
        raise ContractError("schema_version must be the integer 1")

    categories = data["allowed_categories"]
    if type(categories) is not list or not categories:
        raise ContractError("allowed_categories must be a non-empty array")
    if any(
        type(item) is not str
        or not item
        or item != item.strip()
        or has_surrogate(item)
        for item in categories
    ):
        raise ContractError("each category must be a trimmed Unicode string")
    if len(set(categories)) != len(categories):
        raise ContractError("allowed_categories contains duplicates")

    limits: dict[str, int] = {}
    for name in (
        "max_record_id_chars",
        "max_reason_chars",
        "max_line_bytes",
    ):
        value = data[name]
        if type(value) is not int or value <= 0:
            raise ContractError(f"{name} must be a positive integer")
        limits[name] = value

    return (
        Policy(
            allowed_categories=frozenset(categories),
            max_record_id_chars=limits["max_record_id_chars"],
            max_reason_chars=limits["max_reason_chars"],
            max_line_bytes=limits["max_line_bytes"],
        ),
        digest,
    )


def limited_lines(
    stream: BinaryIO,
    limit: int,
    digest: Any,
) -> Iterator[tuple[int, bytes | None]]:
    line_number = 0
    while True:
        chunk = stream.readline(limit + 1)
        if chunk == b"":
            return

        line_number += 1
        digest.update(chunk)
        too_long = len(chunk) > limit

        if too_long and not chunk.endswith(b"\n"):
            while True:
                rest = stream.readline(limit + 1)
                if rest == b"":
                    break
                digest.update(rest)
                if rest.endswith(b"\n"):
                    break

        yield line_number, None if too_long else chunk


def validate_record(value: object, policy: Policy) -> dict[str, object]:
    if type(value) is not dict:
        raise RowError("NOT_OBJECT")
    if set(value) != FIELDS:
        raise RowError("FIELDS_MISMATCH")

    record_id = value["record_id"]
    if type(record_id) is not str:
        raise RowError("RECORD_ID_TYPE")
    if not record_id or record_id != record_id.strip():
        raise RowError("RECORD_ID_FORMAT")
    if len(record_id) > policy.max_record_id_chars:
        raise RowError("RECORD_ID_TOO_LONG")
    if has_surrogate(record_id):
        raise RowError("RECORD_ID_UNICODE")
    if any(ord(char) < 32 or ord(char) == 127 for char in record_id):
        raise RowError("RECORD_ID_CONTROL_CHAR")

    category = value["category"]
    if type(category) is not str:
        raise RowError("CATEGORY_TYPE")
    if category not in policy.allowed_categories:
        raise RowError("CATEGORY_NOT_ALLOWED")

    needs_review = value["needs_review"]
    if type(needs_review) is not bool:
        raise RowError("NEEDS_REVIEW_TYPE")

    reason = value["reason"]
    if type(reason) is not str:
        raise RowError("REASON_TYPE")
    if not reason.strip():
        raise RowError("REASON_EMPTY")
    if len(reason) > policy.max_reason_chars:
        raise RowError("REASON_TOO_LONG")
    if "\x00" in reason or has_surrogate(reason):
        raise RowError("REASON_UNICODE")

    return {
        "record_id": record_id,
        "category": category,
        "needs_review": needs_review,
        "reason": reason,
    }


def json_line(value: object) -> bytes:
    text = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
    return (text + "\n").encode("utf-8")


def write_rejection(
    stream: BinaryIO,
    line_number: int,
    code: str,
) -> None:
    stream.write(json_line({"line": line_number, "code": code}))


def process(
    input_path: Path,
    output_dir: Path,
    policy: Policy,
    policy_sha256: str,
) -> dict[str, object]:
    input_digest = hashlib.sha256()
    seen_ids: set[str] = set()
    total_lines = 0
    valid_records = 0
    rejected_records = 0

    with (
        input_path.open("rb") as source,
        (output_dir / "valid.jsonl").open("xb") as valid_stream,
        (output_dir / "errors.jsonl").open("xb") as error_stream,
    ):
        for line_number, raw in limited_lines(
            source,
            policy.max_line_bytes,
            input_digest,
        ):
            total_lines += 1
            error_code: str | None = None
            record: dict[str, object] | None = None

            if raw is None:
                error_code = "LINE_TOO_LONG"
            else:
                try:
                    encoding = "utf-8-sig" if line_number == 1 else "utf-8"
                    text = raw.decode(encoding)
                except UnicodeDecodeError:
                    error_code = "INVALID_UTF8"
                else:
                    if not text.strip():
                        error_code = "BLANK_LINE"
                    else:
                        try:
                            value = json.loads(
                                text,
                                object_pairs_hook=strict_object,
                                parse_constant=reject_constant,
                            )
                            record = validate_record(value, policy)
                        except (ValueError, RecursionError):
                            error_code = "JSON_SYNTAX"
                        except RowError as exc:
                            error_code = exc.code

            if error_code is None and record is not None:
                record_id = str(record["record_id"])
                if record_id in seen_ids:
                    error_code = "DUPLICATE_RECORD_ID"
                else:
                    seen_ids.add(record_id)
                    valid_stream.write(json_line(record))
                    valid_records += 1

            if error_code is not None:
                write_rejection(error_stream, line_number, error_code)
                rejected_records += 1

    if total_lines == 0:
        raise ContractError("input JSONL is empty")

    summary: dict[str, object] = {
        "schema_version": 1,
        "status": "ok" if rejected_records == 0 else "rejected",
        "input_file": input_path.name,
        "input_sha256": input_digest.hexdigest(),
        "policy_sha256": policy_sha256,
        "total_lines": total_lines,
        "valid_records": valid_records,
        "rejected_records": rejected_records,
    }
    with (output_dir / "summary.json").open("xb") as stream:
        stream.write(json_line(summary))
    return summary


def emit(stream: Any, value: object) -> None:
    text = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
    stream.write(text + "\n")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Validate provider-neutral AI result JSONL."
    )
    parser.add_argument("--input", required=True, type=Path)
    parser.add_argument("--policy", required=True, type=Path)
    parser.add_argument("--output-dir", required=True, type=Path)
    args = parser.parse_args()

    created_output = False
    try:
        if not args.input.is_file():
            raise ContractError("input must be an existing file")
        if not args.policy.is_file():
            raise ContractError("policy must be an existing file")
        if args.output_dir.exists():
            raise ContractError("output directory already exists")
        if not args.output_dir.parent.is_dir():
            raise ContractError("output parent directory does not exist")

        policy, policy_sha256 = load_policy(args.policy)

        try:
            args.output_dir.mkdir()
        except FileExistsError as exc:
            raise ContractError("output directory already exists") from exc
        created_output = True

        marker = args.output_dir / ".incomplete"
        marker.write_bytes(b"incomplete\n")
        summary = process(
            args.input,
            args.output_dir,
            policy,
            policy_sha256,
        )
        marker.unlink()

        result = {
            "status": summary["status"],
            "total_lines": summary["total_lines"],
            "valid_records": summary["valid_records"],
            "rejected_records": summary["rejected_records"],
            "output_dir": str(args.output_dir),
        }
        emit(sys.stdout, result)
        return 0 if summary["status"] == "ok" else 1

    except ContractError as exc:
        if created_output:
            shutil.rmtree(args.output_dir, ignore_errors=True)
        emit(
            sys.stderr,
            {
                "status": "error",
                "code": "CONTRACT_ERROR",
                "message": str(exc),
            },
        )
        return 2
    except KeyboardInterrupt:
        if created_output:
            shutil.rmtree(args.output_dir, ignore_errors=True)
        emit(sys.stderr, {"status": "error", "code": "INTERRUPTED"})
        return 130
    except OSError as exc:
        if created_output:
            shutil.rmtree(args.output_dir, ignore_errors=True)
        emit(
            sys.stderr,
            {
                "status": "error",
                "code": "IO_ERROR",
                "message": str(exc),
            },
        )
        return 3
    except Exception:
        if created_output:
            shutil.rmtree(args.output_dir, ignore_errors=True)
        emit(
            sys.stderr,
            {"status": "error", "code": "UNEXPECTED_ERROR"},
        )
        return 3


if __name__ == "__main__":
    raise SystemExit(main())

処理順と失敗時の扱い

  1. 入力ファイル、ポリシー、未作成の出力先を確認する
  2. ポリシーJSONの重複キー、項目、型、値を確認する
  3. 出力ディレクトリを排他的に作り、.incompleteを置く
  4. 入力をバイト列で1行ずつ読み、行サイズ、UTF-8、JSON構文、項目、型、値、ID重複を確認する
  5. 合格行とエラー情報を分けて書き、最後にsummary.jsonを作る
  6. 完了後だけ.incompleteを削除し、判定に応じた終了コードを返す

行が上限を超えた場合は、その行全体をメモリへためず、改行まで読み捨てて次の行へ進みます。入力全体もメモリへ載せません。ただし、重複検査用のrecord_id集合は保持するため、ユニークID数に応じてメモリ使用量が増えます。

通常の例外とCtrl+Cでは、この実行が作った出力ディレクトリの削除を試みます。プロセスの強制終了や電源断では、途中のディレクトリが残る場合があります。同じ出力先へ複数のプロセスを同時実行しない運用も必要です。

4.実行して終了コードを確認する

次はbash系シェルでの実行例です。

python3 validate_ai_jsonl.py \
  --input ai_results.jsonl \
  --policy result_policy.json \
  --output-dir checked
echo $?

標準出力と終了コードは次のようになります。

{"status":"rejected","total_lines":3,"valid_records":1,"rejected_records":2,"output_dir":"checked"}
1

終了コード1は、検査処理は完了したものの、少なくとも1行を拒否したことを表します。後続のCSV更新や登録処理は止めます。

終了コード 意味
0 全行が契約を通過
1 1行以上を拒否。検査結果は出力済み
2 引数、入力、ポリシー、出力先の契約違反
3 ファイル操作または予期しない実行エラー
130 Ctrl+Cによる中断

argparseが検出する必須引数の不足や未知の引数も終了コード2です。この場合はスクリプト独自のJSONではなく、使用法が標準エラー出力へ表示されます。

5.合格行とエラーを確認する

checked/valid.jsonlには1行目だけが入ります。

{"record_id":"R-001","category":"配送","needs_review":false,"reason":"到着日の確認依頼"}

checked/errors.jsonlには、元の理由文やカテゴリ値を複製せず、位置とコードだけを記録します。

{"line":2,"code":"CATEGORY_NOT_ALLOWED"}
{"line":3,"code":"NEEDS_REVIEW_TYPE"}

summary.jsonには入力とポリシーのSHA-256も入ります。ハッシュは同じ内容かを照合するための値であり、暗号化や匿名化ではありません。

主なエラーコードは次のとおりです。

  • JSON_SYNTAX:JSONとして解析できない
  • DUPLICATE_KEY:同じオブジェクト内に同名キーがある
  • NON_STANDARD_NUMBER:NaNやInfinityがある
  • FIELDS_MISMATCH:必須項目の不足または未知項目の追加
  • CATEGORY_NOT_ALLOWED:許可されていない分類
  • NEEDS_REVIEW_TYPE:真偽値の型が違う
  • DUPLICATE_RECORD_ID:検査を通過した行のIDが重複
  • LINE_TOO_LONG、INVALID_UTF8:行サイズまたは文字コードの違反

6.全行が合格する経路を確認する

次をai_results_ok.jsonlとして保存します。

{"record_id":"R-001","category":"配送","needs_review":false,"reason":"到着日の確認依頼"}
{"record_id":"R-002","category":"請求","needs_review":true,"reason":"請求内容を担当者が確認する必要あり"}

既存ディレクトリを上書きしないため、別の出力先を指定します。

python3 validate_ai_jsonl.py \
  --input ai_results_ok.jsonl \
  --policy result_policy.json \
  --output-dir checked-ok
echo $?

この入力では次の結果になります。

{"status":"ok","total_lines":2,"valid_records":2,"rejected_records":0,"output_dir":"checked-ok"}
0

自動処理へ組み込む場合は、終了コードが0で、.incompleteがなく、summary.jsonのstatusがokであることを確認してから次へ進めます。終了コード1で作られたvalid.jsonlは調査用の部分結果です。完全な結果として登録してはいけません。

このゲートが判定しないこと

  • 分類や理由が業務上正しいか
  • reasonに事実誤認、機密情報、不適切な文言がないか
  • 依頼したIDがすべて返ったか、依頼していないIDが追加されたか
  • needs_reviewの判断基準が妥当か
  • AIサービス固有の利用規約、保存条件、データ送信可否

このスクリプトが確認するのは、構文と決めたデータ契約です。意味の正しさは保証しません。反映前には依頼時と結果のID一覧を突合し、欠落と余分なIDを検出します。そのうえで、needs_reviewが真の行や業務上重要な分類を人が確認します。

導入前の確認

正常な1行、未知カテゴリの1行、型違いの1行を用意し、終了コード0と1の両方を確認します。確認できたら、「AI結果を受け取る処理」と「CSVや業務システムへ反映する処理」の間に、このゲートを置きます。