#PythonのCSV処理のメモ
指定した列の値をKV型辞書ファイルの値で置換する
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
replace_by_kv.py
CSVファイルの指定した列の値を、KVファイル(ヘッダー無し・2列)の
1列目とマッチしたら2列目の値に置換して出力する。
使用例:
python replace_by_kv.py -i data.csv -k kv.csv -c 3 -o out.csv
python replace_by_kv.py -i data.csv -k kv.csv -c 商品コード -o out.csv --has-header
"""
import argparse
import codecs
import csv
import os
import sys
# BOM無しの場合に試す順(UTF-8を先に判定し、失敗したらcp932)
ENCODING_CANDIDATES = ["utf-8", "cp932"]
def detect_encoding(path, specified=None):
"""指定が無ければ候補を順に試して読めるエンコーディングを返す"""
if specified and specified.lower() != "auto":
return specified
with open(path, "rb") as f:
raw = f.read()
if raw.startswith(codecs.BOM_UTF8):
return "utf-8-sig"
for enc in ENCODING_CANDIDATES:
try:
raw.decode(enc)
return enc
except UnicodeDecodeError:
continue
raise UnicodeDecodeError(
"auto", b"", 0, 1,
"エンコーディングを判定できません。--encoding で明示してください: " + path
)
def load_kv(path, encoding, strip_values=True):
"""KVファイル(ヘッダー無し2列)を辞書として読み込む"""
mapping = {}
with open(path, "r", encoding=encoding, newline="") as f:
for lineno, row in enumerate(csv.reader(f), start=1):
if not row or all(c.strip() == "" for c in row):
continue # 空行はスキップ
if len(row) < 2:
print(f"[警告] KVファイル {lineno} 行目: 列が2つ未満のためスキップ: {row}",
file=sys.stderr)
continue
key, value = row[0], row[1]
if strip_values:
key, value = key.strip(), value.strip()
if key in mapping and mapping[key] != value:
print(f"[警告] KVファイル {lineno} 行目: キー '{key}' が重複。"
f"後の値 '{value}' で上書きします。", file=sys.stderr)
mapping[key] = value
return mapping
def resolve_column(col_spec, header):
"""
列指定を0始まりのインデックスに変換する。
数字なら1始まりの列番号、それ以外はヘッダー名として扱う。
"""
if col_spec.isdigit():
idx = int(col_spec) - 1
if idx < 0:
raise ValueError("列番号は1以上を指定してください。")
return idx
if header is None:
raise ValueError("列名で指定する場合は --has-header が必要です。")
if col_spec not in header:
raise ValueError(f"列名 '{col_spec}' がヘッダーに見つかりません。ヘッダー: {header}")
return header.index(col_spec)
def main():
parser = argparse.ArgumentParser(
description="CSVの指定列をKVファイル(ヘッダー無し2列)で置換します。")
parser.add_argument("-i", "--input", required=True, help="入力CSVファイル")
parser.add_argument("-k", "--kv", required=True, help="KVファイル(ヘッダー無し2列)")
parser.add_argument("-o", "--output", required=True, help="出力CSVファイル")
parser.add_argument("-c", "--column", required=True,
help="置換対象の列。1始まりの列番号、または列名(--has-header時)")
parser.add_argument("--has-header", action="store_true",
help="入力CSVの1行目をヘッダーとして扱う(そのまま出力)")
parser.add_argument("--encoding", default="auto",
help="入出力の文字コード。既定はauto(utf-8-sig/cp932を自動判定)")
parser.add_argument("--output-encoding", default=None,
help="出力の文字コード。未指定なら入力と同じ")
parser.add_argument("--no-strip", action="store_true",
help="比較時に前後の空白を除去しない")
parser.add_argument("--report-unmatched", action="store_true",
help="マッチしなかった値を標準エラーに一覧表示する")
args = parser.parse_args()
for path in (args.input, args.kv):
if not os.path.isfile(path):
print(f"[エラー] ファイルが見つかりません: {path}", file=sys.stderr)
return 1
strip = not args.no_strip
try:
in_enc = detect_encoding(args.input, args.encoding)
kv_enc = detect_encoding(args.kv, args.encoding)
except UnicodeDecodeError as e:
print(f"[エラー] {e}", file=sys.stderr)
return 1
out_enc = args.output_encoding or in_enc
mapping = load_kv(args.kv, kv_enc, strip_values=strip)
if not mapping:
print("[エラー] KVファイルから有効な行を読み込めませんでした。", file=sys.stderr)
return 1
replaced = 0
total = 0
unmatched = set()
# newline="" はCSV読み書きの必須指定(Windowsでの改行崩れ防止)
with open(args.input, "r", encoding=in_enc, newline="") as fin, \
open(args.output, "w", encoding=out_enc, newline="") as fout:
reader = csv.reader(fin)
writer = csv.writer(fout)
header = None
if args.has_header:
try:
header = next(reader)
except StopIteration:
print("[エラー] 入力CSVが空です。", file=sys.stderr)
return 1
try:
col = resolve_column(args.column, header)
except ValueError as e:
print(f"[エラー] {e}", file=sys.stderr)
return 1
if header is not None:
writer.writerow(header)
for lineno, row in enumerate(reader, start=2 if header else 1):
if len(row) <= col:
print(f"[警告] {lineno} 行目: 列数が足りないためそのまま出力します。",
file=sys.stderr)
writer.writerow(row)
continue
total += 1
key = row[col].strip() if strip else row[col]
if key in mapping:
row[col] = mapping[key]
replaced += 1
else:
unmatched.add(key)
writer.writerow(row)
print(f"完了: {args.output}")
print(f" 対象行数 : {total}")
print(f" 置換件数 : {replaced}")
print(f" 未一致 : {total - replaced}")
print(f" 文字コード: 入力={in_enc} / 出力={out_enc}")
if args.report_unmatched and unmatched:
print("--- 未一致の値 ---", file=sys.stderr)
for v in sorted(unmatched):
print(v, file=sys.stderr)
return 0
if __name__ == "__main__":
sys.exit(main())
2つのCSVファイルを交互に挟み込む形でマージする
# -*- coding: utf-8 -*-
"""
ヘッダー無しCSVファイル2つを、1行ずつ交互に挟み込んでマージする。
出力の並び:
A_1行目, B_1行目, A_2行目, B_2行目, ...
使い方:
1) 下の設定を書き換えてダブルクリック実行 or python merge_csv.py
2) コマンドライン指定:
python merge_csv.py A.csv B.csv merged.csv
"""
import csv
import sys
from itertools import zip_longest
# ----------------- 設定（ここを書き換える） -----------------
FILE_A = r"C:\data\A.csv"
FILE_B = r"C:\data\B.csv"
FILE_OUT = r"C:\data\merged.csv"
# Excelで作った日本語CSVなら "cp932"、UTF-8のCSVなら "utf-8-sig"
ENCODING_IN = "cp932"
ENCODING_OUT = "cp932"
# ------------------------------------------------------------
_MISSING = object()
def read_rows(path, encoding):
"""CSVを読み込んで行のリストを返す。"""
with open(path, "r", encoding=encoding, newline="") as f:
return list(csv.reader(f))
def interleave(rows_a, rows_b):
"""2つの行リストを1行ずつ交互に並べる。行数が違っても余りはそのまま出力。"""
merged = []
for row_a, row_b in zip_longest(rows_a, rows_b, fillvalue=_MISSING):
if row_a is not _MISSING:
merged.append(row_a)
if row_b is not _MISSING:
merged.append(row_b)
return merged
def write_rows(path, rows, encoding):
"""CSVとして書き出す。newline="" は空行混入を防ぐため必須。"""
with open(path, "w", encoding=encoding, newline="") as f:
csv.writer(f).writerows(rows)
def main():
if len(sys.argv) == 4:
file_a, file_b, file_out = sys.argv[1], sys.argv[2], sys.argv[3]
elif len(sys.argv) == 1:
file_a, file_b, file_out = FILE_A, FILE_B, FILE_OUT
else:
print("使い方: python merge_csv.py A.csv B.csv merged.csv")
return 1
try:
rows_a = read_rows(file_a, ENCODING_IN)
rows_b = read_rows(file_b, ENCODING_IN)
except FileNotFoundError as e:
print("ファイルが見つかりません:", e.filename)
return 1
except UnicodeDecodeError:
print("文字コードが違う可能性があります。ENCODING_IN を 'utf-8-sig' に変えて再実行してください。")
return 1
if len(rows_a) != len(rows_b):
print("警告: 行数が異なります (A={}行, B={}行)。余った行はそのまま末尾に出力します。"
.format(len(rows_a), len(rows_b)))
merged = interleave(rows_a, rows_b)
write_rows(file_out, merged, ENCODING_OUT)
print("完了: {} 行を書き出しました -> {}".format(len(merged), file_out))
return 0
if __name__ == "__main__":
sys.exit(main())
指定した列を指定した桁数で0パディング
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
zero_pad_csv.py
ヘッダー無しCSVファイルの指定した列を、指定した桁数になるまで
先頭に "0" を詰めて（ゼロパディングして）出力するツール。
例) 値が "3" で桁数が 5 のとき -> "00003"
使い方:
python zero_pad_csv.py input.csv output.csv --column 2 --width 5
python zero_pad_csv.py input.csv --column 1,3 --width 8 --encoding cp932
"""
import argparse
import csv
import os
import sys
def parse_columns(text):
""" "2" や "1,3,5" のような文字列を 1 始まりの列番号リストに変換する。"""
columns = []
for part in text.split(","):
part = part.strip()
if not part:
continue
if not part.isdigit():
raise argparse.ArgumentTypeError(
"列番号は 1 以上の整数（カンマ区切り）で指定してください: %r" % part
)
number = int(part)
if number < 1:
raise argparse.ArgumentTypeError("列番号は 1 以上で指定してください: %d" % number)
columns.append(number)
if not columns:
raise argparse.ArgumentTypeError("列番号が指定されていません。")
return columns
def pad_value(value, width, strip_spaces):
"""1 つのセルの値をゼロパディングして返す。"""
if strip_spaces:
value = value.strip()
# 空文字はそのまま（必要なら下の 1 行を有効にして "00000" にできます）
# if value == "":
# return "0" * width
if value == "" or len(value) >= width:
return value
return value.rjust(width, "0")
def convert(src, dst, columns, width, encoding, delimiter, strip_spaces, quote_all):
"""CSV を 1 行ずつ読み込み、指定列をパディングして書き出す。"""
zero_based = [c - 1 for c in columns]
quoting = csv.QUOTE_ALL if quote_all else csv.QUOTE_MINIMAL
changed = 0
total = 0
short_rows = 0
# 読み込み時のみ BOM を許容する（出力はBOM 無し）
read_encoding = encoding
if encoding.lower().replace("_", "-") in ("utf-8", "utf8"):
read_encoding = "utf-8-sig"
# newline="" は csv モジュール使用時の必須指定（改行の二重出力を防ぐ）
with open(src, "r", encoding=read_encoding, newline="") as fin, \
open(dst, "w", encoding=encoding, newline="") as fout:
reader = csv.reader(fin, delimiter=delimiter)
writer = csv.writer(fout, delimiter=delimiter, quoting=quoting,
lineterminator="\r\n")
for row_number, row in enumerate(reader, start=1):
total += 1
for index in zero_based:
if index >= len(row):
short_rows += 1
print("警告: %d 行目に %d 列目がありません。スキップします。"
% (row_number, index + 1), file=sys.stderr)
continue
before = row[index]
after = pad_value(before, width, strip_spaces)
if before != after:
row[index] = after
changed += 1
writer.writerow(row)
return total, changed, short_rows
def main():
parser = argparse.ArgumentParser(
description="ヘッダー無しCSVの指定列を指定桁数までゼロパディングします。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="例: python zero_pad_csv.py in.csv out.csv -c 2 -w 5",
)
parser.add_argument("input", help="入力CSVファイル（ヘッダー無し）")
parser.add_argument("output", nargs="?",
help="出力CSVファイル（省略時は <入力名>_padded.csv）")
parser.add_argument("-c", "--column", required=True, type=parse_columns,
help="対象の列番号。1 始まり。カンマ区切りで複数指定可 (例: 2 または 1,3)")
parser.add_argument("-w", "--width", required=True, type=int,
help="桁数（この桁数になるまで先頭に 0 を詰める）")
parser.add_argument("-e", "--encoding", default="utf-8",
help="文字コード。既定: utf-8（BOM無し）。"
"Excel由来の日本語CSVは cp932、BOM付きは utf-8-sig を指定")
parser.add_argument("-d", "--delimiter", default=",",
help="区切り文字。既定: ,（タブは \\t と指定）")
parser.add_argument("--strip", action="store_true",
help="値の前後の空白を除去してから判定・パディングする")
parser.add_argument("--quote-all", action="store_true",
help="出力時に全項目をダブルクォートで囲む")
args = parser.parse_args()
if args.width < 1:
parser.error("--width は 1 以上を指定してください。")
if args.delimiter == "\\t":
args.delimiter = "\t"
if not os.path.isfile(args.input):
print("エラー: 入力ファイルが見つかりません: %s" % args.input, file=sys.stderr)
return 1
output = args.output
if not output:
base, ext = os.path.splitext(args.input)
output = base + "_padded" + (ext or ".csv")
if os.path.abspath(output) == os.path.abspath(args.input):
print("エラー: 入力ファイルと出力ファイルが同一です。", file=sys.stderr)
return 1
try:
total, changed, short_rows = convert(
args.input, output, args.column, args.width,
args.encoding, args.delimiter, args.strip, args.quote_all,
)
except UnicodeDecodeError:
print("エラー: 文字コードを読み取れませんでした。"
"--encoding cp932 などを試してください。", file=sys.stderr)
return 1
except OSError as e:
print("エラー: ファイル操作に失敗しました: %s" % e, file=sys.stderr)
return 1
print("完了: %s -> %s" % (args.input, output))
print(" 処理行数 : %d" % total)
print(" パディングした件数: %d" % changed)
if short_rows:
print(" 列が足りない箇所 : %d" % short_rows)
return 0
if __name__ == "__main__":
sys.exit(main())
CSVソート
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
sort_csv.py
CSVファイルを、指定した複数の列をソートキーとして並べ替えます。
【使い方】
python sort_csv.py 入力.csv 出力.csv --key 列名1 --key 列名2:desc
例:
# "部署"列を昇順、その中で"売上"列を降順にソート
python sort_csv.py input.csv output.csv --key 部署 --key 売上:desc
# 列を数値として比較したい場合は :num を付ける (例: 売上:desc:num)
python sort_csv.py input.csv output.csv --key 売上:desc:num
# ヘッダーが無いCSVで、列番号(0始まり)を指定する場合
python sort_csv.py input.csv output.csv --key 0 --key 2:desc --no-header
# 文字コードやCSVの区切り文字を指定する場合
python sort_csv.py input.csv output.csv --key 列名1 --encoding cp932 --delimiter ";"
【文字コードの既定動作】
入力: utf-8-sig … BOM付きUTF-8のBOMを自動で取り除いて読み込みます(BOMなしUTF-8もそのまま読めます)
出力: utf-8 … BOMなしUTF-8で書き出します
変更したい場合は --encoding / --output-encoding を使用してください。
【--key の書式】
列名 または 列番号 [ : asc|desc ] [ : str|num ]
- asc/desc を省略すると昇順(asc)
- str/num を省略すると文字列として比較(str)。数値列は num を付けると数値として正しくソートされます。
--key は複数回指定でき、先に指定したものが優先される主キーになります。
標準ライブラリ(csv, operator)のみを使用しているため、pandasのインストールは不要です。
"""
import argparse
import csv
import functools
import sys
def parse_key_spec(spec):
""""列名:asc|desc:str|num" 形式の文字列を解析する"""
parts = spec.split(":")
col = parts[0]
order = "asc"
kind = "str"
for extra in parts[1:]:
extra_lower = extra.lower()
if extra_lower in ("asc", "desc"):
order = extra_lower
elif extra_lower in ("str", "num"):
kind = extra_lower
else:
raise ValueError(f"--key の指定が不正です: {spec!r} (asc/desc/str/num のみ指定可能)")
return {"col": col, "order": order, "kind": kind}
def resolve_column_index(col, header, no_header):
"""列名または列番号(文字列)を実際のインデックスに変換する"""
if no_header or header is None:
try:
return int(col)
except ValueError:
raise ValueError(
f"--no-header 指定時の --key には列番号(0始まりの整数)を指定してください: {col!r}"
)
if col in header:
return header.index(col)
# ヘッダーに見つからない場合、数値指定として扱えるか試す
try:
idx = int(col)
if 0 <= idx < len(header):
return idx
except ValueError:
pass
raise ValueError(f"列 {col!r} がヘッダーに見つかりません。ヘッダー: {header}")
def make_sort_value(value, kind):
if kind == "num":
try:
return float(value)
except (TypeError, ValueError):
# 数値に変換できない場合は最小値扱いにして末尾/先頭に寄せる
return float("-inf")
return value if value is not None else ""
def build_comparator(keys_resolved):
"""複数キー・列ごとの昇順/降順混在に対応した比較関数を作る"""
def compare(row_a, row_b):
for k in keys_resolved:
idx = k["idx"]
va = make_sort_value(row_a[idx] if idx < len(row_a) else "", k["kind"])
vb = make_sort_value(row_b[idx] if idx < len(row_b) else "", k["kind"])
if va == vb:
continue
result = -1 if va < vb else 1
return -result if k["order"] == "desc" else result
return 0
return compare
def main():
parser = argparse.ArgumentParser(
description="CSVファイルを指定した複数列でソートします。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
parser.add_argument("input", help="入力CSVファイルのパス")
parser.add_argument("output", help="出力CSVファイルのパス")
parser.add_argument(
"--key",
action="append",
required=True,
dest="keys",
metavar="列名[:asc|desc][:str|num]",
help="ソートキーとする列。複数回指定可能(先に指定した方が優先)",
)
parser.add_argument(
"--no-header",
action="store_true",
help="入力CSVにヘッダー行が無い場合に指定。--key には列番号(0始まり)を使用",
)
parser.add_argument(
"--encoding",
default="utf-8-sig",
help="入力ファイルの文字コード (既定値: utf-8-sig = BOM付き/なしUTF-8のどちらも読める。"
"日本語ExcelのCSVなら cp932)",
)
parser.add_argument(
"--output-encoding",
default="utf-8",
help="出力ファイルの文字コード (既定値: utf-8 = BOMなし。BOM付きにしたい場合は utf-8-sig)",
)
parser.add_argument(
"--delimiter",
default=",",
help="CSVの区切り文字 (既定値: ',')",
)
args = parser.parse_args()
try:
key_specs = [parse_key_spec(s) for s in args.keys]
except ValueError as e:
print(f"エラー: {e}", file=sys.stderr)
sys.exit(1)
try:
with open(args.input, "r", encoding=args.encoding, newline="") as f:
reader = csv.reader(f, delimiter=args.delimiter)
rows = list(reader)
except FileNotFoundError:
print(f"エラー: 入力ファイルが見つかりません: {args.input}", file=sys.stderr)
sys.exit(1)
except UnicodeDecodeError as e:
print(
f"エラー: 文字コードの読み込みに失敗しました ({args.encoding})。"
f" --encoding cp932 などを試してください。詳細: {e}",
file=sys.stderr,
)
sys.exit(1)
if not rows:
print("エラー: CSVが空です。", file=sys.stderr)
sys.exit(1)
if args.no_header:
header = None
data_rows = rows
else:
header = rows[0]
data_rows = rows[1:]
try:
keys_resolved = []
for spec in key_specs:
idx = resolve_column_index(spec["col"], header, args.no_header)
keys_resolved.append({"idx": idx, "order": spec["order"], "kind": spec["kind"]})
except ValueError as e:
print(f"エラー: {e}", file=sys.stderr)
sys.exit(1)
comparator = build_comparator(keys_resolved)
data_rows.sort(key=functools.cmp_to_key(comparator))
with open(args.output, "w", encoding=args.output_encoding, newline="") as f:
writer = csv.writer(f, delimiter=args.delimiter)
if header is not None:
writer.writerow(header)
writer.writerows(data_rows)
print(f"完了: {len(data_rows)} 行を並べ替えて {args.output} に書き出しました。")
if __name__ == "__main__":
main()
指定した行を抽出
# -*- coding: utf-8 -*-
"""
BOM付きUTF-8のCSVから、行番号リストのテキストファイルで指定した行だけを
抜き出し、BOMなしUTF-8のCSVとして書き出す。
使い方（コマンドプロンプト / PowerShell）:
python extract_rows.py 入力.csv 出力.csv 行番号リスト.txt
行番号リスト.txt は1行に1つ数字を並べたテキスト。
例)
2
5
10
"10-20" のような範囲指定や、"2,5,10" のようなカンマ区切りも書ける。
空行と # から始まる行は無視する。
文字コードは UTF-8（BOM有無どちらでも可）/ シフトJIS 自動判定。
オプション:
--no-header ヘッダー行を持たないCSVとして扱う（先頭行から1で数える）
--lf 出力の改行コードをLFにする（既定はCRLF）
"""
import argparse
import csv
import sys
def read_row_numbers(path):
"""行番号リストのテキストファイルを読み、行番号の集合を返す。"""
text = None
for enc in ("utf-8-sig", "cp932"):
try:
with open(path, "r", encoding=enc) as f:
text = f.read()
break
except UnicodeDecodeError:
continue
if text is None:
raise SystemExit(f"行番号リストの文字コードを判別できません: {path}")
rows = set()
for lineno, line in enumerate(text.splitlines(), start=1):
line = line.split("#", 1)[0].strip()
if not line:
continue
for part in line.replace("\t", ",").split(","):
part = part.strip()
if not part:
continue
try:
if "-" in part:
start, end = part.split("-", 1)
rows.update(range(int(start), int(end) + 1))
else:
rows.add(int(part))
except ValueError:
raise SystemExit(
f"行番号リストの {lineno} 行目を数値として読めません: {part!r}"
)
return rows
def main():
parser = argparse.ArgumentParser()
parser.add_argument("src", help="入力CSV（BOM付きUTF-8）")
parser.add_argument("dst", help="出力CSV（BOMなしUTF-8）")
parser.add_argument("list", help="抽出する行番号を並べたテキストファイル")
parser.add_argument("--no-header", action="store_true",
help="ヘッダー行を持たないCSVとして扱う")
parser.add_argument("--lf", action="store_true",
help="改行コードをLFにする（既定はCRLF）")
args = parser.parse_args()
targets = read_row_numbers(args.list)
if not targets:
raise SystemExit("行番号リストに有効な行番号がありません。")
newline_code = "\n" if args.lf else "\r\n"
# encoding="utf-8-sig" で読むとBOMは自動的に取り除かれる
# encoding="utf-8" で書くとBOMは付かない
# newline="" はcsvモジュールを使うときのお約束（改行の二重化を防ぐ）
with open(args.src, "r", encoding="utf-8-sig", newline="") as f_in, \
open(args.dst, "w", encoding="utf-8", newline="") as f_out:
reader = csv.reader(f_in)
writer = csv.writer(f_out, lineterminator=newline_code)
if not args.no_header:
try:
writer.writerow(next(reader))
except StopIteration:
raise SystemExit("入力ファイルが空です。")
found = set()
total = 0
for i, row in enumerate(reader, start=1):
total = i
if i in targets:
writer.writerow(row)
found.add(i)
print(f"指定 {len(targets)} 行 / 抽出 {len(found)} 行 -> {args.dst}")
missing = sorted(targets - found)
if missing:
head = ", ".join(str(n) for n in missing[:10])
more = f" ほか{len(missing) - 10}件" if len(missing) > 10 else ""
print(f"注意: データ行は {total} 行しかないため、"
f"見つからなかった行番号があります: {head}{more}",
file=sys.stderr)
if __name__ == "__main__":
main()
# -*- coding: utf-8 -*-
"""
CSV セット行抽出スクリプト（Windows / pandas）
【処理内容】
KEY_COLS で指定した 3 列の値がすべて同じ行どうしをグループとみなし、
そのグループ内で FLAG_COL の値が VAL_KEEP("2") の行と VAL_DROP("6") の行を
出現順に 1 対 1 でペアにする。
ペアが成立した分だけ "6" の行を削除し、"2" の行はそのまま残す。
・2 だけの行 → そのまま残る
・6 だけの行 → そのまま残る
・ペアが余った 6 → そのまま残る（1 対 1 でペアにするため）
・2 でも 6 でもない行 → そのまま残る
・行の並び順は元の CSV のまま
【使い方】
1) 下の「設定」を自分の CSV に合わせて書き換える
2) コマンドプロンプトで:
python filter_pairs.py
引数で指定することもできる:
python filter_pairs.py 入力.csv 出力.csv
"""
import sys
import os
import pandas as pd
# ============================ 設定 ============================
# 入力・出力ファイル（コマンドライン引数があればそちらが優先）
INPUT_FILE = r"C:\data\input.csv"
OUTPUT_FILE = r"C:\data\output.csv"
# 同一かどうかを判定する 3 つの列（ヘッダー名。HAS_HEADER=False なら 0 始まりの列番号）
KEY_COLS = ["列A", "列B", "列C"]
# 2 / 6 が入っている列
FLAG_COL = "区分"
# 残す値 / セット成立時に削除する値
VAL_KEEP = "2"
VAL_DROP = "6"
# CSV に見出し行があるか
HAS_HEADER = True
# 文字コード（None なら自動判定）。固定したい場合は "cp932" や "utf-8-sig" を直接指定
ENCODING = None
# 削除した 6 の行を別ファイルに保存したい場合はパスを指定（不要なら None）
REMOVED_FILE = None # 例: r"C:\data\removed.csv"
# ==============================================================
def detect_encoding(path):
"""UTF-8(BOM) → UTF-8 → cp932 の順に試して、読めた文字コードを返す"""
for enc in ("utf-8-sig", "utf-8", "cp932"):
try:
with open(path, "r", encoding=enc) as f:
f.read()
return enc
except UnicodeDecodeError:
continue
raise UnicodeDecodeError(
"auto", b"", 0, 1,
"文字コードを自動判定できませんでした。ENCODING に直接指定してください。"
)
def norm(series):
"""比較用に文字列化して前後の空白を除去（'2' と 2 と ' 2 ' を同じ扱いにする）"""
return series.astype(str).str.strip()
def main():
in_path = sys.argv[1] if len(sys.argv) > 1 else INPUT_FILE
out_path = sys.argv[2] if len(sys.argv) > 2 else OUTPUT_FILE
if not os.path.exists(in_path):
print(f"[エラー] 入力ファイルが見つかりません: {in_path}")
sys.exit(1)
enc = ENCODING or detect_encoding(in_path)
print(f"文字コード: {enc}")
# すべて文字列として読み込む（先頭ゼロや書式を壊さないため）
df = pd.read_csv(
in_path,
dtype=str,
keep_default_na=False,
header=0 if HAS_HEADER else None,
encoding=enc,
)
# 列指定のチェック
for col in list(KEY_COLS) + [FLAG_COL]:
if col not in df.columns:
print(f"[エラー] 列が見つかりません: {col}")
print(f" 実際の列: {list(df.columns)}")
sys.exit(1)
# --- 判定用の正規化した一時列を作る ---
work = pd.DataFrame(index=df.index)
for i, col in enumerate(KEY_COLS):
work[f"_k{i}"] = norm(df[col])
work["_flag"] = norm(df[FLAG_COL])
key_cols_tmp = [f"_k{i}" for i in range(len(KEY_COLS))]
# 削除対象の行（インデックス）を集める
drop_index = []
for _, grp in work.groupby(key_cols_tmp, sort=False, dropna=False):
keep_rows = grp.index[grp["_flag"] == VAL_KEEP] # 2 の行（出現順）
drop_rows = grp.index[grp["_flag"] == VAL_DROP] # 6 の行（出現順）
n_pair = min(len(keep_rows), len(drop_rows)) # 1 対 1 でペアにできる数
if n_pair > 0:
# ペアになった分だけ 6 の行を削除（余った 6 は残す）
drop_index.extend(drop_rows[:n_pair])
result = df.drop(index=drop_index)
# --- 出力 ---
result.to_csv(out_path, index=False, header=HAS_HEADER, encoding=enc)
if REMOVED_FILE and drop_index:
df.loc[drop_index].to_csv(
REMOVED_FILE, index=False, header=HAS_HEADER, encoding=enc
)
print(f"入力行数 : {len(df)}")
print(f"削除行数 : {len(drop_index)} （セット成立した {VAL_DROP} の行）")
print(f"出力行数 : {len(result)}")
print(f"出力先 : {out_path}")
if __name__ == "__main__":
main()
CSV 2列置換
# -*- coding: utf-8 -*-
"""
CSV の 2列目・3列目のペアを、変換表にしたがって置換するスクリプト。
使い方（どちらでもOK）:
1) 下の「設定」の3つのパスを書き換えて、ダブルクリック / python replace_pairs.py
2) コマンドラインから:
python replace_pairs.py 入力.csv 変換表.csv 出力.csv
変換表 (mapping.csv) の形式: ヘッダー無し・4列
旧2列目,旧3列目,新2列目,新3列目
例:
3310245,09021234,3589134,52384971
"""
import csv
import os
import sys
# ============================ 設定 ============================
INPUT_CSV = r"C:\work\input.csv" # 置換対象のCSV（ヘッダー無し）
MAPPING_CSV = r"C:\work\mapping.csv" # 変換表（4列・ヘッダー無し）
OUTPUT_CSV = r"C:\work\output.csv" # 出力先（新規作成 / 上書き）
ENCODING = "utf-8" # 入出力の文字コード（BOM無しUTF-8）
NEWLINE = "\r\n" # 出力の改行コード（CRLF）
KEY_COL_A = 1 # 2列目 → 0始まりで 1
KEY_COL_B = 2 # 3列目 → 0始まりで 2
STRIP_FIELDS = False # True にするとキー比較時に前後の空白を無視
# ==============================================================
def load_mapping(path):
"""変換表を読み込んで {(旧2, 旧3): (新2, 新3)} を返す。"""
mapping = {}
with open(path, "r", encoding="utf-8-sig", newline="") as f:
for lineno, row in enumerate(csv.reader(f), start=1):
if not row or all(c.strip() == "" for c in row):
continue # 空行はスキップ
if len(row) < 4:
raise ValueError(
"変換表 %s の %d 行目の列数が %d です（4列必要）: %r"
% (path, lineno, len(row), row)
)
old_a, old_b, new_a, new_b = (c.strip() for c in row[:4])
key = (old_a, old_b)
if key in mapping and mapping[key] != (new_a, new_b):
raise ValueError(
"変換表 %s の %d 行目: キー %s,%s が重複していて変換先が違います"
% (path, lineno, old_a, old_b)
)
mapping[key] = (new_a, new_b)
if not mapping:
raise ValueError("変換表 %s に有効な行がありません" % path)
return mapping
def main():
args = sys.argv[1:]
if len(args) == 3:
input_csv, mapping_csv, output_csv = args
elif len(args) == 0:
input_csv, mapping_csv, output_csv = INPUT_CSV, MAPPING_CSV, OUTPUT_CSV
else:
print("使い方: python replace_pairs.py [入力.csv 変換表.csv 出力.csv]")
return 2
for label, path in (("入力CSV", input_csv), ("変換表", mapping_csv)):
if not os.path.isfile(path):
print("エラー: %s が見つかりません -> %s" % (label, path))
return 1
if os.path.abspath(input_csv) == os.path.abspath(output_csv):
print("エラー: 入力CSVと出力CSVが同じファイルです。別名にしてください。")
return 1
mapping = load_mapping(mapping_csv)
hit_count = {key: 0 for key in mapping}
total_rows = 0
replaced_rows = 0
short_rows = [] # 列数が足りない行
with open(input_csv, "r", encoding=ENCODING, newline="") as fin, \
open(output_csv, "w", encoding=ENCODING, newline="") as fout:
reader = csv.reader(fin)
writer = csv.writer(fout, lineterminator=NEWLINE)
for lineno, row in enumerate(reader, start=1):
if not row:
writer.writerow(row)
continue
total_rows += 1
if len(row) <= KEY_COL_B:
short_rows.append(lineno)
writer.writerow(row)
continue
a, b = row[KEY_COL_A], row[KEY_COL_B]
key = (a.strip(), b.strip()) if STRIP_FIELDS else (a, b)
new_pair = mapping.get(key)
if new_pair is not None:
row[KEY_COL_A], row[KEY_COL_B] = new_pair
replaced_rows += 1
hit_count[key] += 1
writer.writerow(row)
# ------------------------- 結果レポート -------------------------
unused = [k for k, n in hit_count.items() if n == 0]
print("完了しました。")
print(" 出力ファイル : %s" % os.path.abspath(output_csv))
print(" 読み込み行数 : %d" % total_rows)
print(" 置換した行数 : %d" % replaced_rows)
print(" 変換表ペア数 : %d （うち未使用 %d）" % (len(mapping), len(unused)))
if short_rows:
head = ", ".join(str(n) for n in short_rows[:10])
more = " ほか" if len(short_rows) > 10 else ""
print(" ※ 列数が3未満のため素通しした行: %d 行 (行番号: %s%s)"
% (len(short_rows), head, more))
if unused:
print(" ※ 一度も一致しなかった変換表のペア: %d 件" % len(unused))
for a, b in unused[:20]:
print(" %s,%s" % (a, b))
if len(unused) > 20:
print(" ... 他 %d 件" % (len(unused) - 20))
return 0
if __name__ == "__main__":
try:
code = main()
except Exception as e:
print("エラー: %s" % e)
code = 1
if not sys.argv[1:]:
# ダブルクリック起動時に画面が消えないように
try:
input("Enterキーで終了します...")
except EOFError:
pass
sys.exit(code)
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
ペアCSV検証スクリプト (Windows / Python 3.8+)
■ 概要
ペアCSV（2行1セット）を検証し、すべてのペアがデータCSVに存在するかを確認します。
結果はテキストログファイルに出力します。
■ 使い方
1) コマンドライン引数で指定する場合
python verify_pairs.py ペアCSV データCSV [ログファイル]
例: python verify_pairs.py pair.csv data.csv result.log
2) 引数を省略する場合
下の【設定】の PAIR_CSV / DATA_CSV / LOG_FILE を書き換えて、
ダブルクリック or `python verify_pairs.py` で実行してください。
■ 終了コード
0 = エラーなし / 1 = エラーあり / 2 = 実行時エラー（ファイルが無い等）
"""
import csv
import os
import re
import sys
from datetime import datetime
# ============================================================================
# 【設定】コマンドライン引数を指定しない場合は、ここを書き換えてください
# ============================================================================
PAIR_CSV = r"pair.csv" # 検証対象のペアCSV（2行1セット）
DATA_CSV = r"data.csv" # 定義データCSV（4列）
LOG_FILE = r"verify_result.log" # 出力するログファイル
# --- 入出力の文字コード -------------------------------------------------------
# 入力: UTF-8(BOM無し)想定。"utf-8-sig" にしておくとBOM付きでも読めます。
INPUT_ENCODING = "utf-8-sig"
# 出力: メモ帳・Excelで文字化けしないようBOM付きUTF-8。改行はCRLF。
OUTPUT_ENCODING = "utf-8-sig"
OUTPUT_NEWLINE = "\r\n"
# --- 列位置（0始まり）--------------------------------------------------------
COL_FLAG = 0 # 1列目: 0 / 1
COL_VAL1 = 1 # 2列目: 値（末尾に半角スペースあり → トリムして比較）
COL_VAL2 = 2 # 3列目: 値
COLS_SAME = (3, 4, 5, 6) # 4〜7列目: ペア内で同一であるべき列
COL_SETNO = 7 # 8列目: セットNo（半角数字4桁）
PAIR_MIN_COLS = 8 # ペアCSVに必要な最小列数
DATA_COLS = 4 # データCSVの列数
# --- トリム対象列（0始まり）---------------------------------------------------
# 仕様上トリムが必要なのはペアCSVの2列目のみ。
# 他の列もトリムしたい場合は、下のタプルに列番号を追加してください。
PAIR_TRIM_COLS = (COL_VAL1,)
DATA_TRIM_COLS = () # 例: 全列トリムするなら (0, 1, 2, 3)
# --- ログの詳細出力件数の上限（0 = 無制限）------------------------------------
MAX_DETAIL = 0
SETNO_PATTERN = re.compile(r"^[0-9]{4}$")
# ============================================================================
class Report(object):
"""ログ本文を組み立てるヘルパ"""
def __init__(self):
self.lines = []
def add(self, text=""):
self.lines.append(text)
def section(self, title):
self.add("")
self.add("-" * 78)
self.add(title)
self.add("-" * 78)
def details(self, items):
"""詳細行を出力（MAX_DETAIL で打ち切り）"""
if not items:
self.add(" （該当なし）")
return
shown = items if MAX_DETAIL <= 0 else items[:MAX_DETAIL]
for item in shown:
self.add(" " + item)
rest = len(items) - len(shown)
if rest > 0:
self.add(" ... 他 %d 件（MAX_DETAIL で省略）" % rest)
def text(self):
return OUTPUT_NEWLINE.join(self.lines) + OUTPUT_NEWLINE
def read_csv(path, encoding):
"""CSVを [(物理行番号, [値, ...]), ...] で返す。空行はスキップ。"""
rows = []
with open(path, "r", encoding=encoding, newline="") as f:
reader = csv.reader(f)
for fields in reader:
if not fields or all(v.strip() == "" for v in fields):
continue
rows.append((reader.line_num, fields))
return rows
def trim(fields, trim_cols):
"""指定列の前後空白を除去したリストを返す"""
out = list(fields)
for i in trim_cols:
if i < len(out):
out[i] = out[i].strip()
return out
def load_data_csv(path, report):
"""
データCSVを読み込み、(4値のタプル) -> [行番号, ...] の辞書を返す。
戻り値: (キー辞書, 行数, 形式エラー件数)
"""
rows = read_csv(path, INPUT_ENCODING)
keys = {}
bad_cols = []
for lineno, fields in rows:
if len(fields) != DATA_COLS:
bad_cols.append(
"行 %d: 列数が %d 列です（期待値 %d 列）: %s"
% (lineno, len(fields), DATA_COLS, ",".join(fields))
)
if len(fields) < DATA_COLS:
continue
f = trim(fields, DATA_TRIM_COLS)
key = (f[0], f[1], f[2], f[3])
keys.setdefault(key, []).append(lineno)
if bad_cols:
report.section("[エラー] データCSV: 列数不正")
report.add("件数: %d 件" % len(bad_cols))
report.details(bad_cols)
return keys, len(rows), len(bad_cols)
def main():
# ---- パス決定（引数優先、無ければ設定値）--------------------------------
args = sys.argv[1:]
pair_path = args[0] if len(args) >= 1 else PAIR_CSV
data_path = args[1] if len(args) >= 2 else DATA_CSV
log_path = args[2] if len(args) >= 3 else LOG_FILE
for p, label in ((pair_path, "ペアCSV"), (data_path, "データCSV")):
if not os.path.isfile(p):
sys.stderr.write("%s が見つかりません: %s\n" % (label, p))
return 2
started = datetime.now()
report = Report()
report.add("=" * 78)
report.add("ペアCSV検証結果")
report.add("=" * 78)
report.add("実行日時 : %s" % started.strftime("%Y-%m-%d %H:%M:%S"))
report.add("ペアCSV : %s" % os.path.abspath(pair_path))
report.add("データCSV : %s" % os.path.abspath(data_path))
report.add("ログファイル : %s" % os.path.abspath(log_path))
# ---- データCSV読み込み --------------------------------------------------
data_keys, data_row_count, data_col_errors = load_data_csv(data_path, report)
# データCSVの重複行
data_dups = []
for key, linenos in data_keys.items():
if len(linenos) > 1:
data_dups.append(
"%s (行 %s)" % (",".join(key), ", ".join(str(n) for n in linenos))
)
# ---- ペアCSV読み込み ----------------------------------------------------
pair_rows = read_csv(pair_path, INPUT_ENCODING)
col_errors = [] # 列数不足
groups = {} # セットNo -> [(行番号, フィールド), ...]
for lineno, fields in pair_rows:
if len(fields) < PAIR_MIN_COLS:
col_errors.append(
"行 %d: 列数が %d 列です（最低 %d 列必要）: %s"
% (lineno, len(fields), PAIR_MIN_COLS, ",".join(fields))
)
continue
f = trim(fields, PAIR_TRIM_COLS)
groups.setdefault(f[COL_SETNO], []).append((lineno, f))
# ---- ペアCSV自体の整合性チェック ---------------------------------------
setno_format_errors = []
group_size_errors = []
flag_errors = []
same_col_errors = []
valid_pairs = [] # (セットNo, 行0情報, 行1情報)
for setno, members in groups.items():
linenos = ", ".join(str(n) for n, _ in members)
# セットNo書式（半角数字4桁・先頭ゼロパディング）
if not SETNO_PATTERN.match(setno):
setno_format_errors.append(
"セットNo '%s': 半角数字4桁ではありません (行 %s)" % (setno, linenos)
)
# 1セット = 2行
if len(members) != 2:
group_size_errors.append(
"セットNo '%s': %d 行あります（期待値 2 行） (行 %s)"
% (setno, len(members), linenos)
)
continue
(ln_a, fa), (ln_b, fb) = members
# 1列目は 0 と 1 の組合せ
flags = (fa[COL_FLAG], fb[COL_FLAG])
if set(flags) != set(("0", "1")):
flag_errors.append(
"セットNo '%s': 1列目が ('%s','%s') です（0と1の組合せが必要） (行 %s)"
% (setno, flags[0], flags[1], linenos)
)
continue
# 4〜7列目がペア内で同一
diff = []
for i in COLS_SAME:
if fa[i] != fb[i]:
diff.append("%d列目 '%s' != '%s'" % (i + 1, fa[i], fb[i]))
if diff:
same_col_errors.append(
"セットNo '%s': %s (行 %s)" % (setno, " / ".join(diff), linenos)
)
row0, row1 = (fa, fb) if fa[COL_FLAG] == "0" else (fb, fa)
ln0, ln1 = (ln_a, ln_b) if fa[COL_FLAG] == "0" else (ln_b, ln_a)
valid_pairs.append((setno, ln0, row0, ln1, row1))
# ---- ペアCSV → データCSV 照合 ------------------------------------------
not_found = []
matched_data_keys = set()
matched_count = 0
for setno, ln0, row0, ln1, row1 in valid_pairs:
key = (row0[COL_VAL1], row0[COL_VAL2], row1[COL_VAL1], row1[COL_VAL2])
if key in data_keys:
matched_count += 1
matched_data_keys.add(key)
else:
not_found.append(
"セットNo '%s' (行 %d, %d): [%s]"
% (setno, ln0, ln1, ", ".join("'%s'" % v for v in key))
)
# データCSVの未使用行
unused = []
for key, linenos in sorted(data_keys.items(), key=lambda kv: kv[1][0]):
if key not in matched_data_keys:
unused.append(
"行 %s: %s" % (", ".join(str(n) for n in linenos), ",".join(key))
)
# ========================================================================
# ログ出力
# ========================================================================
error_count = (
data_col_errors
+ len(col_errors)
+ len(setno_format_errors)
+ len(group_size_errors)
+ len(flag_errors)
+ len(same_col_errors)
+ len(not_found)
)
report.section("■ サマリ")
report.add("データCSV 読込行数 : %d 行" % data_row_count)
report.add("データCSV ユニークキー数 : %d 件" % len(data_keys))
report.add("ペアCSV 読込行数 : %d 行" % len(pair_rows))
report.add("ペアCSV セットNo数 : %d 件" % len(groups))
report.add("ペアCSV 検証対象ペア数 : %d 件" % len(valid_pairs))
report.add("")
report.add("[照合結果]")
report.add(" データCSVに存在した : %d 件" % matched_count)
report.add(" データCSVに存在しない : %d 件" % len(not_found))
report.add("")
report.add("[ペアCSV整合性]")
report.add(" 列数不足 : %d 件" % len(col_errors))
report.add(" セットNo書式不正 : %d 件" % len(setno_format_errors))
report.add(" ペア行数不正(≠2行) : %d 件" % len(group_size_errors))
report.add(" 1列目0/1の組合せ不正 : %d 件" % len(flag_errors))
report.add(" 4〜7列目の不一致 : %d 件" % len(same_col_errors))
report.add("")
report.add("[データCSV]")
report.add(" 列数不正 : %d 件" % data_col_errors)
report.add(" 重複行 : %d 件（参考）" % len(data_dups))
report.add(" 未使用行 : %d 件（参考）" % len(unused))
report.add("")
report.add("判定: %s" % ("OK（エラーなし）" if error_count == 0 else "NG（エラー %d 件）" % error_count))
report.section("[エラー] データCSVに存在しないペア")
report.add("件数: %d 件" % len(not_found))
report.add("形式: セットNo (行番号) : [1列目0の行の2列目, 同3列目, 1列目1の行の2列目, 同3列目]")
report.details(not_found)
report.section("[エラー] ペアCSV: 列数不足")
report.add("件数: %d 件" % len(col_errors))
report.details(col_errors)
report.section("[エラー] ペアCSV: セットNo書式不正（半角数字4桁）")
report.add("件数: %d 件" % len(setno_format_errors))
report.details(setno_format_errors)
report.section("[エラー] ペアCSV: セットNoごとの行数が2行でない")
report.add("件数: %d 件" % len(group_size_errors))
report.details(group_size_errors)
report.section("[エラー] ペアCSV: 1列目が0と1の組合せでない")
report.add("件数: %d 件" % len(flag_errors))
report.details(flag_errors)
report.section("[エラー] ペアCSV: 4〜7列目がペア内で不一致")
report.add("件数: %d 件" % len(same_col_errors))
report.details(same_col_errors)
report.section("[参考] データCSV: 重複行")
report.add("件数: %d 件" % len(data_dups))
report.details(data_dups)
report.section("[参考] データCSV: どのペアからも参照されなかった行")
report.add("件数: %d 件" % len(unused))
report.details(unused)
report.add("")
report.add("=" * 78)
report.add("処理時間: %.2f 秒" % (datetime.now() - started).total_seconds())
report.add("=" * 78)
with open(log_path, "w", encoding=OUTPUT_ENCODING, newline="") as f:
f.write(report.text())
# ---- コンソールにも要約を表示 -------------------------------------------
print("検証完了: %s" % ("OK（エラーなし）" if error_count == 0 else "NG（エラー %d 件）" % error_count))
print(" ペア数 %d / 一致 %d / 不一致 %d" % (len(valid_pairs), matched_count, len(not_found)))
print(" ログ: %s" % os.path.abspath(log_path))
return 0 if error_count == 0 else 1
if __name__ == "__main__":
try:
sys.exit(main())
except Exception as e: # noqa: BLE001
sys.stderr.write("実行時エラー: %s\n" % e)
sys.exit(2)