CSVデータクリーナー

1. アプリ概要

CSVを読み込んで欠損値・重複行を検出し、列の型を数値・整数・日付に自動変換するデータクリーニングツール。pandas活用。

このアプリはdbカテゴリの実践的なPythonアプリです。使用ライブラリは tkinter（標準ライブラリ）・pandas、難易度は ★★★ です。

読み込んだ CSV は pandas.DataFrame として保持し、クリーニングの各処理も pandas のメソッド（read_csv()・isna()・fillna()・dropna()・drop_duplicates()・to_numeric()・to_datetime()・describe()・to_csv()）で書いています。pandas が入っていない環境では起動しません（先に pip install pandas が必要です。標準ライブラリだけで動く代替処理は用意していません）。

画面は「元データ」と「クリーン後」を左右に並べ、下の統計プレビューに 行数・列数・列ごとの dtype と欠損数・describe() の要約を表示します。どのオプションが何行・何セルに効いたのかは、実行のたびにステータスバー（例:「クリーン完了: 5 行（元 7 行・2 行削除） / 欠損 10 → 4 セル｜列名を正規化、文字列をトリム・全角→半角、…」）で確認できます。

動かしながら読むことが理解の最短経路です。まずはコードをコピーして実行し、想定どおりに動くことを確認したうえで解説と照らし合わせてください。

カスタマイズでは「機能追加」「UI 改善」「エラー耐性」の三方向で考えると視野が広がります。練習問題にもそれぞれの方向の具体例を用意しています。

2. 機能一覧

• 「📂 CSV読込」ボタン（_load()）・「▶ クリーン実行」ボタン（_clean()）・「💾 保存」ボタン（_save()）で操作
• CSV 読み込み：pd.read_csv() を utf-8-sig で試し、UnicodeDecodeError なら cp932（Excel が保存する Shift_JIS 系）で読み直す
• 欠損・重複の検出：読込直後に df.isna().sum() で欠損セル数、df.duplicated().sum() で重複行数を数えてステータスバーに表示。表では欠損セルを「(欠損)」と表示する
• 列名の正規化：NFKC 正規化 → 前後の空白除去 → 小文字化 → 連続する空白を _ に置換（同名になった列には連番を付ける）
• 文字列のトリム・全角→半角：Series.str.normalize("NFKC") と .str.strip() を文字列列だけに適用
• 型の自動変換：pd.to_numeric()・pd.to_datetime() で欠損以外の全件を変換できた列だけ、数値・整数（Int64）・日付に変換する
• 欠損値の補完：数値列は中央値、日付列は直前の値（ffill()）、文字列列は入力欄の値（既定「不明」）で fillna()
• 欠損値の削除：dropna(how="all") で全欠損の行・列を、dropna(how="any") で欠損を含む行を削除（オプションで個別に切り替え）
• 重複行の削除：drop_duplicates()
• 統計プレビュー：列ごとの dtype・欠損数と df.describe(include="all") の要約を下部パネルに表示
• クリーン結果の保存：to_csv(index=False, encoding="utf-8-sig") で Excel でも文字化けしない CSV に書き出し
• ttk.Treeview による表形式の一覧表示（先頭 MAX_PREVIEW_ROWS = 500 行）
• filedialog.askopenfilename() によるファイル選択ダイアログ
• filedialog.asksaveasfilename() によるファイル選択ダイアログ
• messagebox.showerror()・messagebox.showinfo()・messagebox.showwarning() によるダイアログ通知
• ウィンドウはタイトル「CSVデータクリーナー」・サイズ 1000x629 で起動

3. 事前準備・環境

Windows 11（実機・Python 3.12.10 / pandas 3.0.3）で、読み込み → クリーン実行 → 保存まで実際の CSV で確認しています（全機能の網羅テストではありません）。Linux Mint 22.3（仮想マシン）では起動と画面表示を確認しました。CSV の読み込み以降は Windows 11 での確認です。macOS は未検証です。

• Python 3.10 以上
• OS: Windows 11（実機で起動・基本操作を確認）／Linux Mint 22.3（仮想マシンで起動を確認）

このアプリは pandas が必須です（pandas が無い環境向けの代替処理は用意していません）。手元では Python 3.12・pandas 3.0.3・Windows 11 で、読み込み → クリーン実行 → 保存までを実際の CSV で通して確認しました。pandas 2 系では検証していません。

インストールが必要なライブラリ

pip install pandas

4. 完全なソースコード

import tkinter as tk
from tkinter import ttk, messagebox, filedialog
import os
import re
import unicodedata

import pandas as pd

# 日付として解釈を試すフォーマット（すべての値が一致した列だけ datetime に変換する）
DATE_FORMATS = ["%Y-%m-%d", "%Y/%m/%d", "%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M:%S"]


class App056:
    """CSVデータクリーナー（pandas でクリーニングする）"""

    MAX_PREVIEW_ROWS = 500

    def __init__(self, root):
        self.root = root
        self.root.title("CSVデータクリーナー")
        self.root.geometry("1000x629")
        self.root.minsize(1000, 629)
        self.root.configure(bg="#1e1e1e")
        self.df_original = pd.DataFrame()
        self.df_cleaned = pd.DataFrame()
        self._build_ui()

    def _build_ui(self):
        header = tk.Frame(self.root, bg="#252526", pady=6)
        header.pack(fill=tk.X)
        tk.Label(header, text="🧹 CSVデータクリーナー",
                 font=("Noto Sans JP", 12, "bold"),
                 bg="#252526", fg="#4fc3f7").pack(side=tk.LEFT, padx=12)

        # ツールバー
        tb = tk.Frame(self.root, bg="#2d2d2d", pady=4)
        tb.pack(fill=tk.X)
        ttk.Button(tb, text="📂 CSV読込",    command=self._load).pack(side=tk.LEFT, padx=4)
        ttk.Button(tb, text="▶ クリーン実行", command=self._clean).pack(side=tk.LEFT, padx=4)
        ttk.Button(tb, text="💾 保存",        command=self._save).pack(side=tk.LEFT, padx=4)

        # オプションパネル
        opt_f = tk.LabelFrame(self.root, text="クリーンオプション",
                              bg="#252526", fg="#ccc", font=("Arial", 9),
                              padx=8, pady=4)
        opt_f.pack(fill=tk.X, padx=8, pady=4)

        self.opt_normalize_cols = tk.BooleanVar(value=True)
        self.opt_trim           = tk.BooleanVar(value=True)
        self.opt_convert_type   = tk.BooleanVar(value=True)
        self.opt_fill_na        = tk.BooleanVar(value=False)
        self.opt_drop_all_na    = tk.BooleanVar(value=True)
        self.opt_drop_any_na    = tk.BooleanVar(value=False)
        self.opt_drop_dup       = tk.BooleanVar(value=True)
        self.fill_val_var       = tk.StringVar(value="不明")

        opts = [
            ("列名を正規化",             self.opt_normalize_cols),
            ("文字列をトリム・全角→半角", self.opt_trim),
            ("型を自動変換",             self.opt_convert_type),
            ("欠損値を補完",             self.opt_fill_na),
            ("全欠損の行・列を削除",     self.opt_drop_all_na),
            ("欠損のある行を削除",       self.opt_drop_any_na),
            ("重複行を削除",             self.opt_drop_dup),
        ]
        for i, (label, var) in enumerate(opts):
            tk.Checkbutton(opt_f, text=label, variable=var,
                           bg="#252526", fg="#ccc", selectcolor="#3c3c3c",
                           activebackground="#252526",
                           font=("Arial", 9)).grid(
                row=i // 3, column=i % 3, sticky="w", padx=6)

        tk.Label(opt_f, text="文字列の補完値:", bg="#252526", fg="#ccc",
                 font=("Arial", 9)).grid(row=2, column=1, sticky="e", padx=6)
        ttk.Entry(opt_f, textvariable=self.fill_val_var,
                  width=10).grid(row=2, column=2, sticky="w")

        # 左右プレビュー
        paned = ttk.PanedWindow(self.root, orient=tk.HORIZONTAL)
        paned.pack(fill=tk.BOTH, expand=True, padx=8, pady=4)

        left = tk.Frame(paned, bg="#1e1e1e")
        paned.add(left, weight=1)
        tk.Label(left, text="元データ", bg="#1e1e1e", fg="#888",
                 font=("Arial", 9)).pack(anchor="w")
        self.orig_tree = self._make_tree(left)

        right = tk.Frame(paned, bg="#1e1e1e")
        paned.add(right, weight=1)
        tk.Label(right, text="クリーン後", bg="#1e1e1e", fg="#888",
                 font=("Arial", 9)).pack(anchor="w")
        self.clean_tree = self._make_tree(right)

        # 統計プレビュー
        stat_f = tk.LabelFrame(self.root, text="統計プレビュー",
                               bg="#252526", fg="#ccc", font=("Arial", 9))
        stat_f.pack(fill=tk.X, padx=8, pady=4)
        self.stat_text = tk.Text(stat_f, height=8, bg="#1e1e1e", fg="#d4d4d4",
                                 font=("Consolas", 9), wrap="none",
                                 insertbackground="#d4d4d4")
        self.stat_text.pack(fill=tk.X, padx=4, pady=4)

        self.status_var = tk.StringVar(value="CSVファイルを読み込んでください")
        tk.Label(self.root, textvariable=self.status_var,
                 bg="#252526", fg="#858585", font=("Arial", 9),
                 anchor="w", padx=8).pack(fill=tk.X, side=tk.BOTTOM)

    def _make_tree(self, parent):
        frame = tk.Frame(parent, bg="#1e1e1e")
        frame.pack(fill=tk.BOTH, expand=True)
        tree = ttk.Treeview(frame, show="headings", selectmode="browse")
        sbx = ttk.Scrollbar(frame, orient=tk.HORIZONTAL, command=tree.xview)
        sby = ttk.Scrollbar(frame, command=tree.yview)
        tree.configure(xscrollcommand=sbx.set, yscrollcommand=sby.set)
        sby.pack(side=tk.RIGHT, fill=tk.Y)
        tree.pack(fill=tk.BOTH, expand=True)
        sbx.pack(fill=tk.X)
        return tree

    # ------------------------------------------------------------------
    # 読み込み・表示
    # ------------------------------------------------------------------
    def _load(self):
        path = filedialog.askopenfilename(
            filetypes=[("CSV", "*.csv"), ("TSV", "*.tsv"), ("すべて", "*.*")])
        if not path:
            return
        try:
            try:
                df = pd.read_csv(path, encoding="utf-8-sig")
            except UnicodeDecodeError:
                # Excel が保存した Shift_JIS 系の CSV を読み直す
                df = pd.read_csv(path, encoding="cp932")
        except pd.errors.EmptyDataError:
            messagebox.showinfo("情報", "データが空です")
            return
        except Exception as e:
            messagebox.showerror("エラー", str(e))
            return

        self.df_original = df
        self.df_cleaned = df.copy()
        self._show_tree(self.orig_tree, df)
        self._show_tree(self.clean_tree, df)
        self._show_stats(df, "読込直後")
        na = int(df.isna().sum().sum())
        dup = int(df.duplicated().sum())
        self.status_var.set(
            f"読込完了: {len(df)} 行 × {len(df.columns)} 列 / "
            f"欠損 {na} セル・重複 {dup} 行を検出")

    def _show_tree(self, tree, df):
        cols = [f"c{i}" for i in range(len(df.columns))]
        tree.configure(columns=cols)
        for i, name in enumerate(df.columns):
            tree.heading(f"c{i}", text=str(name))
            tree.column(f"c{i}", width=100, anchor="w")
        tree.delete(*tree.get_children())
        for row in df.head(self.MAX_PREVIEW_ROWS).itertuples(index=False):
            values = ["(欠損)" if pd.isna(v) else str(v) for v in row]
            tree.insert("", tk.END, values=values)

    def _show_stats(self, df, label):
        """行数・列数・dtype・列ごとの欠損数と describe() をテキストで表示する。"""
        lines = [f"■ {label}: {len(df)} 行 × {len(df.columns)} 列"]
        na_counts = df.isna().sum()
        lines.append("列名 / dtype / 欠損数")
        for name in df.columns:
            lines.append(f"  {name} / {df[name].dtype} / {int(na_counts[name])}")
        if len(df.columns):
            try:
                lines.append("")
                lines.append(df.describe(include="all").to_string())
            except ValueError as e:
                lines.append(f"（統計を計算できませんでした: {e}）")
        self.stat_text.delete("1.0", tk.END)
        self.stat_text.insert("1.0", "\n".join(lines))

    # ------------------------------------------------------------------
    # クリーニング処理（各ステップは pandas の DataFrame を返す）
    # ------------------------------------------------------------------
    def _string_columns(self, df):
        """文字列として扱える列名の一覧（pandas 3 の str dtype と object の両方）。"""
        return [c for c in df.columns
                if pd.api.types.is_string_dtype(df[c]) or df[c].dtype == "object"]

    def _normalize_columns(self, df):
        """列名を NFKC 正規化 → 前後空白除去 → 小文字化 → 空白を _ に置換。"""
        names, used = [], {}
        for name in df.columns:
            new = unicodedata.normalize("NFKC", str(name)).strip().lower()
            new = re.sub(r"\s+", "_", new)
            if new == "":
                new = "column"
            if new in used:                      # 重複した列名に連番を付ける
                used[new] += 1
                new = f"{new}_{used[new]}"
            else:
                used[new] = 1
            names.append(new)
        df.columns = names
        return df

    def _trim_strings(self, df):
        """文字列列を NFKC 正規化（全角→半角）して前後の空白を落とす。"""
        for col in self._string_columns(df):
            df[col] = df[col].str.normalize("NFKC").str.strip()
        return df

    def _as_integer(self, s):
        """小数点以下が無い数値列を Int64（欠損を保持できる整数型）にして返す。"""
        valid = s.dropna()
        if len(valid) and (valid % 1 == 0).all():
            return s.astype("Int64")
        return None

    def _convert_types(self, df):
        """数値・日付に「全件」変換できる列だけ型変換する（変換できない値は作らない）。"""
        converted = []
        str_cols = self._string_columns(df)
        for col in df.columns:
            filled = df[col].notna()
            if not filled.any():
                continue
            if pd.api.types.is_numeric_dtype(df[col]):
                as_int = self._as_integer(df[col])
                if as_int is not None and str(df[col].dtype) != "Int64":
                    df[col] = as_int
                    converted.append(f"{col}→整数")
                continue
            if col not in str_cols:
                continue
            num = pd.to_numeric(df[col], errors="coerce")
            if not (num.isna() & filled).any():
                as_int = self._as_integer(num)
                df[col] = num if as_int is None else as_int
                converted.append(f"{col}→数値" if as_int is None else f"{col}→整数")
                continue
            dt = None
            for fmt in DATE_FORMATS:
                parsed = pd.to_datetime(df[col], format=fmt, errors="coerce")
                dt = parsed if dt is None else dt.fillna(parsed)
            if dt is not None and not (dt.isna() & filled).any():
                df[col] = dt
                converted.append(f"{col}→日付")
        return df, converted

    def _fill_missing(self, df):
        """数値列は中央値、日付列は前の値、文字列列は入力欄の値で欠損を埋める。"""
        fill_val = self.fill_val_var.get()
        for col in df.columns:
            if not df[col].isna().any():
                continue
            if pd.api.types.is_numeric_dtype(df[col]):
                median = df[col].median()
                if pd.notna(median):
                    if pd.api.types.is_integer_dtype(df[col]):
                        median = int(round(median))
                    df[col] = df[col].fillna(median)
            elif pd.api.types.is_datetime64_any_dtype(df[col]):
                df[col] = df[col].ffill()
            else:
                df[col] = df[col].fillna(fill_val)
        return df

    def _clean(self):
        if self.df_original.empty and not len(self.df_original.columns):
            messagebox.showwarning("警告", "CSVを読み込んでください")
            return

        df = self.df_original.copy()
        before_rows = len(df)
        before_na = int(df.isna().sum().sum())
        log = []

        if self.opt_normalize_cols.get():
            df = self._normalize_columns(df)
            log.append("列名を正規化")

        if self.opt_trim.get():
            df = self._trim_strings(df)
            log.append("文字列をトリム・全角→半角")

        if self.opt_convert_type.get():
            df, converted = self._convert_types(df)
            if converted:
                log.append("型変換 " + "・".join(converted))

        if self.opt_fill_na.get():
            df = self._fill_missing(df)
            log.append("欠損値を補完")

        if self.opt_drop_all_na.get():
            df = df.dropna(axis=0, how="all").dropna(axis=1, how="all")
            log.append("全欠損の行・列を削除")

        if self.opt_drop_any_na.get():
            df = df.dropna(axis=0, how="any")
            log.append("欠損のある行を削除")

        if self.opt_drop_dup.get():
            df = df.drop_duplicates()
            log.append("重複行を削除")

        df = df.reset_index(drop=True)
        self.df_cleaned = df
        self._show_tree(self.clean_tree, df)
        self._show_stats(df, "クリーン後")
        after_na = int(df.isna().sum().sum())
        self.status_var.set(
            f"クリーン完了: {len(df)} 行（元 {before_rows} 行・"
            f"{before_rows - len(df)} 行削除） / 欠損 {before_na} → {after_na} セル"
            + ("｜" + "、".join(log) if log else ""))

    # ------------------------------------------------------------------
    # 保存
    # ------------------------------------------------------------------
    def _save(self):
        if self.df_cleaned.empty and not len(self.df_cleaned.columns):
            messagebox.showwarning("警告", "データがありません")
            return
        path = filedialog.asksaveasfilename(
            defaultextension=".csv",
            filetypes=[("CSV", "*.csv")])
        if not path:
            return
        try:
            self.df_cleaned.to_csv(path, index=False, encoding="utf-8-sig")
            messagebox.showinfo(
                "完了", f"{len(self.df_cleaned)} 行を保存しました:\n{path}")
            self.status_var.set(f"保存完了: {os.path.basename(path)}")
        except Exception as e:
            messagebox.showerror("エラー", str(e))


if __name__ == "__main__":
    root = tk.Tk()
    app = App056(root)
    root.mainloop()

5. コード解説

CSVデータクリーナー（pandas でクリーニングする）のコードを、実際に書かれている実装に沿って解説します。

6. ステップバイステップガイド

このアプリをゼロから自分で作る手順を解説します。コードをコピーするだけでなく、実際に手順を追って自分で書いてみましょう。

1. 1
   ファイルを作成する

   新しいファイルを作成して app056.py と保存します。外部ライブラリ pandas を使います。先に pip install pandas を実行してください。

2. 2
   クラスの骨格を作る

   App056 クラスを定義し、__init__ と、末尾の root = tk.Tk() ～ mainloop() の最小構成を書いて、まず空のウィンドウが出ることを確認します。

3. 3
   ウィンドウを設定する

   title("CSVデータクリーナー")・geometry("1000x629")・configure(bg="#1e1e1e")・minsize(1000, 629) をコンストラクタで設定します。

4. 4
   画面部品を並べる

   _build_ui() の中で、tk.Frame×5・tk.Label×5・ttk.Button×3・tk.LabelFrame×2・tk.Checkbutton（ループで生成）・ttk.Entry・ttk.PanedWindow・tk.Text・ttk.Treeview・ttk.Scrollbar×2 を作って配置します（掲載コードと同じ並び順で書くとレイアウトが一致します）。まず表示だけ確認しましょう。

5. 5
   イベントを接続する

   「2. 機能一覧」で挙げた各ボタンを command= で対応するメソッド（_load()・_clean()・_save()）につなぎます。

6. 6
   中心になるメソッドを実装する

   アプリの本体である _clean()（48行）・_load()（28行）・_save()（16行） を実装します。

7. 7
   動作確認する

   python app056.py で起動し、各ボタンが反応することを確認します。

7. カスタマイズアイデア

基本機能を習得したら、以下のカスタマイズに挑戦してみましょう。

8. よくある問題と解決法

9. 練習問題

アプリの理解を深めるための練習問題です。

1. 課題1：機能拡張

   CSVデータクリーナーに新しい機能を1つ追加してみましょう。

2. 課題2：UIの改善

   色・フォント・レイアウトを変更して、より使いやすいUIにカスタマイズしましょう。

3. 課題3：キー操作に対応させる

   ボタンから呼んでいる _load() を、キーボードのキーからも呼べるようにしてみましょう。tkinter では bind を使ってキーイベントとメソッドを結び付けられます。

4. 課題4：列を選んで重複を判定する

   drop_duplicates() は既定で全列が一致した行だけを重複とみなします。「この列だけで重複を判定する」チェックボックス（または列選択のコンボボックス）を追加し、drop_duplicates(subset=[列名], keep="first") を呼べるようにしてみましょう。keep="last" にすると残る行がどう変わるかも確かめてください。

5. 課題5：クリーニングの前後を比べるレポートを出す

   クリーン実行のたびに、列ごとの「欠損数の変化」「dtype の変化」を DataFrame にまとめて統計プレビューに表示してみましょう。df.isna().sum() の結果は Series なので、pd.concat([before, after], axis=1) で横に並べると差分表になります。