第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秒です。別プロセスが書き込みを保持し続けた場合は、データベースエラーとして終了します。実行中の別作業を確認してから再試行します。
運用へ入れる前の確認
- 第162回の検証ゲートを通過したJSONLだけを入力にする
- カテゴリ一覧を業務担当者と確認してからデータベースを作る
- 処理単位ごとに重複しない
batch_idを付ける - AIの理由だけで判断せず、権限のある元データを照合する
pending_countが0になったことを確認する- データベースを元データと同じ機密区分で保護し、必要な期間だけ保存する
pending_count=0が示すのは、そのバッチの要確認行に判定が付いたことだけです。自動判定された行の正しさや、後続システムへの反映完了までは保証しません。
まずはサンプル用の新しいデータベースで、取り込み、一覧取得、判定、再判定の拒否までを確認します。その後、業務データへ適用してください。