【初心者歓迎】CSVファイル結合ツールを作ろう

今回の自作ツールは、「CSVファイル結合ツール」です。ファイルを単純に結合するだけならWindowsのcopy コマンドが使えますが、CSVファイルの場合はヘッダの考慮(2ファイル目以降のヘッダは無視する)が必要です。

今回紹介する「CSVファイル結合ツール」は、ヘッダを考慮した結合に対応しているのはもちろんのこと、異なるヘッダを持つCSVを結合したり、指定キーワードを含む行の抽出/削除ができるようになっています。

PythonでCSVファイルを結合するツールを作りたい方は、本記事を参考にトライしてみてください。

自作ツールを作るためのポイントと今後公開予定の自作ツール一覧を「【実践】ちょっと便利なツールをPythonで作ろう!」で紹介していますので、併せてご覧ください。

目次

CSVファイル結合ツールの紹介

結合したいファイルを格納したフォルダと、出力ファイル名を指定し、「CSVファイルの結合」ボタンをクリックすれば、1つのファイルに結合してくれます。

  • 結合の最初に読み込んだファイルのヘッダを引き継ぎ、2ファイル目以降のヘッダは無視します。
  • ヘッダの行数は任意の行数の指定が可能です。
  • 結合元フォルダの入力欄にファイルのワイルドカードを指定することが可能です。
  • 結合対象のCSVファイルと、結合後のCSVファイルの文字コードを指定可能(文字コード変換機能)です。
  • 指定したワードを含む行のみ取り出すことが可能(ワードはカンマ区切りで複数指定可能)です。
  • 指定したワードを含む行を除外することが可能(ワードはカンマ区切りで複数指定可能)です。
  • 画面から入力した内容は保持され、次回起動時に復元されます(チェックボックスは除く)。
  • 結合元フォルダ名と結合ファイル名欄はドラッグ&ドロップでの指定が可能です。

画面レイアウト

使い方

CSVファイルの結合には2種類あります。1つはCSVファイルを単純に結合する「単純結合」モードで、全てのファイルが同じフォーマット(ヘッダ数、区切り文字)であることが前提です。

もう1つは、異なる列名を持つCSVファイルを結合する「マージ」モードで、フォルダ内にあるすべてのCSVファイルを最初に読み込み、最大公約数の列名を持つヘッダを作成してからCSVファイルを結合します。

「単純結合」モードはテキストファイルを結合するものであり、CSVのフォーマット(区切り文字、ダブルクォーテーション有無)は意識していません。一方「マージ」モードはCSVとしてのフォーマットを意識しているため、区切り文字を指定する必要があります。

単純結合モードによるCSVファイルの結合

「異なるカラムを持つCSVを結合」チェックボックスのチェックが外れていると「単純結合」モードとなります。起動時はこのモードになっています。

  • 結合対象のCSVファイルが置かれたフォルダを指定する
  • 「異なるカラムを持つCSVを結合」チェックボックスのチェックを外す
  • 出力ファイル名を指定する
  • 「CSVファイルの結合」ボタンをクリックする

マージモードによるCSVファイルの結合

「異なるカラムを持つCSVを結合」チェックボックスのチェックが入っていると「マージ」モードとなります。マージモードではキーワードを含む行の抽出や除外が出来ません。

  • 結合対象のCSVファイルが置かれたフォルダを指定する
  • 「異なるカラムを持つCSVを結合」チェックボックスにチェックを入れる
  • 区切り文字を指定する
  • 出力ファイル名を指定する
  • 「CSVファイルの結合」ボタンをクリックする

文字コード変換

文字コードを指定することで、CSVファイルを結合しながら文字コード変換することが可能です。この機能は「単純結合」モード、「マージ」モードのどちらでも有効です。

結合時に指定したキーワードを含む行を抽出/除外する

「キーワードを含む行を選択」、「キーワードを含む行を除外」を指定することで、そのキーワードを含む行だけを取り出したり、削除することができます。「単純結合」モードでのみ使用可能です。

尚、両方に同じキーワードは指定しないでください。

プログラムのダウンロードと動作環境の設定

STEP
Pythonのインストール

Pythonのインストールが必要です。既に構築済みの方は読み飛ばしてください。プログラムを修正する場合は、VSCode Portable も併せてインストールしてください。

自作ツールのためのポータブルPython開発・実行環境を作ろう!プログラム実行時に必要
WinPythonにVSCode Portable版を入れよう!プログラム修正時に必要
STEP
ライブラリのインストール

Python 環境に下記のライブラリをインストールします。 command.bat を実行後、下記のコマンドを実行してください。

pip install customtkinter
pip install tkinterdnd2

STEP
ソースコードのダウンロードと展開

下記のリンクからZipファイルをダウンロード&解凍します。次に、Python環境の任意の場所にフォルダを作成し、そこに解凍したファイルを丸ごとコピーしてください。

実行方法

次の手順で実行してください。

  • WinPython のインストールフォルダ内にある command.bat を実行
  • ダウンロードしたプログラムファイルをコピーした場所にカレントディレクトリを移動
  • Python ui.py を実行

Python ui.py

下記の画面が表示されます。本記事の最初で説明した手順に従って操作してください。

何らかのエラーが発生すると、コマンドプロンプトにその内容が表示されます。

CSVファイル結合ツールの設計

以下は、プログラム作成の際に使用を取りまとめた内容(=設計)です。単にCSVファイル結合ツールをダウンロードして使いたい方は無視してください。
DIYプログラミングの参考になる内容ですので、興味のある方は最後までお読みいただければと思います。

仕様

画面項目入力内容
結合元フォルダ名結合対象のファイルが保存されているファイルを選択するかドラッグ&ドロップします
文字コード結合対象のCSVファイルの文字コードを、SHIFT-JIS と UTF-8 から指定します。Pythonがサポートしている文字コードを直接入力することも可能です。
ヘッダ行数CSVファイルのヘッダ数を指定します。
異なるカラムを持つCSVを結合チェックを入れると「マージ」モードになり、チェックを外すと「単純結合」モードになります。
区切り文字「マージ」モードで結合したいCSVの区切り文字を、カンマ又はタブから選択します。手入力で任意の区切り文字を入力することも可能です。
結合ファイル名結合後のCSVファイル名のパスを指定します。
文字コード結合後のCSVファイルの文字コードを指定します。
キーワードを含む行を抽出「単純結合」モードの時だけ有効になります。
キーワードをカンマ区切りで列挙すると、そのキーワードのいずれかが含まれている行だけが出力対象となります。
キーワードを含む行を除外「単純結合」モードの時だけ有効になります。
キーワードをカンマ区切りで列挙すると、そのキーワードのいずれかが含まれている行は出力されなくなります。

設計段階で考えた画面レイアウトは、実際に出来上がった画面レイアウトと若干異なっています。当初はマージモードという表現にしていましたが、最終的には「異なるカラムを持つCSVを結合」という文言に変えました。

実現方法

  • 画面はCustomTkinter で作成
  • 「単純結合」モードにおいて、巨大なCSVファイルが結合できるよう、1行づつ読み込んで処理する(pandas の read_csv は使わない)
  • 「マージ」モードにおいて、全てのCSVの列をマージする方法は pandas の DataFrame を使う。結合元フォルダの全ファイルからヘッダだけを読み込み、列を統合して空のDataFrameを作成。このDataFrameに1ファイル筒読み込んでは出力CSVに追加していく。
    入力のCSVが巨大な場合はメモリ不足になるが、それは仕方ないものとする。

プログラムの構成

本プログラムは4つの py ファイルで構成しています。csv_spliter.py には split_file関数のみ記述していますが、これは他のプログラムで流用することを考慮しているからです。

モジュール名役割
ui.pyCSVファイル結合ツールのユーザインターフェース
csv_joinner.pyCSVファイルを結合するための処理(関数)
customtkintercontrols.pyファイル選択やフォルダ選択、ドラッグ&ドロップを実現するためのクラス
appconfig.py画面に入力された値をJson形式のファイルに保存/読込するためのクラス

個々の処理手順

下図は、「CSVファイルの結合」ボタンがクリックされた時の「単純結合」モードと「マージ」モードの簡易的なフローチャートです。

ソースコード

ui.py

import os
import customtkinter as ctk
from customtkintercontrols import App
from customtkintercontrols import FileSelectionControl
from customtkintercontrols import FolderSelectionControl
from customtkintercontrols import EntryControl
import csv_joinner as cj
from appconfig import AppConfig 
# -----------------------------------------------
# コールバック関数の定義 
# -----------------------------------------------

# ウィンドウを閉じる時に呼ばれる関数
def close_window():
    # 画面入力値を保存
    conf.set("file_selection",file_selection.get())
    conf.set("input_encoding_name",input_encoding_name.get())
    conf.set("folder",folder_selection.get())
    conf.set("output_encoding_name",output_encoding_name.get())
    conf.set("input_header_cnt",input_header_cnt.get())
    conf.set("contain_words",contain_words.get())
    conf.set("not_contain_words",not_contain_words.get())
    conf.write()
    # プログラムの終了
    app.quit()

# 変換ボタンが押された時に呼ばれる関数
def join():
    csv_file = file_selection.get()
    in_encoder = input_encoding_name.get()
    in_folder = folder_selection.get()
    out_encoder = output_encoding_name.get()
    head_cnt = input_header_cnt.get()
    merge = is_merge.get()
    sepa = delimiter.get().strip()
    varid_words = contain_words.get() if is_contain.get() != 0 else ""
    omit_wors = not_contain_words.get() if is_not_contain.get() != 0 else ""
    if merge == 0:
        cj.join_csv(in_folder,csv_file,int(head_cnt),in_encoder,out_encoder,varid_words,omit_wors)
    else:
        cj.merge_csv(in_folder,csv_file,sepa,in_encoder,out_encoder,varid_words,omit_wors)
        

def input_control():
    if is_merge.get() == 0:
        is_contain.configure(state="normal")
        is_not_contain.configure(state="normal")
        contain_words.entry.configure(state="normal")
        not_contain_words.entry.configure(state="normal")
        delimiter.configure(state="disabled") 
    else:
        is_contain.configure(state="disabled")
        is_not_contain.configure(state="disabled")
        contain_words.entry.configure(state="disabled")  
        not_contain_words.entry.configure(state="disabled")  
        delimiter.configure(state="normal") 
           
# -----------------------------------------------
# メインウインドウの定義
# -----------------------------------------------

# 前回入力値の保存・復元用インスタンスの生成
conf = AppConfig("./config.json")

# メインウィンドウの作成
app = App()
app.minsize(width=600, height=400)
app.geometry(f"800x480")
app.title("CSVファイル結合")

# customtkinterをダークモードに設定
ctk.set_appearance_mode("light")

# ウィンドウを閉じる時に呼びたい関数を登録
app.protocol("WM_DELETE_WINDOW", close_window)

# -----------------------------------------------
# 画面レイアウト(ウィジェット)の定義 
# -----------------------------------------------

# フォルダ選択ダイアログを配置
folder_selection = FolderSelectionControl(app)
folder_selection.label.configure(text="結合元フォルダ名")
folder_selection.pack(padx=10,pady=10,fill="x")

# 文字コードとヘッダ行数の指定ウィジェットを配置
group = ctk.CTkFrame(app)
label = ctk.CTkLabel(group , text="      文字コード")
label.pack(padx=10,pady=10,side="left")

# 入力ファイル文字コードコンボボックスを配置
input_encoding_name = ctk.CTkComboBox(group , values=["SHIFT-JIS","UTF-8"])
input_encoding_name.pack(padx=10,pady=10,side="left")

# ヘッダ行数の選択コンボボックスを配置
label = ctk.CTkLabel(group , text="ヘッダ行数")
label.pack(padx=10,pady=10,side="left")
input_header_cnt = ctk.CTkComboBox(group , values=["0","1","2","3","4","5"],width=60)
input_header_cnt.set("1")
input_header_cnt.pack(padx=10,pady=10,side="left")

# 異なるカラムを持つCSVを結合するオプションを配置
is_merge = ctk.CTkCheckBox(group,text="異なるカラムを持つCSVを結合",command=input_control)
is_merge.pack(padx=30,pady=10,side="left")
delimiter = ctk.CTkComboBox(group , values=[",","\\t"],width=60)
delimiter.set(",")
delimiter.configure(state="disabled") 
delimiter.pack(padx=10,pady=10,side="left")
group.pack(padx=10,pady=10,fill="x")
group.pack(padx=10,pady=10,fill="x")

# ファイル選択ダイアログを配置
file_selection = FileSelectionControl(app)
file_selection.label.configure(text="   結合ファイル名")
file_selection.pack(padx=10,pady=10,fill="x")

# 出力ファイル文字コードコンボボックスを配置
group = ctk.CTkFrame(app)
label = ctk.CTkLabel(group , text="      文字コード")
label.pack(padx=10,pady=10,side="left")
output_encoding_name = ctk.CTkComboBox(group , values=["SHIFT-JIS","UTF-8"])
output_encoding_name.pack(padx=10,pady=10,side="left")
group.pack(padx=10,pady=10,fill="x")


# 抽出対象のキーワード入力欄を配置
group = ctk.CTkFrame(app)
is_contain = ctk.CTkCheckBox(group,text="キーワードを含む行を抽出",width=90)
is_contain.pack(padx=10,pady=10,side="left")
contain_words = EntryControl(group)
contain_words.pack(padx=10,pady=10,fill="x", expand=True)
group.pack(padx=10,pady=10,fill="x")

# 除外対象のキーワード入力欄を配置
group = ctk.CTkFrame(app)
is_not_contain = ctk.CTkCheckBox(group,text="キーワードを含む行を除外",width=90)
is_not_contain.pack(padx=10,pady=10,side="left")
not_contain_words = EntryControl(group)
not_contain_words.pack(padx=10,pady=10,fill="x", expand=True)
group.pack(padx=10,pady=10,fill="x")

# 実行ボタンの配置
split_button = ctk.CTkButton(app,text="CSVファイルの結合",height=40,width=300,command=join)
split_button.pack(padx=20,pady=10,side="top", anchor="e")


# -----------------------------------------------
# 前回実行時の画面入力値を復元
# -----------------------------------------------
file_selection.set(conf.get("file_selection",""))
input_encoding_name.set(conf.get("input_encoding_name","SHIFT-JIS"))
input_header_cnt.set(conf.get("input_header_cnt","1"))
output_encoding_name.set(conf.get("output_encoding_name","SHIFT-JIS"))
is_merge.setvar(conf.get("is_merge",""))
delimiter.set(conf.get("delimiter",","))
contain_words.set(conf.get("contain_words",""))
not_contain_words.set(conf.get("not_contain_words",""))

# CSVファイル名を復元
folder_selection.set(conf.get("csvfile"))

# 保存先フォルダの値を復元
folder_selection.set(conf.get("folder"))
if folder_selection.get() == "":
    folder_selection.set(os.getcwd())


# -----------------------------------------------
# CustomTkinterの起動 
# -----------------------------------------------

# メインループを開始
app.mainloop()

csv_joinner.py

関数の中に、 matchというローカル関数を用意しています。
match は 読み込んだ行が抽出対象か、除外対象化を判定していますので、もっと複雑な条件を指定したい場合は、この関数を修正してください。

get_filesは指定したフォルダからファイルの一覧を取得する関数です。フォルダ名に続けてワイルドカードの指定です。

get_files("d:/hoge/*.csv")

get_files には、フォルダ階層を含めてファイル一覧を取得する recursive 引数と、隠しファイルも含めて取得する include_hidden 引数を用意していますので、必要に応じてご利用ください。

import os
import glob
import pandas as pd


def get_files(folder,recursive=False,include_hidden=False):
    """
    指定されたフォルダ内のファイルを再帰的に取得する関数
    :param folder: フォルダのパス
    :param recursive: 再帰的に取得するかどうかのフラグ
    :param include_hidden: 隠しファイルを含めるかどうかのフラグ
    :return: ファイルのリスト
    """
    if os.path.isdir(folder):
        folder = os.path.join(folder,"*")
    if recursive:
        dir,name = os.path.split(folder)
        folder = os.path.join(dir,"**",name)
    return glob.glob(folder,recursive=recursive,include_hidden=include_hidden)

def join_csv(folder, filename, header_count=1, input_encoder="shift-jis", output_encoder="shift-jis", valid_words="", omit_words=""):
    """
    指定されたフォルダ内のCSVファイルを結合して新しいファイルに書き込む関数。

    Args:
    folder (str): CSVファイルが格納されたフォルダのパスまたはファイルパス。ワイルドカードを使用できる。
    filename (str): 結合したCSVファイルの出力先となるファイル名。
    header_count (int): ヘッダ行数。2個目以上のファイルはこの行数分を読み飛ばす。
    input_encoder (str): 入力ファイルの文字エンコーディング。
    output_encoder (str): 出力ファイルの文字エンコーディング。
    valid_words (str): 有効な単語のリスト。
    omit_words (str): 省略する単語のリスト。
    """

    # 1行の中に指定したキーワードが含まれているかをチェックする関数
    def match(line,words):
        return any(word in line for word in words.split(','))

    # フォルダ内のCSVファイルを取得
    files = get_files(folder)

    # 結合した結果を書き込むファイルをオープン
    with open(filename, 'w', encoding=output_encoder) as outfile:

        # 各ファイルを順番に処理
        for file_no, file in enumerate(files):
            # ファイルをオープン
            with open(file, 'r', encoding=input_encoder) as infile:
                # 各行を順番に処理
                for line_no, line in enumerate(infile):
                                
                    # ヘッダ行か否かで処理を分岐
                    if line_no < header_count:
                        # 最初のファイルのヘッダ行を出力ファイルに書き込む 
                        if file_no == 0:                      
                            outfile.write(line)
                    else:
                        # 指定したキーワードが含まれていなければ読み飛ばす
                        if valid_words != "" and not match(line,valid_words):
                            continue
                        # 指定したキーワードが含まれていたら読み飛ばす
                        if omit_words != "" and match(line,omit_words):
                            continue

                        # 行を出力ファイルに書き込む
                        outfile.write(line)




def merge_csv(folder, filename, delimiter=',', input_encoder="shift-jis", output_encoder="shift-jis"):
    """
    指定された複数のファイルに対して、1行目を読み取り、それをカラムとして持つDataFrameを作成し、指定されたファイルを読み込む処理を繰り返し、最終的にファイルに出力する関数。

    Args:
    file_list (list): 読み込むファイルのリスト
    filename (str): 出力ファイル名
    delimiter (str): 区切り文字
    """
    # 1行目を読み取って、カラムのリストを作成
    files = get_files(folder)
    columns = set()
    for file in files:
        with open(file, 'r',encoding=input_encoder) as infile:
            line = infile.readline().strip()
            columns.update(line.split(delimiter))

    combined_df = pd.DataFrame(columns=list(columns))

    # ファイルを削除
    os.remove(filename)

    # ファイルごとにDataFrameを作成し、CSVに出力
    for file in files:
        sour_df = pd.read_csv(file, delimiter=delimiter,encoding=input_encoder)
        dest_df = pd.concat([combined_df,sour_df])

        with open(filename, 'a') as f:
            dest_df.to_csv(f, header=False, index=False,encoding=output_encoder)

まとめ

今回は、「CSVファイル結合ツール」について、使い方から環境構築に至るまでの手順と、ツールを作るにあたって事前に検討した内容(設計)を紹介しました。

今回用意した2つの結合モードのうち、「単純結合」モードはメモリに乗りきらないくらいの巨大なファイルでも結合可能です。一方「マージ」モードでは、ヘッダ項目が異なるCSVファイルが結合できるようになっています。

再利用できるよう2つとも関数化していますので、ご自身の用途にカスタマイズして使っていただければ幸いです。

よかったらシェアしてね!
  • URLをコピーしました!
  • URLをコピーしました!

この記事を書いた人

コメント

コメントする

目次