CSVファイルの結合

目的

定期的にDBを検索して検索結果のCSVファイルが出力されるときに、過去の検索結果と今回の検索結果を1つのCSVファイルとして保存する処理を作成する。

今回の処理では、以下の動作を想定しています。

• 初回は過去の結果がないため、検索結果をヘッダーを含めて保存する
• 2回目以降は、検索結果のヘッダーを除いてデータのみ追記する
• ヘッダーが1行だけでなく、複数行の場合にも対応する
• 保存済みの検索結果CSVは trash フォルダへ移動する

想定動作

1. DBを検索
2. 検索結果のCSVファイルをoutputフォルダに保存
3. outputフォルダのCSVファイルをmergeフォルダのCSVファイルに結合
4. 保存済みのCSVファイルをtrushフォルダに移動

今回は3と4を動作するプログラムを作成する。
DB検索処理については今回の記事では扱わない。

設計について

DB検索処理とCSVのmerge処理を分離して考えています。
DB検索側では、検索結果を output フォルダにCSVとして出力するだけにします。
その後、merge_csv(merge_filename, header_rows)を呼び出すことで、outputフォルダに存在するCSVを、指定したmergeファイルへ追加することを想定しています。

フォルダ構成

プロジェクト/
├─ mergeCSV.py
├─ output/
│   └─ test_20260812.csv
├─ merge/
│   └─ test_merge.csv
└─ trash/

各フォルダの役割

フォルダ	役割
output	DB検索によって出力されたCSVを一時保存
merge	過去の検索結果を蓄積したCSVを保存
trash	merge処理が完了したCSVを退避

プログラム

from pathlib import Path
import csv
import shutil


# ==========================================
# フォルダ設定
# ==========================================

BASE_DIR = Path(__file__).resolve().parent

OUTPUT_DIR = BASE_DIR / "output"
MERGE_DIR = BASE_DIR / "merge"
TRASH_DIR = BASE_DIR / "trash"


# ==========================================
# CSVをmergeする
# ==========================================

def merge_csv(merge_filename, header_rows):
    """
    outputフォルダ内のCSVを
    mergeフォルダの指定ファイルへ追記する。

    Parameters
    ----------
    merge_filename : str
        merge先のファイル名（拡張子なし）　csv固定

    header_rows : int
        CSVのヘッダー行数
        1 = ヘッダー1行
        2 = ヘッダー2行
    """

    # --------------------------------------
    # 引数チェック
    # --------------------------------------

    if header_rows < 1:
        raise ValueError(
            "header_rowsは1以上を指定してください"
        )

    # --------------------------------------
    # フォルダを作成
    # --------------------------------------

    OUTPUT_DIR.mkdir(exist_ok=True)
    MERGE_DIR.mkdir(exist_ok=True)
    TRASH_DIR.mkdir(exist_ok=True)

    # --------------------------------------
    # merge先ファイル
    # --------------------------------------

    merge_path = MERGE_DIR / f"{merge_filename}.csv"

    # --------------------------------------
    # outputフォルダ内のCSVを取得
    # --------------------------------------

    csv_files = sorted(
        OUTPUT_DIR.glob("*.csv")
    )

    if not csv_files:
        print("outputフォルダにCSVファイルがありません。")
        return

    print()
    print("==========================================")
    print(f"merge開始: {merge_filename}.csv")
    print(f"ヘッダー行数: {header_rows}")
    print("==========================================")

    # --------------------------------------
    # CSVを順番に処理
    # --------------------------------------

    for csv_path in csv_files:

        print()
        print(f"処理: {csv_path.name}")

        # ----------------------------------
        # CSV読み込み
        # ----------------------------------

        with open(
            csv_path,
            "r",
            encoding="utf-8-sig",
            newline=""
        ) as f:

            rows = list(csv.reader(f))

        # ----------------------------------
        # CSVの行数チェック
        # ----------------------------------

        if len(rows) < header_rows:

            raise ValueError(
                f"{csv_path.name} の行数が "
                f"header_rows={header_rows} "
                f"より少ないです。"
            )

        # ----------------------------------
        # mergeファイルが存在しない場合
        # ----------------------------------

        if not merge_path.exists():

            print(
                f"  → {merge_path.name} を新規作成"
            )

            with open(
                merge_path,
                "w",
                encoding="utf-8-sig",
                newline=""
            ) as f:

                writer = csv.writer(f)

                # 最初のCSVなので
                # ヘッダーを含めて保存
                writer.writerows(rows)

        # ----------------------------------
        # mergeファイルが存在する場合
        # ----------------------------------

        else:

            print(
                f"  → {merge_path.name} に追記"
            )

            with open(
                merge_path,
                "a",
                encoding="utf-8-sig",
                newline=""
            ) as f:

                writer = csv.writer(f)

                # ヘッダーを除いて追記
                writer.writerows(
                    rows[header_rows:]
                )

        # ----------------------------------
        # 処理済みCSVをtrashへ移動
        # ----------------------------------

        trash_path = TRASH_DIR / csv_path.name

        # 同名ファイルが存在する場合は連番を付ける
        if trash_path.exists():

            stem = csv_path.stem
            suffix = csv_path.suffix

            counter = 1

            while True:

                trash_path = (
                    TRASH_DIR
                    / f"{stem}_{counter}{suffix}"
                )

                if not trash_path.exists():
                    break

                counter += 1

        shutil.move(
            str(csv_path),
            str(trash_path)
        )

        print(
            f"  → {trash_path.relative_to(BASE_DIR)} に移動"
        )

    print()
    print("==========================================")
    print("merge完了")
    print(f"出力: {merge_path}")
    print("==========================================")


# ==========================================
# テスト
# ==========================================

def main():

　   merge_filename = "test"　#結合後のファイル名指定
    header_rows = 1 #ヘッダーの行数を指定

    # --------------------------------------
    # merge実行
    # --------------------------------------

    merge_csv(
        merge_filename,
        header_rows
    )



if __name__ == "__main__":
    main()