14日前のcsvファイルのみを取得するという処理

現在作成しているPythonETLで、連携元フォルダに3万件のcsvファイルが溜まっていて、その中から14日前のファイルのみを取得して、後続の処理に回すという処理を作成する必要があった。

この記事では、そのミニチュア版を作成していこう。

3万件のcsvファイルを生成する

まずは、create_dummy_csv_filesという関数で、3万件のCSVファイルを作成する。

import os
import datetime
import random

# フォルダのパス
DATA_DIR = "dummy_csv_files"

# ダミーの3万件の空のCSVファイルを作成
def create_dummy_csv_files(num_files=30000):
    os.makedirs(DATA_DIR, exist_ok=True)

    base_date = datetime.datetime.today()

    for _ in range(num_files):
        # 過去30日以内の日付をランダムに生成
        random_days = random.randint(0, 30)
        file_date = (base_date - datetime.timedelta(days=random_days)).strftime("%Y-%m-%d")
        file_name = f"{file_date}_data_{random.randint(1000, 9999)}.csv"
        file_path = os.path.join(DATA_DIR, file_name)

        # 空のCSVファイルを作成
        with open(file_path, "w") as f:
            pass  # 中身なし

    print(f"ダミーCSVファイル {num_files} 件を作成しました。")

# 14日前のファイルを取得
def get_target_files():
    base_date = datetime.datetime.today()
    target_date = (base_date - datetime.timedelta(days=14)).strftime("%Y-%m-%d")

    # 14日前のファイルのみ取得
    target_files = [f for f in os.listdir(DATA_DIR) if f.startswith(target_date)]

    print(f"14日前のファイル: {len(target_files)} 件")
    print(f"例: {target_files[0]}")
    return target_files

# 実行
create_dummy_csv_files()
target_files = get_target_files()

14日前のcsvファイルのみを取得する

次に、get_target_filesという関数で、14日前のファイルのみを取得する。

import os
import datetime
import random

# フォルダのパス
DATA_DIR = "dummy_csv_files"

# ダミーの3万件の空のCSVファイルを作成
def create_dummy_csv_files(num_files=30000):
    os.makedirs(DATA_DIR, exist_ok=True)

    base_date = datetime.datetime.today()

    for _ in range(num_files):
        # 過去30日以内の日付をランダムに生成
        random_days = random.randint(0, 30)
        file_date = (base_date - datetime.timedelta(days=random_days)).strftime("%Y-%m-%d")
        file_name = f"{file_date}_data_{random.randint(1000, 9999)}.csv"
        file_path = os.path.join(DATA_DIR, file_name)

        # 空のCSVファイルを作成
        with open(file_path, "w") as f:
            pass  # 中身なし

    print(f"ダミーCSVファイル {num_files} 件を作成しました。")

# 14日前のファイルを取得
def get_target_files():
    base_date = datetime.datetime.today()
    target_date = (base_date - datetime.timedelta(days=14)).strftime("%Y-%m-%d")

    # 14日前のファイルのみ取得
    target_files = [f for f in os.listdir(DATA_DIR) if f.startswith(target_date)]

    print(f"14日前のファイル: {len(target_files)} 件")
    print(f"例: {target_files[0]}")
    return target_files

# 実行
create_dummy_csv_files()
target_files = get_target_files()