Merge pull request #3 from acompany-develop/feature/hiraoka/CC-856/reformat-handler.py

[CC-856] handler.pyの可読性を上げる

Author: tak-ka3

functions/cross_table/README.md

@@ -0,0 +1,113 @@
+# Cross Table Function
+
+## 概要
+
+Cross Table Functionは、2つのCSVファイルを結合し、クロス集計表を作成する関数です。両方のデータフレームの最左列をキーとして使用し、内部結合（INNER JOIN）を実行した後、属性列の組み合わせごとに集計を行います。Polarsライブラリを使用して高速なデータ処理を実現します。
+
+## 機能
+
+- **データ結合**: 2つのCSVファイルを最左列をキーとして結合
+- **列名リネーム**: 結合前に列名にプレフィックス（0:, 1:）を付与
+- **クロス集計**: 属性列の組み合わせごとに件数を集計
+- **閾値フィルタリング**: 集計数が閾値未満の行を除外
+- **ログ出力**: 処理経過とエラー情報をログファイルに記録
+- **処理結果**: 結果をcsvファイルに格納して返す
+
+## 入力データ
+
+### input_a.csv (user1専用入力データ)
+- **パス**: `/work/inputs/input_1/input_a.csv`
+- **形式**: CSVファイル
+- **要件**: 最左列が結合キーとして使用される
+
+### input_b.csv (user2専用入力データ)
+- **パス**: `/work/inputs/input_2/input_b.csv`
+- **形式**: CSVファイル
+- **要件**: 最左列が結合キーとして使用される
+
+## 出力データ
+
+### output.csv (user1専用出力データ)
+- **パス**: `/work/outputs/output_1/output.csv`
+- **形式**: CSVファイル
+- **内容**: クロス集計結果（number_of_rows列が先頭、属性列がソート済み）
+
+### output.csv (user2専用出力データ)
+- **パス**: `/work/outputs/output_2/output.csv`
+- **形式**: CSVファイル
+- **内容**: クロス集計結果（number_of_rows列が先頭、属性列がソート済み）
+
+### ログファイル
+- **app.log**: 処理ログ（`/work/outputs/output_1/app.log`、`/work/outputs/output_2/app.log`）
+
+## アルゴリズム
+
+1. **データ読み込み**
+   - `input_a.csv`をLazyFrameとして読み込み
+   - `input_b.csv`をLazyFrameとして読み込み
+
+2. **キー抽出**
+   - 両データフレームの最左列を結合キーとして抽出
+
+3. **列名リネーム**
+   - input_aの列名に`0:`プレフィックスを付与（キー列以外）
+   - input_bの列名に`1:`プレフィックスを付与（キー列以外）
+
+4. **データ結合**
+   - `polars.join()`を使用して内部結合を実行
+   - `left_on`: input_aの最左列
+   - `right_on`: input_bの最左列
+
+5. **クロス集計**
+   - 属性列（キー列以外）でグループ化
+   - 各組み合わせの件数を集計
+
+6. **閾値フィルタリング**
+   - 集計数が閾値（THRESHOLD=2）未満の行を除外
+
+7. **列順序整理**
+   - `number_of_rows`列を先頭に配置
+   - 属性列をアルファベット順にソート
+
+8. **結果保存**
+   - クロス集計結果を両方の出力ディレクトリに保存
+
+## 使用例
+
+### 入力ファイル例
+
+**input_a.csv**:
+```csv
+id,age,gender
+1,25,F
+2,30,M
+3,35,M
+4,28,F
+5,25,F
+6,30,M
+```
+
+**input_b.csv**:
+```csv
+id,city,department
+1,Tokyo,Sales
+2,Osaka,Marketing
+3,Tokyo,Engineering
+4,Kyoto,HR
+5,Tokyo,Sales
+6,Osaka,Marketing
+```
+
+### 出力ファイル例
+
+**output.csv** (user1用・user2用):
+```csv
+number_of_rows,0:age,0:gender,1:city,1:department
+2,25,F,Tokyo,Sales
+2,30,M,Osaka,Marketing
+```
+
+## 設定パラメータ
+
+- **THRESHOLD**: 集計数がこの値未満の行は出力されない（デフォルト: 2）
+

functions/cross_table/function/handler.py

@@ -3,144 +3,100 @@
 import traceback
 from datetime import datetime
 
-sys.path.insert(0, "/work/function/packages")  # functionが依存するパッケージのパス
+# 依存ライブラリがある場合は、packagesディレクトリをパスに追加
+sys.path.append("/work/function/packages")
 
+import polars as pl
+
+# 作業ディレクトリとI/Oディレクトリのパス設定
 WORK_DIR = "/work"
-INPUT_A_PATH = f"{WORK_DIR}/inputs/input_1"
-INPUT_B_PATH = f"{WORK_DIR}/inputs/input_2"
-OUTPUT_A_PATH = f"{WORK_DIR}/outputs/output_1"
-OUTPUT_B_PATH = f"{WORK_DIR}/outputs/output_2"
-DOWNLOAD_DIR = "downloads/"
+INPUT_DIR = f"{WORK_DIR}/inputs"
+OUTPUT_DIR = f"{WORK_DIR}/outputs"
+INPUT_1_DIR = f"{INPUT_DIR}/input_1"
+INPUT_2_DIR = f"{INPUT_DIR}/input_2"
+OUTPUT_1_DIR = f"{OUTPUT_DIR}/output_1"
+OUTPUT_2_DIR = f"{OUTPUT_DIR}/output_2"
 
 THRESHOLD = 2  # 集計数がこの値未満の行は出力されない
 
 
-def print_log(msg):
+def print_log(msg: str):
     current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-    try:
-        os.makedirs(DOWNLOAD_DIR, exist_ok=True)
-        with open(os.path.join(DOWNLOAD_DIR, "app.log"), "a") as log_file:
-            log_file.write(f"[{current_time}]:[handler.py]: {msg}\n")
-        os.makedirs(OUTPUT_A_PATH, exist_ok=True)
-        with open(os.path.join(OUTPUT_A_PATH, "app.log"), "a") as log_file:
-            log_file.write(f"[{current_time}]:[handler.py]: {msg}\n")
-        os.makedirs(OUTPUT_B_PATH, exist_ok=True)
-        with open(os.path.join(OUTPUT_B_PATH, "app.log"), "a") as log_file:
-            log_file.write(f"[{current_time}]:[handler.py]: {msg}\n")
-    except Exception:
-        pass  # 出力ディレクトリへの書き込みに失敗しても続行
-
-
-# メモリ使用量と実行時間を計測するための関数
-def get_memory_usage():
-    """現在のメモリ使用量を取得"""
-    import psutil
-
-    process = psutil.Process(os.getpid())
-    memory_info = process.memory_info()
-    return memory_info.rss / 1024 / 1024  # MB単位
-
-
-def print_memory_usage(stage_name):
-    """メモリ使用量を表示"""
-    try:
-        memory_mb = get_memory_usage()
-        print_log(f"[{stage_name}] メモリ使用量: {memory_mb:.2f} MB")
-    except Exception as e:
-        print_log(f"Failed to get memory usage: {e.__class__.__name__}")
-        return 0
+    for path in [OUTPUT_1_DIR, OUTPUT_2_DIR]:
+        os.makedirs(path, exist_ok=True)
+        with open(os.path.join(path, "app.log"), "a") as log_file:
+            log_file.write(f"{current_time}:{msg}\n")
+
+
+def cross_table_data(input_1_path: str, input_2_path: str):
+    print_log("cross_table_data: Started.")
+
+    # 入力データを読み込む（LazyFrameとして読み込み）
+    lf_a = pl.scan_csv(input_1_path)
+    lf_b = pl.scan_csv(input_2_path)
+
+    # キー列を特定
+    key_a = lf_a.columns[0]
+    key_b = lf_b.columns[0]
+
+    # 列名をリネーム
+    cols_to_rename_a = [col for col in lf_a.columns if col != key_a]
+    rename_map_a = {col: f"0:{col}" for col in cols_to_rename_a}
+    lf_a_renamed = lf_a.rename(rename_map_a)
+
+    cols_to_rename_b = [col for col in lf_b.columns if col != key_b]
+    rename_map_b = {col: f"1:{col}" for col in cols_to_rename_b}
+    lf_b_renamed = lf_b.rename(rename_map_b)
+
+    # データを結合
+    lf_joined = lf_a_renamed.join(
+        lf_b_renamed, left_on=key_a, right_on=key_b, how="inner"
+    )
+
+    # 属性列を特定（キー列以外）
+    attribute_cols = [col for col in lf_joined.columns if col != key_a and col != key_b]
+
+    # グループ化して集計
+    lf_summary = lf_joined.group_by(attribute_cols).agg(
+        pl.count().alias("number_of_rows")
+    )
+
+    # 閾値でフィルタリング
+    lf_filtered = lf_summary.filter(pl.col("number_of_rows") >= THRESHOLD)
+
+    # 列の順序を整理（number_of_rowsを先頭に）
+    sorted_cols = ["number_of_rows"] + sorted(attribute_cols)
+    lf_final = lf_filtered.select(sorted_cols)
+
+    # 計算を実行して結果を取得
+    final_result = lf_final.collect(streaming=True)
+
+    print_log("cross_table_data: Completed.")
+    return final_result
 
 
 def run():
     try:
-        print_memory_usage("開始時")
-        print_log("handler.run: Started.")
-
-        import polars as pl
-
-        print_log("handler.run: Imported successfully.")
-
-        # 入力データを読み込む
-        lf_a = pl.scan_csv(os.path.join(INPUT_A_PATH, "input_a.csv"))
-        print_log("handler.run: Read input_a.csv successfully.")
-        print_memory_usage("input_a.csv読み込み後")
-        lf_b = pl.scan_csv(os.path.join(INPUT_B_PATH, "input_b.csv"))
-        print_log("handler.run: Read input_b.csv successfully.")
-        print_memory_usage("input_b.csv読み込み後")
-
-        # キー列を特定
-        key_a = lf_a.columns[0]
-        key_b = lf_b.columns[0]
-
-        # 2. Join前のリネーム処理
-        # dataset_a の列名をリネーム (id以外)
-        cols_to_rename_a = [col for col in lf_a.columns if col != key_a]
-        rename_map_a = {col: f"0:{col}" for col in cols_to_rename_a}
-        lf_a_renamed = lf_a.rename(rename_map_a)
-        print_memory_usage("0_列名リネーム後")
-
-        # dataset_b の列名をリネーム (id以外)
-        cols_to_rename_b = [col for col in lf_b.columns if col != key_b]
-        rename_map_b = {col: f"1:{col}" for col in cols_to_rename_b}
-        lf_b_renamed = lf_b.rename(rename_map_b)
-        print_memory_usage("1_列名リネーム後")
-        print_log("handler.run: Renamed columns successfully.")
-
-        # 3. リネーム済みのLazyFrameをJoin
-        lf_joined = lf_a_renamed.join(
-            lf_b_renamed, left_on=key_a, right_on=key_b, how="inner"
+        print_log("run: Started.")
+
+        df_cross_table = cross_table_data(
+            f"{INPUT_1_DIR}/input_a.csv", f"{INPUT_2_DIR}/input_b.csv"
         )
-        print_log("handler.run: Merged successfully with leftmost columns.")
-        print_memory_usage("Join後")
+        print_log("run: Cross table data created.")
 
-        # 4. 全Attribute列でGroup By & Count
-        # id以外の全ての列（a_...とb_...）をグループ化のキーに指定
-        attribute_cols = [
-            col for col in lf_joined.columns if col != key_a and col != key_b
-        ]
+        # polarsのDataFrameを直接保存
+        df_cross_table.write_csv(f"{OUTPUT_1_DIR}/output.csv")
+        print_log("run: Saved output.csv to output_1.")
 
-        lf_summary = lf_joined.group_by(attribute_cols).agg(
-            pl.count().alias("number_of_rows")
-        )
-        print_memory_usage("Group By & Count後")
-
-        # 5. 列の整形
-        # number_of_rows を先頭に持ってくる
-        # 列名をソートしてから指定
-        sorted_cols = ["number_of_rows"] + sorted(attribute_cols)
-        lf_final = lf_summary.select(sorted_cols)
-
-        filtered = lf_final.filter(pl.col("number_of_rows") >= THRESHOLD)
-        print_log(f"handler.run: Filtered successfully with threshold {THRESHOLD}.")
-        print_memory_usage("Filter後")
-
-        # 計算を実行して結果を表示
-        final_result = filtered.collect(streaming=True)
-        print_memory_usage("Collect後")
-
-        # CSV形式で出力
-        try:
-            os.makedirs(OUTPUT_A_PATH, exist_ok=True)
-            os.makedirs(OUTPUT_B_PATH, exist_ok=True)
-
-            final_result.write_csv(os.path.join(OUTPUT_A_PATH, "output.csv"))
-            print_log("handler.run: Saved a's output.csv successfully.")
-            print_memory_usage("a's output.csv保存後")
-            final_result.write_csv(os.path.join(OUTPUT_B_PATH, "output.csv"))
-            print_log("handler.run: Saved b's output.csv successfully.")
-            print_memory_usage("b's output.csv保存後")
-        except Exception as e:
-            print_log(f"handler.run: Error saving results: {str(e)}")
-
-        print_log("handler.run: DONE.")
-        print_memory_usage("終了時")
+        df_cross_table.write_csv(f"{OUTPUT_2_DIR}/output.csv")
+        print_log("run: Saved output.csv to output_2.")
 
+        print_log("run: Completed.")
     except BaseException as e:
-        print_log(f"handler.run: ERROR: {str(e)}")
-        try:
-            os.makedirs(DOWNLOAD_DIR, exist_ok=True)
-            with open(os.path.join(DOWNLOAD_DIR, "error.log"), "w") as error_file:
-                traceback.print_exc(file=error_file)
-        except Exception:
-            pass  # エラーログの書き込みに失敗しても続行
+        print_log(f"error type: {type(e).__name__}")
+
+        tb = traceback.extract_tb(e.__traceback__)
+        if tb:
+            for i, frame in enumerate(tb):
+                print_log(f"error location {i + 1}: {frame.filename}:{frame.lineno}")
         raise e

functions/cross_table/requirements.txt

@@ -1,8 +1,5 @@
 numpy==1.26.4
-python-dateutil==2.8.2
 pytz==2023.3
-six==1.16.0
 tzdata==2023.3
 polars==0.19.19
-psutil==5.9.5
 pyarrow==14.0.1