第159回では、CSVをUTF-8へ正規化する入口を作りました。しかし、文字コードが正しくても、表計算ソフトへ渡す前に確認したいことがあります。外部から受け取った文字列が特定の記号で始まると、文字列ではなく数式として解釈される可能性があるためです。
Pythonのcsvモジュールが数式を実行するわけではありません。問題になるのは、担当者や取引先がCSVを表計算ソフトで開く段階です。今回は、外部入力が入るテキスト列を検査し、数式候補があれば終了コード3で配布を止めるCLIを作ります。入力ファイルは変更しません。
CSVの引用符だけでは防げない
CSVの引用符は、カンマや改行を含む値を一つのセルとして扱うための構文です。たとえばCSV内で"=1+1"と書かれていても、読み込み後の値は=1+1です。引用符で囲むことと、数式としての解釈を止めることは別の処理です。
このスクリプトでは、先頭の半角スペースを除いた最初の文字が、次のいずれかなら数式候補として検出します。
- 半角の
=、+、-、@ - タブ、CR、LF
- 全角の
＝、＋、－、＠
全角文字は、日本語環境を考慮した保守的な検査です。実際の解釈は、表計算ソフト、バージョン、地域設定、読み込み方法によって変わります。この検査だけで、すべての環境における安全性を保証することはできません。
入出力の決まり
- 入力はヘッダー付き、カンマ区切りのUTF-8 CSVとする。UTF-8 BOMの有無はどちらも受け付ける。
--text-columnで指定した列だけを検査する。- 外部入力が入るテキスト列は漏れなく指定する。
- 数値列は事前のスキーマ検査で数値だけに制限する。負数を許可するためだけに、未検証の列を検査対象から外さない。
- セルの元の値はJSONへ出さず、論理レコード番号、列名、検出文字だけを記録する。
- 入力ファイルは変更せず、変換ファイルも作らない。
数式候補を自動的に書き換えないのは、先頭へのアポストロフィやタブの追加が、すべての表計算ソフトと後続処理で同じ意味になるとは限らないためです。検出時は配布を止め、元データまたはCSVを作る処理を修正します。
検査スクリプト
次のコードをcsv_formula_guard.pyとして保存します。Python 3.8以降の標準ライブラリだけを使います。
#!/usr/bin/env python3
import argparse
import csv
import json
import sys
from pathlib import Path
DANGEROUS_PREFIXES = frozenset("=+-@\t\r\n＝＋－＠")
LEADING_SPACES = " "
MAX_DETAILS = 100
class InputError(Exception):
pass
def formula_trigger(value):
candidate = value.lstrip(LEADING_SPACES)
if not candidate:
return None
first = candidate[0]
if first in DANGEROUS_PREFIXES:
return first
return None
def validate_header(header):
if not header:
raise InputError("ヘッダーがありません")
if any(not name.strip() for name in header):
raise InputError("空の列名があります")
if any(name != name.strip() for name in header):
raise InputError("列名の前後に空白があります")
if len(header) != len(set(header)):
raise InputError("列名が重複しています")
def unique_columns(names):
result = []
for name in names:
if name not in result:
result.append(name)
return result
def select_columns(header, names):
missing = [name for name in names if name not in header]
if missing:
raise InputError("対象列がありません: " + ", ".join(missing))
return [(header.index(name), name) for name in names]
def scan_csv(path, columns):
finding_count = 0
details = []
try:
with path.open("r", encoding="utf-8-sig", newline="") as source:
reader = csv.reader(source, strict=True)
try:
header = next(reader)
except StopIteration as exc:
raise InputError("CSVが空です") from exc
validate_header(header)
selected = select_columns(header, columns)
for record_number, row in enumerate(reader, start=2):
if len(row) != len(header):
raise InputError(
f"レコード{record_number}: 列数が{len(row)}です"
f"（期待値: {len(header)}）"
)
for index, column_name in selected:
trigger = formula_trigger(row[index])
if trigger is None:
continue
finding_count += 1
if len(details) < MAX_DETAILS:
details.append({
"record": record_number,
"column": column_name,
"trigger": trigger,
})
except UnicodeDecodeError as exc:
raise InputError(f"{path}: UTF-8として読めません") from exc
except csv.Error as exc:
raise InputError(f"{path}: CSV構文エラー: {exc}") from exc
return finding_count, details
def build_parser():
parser = argparse.ArgumentParser(
description="CSVの指定列にある数式候補を検査します"
)
parser.add_argument("input", type=Path)
parser.add_argument(
"--text-column",
dest="text_columns",
action="append",
required=True,
help="検査する列名。複数回指定できます",
)
return parser
def print_report(report):
print(json.dumps(
report,
ensure_ascii=False,
separators=(",", ":"),
))
def main():
args = build_parser().parse_args()
columns = unique_columns(args.text_columns)
try:
finding_count, details = scan_csv(args.input, columns)
except (InputError, OSError) as exc:
print(f"error: {exc}", file=sys.stderr)
return 1
report = {
"status": "blocked" if finding_count else "ok",
"input": str(args.input),
"checked_columns": columns,
"finding_count": finding_count,
"findings": details,
"findings_truncated": finding_count > len(details),
}
print_report(report)
return 3 if finding_count else 0
if __name__ == "__main__":
sys.exit(main())
全レコードをリストへ保存せず、1レコードずつ検査します。詳細は最大100件に制限しますが、finding_countは最後まで数えます。
検出結果を再現する
次の内容をUTF-8のsample.csvとして保存します。amount列は、別工程で数値として検証済みという前提です。
id,customer_name,memo,amount
1,株式会社A,折り返し連絡,1200
2,=1+1,通常,900
3,株式会社C," +SUM(1,2)",-500
外部入力が入るcustomer_name列とmemo列を検査します。
python3 csv_formula_guard.py sample.csv --text-column customer_name --text-column memo標準出力は次の1行です。
{"status":"blocked","input":"sample.csv","checked_columns":["customer_name","memo"],"finding_count":2,"findings":[{"record":3,"column":"customer_name","trigger":"="},{"record":4,"column":"memo","trigger":"+"}],"findings_truncated":false}終了コードは3です。recordは物理的な行番号ではなく、ヘッダーを1番としたCSV上の論理レコード番号です。セル内に改行があっても、一つのレコードとして数えます。
amount列の-500は検査対象ではないため、数式候補には数えません。ただし、数値列へ文字列を混入できる状態であれば、この運用は安全ではありません。先に整数や小数として解析できることを検査してください。
終了コードを後続処理へ渡す
| 終了コード | 意味 | 後続処理 |
|---|---|---|
| 0 | 数式候補がない | 次の工程へ進める |
| 1 | ファイル、文字コード、CSV構文、ヘッダーまたは列数のエラー | 入力を修正して再実行する |
| 2 | コマンド引数の誤り | ファイル名と列指定を確認する |
| 3 | 数式候補を検出した | 配布を止め、元データを確認する |
入力エラーは標準エラー出力へ書き、JSONは出力しません。後続処理ではJSONの文言だけで判断せず、終了コードを確認します。
業務フローへ入れる順番
- 第159回の入口で文字コードをUTF-8へそろえる。
- 固定ヘッダー、列数、必須項目、数値型などをスキーマ検査する。
- 外部入力が入るテキスト列をすべて指定して、このスクリプトを実行する。
- 終了コード3なら配布を止め、該当レコードの元データを修正する。
- 修正後に同じ条件で再実行する。
- 終了コード0を確認したうえで、実際の受け渡し先と同じ方法でも読み込み結果を確認する。
この検査の制約
- 対象は指定した列だけです。列指定の漏れは検出できません。
- ヘッダー自体は数式候補として検査しません。外部入力から列名を作らず、固定スキーマと照合してください。
- CSVの区切り文字はカンマ、引用符はダブルクォートという
csvモジュールの既定設定です。 strict=Trueでも、あらゆる不正形式を判定できるわけではありません。列数と固定スキーマの検査を併用します。- 検出文字の集合は保守的な規則です。表計算ソフト固有の挙動や将来の変更までは判定しません。
- 終了コード0は、この規則に一致する値がなかったことを示すだけです。受け渡し先での安全性を証明するものではありません。
最初に行うのは、顧客名、備考、問い合わせ内容など、外部入力が入る列の洗い出しです。検出後に自動変換するのではなく、終了コード3で配布を止めるところまで業務手順に組み込みます。