問い合わせCSVをAIで分類するとき、氏名やメールアドレスまで入力に残す必要がない場合があります。手作業で列を削除する運用では、削除漏れや担当者ごとの手順差を防げません。
この記事では、ポリシーで許可した列だけを別のCSVへ書き出すゲートを、Python標準ライブラリだけで作ります。このスクリプトはAIサービスへ通信しません。生成したCSVを外部へ渡してよいかは、組織の規則、契約、データの内容に基づいて別途確認します。
今回作る処理の契約
- 出力列は
allowed_columnsに記載した列だけとし、記載順も維持する - 禁止列が許可リストにも含まれていたら、ポリシーの誤りとして拒否する
- ヘッダーの重複、許可列の不足、行の列数不一致を拒否する
- 既存の出力ファイルを上書きしない
- ヘルプ表示を除き、標準出力には1行のJSONを出す
- データ行のセル値は結果JSONへ出さない
forbidden_columns は、ポリシー編集時の事故を防ぐための二重チェックです。入力CSVに禁止列が存在すること自体はエラーではありません。その列を出力対象へ含めようとした場合に停止します。
| 終了コード | 意味 |
|---|---|
| 0 | 出力完了 |
| 2 | 引数、入力CSV、またはポリシーを拒否 |
| 3 | ファイル操作または予期しない実行エラー |
| 130 | Ctrl+Cによる中断 |
1. サンプルCSVを用意する
動作確認用の架空データを source.csv として保存します。文字コードはUTF-8です。
case_id,customer_name,email,category,request_text,internal_note
C-001,山田花子,hanako@example.com,配送,到着日を確認したい,返金履歴あり
C-002,佐藤太郎,taro@example.com,請求,領収書を再発行したい,本人確認済み
2. 許可列と禁止列を定義する
次の内容を export-policy.json として保存します。出力CSVの列順は allowed_columns の順番になります。
{
"allowed_columns": [
"case_id",
"category",
"request_text"
],
"forbidden_columns": [
"customer_name",
"email",
"phone",
"address",
"internal_note"
]
}
ここでの列名は例です。実際には、自社のCSV仕様と利用目的に合わせて明示的に決めます。自由記述欄には氏名や連絡先が入力される可能性があるため、列名だけの検査では安全性を保証できません。
3. エクスポートゲートを実装する
次のコードを export_allowed_csv.py として保存します。Python 3.10以降を想定し、外部ライブラリは使いません。
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import csv
import json
import os
import sys
from pathlib import Path
from typing import Any
EXIT_OK = 0
EXIT_REJECTED = 2
EXIT_OPERATIONAL_ERROR = 3
EXIT_INTERRUPTED = 130
class RejectedError(Exception):
"""引数、入力、またはポリシーを拒否した。"""
class OperationalError(Exception):
"""ファイル操作など、実行環境のエラー。"""
class JsonArgumentParser(argparse.ArgumentParser):
def error(self, message: str) -> None:
raise RejectedError(f"引数エラー: {message}")
def emit(payload: dict[str, Any]) -> None:
print(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))
def find_duplicates(names: list[str]) -> list[str]:
seen: set[str] = set()
duplicates: set[str] = set()
for name in names:
if name in seen:
duplicates.add(name)
seen.add(name)
return sorted(duplicates)
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 RejectedError(
f"ポリシーJSONに重複キーがあります: {key}"
)
result[key] = value
return result
def read_name_list(
data: dict[str, Any], key: str, *, allow_empty: bool
) -> list[str]:
value = data[key]
if not isinstance(value, list):
raise RejectedError(f"{key} は配列で指定してください")
if not allow_empty and not value:
raise RejectedError(f"{key} を1件以上指定してください")
names: list[str] = []
for item in value:
if (
not isinstance(item, str)
or not item
or item != item.strip()
or "\r" in item
or "\n" in item
):
raise RejectedError(
f"{key} の列名に空欄、前後空白、または改行があります"
)
names.append(item)
duplicates = find_duplicates(names)
if duplicates:
raise RejectedError(
f"{key} に重複があります: {', '.join(duplicates)}"
)
return names
def load_policy(path: Path) -> list[str]:
try:
text = path.read_text(encoding="utf-8-sig")
except UnicodeDecodeError as exc:
raise RejectedError("ポリシーはUTF-8で読めません") from exc
except OSError as exc:
raise OperationalError(f"ポリシーを読めません: {path}") from exc
try:
raw = json.loads(text, object_pairs_hook=unique_object)
except json.JSONDecodeError as exc:
raise RejectedError(
f"ポリシーJSONの構文エラーです（行{exc.lineno}、列{exc.colno}）"
) from exc
if not isinstance(raw, dict):
raise RejectedError("ポリシーの最上位はオブジェクトにしてください")
expected_keys = {"allowed_columns", "forbidden_columns"}
actual_keys = set(raw)
if actual_keys != expected_keys:
details: list[str] = []
missing = sorted(expected_keys - actual_keys)
extra = sorted(actual_keys - expected_keys)
if missing:
details.append(f"不足: {', '.join(missing)}")
if extra:
details.append(f"未定義: {', '.join(extra)}")
raise RejectedError(
f"ポリシーのキーが不正です（{'; '.join(details)}）"
)
allowed = read_name_list(raw, "allowed_columns", allow_empty=False)
forbidden = set(
read_name_list(raw, "forbidden_columns", allow_empty=True)
)
overlap = sorted(set(allowed) & forbidden)
if overlap:
raise RejectedError(
"allowed_columns と forbidden_columns が重複しています: "
+ ", ".join(overlap)
)
return allowed
def validate_header(
header: list[str] | None, allowed: list[str]
) -> list[str]:
if header is None:
raise RejectedError("入力CSVが空です")
invalid = [
name
for name in header
if not name
or name != name.strip()
or "\r" in name
or "\n" in name
]
if invalid:
raise RejectedError(
"CSVヘッダーに空欄、前後空白、または改行があります"
)
duplicates = find_duplicates(header)
if duplicates:
raise RejectedError(
f"CSVヘッダーに重複があります: {', '.join(duplicates)}"
)
missing = [name for name in allowed if name not in header]
if missing:
raise RejectedError(
f"入力CSVに許可列がありません: {', '.join(missing)}"
)
return header
def same_path(left: Path, right: Path) -> bool:
try:
return left.resolve(strict=False) == right.resolve(strict=False)
except (OSError, RuntimeError) as exc:
raise OperationalError("入出力パスを解決できません") from exc
def cleanup_partial(path: Path) -> None:
try:
path.unlink()
except FileNotFoundError:
return
except OSError as exc:
raise OperationalError(
f"不完全な出力ファイルを削除できません: {path}"
) from exc
def export_allowed_csv(
source_path: Path, policy_path: Path, output_path: Path
) -> dict[str, Any]:
if same_path(source_path, output_path):
raise RejectedError("入力元と出力先に同じパスは指定できません")
if output_path.exists() or output_path.is_symlink():
raise RejectedError(f"出力先が既に存在します: {output_path}")
if not output_path.parent.exists() or not output_path.parent.is_dir():
raise OperationalError(
f"出力先ディレクトリを利用できません: {output_path.parent}"
)
allowed = load_policy(policy_path)
output_created = False
completed = False
reader = None
header: list[str] = []
row_count = 0
try:
with source_path.open(
"r", encoding="utf-8-sig", newline=""
) as source:
reader = csv.DictReader(source, strict=True)
header = validate_header(reader.fieldnames, allowed)
with output_path.open(
"x", encoding="utf-8", newline=""
) as target:
output_created = True
writer = csv.DictWriter(
target,
fieldnames=allowed,
extrasaction="raise",
lineterminator="\n",
)
writer.writeheader()
for row in reader:
if None in row:
raise RejectedError(
"列数がヘッダーより多い行があります: "
f"物理行{reader.line_num}"
)
missing_values = [
name for name in header if row.get(name) is None
]
if missing_values:
raise RejectedError(
"列数がヘッダーより少ない行があります: "
f"物理行{reader.line_num}"
)
writer.writerow(
{name: row[name] for name in allowed}
)
row_count += 1
target.flush()
os.fsync(target.fileno())
completed = True
except FileExistsError as exc:
raise RejectedError(
f"出力先が既に存在します: {output_path}"
) from exc
except UnicodeDecodeError as exc:
raise RejectedError(
"入力CSVはUTF-8（BOM可）として読めません"
) from exc
except csv.Error as exc:
line_number = reader.line_num if reader is not None else 0
raise RejectedError(
f"CSV構文エラーです（物理行{line_number}）"
) from exc
except OSError as exc:
target_name = exc.filename or "対象不明"
raise OperationalError(
f"ファイル操作に失敗しました: {target_name}"
) from exc
finally:
if output_created and not completed:
cleanup_partial(output_path)
return {
"rows": row_count,
"input_columns": len(header),
"output_columns": allowed,
"dropped_columns": len(header) - len(allowed),
"output": str(output_path),
}
def build_parser() -> argparse.ArgumentParser:
parser = JsonArgumentParser(
description="許可した列だけをCSVへ書き出します"
)
parser.add_argument("source", type=Path, help="入力CSV")
parser.add_argument("policy", type=Path, help="ポリシーJSON")
parser.add_argument("output", type=Path, help="新規出力CSV")
return parser
def main(argv: list[str] | None = None) -> int:
try:
args = build_parser().parse_args(argv)
report = export_allowed_csv(
args.source, args.policy, args.output
)
except RejectedError as exc:
emit({"status": "rejected", "reason": str(exc)})
return EXIT_REJECTED
except OperationalError as exc:
emit({"status": "error", "reason": str(exc)})
return EXIT_OPERATIONAL_ERROR
except KeyboardInterrupt:
emit({"status": "error", "reason": "処理が中断されました"})
return EXIT_INTERRUPTED
except Exception as exc:
emit(
{
"status": "error",
"reason": f"予期しないエラー: {type(exc).__name__}",
}
)
return EXIT_OPERATIONAL_ERROR
emit({"status": "ok", **report})
return EXIT_OK
if __name__ == "__main__":
sys.exit(main())
4. 実行結果を確認する
出力先には、まだ存在しないファイル名を指定します。
python3 export_allowed_csv.py source.csv export-policy.json ai_input.csv
rc=$?
echo "exit=$rc"
スクリプトが標準出力へ出すJSONは次の1行です。
{"status":"ok","rows":2,"input_columns":6,"output_columns":["case_id","category","request_text"],"dropped_columns":3,"output":"ai_input.csv"}
続いて、シェルの echo が exit=0 を表示します。生成された ai_input.csv は次の内容です。
case_id,category,request_text
C-001,配送,到着日を確認したい
C-002,請求,領収書を再発行したい
元の行順は維持されます。空文字のセルは許可しますが、列自体が不足している行は拒否します。データ行が0件でも、ヘッダーが正しければ正常終了します。
5. ポリシーの誤りを止められるか確認する
email を許可列と禁止列の両方へ入れた bad-policy.json を用意します。
{
"allowed_columns": [
"case_id",
"email",
"category",
"request_text"
],
"forbidden_columns": [
"customer_name",
"email",
"phone",
"address",
"internal_note"
]
}
既存ファイルと衝突しない出力名で実行します。
python3 export_allowed_csv.py source.csv bad-policy.json bad.csv
rc=$?
echo "exit=$rc"
test ! -e bad.csv
echo "output_absent=$?"
標準出力と終了コードは次のようになります。output_absent=0 は、bad.csv が存在しないことを確認できたという意味です。
{"status":"rejected","reason":"allowed_columns と forbidden_columns が重複しています: email"}
exit=2
output_absent=0
運用へ組み込む順序
- 処理目的に必要な列を担当者とデータ管理者が決める
- 許可列と禁止列をポリシーJSONでレビューする
- 元CSVから別名のCSVを生成する
- 終了コードが0であることと、結果JSONの件数・列名を確認する
- 生成ファイルの内容を確認してから、承認済みの処理へ渡す
- 生成ファイルは組織の保管・削除ルールに従って扱う
後続処理は、ファイルの存在だけではなく終了コード0を開始条件にします。処理中も出力ファイルは見えるため、同じ出力先を監視する別処理がある場合は、完了前に読み込ませない設計が必要です。
この実装で保証しないこと
- 許可した自由記述欄に含まれる氏名、電話番号、秘密情報の検出
case_idなど、別データと照合できる識別子の匿名化- AIサービスへの送信可否や法令・契約への適合判断
- 表計算ソフトで開く場合の数式注入対策
- タブ区切りやセミコロン区切りなど、カンマ区切り以外の形式
- 同じ出力先へ複数プロセスが同時に書き込む運用
- 強制終了や電源断のあとに不完全なファイルが残らないこと
通常の検証エラーやCtrl+Cでは、作成途中と認識できた出力を削除します。ただし、プロセスの強制終了や電源断では不完全なファイルが残る可能性があります。再実行前に出力ファイルを確認し、不要と判断できた場合だけ明示的に削除してください。
まとめ
AIへ渡すデータを減らすときは、「削除する列」を毎回選ぶより、「出してよい列」を固定する方が手順を再現しやすくなります。このゲートは完全な匿名化ツールではありません。それでも、許可列、ポリシーの矛盾、CSV構造、上書き防止、終了コードを検査し、列の削除漏れを機械的に止められます。