【Python】subprocess入門：非同期処理で複数タスクを効率よく管理しよう

import time

print('[sample_1]Start!')
time.sleep(10) # 10秒スリープ.
print('[sample_1]wait 10 sec End!')

import time

print('[sample_2]Start!')
time.sleep(15) # 15秒スリープ.
print('[sample_2]wait 15 sec End!')

import time

print('[sample_3]Start!')
time.sleep(5) # 5秒スリープ.
print('[sample_3]wait 5 sec End!')

import os
import sys
import csv
from datetime import datetime
from collections import defaultdict
import time

# 引数: 入力CSV, 出力CSV
input_csv, output_csv = sys.argv[1], sys.argv[2]
print("file:" + input_csv)
totals = defaultdict(lambda : defaultdict(int))

# 売上のcsvを基に集計.
with open(input_csv, newline="", encoding="cp932") as f:
    reader = csv.DictReader(f)
    for row in reader:
        store = row["store"]
        base_date = datetime.strptime(row["date"], "%Y/%m/%d").date()
        ym_str = f"{base_date.year}-{base_date.month:02d}"
       
        # 店舗の月ごとの合計値を計算.
        totals[store][ym_str] += int(row["sales"])

# 集計結果を出力.
with open(output_csv, "w", newline="", encoding="cp932") as f:
    writer = csv.DictWriter(f, fieldnames=["store", "date", "total_sales"])
    writer.writeheader()
    for k, v in totals.items():
        for k2, v2 in v.items():
            writer.writerow({"store":k, "date": k2, "total_sales": v2})

print("> output:" + output_csv)

import os
import glob
import subprocess
import csv

# 出力先フォルダ.
RESULT_DIR = "result"
os.makedirs(RESULT_DIR, exist_ok=True)

# 処理対象のCSV一覧.
input_files = glob.glob("data/sales_*.csv")
output_files = []

# Popenを使って非同期でサブプロセスを起動.
print("============== Start store_monthly_sales.py ==============\n")
processes = []
for f in input_files:
    out = os.path.join(RESULT_DIR, os.path.basename(f).replace(".csv", "_summary.csv"))
    print("file:" + out)
    output_files.append(out)
    p = subprocess.Popen(
        ["python", "store_monthly_sales.py", f, out],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    processes.append((f, p))

# 全プロセスの完了を待機.
for f, p in processes:
    stdout, stderr = p.communicate()
    if p.returncode == 0:
        print(f"[OK] {f} → {stdout.strip()}")
    else:
        print(f"[ERROR] {f}: {stderr.strip()}")

print("============== End store_monthly_sales.py ==============\n")
print("============== Start Calc Company sales ==============\n")

# 全ての結果を統合
output_file = "result/all_sales.csv"
with open(output_file, "w", newline="", encoding="cp932") as fout:
    writer = None
    for of in output_files:
        if not os.path.exists(of):
            continue
        with open(of, newline="", encoding="cp932") as fin:
            reader = csv.DictReader(fin)
            if writer is None:
                writer = csv.DictWriter(fout, fieldnames=reader.fieldnames)
                writer.writeheader()
            for row in reader:
                writer.writerow(row)

print("result >> " + output_file)
print("============== End Calc Company sales ==============\n")

store,date,sales
A,2025/1/1,1200
A,2025/1/2,800
A,2025/2/11,600
A,2025/2/20,500
A,2025/2/21,10
A,2025/2/22,300

store,date,sales
B,2025/1/1,1200
B,2025/1/5,800
B,2025/1/11,600
B,2025/1/20,500
B,2025/1/21,10
B,2025/1/10,300

store,date,sales
C,2025/1/5,1200
C,2025/2/5,1800
C,2025/2/16,20
C,2025/3/2,5
C,2025/3/10,30
C,2025/3/20,200