【Python】APSchedulerとpandasで売上CSVの定期レポート作成CLIアプリを作ってみた
公開:
毎日の売上CSVを開いて合計を数えるだけの作業は、地味なわりに時間を奪います。そこでpandasで集計し、APSchedulerで毎日決まった時刻に走らせる定期レポート作成CLIを作ってみました。
サンプルCSV作成・即時集計・日次スケジュール起動・保存済みレポート一覧という4つのコマンドだけの自己完結構成で、外部サービスに繋がなくても手元で完結します。実際に4回のコマンドを実行し、すべて正常終了(終了コード0)まで確認できました。この記事では、その実装と動かした様子を初心者向けに追っていきます。
APSchedulerの基本概念、要件定義、実装、動作確認までを順番に学べる構成です。動画は目次から確認したい場面へ移動でき、本文だけでも手順と考え方が完結します。
動画の内容をテキストで確認する
オープニング。APSchedulerとpandasを使って売上定期レポート作成CLIを作るカリキュラムを始めます。概要紹介。
APSchedulerとpandasの役割と使い方を学ぶ 売上定期レポート作成CLIの要件を整理する 完成コードと実行結果を確認する 最後に実コマンドとファイル状態で完成挙動を確かめる 具体的にやること。
sampleコマンドでサンプル売上CSVを生成する 生成した明細の先頭5行と注文日の範囲を表示するreportコマンドで売上CSVを1回集計する 売上合計と注文件数と明細行数と対象期間をレポートへ書き出す 商品別の数量と金額と構成比を金額の多い順に並べる 実装環境・必須アプリ。
OS:Windows 11 Pro Python:3.13.3シェル:PowerShell 5.1必須アプリ:コードエディター、ターミナル、エクスプローラー パッケージ:pip、"apscheduler<4"、pandas PythonのAPSchedulerとpandasとは。
APScheduler:APSchedulerは、決まった時刻や間隔でPythonの関数を自動的に呼び出せる定期実行ライブラリpandas:pandasは、表形式のデータをDataFrameとして読み込み、集計や並べ替えをまとめて行えるデータ分析ライブラリ APSchedulerで作る売上レポートCLIの要点。
sampleは商品マスタから明細行を生成するreportはrun_onceで集計とレポート保存を行うscheduleは常駐して毎日の集計を予約する 売上定期レポート作成CLIの要件定義。
sampleコマンドでsales.csvが作成されるsample実行時に明細行数と注文日の範囲が表示されるreportコマンドの出力に売上合計と注文件数が並ぶ 出力先にsales_report_日付.txtが残るscheduleコマンドがジョブIDと次回実行時刻を表示する 常駐秒数の経過後にスケジューラ停止のメッセージが出る INTRO: Monaco Editorで売上定期レポート作成CLIを実装。
コードを1行ずつ入力し、補完と自動インデントを使いながら実行結果を確認します。LINE 002: モジュールの説明文。このスクリプト全体が何をするものかを説明するドキュメント文字列です。
APSchedulerとpandasを使った売上CSVの定期レポート作成CLIであることを示しています。LINE 004: 将来のアノテーション機能を有効化。型ヒントの評価方法を将来のPythonの仕様に合わせるための宣言です。
これにより関数の型注釈をより柔軟に書けるようになります。LINE 006: コマンドライン引数解析ライブラリの読み込み。コマンドライン引数を扱うためのargparseモジュールを読み込んでいます。
後でCLIのサブコマンドや引数を定義する際に使います。LINE 007: 乱数生成ライブラリの読み込み。ランダムな値を生成するためのrandomモジュールを読み込んでいます。
サンプル売上データの生成に利用します。LINE 008: 時間操作ライブラリの読み込み。待機処理などに使うtimeモジュールを読み込んでいます。
スケジューラを常駐させる際のスリープ処理に使われます。LINE 009: 日付・時刻関連クラスの読み込み。日付や日時、期間を扱うためのdate、datetime、timedeltaクラスをまとめて読み込んでいます。
売上データの日付処理やレポート作成時刻の記録に使います。LINE 010: パス操作クラスの読み込み。ファイルやディレクトリのパスを扱いやすくするPathクラスを読み込んでいます。
CSVファイルやレポートファイルの保存先指定に使います。LINE 012: pandasライブラリの読み込み。表形式データを扱うためのpandasライブラリをpdという別名で読み込んでいます。
売上CSVの読み込みや集計処理の中心となるライブラリです。LINE 013: バックグラウンドスケジューラの読み込み。APSchedulerのバックグラウンド実行用スケジューラクラスを読み込んでいます。
定期的にレポート作成処理を実行する仕組みに使います。LINE 014: Cronトリガーの読み込み。cron形式で実行時刻を指定できるトリガークラスを読み込んでいます。
毎日決まった時刻にジョブを実行するために使います。LINE 018: 既定のデータディレクトリ名の定義。売上CSVを置く既定のディレクトリ名を定数として定義しています。
コマンド引数で指定がない場合にこの値が使われます。LINE 019: 既定のレポート出力先ディレクトリ名の定義。作成したレポートを保存する既定のディレクトリ名を定数として定義しています。
コマンド引数で指定がない場合にこの値が使われます。LINE 020: 売上CSVファイル名の定義。売上データを保存するCSVファイルの固定名を定数として定義しています。
サンプル作成時と読み込み時の両方でこの名前を使います。LINE 021: レポートファイル名の接頭辞の定義。保存するレポートファイル名の先頭に付ける文字列を定数として定義しています。
日付と組み合わせてファイル名を作る際に使われます。LINE 022: スケジューラジョブIDの定義。APSchedulerに登録するジョブを識別するためのIDを定数として定義しています。
ジョブの登録や取得の際にこのIDを指定します。LINE 023: 商品マスタリストの定義開始。サンプル売上データ生成に使う商品情報のリストを定義し始めています。
各商品は商品コード、商品名、単価の組み合わせで表されます。LINE 024: 商品データ1件目の定義。商品コードA-101、商品名BlendCoffee、単価1200円の商品情報をリストの1件目として定義しています。
LINE 025: 商品データ2件目の定義。商品コードA-102、商品名DripBag10、単価980円の商品情報をリストの2件目として定義しています。LINE 026: 商品データ3件目の定義。
商品コードB-201、商品名TeaAssort、単価1580円の商品情報をリストの3件目として定義しています。LINE 027: 商品データ4件目の定義。商品コードB-202、商品名HerbTeaCan、単価2200円の商品情報をリストの4件目として定義しています。
LINE 028: 商品データ5件目の定義。商品コードC-301、商品名MugCup、単価1800円の商品情報をリストの5件目として定義しています。LINE 029: 商品データ6件目の定義。
商品コードC-302、商品名SteelBottle、単価3400円の商品情報をリストの6件目として定義しています。LINE 030: 商品マスタリストの定義終了。商品情報リストPRODUCTSの定義を閉じています。
以降のサンプルデータ生成処理でこのリストから商品をランダムに選びます。RUN 1/9: 既定値と商品マスタを確認する。インポートと定数の定義までを入力しました。
商品マスタの件数や既定のディレクトリ名を表示し、後の処理が参照する土台を確かめます。CHECK 1/9: 途中実行に成功。
商品マスタ件数: 6 既定データ先: sales_data / 既定出力先: reports 先頭の商品: ('A-101', 'BlendCoffee', 1200) ジョブID: daily_sales_report RETURN 01: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 034: サンプルCSV作成関数の定義。データディレクトリ、行数、日数、乱数シードを受け取ってサンプル売上CSVを作成する関数を定義しています。戻り値として作成したCSVのパスを返します。
LINE 035: 乱数生成器の初期化。指定されたシード値を使って乱数生成器を作成しています。同じシードを指定すれば毎回同じサンプルデータが生成されるようになります。
LINE 036: 本日の日付の取得。今日の日付を取得して変数todayに格納しています。これを基準に注文日をさかのぼって生成します。
LINE 037: レコード格納用リストの初期化。生成する売上データを一時的にためておくための空リストを用意しています。この後のループでここに1件ずつデータを追加していきます。
LINE 038: 指定行数分の繰り返し処理の開始。指定された行数の分だけ売上データを生成するためのループを開始しています。indexにはループの回数が0から順に入ります。
LINE 039: 商品情報のランダム選択。商品マスタリストからランダムに1件選び、商品コード、商品名、単価を取り出しています。これにより多様な商品の売上データが作られます。
LINE 040: 注文日のランダム生成。今日の日付から指定した日数の範囲内でランダムに日数を引いて、注文日を算出しています。これにより注文日が過去数日間に散らばります。
LINE 041: 売上レコードのリストへの追加開始。1件分の売上データを辞書として組み立て、recordsリストに追加する処理を開始しています。LINE 042: 売上レコードの辞書定義開始。
1件分の売上データを表す辞書の定義を開始しています。この後に注文ID、注文日、商品情報などのキーと値を並べていきます。LINE 043: 注文IDの生成。
連番のindexを使って4桁ゼロ埋めの注文IDを作成し、order_idキーに設定しています。例えばORD-0001のような形式になります。LINE 044: 注文日の文字列化と設定。
先ほど算出した注文日をISO形式の文字列に変換し、order_dateキーに設定しています。LINE 045: 商品コードの設定。ランダムに選んだ商品コードをproduct_codeキーに設定しています。
LINE 046: 商品名の設定。ランダムに選んだ商品名をproduct_nameキーに設定しています。LINE 047: 数量のランダム生成と設定。
1から5の範囲でランダムな注文数量を生成し、quantityキーに設定しています。LINE 048: 単価の設定。ランダムに選んだ商品の単価をunit_priceキーに設定しています。
LINE 049: 売上レコードの辞書定義終了。1件分の売上データを表す辞書の定義を閉じています。この辞書がrecordsリストに追加されます。
LINE 050: レコード追加処理の終了。1件分の売上データをrecordsリストへ追加する処理を締めくくっています。ループが繰り返されるたびにデータが1件ずつ増えていきます。
LINE 051: データフレームの作成と並び替え。生成したレコードのリストからpandasのデータフレームを作成し、注文日順に並び替えてインデックスを振り直しています。この後CSVとして保存されます。
LINE 053: 保存先ディレクトリの指定。引数で渡されたデータディレクトリ名をPathオブジェクトに変換しています。この後ディレクトリの作成やファイルパスの組み立てに使います。
LINE 054: ディレクトリの作成。保存先ディレクトリが存在しない場合に自動的に作成しています。既に存在していてもエラーにならないようにしています。
LINE 055: CSVファイルパスの組み立て。ディレクトリと固定のファイル名を組み合わせて、保存するCSVファイルの完全なパスを作成しています。LINE 056: データフレームのCSV書き出し。
作成した売上データフレームをCSVファイルとして書き出しています。インデックス列は含めず、文字コードはUTF-8で保存しています。LINE 057: 作成したCSVパスの返却。
作成したサンプル売上CSVのファイルパスを呼び出し元に返しています。この値は後続の処理でファイルの場所を示すために使われます。RUN 2/9: サンプルCSVを作る関数を試す。
create_sample_csvが書き終わりました。行数とシードを指定して呼び出し、CSVが作られたことと行数を確認します。CHECK 2/9: 途中実行に成功。
作成先: cp_data/sales.csv / 存在: True 行数: 10 RETURN 02: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 061: 売上読み込み関数の定義。
CSVファイルのパスを受け取り、売上データを読み込んで前処理を行う関数を定義しています。戻り値として整形済みのデータフレームを返します。LINE 062: CSVファイルの読み込み。
指定されたパスの売上CSVファイルをUTF-8で読み込み、データフレームとして取得しています。LINE 063: 注文日列の日付型変換。order_date列の文字列を日付時刻型に変換しています。
これにより後の期間集計や日付の比較が正しく行えるようになります。LINE 064: 数量列の数値型変換。quantity列を数値型に変換しています。
これにより計算や集計で正しく扱えるようになります。LINE 065: 単価列の数値型変換。unit_price列を数値型に変換しています。
これにより金額計算が正しく行えるようになります。LINE 066: 売上金額列の算出。数量と単価を掛け合わせて売上金額を計算し、amount列として新しく追加しています。
この列が後の集計で使われます。LINE 067: 整形済みデータフレームの返却。日付変換や金額計算を終えたデータフレームを呼び出し元に返しています。
この結果がaggregate関数への入力になります。RUN 3/9: 読み込み後の列構成を確認する。load_salesで型変換と金額列の追加まで行う関数が完成しました。
読み込んだDataFrameの列名と先頭行の金額を表示します。CHECK 3/9: 途中実行に成功。
列: ['order_id', 'order_date', 'product_code', 'product_name', 'quantity', 'unit_price', 'amount'] 先頭の金額: 980 RETURN 03: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 071: 集計関数の定義。
整形済みの売上データフレームを受け取り、合計金額や商品別集計などをまとめた辞書を作成する関数を定義しています。LINE 072: 商品別集計処理の開始。商品コードと商品名でグループ化した商品別集計の処理を開始しています。
この後数量と金額の合計を求めていきます。LINE 073: 商品コードと商品名でのグループ化。データフレームを商品コードと商品名の組み合わせでグループ化しています。
as_indexをFalseにすることでグループ化列が通常の列として残ります。LINE 074: 数量と金額の合計集計。グループごとに数量の合計と金額の合計を計算しています。
これにより商品ごとの売れ行きが分かるようになります。LINE 075: 金額の降順での並び替え。集計した金額が多い順にデータを並び替えています。
これにより売上上位の商品が上に表示されます。LINE 076: インデックスの振り直し。並び替え後の行番号を0から振り直しています。
これにより見やすい連番のインデックスになります。LINE 077: 商品別集計処理の終了。商品コード・商品名でのグループ化から並び替えまでの一連の処理を閉じ、by_product変数に結果を格納しています。
LINE 078: 売上合計金額の算出。データフレーム全体のamount列を合計し、整数に変換して売上全体の合計金額を求めています。LINE 079: 合計金額が正かどうかの判定開始。
売上合計金額が0より大きいかどうかを判定する条件分岐を開始しています。0円の場合の割り算を避けるための処理です。LINE 080: 商品別の売上構成比の算出。
各商品の金額を売上合計で割ってパーセントに変換し、小数点第1位に丸めてshare_pct列として追加しています。LINE 081: 合計金額が0の場合の分岐。売上合計金額が0以下の場合の処理に入ります。
この場合は構成比の計算ができないため、代わりに固定値を設定します。LINE 082: 売上ゼロ時の割合を0に設定。売上合計が0円のときに商品別の構成比を0.0で埋めています。
ゼロ除算を避けつつ、割合の列を必ず用意できるようにしています。LINE 084: 集計結果の辞書を返す開始。aggregate関数が呼び出し元へ渡す集計結果の辞書を組み立て始めています。
ここから複数の集計値をまとめて返却します。LINE 085: 売上合計を辞書に格納。事前に計算しておいた売上合計を辞書のtotal_amountキーに設定しています。
レポート本文で売上合計として利用されます。LINE 086: 注文件数を算出して格納。order_id列の重複を除いた件数を数え、注文件数として辞書に格納しています。
int型に変換して扱いやすくしています。LINE 087: 明細行数を算出して格納。データフレーム全体の行数を数え、明細行数として辞書に格納しています。
CSVの明細がどれだけあったかを示す値です。LINE 088: 集計期間の開始日を算出。order_date列の最小値から日付部分を取り出し、文字列に変換して期間の開始日として格納しています。
LINE 089: 集計期間の終了日を算出。order_date列の最大値から日付部分を取り出し、文字列に変換して期間の終了日として格納しています。LINE 090: 商品別集計表を辞書に格納。
事前に作成した商品別集計のデータフレームをそのままby_productキーに格納しています。レポート本文の表として使われます。LINE 091: 集計結果辞書の定義終了。
aggregate関数が返す辞書リテラルの定義を閉じています。この辞書が呼び出し元に返却されます。RUN 4/9: 集計関数の戻り値を確認する。
aggregateが売上合計や商品別の表を返せる状態になりました。辞書の主要な値を取り出して表示します。CHECK 4/9: 途中実行に成功。
売上合計: 45180 注文件数: 10 / 明細行数: 10 商品別の行数: 6 RETURN 04: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 095: レポート本文組み立て関数の定義。
集計結果summaryとCSVパス、生成日時を受け取り、レポートの文字列を組み立てる関数を定義しています。LINE 096: レポート行リストの作成開始。レポート本文を構成する各行の文字列をリストとしてまとめ始めています。
後で改行で連結して1つの文章にします。LINE 097: レポートタイトル行を作成。生成日時から日付を取り出し、レポートのタイトルとして先頭行に表示する文字列を作成しています。
LINE 098: 作成日時行を作成。生成日時を年月日時分秒の形式に整形し、レポートの作成日時として表示する行を作成しています。LINE 099: 対象ファイル行を作成。
集計に使った売上CSVのパスを表示する行を作成しています。どのファイルを元にしたレポートかが分かるようにしています。LINE 100: 対象期間行を作成。
summaryに格納された集計期間の開始日と終了日を組み合わせ、対象期間として表示する行を作成しています。LINE 101: 空行を挿入。見出し部分と数値部分の間に空行を入れ、レポートを読みやすく区切っています。
LINE 102: 売上合計行を作成。summaryの売上合計をカンマ区切りの数値として整形し、円単位で表示する行を作成しています。LINE 103: 注文件数行を作成。
summaryの注文件数を件単位で表示する行を作成しています。LINE 104: 明細行数行を作成。summaryの明細行数を行単位で表示する行を作成しています。
LINE 105: 空行を挿入。数値部分と商品別集計表の間に空行を入れ、レポートの区切りを見やすくしています。LINE 106: 商品別集計の見出しを作成。
これから表示する商品別集計表の前に、見出しとなる文字列を作成しています。LINE 107: 商品別集計表を文字列化。summaryに含まれる商品別集計のデータフレームを、インデックスなしの文字列形式に変換してレポートに組み込んでいます。
LINE 108: 末尾に空行を挿入。レポート本文の最後に空行を追加し、出力の見た目を整えています。LINE 109: レポート行リストの定義終了。
レポート本文を構成する行のリスト定義を閉じています。この後、改行で連結されて1つの文字列になります。LINE 110: レポート行を改行で連結して返す。
組み立てた各行のリストを改行文字で連結し、1つの文字列としてレポート本文を返しています。RUN 5/9: レポート本文の組み立てを確認する。format_reportでレポート1本分の文字列を作れるようになりました。
先頭行と売上合計の行、全体の行数を表示します。CHECK 5/9: 途中実行に成功。# 売上レポート2026-08-18 売上合計: 45,180円 本文の行数: 17 RETURN 05: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 114: レポート保存関数の定義。出力ディレクトリとレポート本文、生成日時を受け取り、ファイルとして保存する関数を定義しています。
LINE 115: 出力先ディレクトリのPathを作成。引数で受け取った出力先ディレクトリの文字列をPathオブジェクトに変換し、後続のファイル操作に使えるようにしています。LINE 116: 出力先ディレクトリを作成。
出力先ディレクトリが存在しない場合に、親ディレクトリも含めて作成しています。既に存在してもエラーにならないようにしています。LINE 117: 日付付きレポートファイル名を作成。
生成日時の日付部分を使い、REPORT_PREFIXと組み合わせてレポートファイルのパスを作成しています。日ごとに別ファイルとして保存されます。LINE 118: レポート本文をファイルへ書き込み。
組み立てたレポート本文の文字列を、UTF-8エンコーディングで指定パスのファイルに書き込んでいます。LINE 119: 保存したレポートのパスを返す。ファイルとして保存したレポートのパスを呼び出し元に返し、保存先を確認できるようにしています。
LINE 123: 1回集計処理の関数定義。売上CSVを読み込んで集計し、日付付きレポートを保存するまでの一連の処理を行う関数を定義しています。LINE 124: 売上CSVのパスを組み立て。
データディレクトリとSALES_FILEを組み合わせ、読み込む売上CSVのパスを作成しています。LINE 125: 売上CSVの存在チェック。組み立てたパスに売上CSVファイルが存在するかどうかを確認しています。
存在しない場合は次の行でエラーを発生させます。LINE 126: ファイルが無い場合の例外送出。売上CSVが見つからない場合に、パスを含めたエラーメッセージとともに例外を発生させて処理を中断しています。
LINE 128: 売上CSVを読み込み前処理。load_sales関数を呼び出し、CSVを読み込んで型変換や金額計算を済ませたデータフレームを取得しています。LINE 129: 集計処理を実行。
aggregate関数を呼び出し、読み込んだデータフレームから売上合計や商品別集計などの結果をまとめて取得しています。LINE 130: レポート生成時刻を取得。現在の日時を取得し、レポートのタイトルやファイル名に使う生成日時として保持しています。
LINE 131: レポート本文を作成。集計結果とCSVパス、生成日時を渡してformat_report関数を呼び出し、レポートの文字列を作成しています。LINE 132: レポートをファイルに保存。
save_report関数を呼び出し、作成したレポート本文を日付付きファイルとして保存し、そのパスを取得しています。LINE 133: 集計結果一式を辞書で返す。保存先パスとレポート本文、集計結果summaryをまとめた辞書を作成し、呼び出し元に返しています。
RUN 6/9: 集計から保存までを通しで動かす。run_onceが読み込みから保存までをまとめる関数として完成しました。保存先のパスと売上合計を表示して通しの流れを確かめます。
CHECK 6/9: 途中実行に成功。保存先: cp_reports/sales_report_2026-08-18.txt 売上合計: 45180 ファイルの存在: True RETURN 06: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 137: 時刻文字列を解釈する関数の定義。HH:MM形式の文字列を受け取り、時と分の整数値に変換する関数を定義しています。LINE 138: 時刻文字列をコロンで分割。
partitionメソッドでコロンを区切りとして文字列を分割し、時の部分と分の部分の文字列を取り出しています。LINE 139: 時と分を整数に変換。分割して取り出した文字列を整数型に変換し、後の範囲チェックや時刻設定で使えるようにしています。
LINE 140: 時刻の範囲を検証。時が0から23、分が0から59の範囲に収まっているかを確認しています。範囲外なら次の行で例外を発生させます。
LINE 141: 不正な時刻形式の例外送出。時刻が正しい範囲に収まっていない場合に、入力値を含めたエラーメッセージとともに例外を発生させています。LINE 142: 解釈した時と分を返す。
検証済みの時と分をタプルとして呼び出し元に返し、CronTriggerの設定などに利用できるようにしています。RUN 7/9: 実行時刻の解釈を確認する。parse_hhmmでHH:MM形式を時と分へ分けられるようになりました。
正しい指定と範囲外の指定の両方を試します。CHECK 7/9: 途中実行に成功。07:30の解釈: (7, 30) 23:59の解釈: (23, 59) 範囲外の指定: 時刻はHH:MM形式で指定してください: 24:00 RETURN 07: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 146: 日次スケジュール開始関数の定義。データディレクトリや出力先、実行時刻、常駐秒数を受け取り、日次のレポート集計を常駐実行する関数を定義しています。
LINE 147: 実行時刻を時と分に分解。parse_hhmm関数を呼び出し、引数atで指定された時刻文字列を時と分の整数値に変換しています。LINE 148: Cronトリガーを作成。
取得した時と分を使い、毎日その時刻に発火するCronTriggerを作成しています。このトリガーがジョブの実行タイミングを決めます。LINE 150: ジョブとして実行する関数を定義。
スケジューラが毎日実行する処理内容を、job_funcという内部関数としてここで定義しています。LINE 151: 実行時刻の文字列を取得。ジョブが実行された時点の現在時刻を取得し、ログ出力用に年月日時分秒の文字列へ整形しています。
LINE 152: 集計処理のtry開始。レポート集計処理を実行する部分をtryブロックで囲み、失敗しても常駐処理全体が止まらないようにしています。LINE 154: 1回分の集計処理を実行。
run_once関数を呼び出し、売上CSVの集計とレポート保存をジョブ実行のたびに行っています。LINE 155: 保存結果をログ出力。集計処理が成功した際に、保存されたレポートのパスを含むメッセージを表示しています。
LINE 156: 例外発生時の捕捉。集計処理中に何らかの例外が発生した場合にそれを捕捉し、errorという変数に格納しています。LINE 157: 失敗時のログ出力。
集計処理が失敗した際に、時刻とエラー内容を含むメッセージを表示し、常駐処理は継続させています。LINE 159: バックグラウンドスケジューラを作成。APSchedulerのBackgroundSchedulerインスタンスを作成し、以降のジョブ登録や起動に使う準備をしています。
LINE 160: ジョブをスケジューラに登録。作成したjob_func、トリガー、ジョブID、名前を指定してスケジューラにジョブを登録し、日次実行の設定を完了させています。LINE 161: スケジューラを起動。
登録したジョブが実際に動作するようにスケジューラを起動し、バックグラウンドでの常駐実行を開始しています。LINE 163: 登録済みジョブを取得。スケジューラからJOB_IDに一致するジョブオブジェクトを取得し、次回実行時刻の確認に使えるようにしています。
LINE 164: 次回実行時刻を取得。取得したジョブオブジェクトからnext_run_time属性を安全に取り出し、存在しない場合はNoneとして扱っています。LINE 165: 次回実行時刻の欠落チェック。
next_runがまだ取得できていない場合に備え、後続の処理で代わりの計算方法を使うための条件分岐を行っています。LINE 166: トリガーから次回発火時刻を算出。next_runが取得できなかった場合に、CronTriggerのget_next_fire_timeメソッドを使って次回の発火時刻を計算しています。
LINE 167: 登録完了メッセージの表示。スケジュールを登録したことと、毎日何時何分に集計するかを画面に表示しています。時刻はゼロ埋めの2桁で見やすく整形しています。
LINE 168: ジョブ情報と次回実行時刻の表示。登録したジョブのIDと、次に実行される日時を表示しています。next_runを日時の文字列形式に変換して分かりやすく出力しています。
LINE 169: 常駐秒数の判定。run_secondsが0より大きいかどうかを判定しています。この結果によって、常駐時間を区切るか無期限に動かし続けるかを決めています。
LINE 170: 常駐終了予告の表示。指定した秒数だけ常駐して終了することを画面に表示しています。0を指定した場合の操作方法も合わせて案内しています。
LINE 173: 例外処理の開始。常駐処理中に発生する可能性のある中断を捉えるため、try文を使って処理範囲を開始しています。LINE 174: 常駐秒数の再判定。
run_secondsが0より大きいかどうかを再度確認し、有限時間の待機を行うか無限ループに入るかを分岐しています。LINE 175: 指定秒数の待機。指定された秒数だけ処理を停止させています。
この間にバックグラウンドのスケジューラがジョブを実行します。LINE 176: 無限待機への分岐。run_secondsが0の場合の処理に入ることを示しています。
この場合はCtrl+Cが押されるまで常駐を続けます。LINE 177: 無限ループの開始。終了条件のないループを開始し、ユーザーが中断するまでプログラムを常駐させ続けています。
LINE 178: 1秒ごとの待機。1秒だけ処理を止めることを繰り返しています。CPUを無駄に使わずに常駐状態を維持するための待機です。
LINE 179: 中断シグナルの捕捉。Ctrl+Cによる中断やシステム終了の要求を受け取った場合に、この例外処理へ入るようにしています。LINE 180: 停止要求メッセージの表示。
ユーザーからの停止操作を受け取ったことを画面に表示し、これから終了処理に移ることを伝えています。LINE 181: 終了処理の開始。正常終了でも例外発生時でも必ず実行される後始末の処理をここから記述しています。
LINE 182: スケジューラの停止。バックグラウンドで動いていたスケジューラを安全に停止させています。これによりジョブの実行が止まります。
LINE 184: 実行結果の辞書返却。ジョブID、実行時刻、次回実行予定時刻をまとめた辞書を呼び出し元へ返しています。この情報は呼び出し元での表示に使われます。
LINE 188: レポート一覧関数の定義。保存済みレポートの一覧をデータフレームとして作成する関数を定義しています。出力先ディレクトリを引数として受け取ります。
LINE 189: パスオブジェクトへの変換。文字列で渡されたディレクトリをPathオブジェクトに変換し、以降のファイル操作をしやすくしています。LINE 190: レポートファイルの検索。
ディレクトリが存在する場合に、レポート接頭辞とtxt拡張子に一致するファイルを検索して並べ替えています。存在しない場合は空のリストにしています。LINE 191: レコードリストの作成開始。
各レポートファイルの情報をまとめるためのリスト内包表記を開始しています。LINE 192: 1件分の辞書の開始。1つのレポートファイルについての情報をまとめる辞書の定義を始めています。
LINE 193: ファイル名の取得。レポートファイルのファイル名部分だけを取り出して辞書に格納しています。LINE 194: ファイルサイズの取得。
ファイルの容量をバイト単位で取得し、辞書に格納しています。LINE 195: 更新日時の取得と整形。ファイルの最終更新時刻を取得し、見やすい日時文字列に変換して辞書に格納しています。
LINE 196: 辞書定義の終了。1件分のレポート情報をまとめた辞書の定義を閉じています。LINE 197: ループ対象の指定。
先ほど取得したファイル一覧の各要素に対して、この辞書作成処理を繰り返し適用しています。LINE 198: リスト内包表記の終了。すべてのレポートファイルについての辞書をまとめたリストの作成を完了させています。
LINE 199: データフレームへの変換。作成したレコードのリストを、指定した列順のデータフレームに変換して呼び出し元へ返しています。RUN 8/9: 保存済みレポートの一覧を確認する。
list_reportsが保存済みファイルを表にまとめる関数として完成しました。レポートを1件作ってから一覧の件数と列を確認します。CHECK 8/9: 途中実行に成功。
一覧の件数: 1 列: ['file', 'size_bytes', 'updated_at'] file size_bytes updated_at sales_report_2026-08-18.txt 615 2026-08-18 00:33:40 RETURN 08: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 203: パーサー構築関数の定義。コマンドライン引数を解析するためのパーサーを組み立てる関数を定義しています。
LINE 204: 共通引数パーサーの作成。複数のサブコマンドで共通して使う引数をまとめるためのパーサーを作成しています。ヘルプ表示は個別に管理するため無効にしています。
LINE 205: データディレクトリ引数の追加。売上CSVを置くディレクトリを指定する引数を追加しています。指定がない場合は既定値が使われます。
LINE 206: 出力ディレクトリ引数の追加。レポートの保存先ディレクトリを指定する引数を追加しています。指定がない場合は既定値が使われます。
LINE 208: メインパーサーの作成開始。プログラム全体のコマンドライン引数を扱うメインパーサーの作成を始めています。LINE 209: プログラム名の設定。
コマンドライン実行時に表示されるプログラム名を設定しています。LINE 210: プログラム説明の設定。このCLIがどのような処理を行うかを説明する文言を設定しています。
ヘルプ表示に使われます。LINE 211: パーサー作成の終了。メインパーサーの設定を閉じています。
LINE 212: サブコマンド機構の追加。sample、report、schedule、listといったサブコマンドを扱うための仕組みをパーサーに追加しています。コマンド指定を必須にしています。
LINE 214: sampleサブコマンドの追加。サンプル売上CSVを作成するためのsampleサブコマンドを追加し、共通引数も引き継いでいます。LINE 215: 行数引数の追加。
生成する明細行数を指定するrows引数を追加しています。指定がない場合は60行になります。LINE 216: 日数引数の追加。
注文日をどれだけの日数に散らすかを指定するdays引数を追加しています。指定がない場合は7日になります。LINE 217: 乱数シード引数の追加。
生成結果を再現できるようにする乱数シード引数を追加しています。指定がない場合は42になります。LINE 219: reportサブコマンドの追加。
売上CSVを1回集計してレポートを保存するためのreportサブコマンドを追加しています。共通引数も引き継いでいます。LINE 221: scheduleサブコマンドの追加。
毎日指定時刻にレポート集計を実行するscheduleサブコマンドを追加し、共通引数も引き継いでいます。LINE 222: 実行時刻引数の追加。レポート集計を実行する時刻をHH:MM形式で指定するat引数を追加しています。
指定がない場合は07:30になります。LINE 223: 常駐秒数引数の追加。プログラムを常駐させる秒数を指定するrun-seconds引数を追加しています。
0を指定するとCtrl+Cまで常駐します。LINE 225: listサブコマンドの追加。保存済みレポートを一覧表示するlistサブコマンドを追加しています。
共通引数も引き継いでいます。LINE 226: パーサーの返却。組み立てが完了したパーサーを呼び出し元へ返しています。
RUN 9/9: サブコマンドの定義を確認する。build_parserで4つのサブコマンドと共通オプションが定義できました。引数を渡して解析結果を表示し、既定値まで確かめます。
CHECK 9/9: 途中実行に成功。コマンド: sample / 行数: 12 / データ先: sales_data 時刻: 07:30 / 常駐秒数: 3 / 出力先: reports RETURN 09: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 230: main関数の定義。プログラム全体のエントリーポイントとなるmain関数を定義しています。終了コードを返す仕組みになっています。
LINE 231: 引数の解析。build_parser関数で作ったパーサーを使って、コマンドライン引数を実際に解析しています。LINE 233: sampleコマンドの判定。
実行されたコマンドがsampleかどうかを判定し、該当する場合の処理へ分岐しています。LINE 234: サンプルCSVの作成呼び出し。指定された引数を使ってサンプル売上CSVを作成し、そのファイルパスを受け取っています。
LINE 235: 作成したCSVの読み込み。作成したばかりのサンプルCSVを読み込み、内容確認のためのデータフレームにしています。LINE 236: 作成完了メッセージの表示。
サンプルCSVを作成したことと、そのファイルパスを画面に表示しています。LINE 237: 明細行数と注文期間の表示。作成したCSVの行数と、注文日の最小値・最大値を画面に表示しています。
LINE 238: 先頭データの表示。作成したCSVの先頭5行を表として画面に表示し、内容を確認できるようにしています。LINE 240: reportコマンドの判定。
実行されたコマンドがreportかどうかを判定し、該当する場合の処理へ分岐しています。LINE 242: 1回集計処理の実行。run_once関数を呼び出し、売上CSVを集計してレポートを保存する処理を実行しています。
LINE 243: レポート本文の表示。生成されたレポートの本文テキストを画面に表示しています。LINE 244: 保存先パスの表示。
レポートが保存されたファイルパスを画面に表示しています。LINE 246: scheduleコマンドの判定。実行されたコマンドがscheduleかどうかを判定し、該当する場合の処理へ分岐しています。
LINE 248: 日次スケジュールの開始呼び出し。start_daily_schedule関数を呼び出し、指定した時刻に毎日レポート集計を行う常駐処理を開始しています。LINE 249: 停止完了メッセージの表示。
スケジューラが停止したことと、対象だったジョブIDや予定時刻を画面に表示しています。LINE 251: listコマンドの判定。実行されたコマンドがlistかどうかを判定し、該当する場合の処理へ分岐しています。
LINE 252: 保存済みレポート一覧の取得。list_reports関数を呼び出し、保存済みレポートの一覧をデータフレームとして取得しています。LINE 253: 一覧が空かどうかの判定。
取得したレポート一覧が空かどうかを判定し、表示内容を分岐させる準備をしています。LINE 254: レポート未保存時のメッセージ表示。table.emptyがTrueのとき、つまり保存済みレポートが1件もないときに、対象ディレクトリのパス付きでその旨をコンソールに表示します。
ユーザーがどのディレクトリを確認すればよいか一目で分かるようにしています。LINE 255: else節の開始。table.emptyがFalse、つまりレポートが1件以上ある場合の処理をここから記述します。
前の行の未保存分岐と対になる処理です。LINE 256: レポート件数の表示。len(table)で一覧表の行数を数え、保存されているレポートの件数と対象ディレクトリを合わせて表示します。
ユーザーは何件のレポートが保存されているかをここで把握できます。LINE 257: レポート一覧表の表示。table.to_string(index=False)で行番号を省いた一覧表を文字列に変換し、print関数でコンソールに出力します。
ファイル名やサイズ、更新日時などの情報が整形されて表示されます。LINE 259: main関数の正常終了。すべてのコマンド処理を終えた後、main関数の戻り値として0を返します。
この0は正常終了を意味する終了コードとして扱われます。LINE 262: スクリプト実行時の判定。このファイルがモジュールとしてインポートされたのではなく、直接実行されたときにだけ以降の処理を行うための定型的な条件式です。
CLIツールとしてコマンドラインから起動されたときに真になります。LINE 263: main関数の呼び出しと終了コード設定。main関数を呼び出してその戻り値を受け取り、SystemExitを発生させることでプログラム全体の終了コードとして設定します。
main関数が返した0や例外発生時のコードがそのままプロセスの終了ステータスになります。実行1/4: サンプル売上CSVを作成する。sampleコマンドで練習用の売上CSVを作ります。
作成先のパスと明細行数、先頭5行の内容が画面に表示されます。確認1/4: サンプル売上CSVを作成する。sampleコマンドで練習用の売上CSVを作ります。
作成先のパスと明細行数、先頭5行の内容が画面に表示されます。RETURN 10: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
実行2/4: 売上CSVを1回集計してレポートを保存する。サンプルCSVを用意したうえでreportコマンドを実行します。売上合計や注文件数を含むレポート本文と保存先が表示されます。
確認2/4: 売上CSVを1回集計してレポートを保存する。サンプルCSVを用意したうえでreportコマンドを実行します。売上合計や注文件数を含むレポート本文と保存先が表示されます。
RETURN 11: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。実行3/4: 毎日の集計スケジュールを登録する。
scheduleコマンドへ実行時刻と常駐秒数を渡します。登録したジョブIDと次回実行時刻が出たあと、指定秒数で常駐が終わります。確認3/4: 毎日の集計スケジュールを登録する。
scheduleコマンドへ実行時刻と常駐秒数を渡します。登録したジョブIDと次回実行時刻が出たあと、指定秒数で常駐が終わります。RETURN 12: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。実行4/4: 保存済みレポートを一覧する。サンプル作成とレポート保存を済ませてからlistコマンドを実行します。
保存済みレポートの件数とファイル一覧の表が表示されます。確認4/4: 保存済みレポートを一覧する。サンプル作成とレポート保存を済ませてからlistコマンドを実行します。
保存済みレポートの件数とファイル一覧の表が表示されます。RETURN 13: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
学習内容のまとめ。BackgroundSchedulerを生成してstartで常駐を開始するread_csvでUTF-8の売上CSVを読み込むsampleは商品マスタから明細行を生成する --data-dirと--out-dirを全サブコマンド共通のオプションにする 小さく実行確認しながら完成状態まで段階的に組み立てる エンディング。
Python研修はCodeCampでご確認ください。
PythonのAPSchedulerとpandasとは
今回使用する主要なライブラリについて、役割と使い分けを順番に確認します。
定期実行を担うAPSchedulerの役割
APSchedulerは、決まった時刻や間隔でPythonの関数を自動的に呼び出せる定期実行ライブラリです。今回のCLIでは、集計処理そのものを書き換えることなく、毎日07:30のような予定をあとから与える役割を担っています。
裏側でジョブを見張るBackgroundSchedulerと、時刻条件を表すCronTriggerを組み合わせることで、常駐しているあいだだけ予定どおりにジョブが動く仕組みを作れました。停止するときにはshutdownで待機スレッドを畳むため、開始から終了までを1つの関数にまとめています。
この売上レポートCLIでAPSchedulerが受け持つ処理を、登録から停止まで順に挙げます
- BackgroundSchedulerを生成してstartで常駐を開始する
- CronTriggerのhourとminuteで毎日の発火時刻を表す
- add_jobにid=daily_sales_reportを渡してジョブを識別する
- get_jobとnext_run_timeで次回実行時刻を取り出す
- job_func内のtryとexceptで集計失敗時も常駐を継続する
- finallyのshutdownでスケジューラを止める
売上データを集計するpandasの役割
pandasは、表形式のデータをDataFrameとして読み込み、集計や並べ替えをまとめて行えるデータ分析ライブラリです。売上CSVのように行と列で表せるデータであれば、ループを自分で書かなくても列同士の掛け算やグループ単位の合計を短い記述で求められます。今回は数量と単価から金額列を作り、商品コードと商品名でグループ化して数量・金額・構成比を並べました。
表示にはto_stringを使い、レポート本文へそのまま差し込める文字列に変換しています。
レポート作成でpandasに任せている処理を、読み込みから表示まで順に挙げます
- read_csvでUTF-8の売上CSVを読み込む
- to_datetimeでorder_dateを日付型へ変換する
- to_numericで数量と単価を数値化する
- 数量と単価の積からamount列を追加する
- groupbyとaggで商品別の数量と金額を合計する
- sort_valuesで金額の多い順に並べ替える
- to_csvでサンプル明細を書き出す
Python・APSchedulerで開発する場合の環境構築
この記事のセットアップ手順と掲載コードは、Windows 11 Pro、PowerShell 5.1、Python 3.13.3で動作確認しています。仮想環境を有効化せず、その中のPythonを直接指定するため、以下のコマンドはPowerShellとコマンドプロンプト(cmd)の両方で使えます。
python -m venv .venv
.\.venv\Scripts\python.exe -m pip install --upgrade pip
.\.venv\Scripts\python.exe -m pip install "apscheduler<4" pandas
macOS・Linuxでは仮想環境内のPythonパスが異なります。今回の動作確認環境とは異なるため、以下は環境差分を補う参考手順です。
python3 -m venv .venv
./.venv/bin/python -m pip install --upgrade pip
./.venv/bin/python -m pip install "apscheduler<4" pandas
- APSchedulerは4系が開発中のため、本記事のBackgroundScheduler・CronTriggerの書き方が通る3系を指定しています。
- スケジュールの時刻はOSのローカルタイムゾーンで解釈されます。サーバがUTC設定の場合は指定時刻の読み替えが必要です。
- scheduleコマンドは常駐します。動作確認では--run-secondsに数秒を渡すと自動で停止して終了できます。
売上定期レポート作成CLIの要件定義
目的は、サンプル売上CSVを作成し、pandasで売上合計と商品別集計を求め、APSchedulerで毎日決まった時刻にレポートを保存するCLIを最後まで作り上げることです。
対象者として、Pythonの基本文法を学び終え、argparseのサブコマンド設計やpandasの集計、定期実行の組み立て方を実際のツール作りで身につけたい人を想定しています。
完成物は、APSchedulerとpandasで作る売上定期レポート作成CLIで、サンプルCSV作成、即時集計と日付付きレポート保存、日次スケジュール登録、保存済みレポート一覧の4コマンドを確認できるツールです。
実装へ入る前に、機能・品質・受け入れ条件を分けて確認します。
機能要件
- sampleコマンドでサンプル売上CSVを生成する
- 生成した明細の先頭5行と注文日の範囲を表示する
- reportコマンドで売上CSVを1回集計する
- 売上合計と注文件数と明細行数と対象期間をレポートへ書き出す
- 商品別の数量と金額と構成比を金額の多い順に並べる
- レポートを日付付きファイル名で出力先ディレクトリへ保存する
- scheduleコマンドで毎日指定時刻に走るジョブを登録する
- 登録したジョブIDと次回実行時刻を表示する
- listコマンドで保存済みレポートのファイル名とサイズと更新日時を一覧する
非機能要件
- --data-dirと--out-dirを全サブコマンド共通のオプションにする
- 保存先ディレクトリをparents=Trueのmkdirで自動作成
- CSVとレポートの読み書きをUTF-8に統一
- --seed指定によるサンプルデータの再現性
- ジョブ内の例外を捕捉して常駐を継続
- --run-seconds 0の場合はCtrl+Cまで常駐
- finallyでのスケジューラ停止
- 売上CSVが無い場合のFileNotFoundError送出
- 外部サービスへ接続しないローカル完結構成
実装方針
今回はAPSchedulerとpandasの基本動作を追いやすくするため、売上定期レポート作成CLI本体を1つのPythonファイルへまとめます。
入力、判定、結果表示の役割を分け、実行結果を確認しながら機能を積み上げます。
売上定期レポート作成CLIを安全に組み立てるための実装方針は次のとおりです。
- --data-dirと--out-dirを全サブコマンド共通のオプションにする
- 保存先ディレクトリをparents=Trueのmkdirで自動作成
- CSVとレポートの読み書きをUTF-8に統一
- --seed指定によるサンプルデータの再現性
- ジョブ内の例外を捕捉して常駐を継続
- --run-seconds 0の場合はCtrl+Cまで常駐
- finallyでのスケジューラ停止
- 売上CSVが無い場合のFileNotFoundError送出
- 外部サービスへ接続しないローカル完結構成
完成と判断する条件
- sampleコマンドでsales.csvが作成される
- sample実行時に明細行数と注文日の範囲が表示される
- reportコマンドの出力に売上合計と注文件数が並ぶ
- 出力先にsales_report_日付.txtが残る
- scheduleコマンドがジョブIDと次回実行時刻を表示する
- 常駐秒数の経過後にスケジューラ停止のメッセージが出る
- listコマンドが保存済みレポートの件数と表を出力する
- 4つのコマンドがいずれも終了コード0で終わる
売上定期レポート作成CLIを作る際の重要ポイント
この売上定期レポート作成CLIは、argparseのサブコマンドでsample・report・schedule・listという4つの入り口を用意しています。sampleが練習用の売上CSVを作り、reportがその場で集計してレポートを保存し、scheduleは同じ集計を毎日決まった時刻に呼び出す構成です。
集計と保存はrun_onceにまとめてあるため、手動実行と定期実行のどちらから呼んでも同じ形式の結果になります。最後のlistは保存済みファイルを一覧し、実行の履歴を確かめる窓口として働いています。
4つのサブコマンドが担当する役割を、実行する順番に沿って挙げます
- sampleは商品マスタから明細行を生成する
- reportはrun_onceで集計とレポート保存を行う
- scheduleは常駐して毎日の集計を予約する
- listは出力先ディレクトリのファイルを表にする
- --data-dirと--out-dirは4コマンド共通のオプション
- mainは各分岐の結果を印字して0を返す
共通オプションを親パーサーで束ねる工夫
build_parserでは、add_help=Falseで作った親パーサーに--data-dirと--out-dirを定義し、各サブコマンドのparentsへ渡しています。同じ指定を4回書かずに済むうえ、既定値を変えたいときも1か所の修正で反映できました。
サブコマンドはrequired=Trueで必須にしてあるため、コマンド名を付けずに実行すると使い方が表示されます。sampleだけは行数・日数・シードの追加オプションを持ち、scheduleは時刻と常駐秒数を受け取る仕組みです。
パーサーに登録されているオプションを整理します
- 共通の--data-dirは既定でsales_data
- 共通の--out-dirは既定でreports
- sampleの--rowsと--daysと--seed
- scheduleの--atと--run-seconds
run_onceに集計と保存をまとめる意味
run_onceは、売上CSVの存在確認から読み込み、集計、レポート整形、ファイル保存までを一続きで行う関数です。CSVが見つからないときはFileNotFoundErrorを送出し、原因になったパスをメッセージへ含めました。
reportサブコマンドとスケジューラのジョブは、どちらもこのrun_onceを呼び出します。処理の本体が1か所に集まっているため、集計内容を直しても手動実行と定期実行の結果がそろいます。
run_onceが呼び出し元へ返す辞書の中身を挙げます
- 保存したレポートのパス
- レポート本文の文字列
- 集計結果をまとめた辞書
保存済みレポートを一覧する仕組み
list_reportsは、出力先ディレクトリからsales_report_で始まるテキストファイルを集め、ファイル名・サイズ・更新日時の3列を持つDataFrameとして返します。ディレクトリがまだ無い場合は空の表になり、mainの分岐が保存済みレポートの無い状態を案内する流れです。
レポートのファイル名は日付で決まるため、同じ日に何度実行しても一覧は1件にまとまります。サイズと更新日時が見えるので、直近の実行がいつだったかも表から読み取れました。
listコマンドの出力に並ぶ列を挙げます
- fileのファイル名
- size_bytesのバイト数
- updated_atの更新日時
argparseの4サブコマンドで機能を切り分ける
最初に決めたのは、機能をコマンド名で分けることでした。argparseのサブコマンド機能を使うと、sampleやreportのように動詞で呼び分けでき、オプションもコマンドごとに独立させられます。
1つのスクリプトにフラグを詰め込む書き方では、--make-sampleと--scheduleが同時に指定されたときの挙動が曖昧になりがちです。サブコマンド構成なら、そもそもその組み合わせが発生しません。
実行の成否は終了コードで判断できるので、あとからシェルスクリプトやOS標準のタスク管理に組み込むときも扱いやすくなります。
このセクションの用語
- argparse
- コマンドラインの引数やオプションを解析するPython標準ライブラリです。ヘルプ表示も自動で生成してくれます。
- 終了コード
- コマンドが終わったときにOSへ返す数値で、0が正常終了を表す値です。自動実行の成否判定に使われます。
| サブコマンド | 役割 | 主なオプション |
|---|---|---|
| sample | 商品マスタからサンプル売上CSVを生成 | --rows / --days / --seed |
| report | CSVを即時集計し日付付きレポートを保存 | 既定のsales_dataとreportsを使用 |
| schedule | 毎日指定時刻に同じ集計を自動実行 | --at / --run-seconds |
| list | 保存済みレポートのファイル名を一覧表示 | なし |
Pythonで売上定期レポート作成CLIの完成コード
実装は1ファイル完結です。サンプル作成のcreate_sample_csv、読み込みと前処理のload_sales、集計のaggregateという3つの関数を軸に組み立てました。
集計をaggregateに寄せたのは、即時実行のreportと定期実行のscheduleがまったく同じ計算を共有するためです。ロジックが一箇所にあると、出力の食い違いが起きにくくなります。
ここからは、押さえておけば全体像がつかめる7か所を抜き出して見ていきましょう。順番はデータが流れる向きに合わせています。
このセクションの用語
- DataFrame
-
pandasが扱う表形式のデータ構造です。ExcelのシートのようにCSVの行と列をそのまま保持します。 - groupby
- 指定した列の値が同じ行をひとまとめにして、合計や平均を求める操作です。商品別・日別の集計に使います。
- 名前付き集計
-
aggに列名=(元の列, 集計方法)の形で渡す書き方です。出力される列名を自分で決められます。 - シード
- 乱数の出方を決める初期値です。同じ値を指定すると、毎回まったく同じランダム結果が再現されます。
- Path
- 標準ライブラリ
pathlibのクラスで、ファイルやフォルダの場所をOS差を意識せず扱えます。
#!/usr/bin/env python
"""APSchedulerとpandasで作る売上CSVの定期レポート作成CLI。"""
from __future__ import annotations
import argparse
import random
import time
from datetime import date, datetime, timedelta
from pathlib import Path
import pandas as pd
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
# 既定値と商品マスタの定義
DEFAULT_DATA_DIR = "sales_data"
DEFAULT_OUT_DIR = "reports"
SALES_FILE = "sales.csv"
REPORT_PREFIX = "sales_report_"
JOB_ID = "daily_sales_report"
PRODUCTS = [
("A-101", "BlendCoffee", 1200),
("A-102", "DripBag10", 980),
("B-201", "TeaAssort", 1580),
("B-202", "HerbTeaCan", 2200),
("C-301", "MugCup", 1800),
("C-302", "SteelBottle", 3400),
]
# サンプル売上CSVの作成
def create_sample_csv(data_dir: str, rows: int, days: int, seed: int) -> Path:
rng = random.Random(seed)
today = date.today()
records = []
for index in range(rows):
code, name, unit_price = PRODUCTS[rng.randrange(len(PRODUCTS))]
order_date = today - timedelta(days=rng.randrange(max(days, 1)))
records.append(
{
"order_id": f"ORD-{index + 1:04d}",
"order_date": order_date.isoformat(),
"product_code": code,
"product_name": name,
"quantity": rng.randint(1, 5),
"unit_price": unit_price,
}
)
frame = pd.DataFrame(records).sort_values("order_date").reset_index(drop=True)
directory = Path(data_dir)
directory.mkdir(parents=True, exist_ok=True)
csv_path = directory / SALES_FILE
frame.to_csv(csv_path, index=False, encoding="utf-8")
return csv_path
# 売上CSVの読み込みと前処理
def load_sales(csv_path: Path) -> pd.DataFrame:
frame = pd.read_csv(csv_path, encoding="utf-8")
frame["order_date"] = pd.to_datetime(frame["order_date"])
frame["quantity"] = pd.to_numeric(frame["quantity"])
frame["unit_price"] = pd.to_numeric(frame["unit_price"])
frame["amount"] = frame["quantity"] * frame["unit_price"]
return frame
# 売上合計・注文件数・商品別集計の算出
def aggregate(frame: pd.DataFrame) -> dict:
by_product = (
frame.groupby(["product_code", "product_name"], as_index=False)
.agg(quantity=("quantity", "sum"), amount=("amount", "sum"))
.sort_values("amount", ascending=False)
.reset_index(drop=True)
)
total_amount = int(frame["amount"].sum())
if total_amount > 0:
by_product["share_pct"] = (by_product["amount"] / total_amount * 100).round(1)
else:
by_product["share_pct"] = 0.0
return {
"total_amount": total_amount,
"order_count": int(frame["order_id"].nunique()),
"row_count": int(len(frame)),
"period_from": frame["order_date"].min().date().isoformat(),
"period_to": frame["order_date"].max().date().isoformat(),
"by_product": by_product,
}
# レポート本文の組み立て
def format_report(summary: dict, csv_path: Path, generated_at: datetime) -> str:
lines = [
f"# 売上レポート {generated_at.date().isoformat()}",
f"作成日時: {generated_at.strftime('%Y-%m-%d %H:%M:%S')}",
f"対象ファイル: {csv_path}",
f"対象期間: {summary['period_from']} - {summary['period_to']}",
"",
f"売上合計: {summary['total_amount']:,} 円",
f"注文件数: {summary['order_count']} 件",
f"明細行数: {summary['row_count']} 行",
"",
"## 商品別集計",
summary["by_product"].to_string(index=False),
"",
]
return "\n".join(lines)
# レポートの保存(ファイル名に日付を付ける)
def save_report(out_dir: str, text: str, generated_at: datetime) -> Path:
directory = Path(out_dir)
directory.mkdir(parents=True, exist_ok=True)
report_path = directory / f"{REPORT_PREFIX}{generated_at.date().isoformat()}.txt"
report_path.write_text(text, encoding="utf-8")
return report_path
# 売上CSVを1回集計して日付付きレポートを保存する(reportサブコマンドの中核処理)
def run_once(data_dir: str, out_dir: str) -> dict:
csv_path = Path(data_dir) / SALES_FILE
if not csv_path.exists():
raise FileNotFoundError(f"売上CSVが見つかりません: {csv_path}")
frame = load_sales(csv_path)
summary = aggregate(frame)
generated_at = datetime.now()
text = format_report(summary, csv_path, generated_at)
report_path = save_report(out_dir, text, generated_at)
return {"report_path": report_path, "text": text, "summary": summary}
# 実行時刻(HH:MM)の解釈
def parse_hhmm(value: str) -> tuple[int, int]:
hour_text, _, minute_text = value.partition(":")
hour, minute = int(hour_text), int(minute_text)
if not (0 <= hour <= 23 and 0 <= minute <= 59):
raise ValueError(f"時刻はHH:MM形式で指定してください: {value}")
return hour, minute
# 毎日指定時刻にレポート集計を実行する(APSchedulerのCronTriggerでscheduleサブコマンドを常駐させる)
def start_daily_schedule(data_dir: str, out_dir: str, at: str, run_seconds: int) -> dict:
hour, minute = parse_hhmm(at)
trigger = CronTrigger(hour=hour, minute=minute)
def job_func() -> None:
stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
# ここで毎日指定時刻にレポート集計を実行する(run_onceを呼び出す)
result = run_once(data_dir, out_dir)
print(f"[{stamp}] レポートを保存しました: {result['report_path']}")
except Exception as error: # 集計が失敗しても常駐は止めない
print(f"[{stamp}] 集計に失敗しました: {error}")
scheduler = BackgroundScheduler()
scheduler.add_job(job_func, trigger, id=JOB_ID, name="日次売上レポート")
scheduler.start()
job = scheduler.get_job(JOB_ID)
next_run = getattr(job, "next_run_time", None)
if next_run is None:
next_run = trigger.get_next_fire_time(None, datetime.now(trigger.timezone))
print(f"日次スケジュールを登録しました: 毎日 {hour:02d}:{minute:02d} に集計します")
print(f"ジョブID: {JOB_ID} / 次回実行: {next_run.strftime('%Y-%m-%d %H:%M:%S %Z')}")
if run_seconds > 0:
print(f"{run_seconds} 秒だけ常駐して終了します(--run-seconds 0 で Ctrl+C まで常駐)")
# スケジューラの常駐と停止
try:
if run_seconds > 0:
time.sleep(run_seconds)
else:
while True:
time.sleep(1)
except (KeyboardInterrupt, SystemExit):
print("停止要求を受け取りました")
finally:
scheduler.shutdown()
return {"job_id": JOB_ID, "at": f"{hour:02d}:{minute:02d}", "next_run_time": next_run}
# 保存済みレポートの一覧作成
def list_reports(out_dir: str) -> pd.DataFrame:
directory = Path(out_dir)
paths = sorted(directory.glob(f"{REPORT_PREFIX}*.txt")) if directory.exists() else []
records = [
{
"file": path.name,
"size_bytes": path.stat().st_size,
"updated_at": datetime.fromtimestamp(path.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S"),
}
for path in paths
]
return pd.DataFrame(records, columns=["file", "size_bytes", "updated_at"])
# コマンドライン定義
def build_parser() -> argparse.ArgumentParser:
common = argparse.ArgumentParser(add_help=False)
common.add_argument("--data-dir", default=DEFAULT_DATA_DIR, help="売上CSVを置くディレクトリ")
common.add_argument("--out-dir", default=DEFAULT_OUT_DIR, help="レポートの保存先ディレクトリ")
parser = argparse.ArgumentParser(
prog="sales_report_scheduler",
description="売上CSVを集計して日付付きレポートを保存する定期レポート作成CLI",
)
sub = parser.add_subparsers(dest="command", required=True, metavar="command")
sample = sub.add_parser("sample", parents=[common], help="サンプル売上CSVを作成する")
sample.add_argument("--rows", type=int, default=60, help="生成する明細行数")
sample.add_argument("--days", type=int, default=7, help="注文日を散らす日数")
sample.add_argument("--seed", type=int, default=42, help="乱数シード")
sub.add_parser("report", parents=[common], help="売上CSVを1回集計して日付付きレポートを保存する")
schedule = sub.add_parser("schedule", parents=[common], help="毎日指定時刻にレポート集計を実行する")
schedule.add_argument("--at", default="07:30", help="実行時刻(HH:MM)")
schedule.add_argument("--run-seconds", type=int, default=0, help="常駐秒数(0はCtrl+Cまで常駐)")
sub.add_parser("list", parents=[common], help="保存済みレポートを一覧表示する")
return parser
# コマンドの実行
def main() -> int:
args = build_parser().parse_args()
if args.command == "sample":
csv_path = create_sample_csv(args.data_dir, args.rows, args.days, args.seed)
frame = pd.read_csv(csv_path)
print(f"サンプル売上CSVを作成しました: {csv_path}")
print(f"明細行数: {len(frame)} 行 / 注文日: {frame['order_date'].min()} - {frame['order_date'].max()}")
print(frame.head(5).to_string(index=False))
elif args.command == "report":
# 売上CSVを1回集計して日付付きレポートを保存する
result = run_once(args.data_dir, args.out_dir)
print(result["text"])
print(f"保存先: {result['report_path']}")
elif args.command == "schedule":
# 毎日指定時刻にレポート集計を実行する(APSchedulerで常駐させる)
info = start_daily_schedule(args.data_dir, args.out_dir, args.at, args.run_seconds)
print(f"スケジューラを停止しました(ジョブ: {info['job_id']} / 予定時刻: 毎日 {info['at']})")
elif args.command == "list":
table = list_reports(args.out_dir)
if table.empty:
print(f"保存済みレポートはありません: {Path(args.out_dir)}")
else:
print(f"保存済みレポート: {len(table)} 件({Path(args.out_dir)})")
print(table.to_string(index=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())
コード全文は上の折り畳みに入れてあるので、全部を上から読む必要はありません。ここでは特に重要な部分だけを抜き出して、何をしているのか順番に見ていきます。
BackgroundSchedulerとCronTriggerの読み込み
import pandas as pd
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger集計用にpandasを、定期実行用にAPSchedulerの2つの部品を読み込んでいます。BackgroundSchedulerはプログラムの裏側でジョブを動かす担当、CronTriggerは「毎日この時刻」という条件を表す担当です。役割が分かれているので、片方だけ差し替える改造もしやすくなります。
PRODUCTSで商品マスタを定義する
PRODUCTS = [
("A-101", "BlendCoffee", 1200),
("A-102", "DripBag10", 980),
("B-201", "TeaAssort", 1580),
("B-202", "HerbTeaCan", 2200),
("C-301", "MugCup", 1800),
("C-302", "SteelBottle", 3400),
]商品コード・商品名・単価の3点セットを6件だけ定義した簡易マスタです。サンプルCSVはここからランダムに選んで作るので、商品別集計が必ず複数行になります。実データに差し替えるときは、この定数を消してCSV側の列だけ合わせれば動きます。
create_sample_csvで再現性のある明細を作る
def create_sample_csv(data_dir: str, rows: int, days: int, seed: int) -> Path:
rng = random.Random(seed)
today = date.today()
records = []
for index in range(rows):
code, name, unit_price = PRODUCTS[rng.randrange(len(PRODUCTS))]
order_date = today - timedelta(days=rng.randrange(max(days, 1)))random.Random(seed)で乱数の種を固定しているため、同じ--seedを渡せば毎回同じ明細が作られます。注文日は今日からdays日前までの範囲でずらし、max(days, 1)で0が渡されても落ちないようにしました。動作確認のたびに数字が変わらないのは、初心者がデバッグするうえでかなり楽です。
to_csvでsales.csvへ書き出す
frame = pd.DataFrame(records).sort_values("order_date").reset_index(drop=True)
directory = Path(data_dir)
directory.mkdir(parents=True, exist_ok=True)
csv_path = directory / SALES_FILE
frame.to_csv(csv_path, index=False, encoding="utf-8")
return csv_path辞書のリストをDataFrameにして注文日で並べ替え、通し番号を振り直してから保存しています。mkdir(parents=True, exist_ok=True)があるので、保存先フォルダが無くても自動で作られ、既にあってもエラーになりません。index=Falseを付けないと余計な連番列がCSVに混ざります。
load_salesでamount列を作る
def load_sales(csv_path: Path) -> pd.DataFrame:
frame = pd.read_csv(csv_path, encoding="utf-8")
frame["order_date"] = pd.to_datetime(frame["order_date"])
frame["quantity"] = pd.to_numeric(frame["quantity"])
frame["unit_price"] = pd.to_numeric(frame["unit_price"])
frame["amount"] = frame["quantity"] * frame["unit_price"]
return frameCSVは中身がすべて文字列になりうるので、日付はto_datetime、数量と単価はto_numericで型をそろえています。そのうえで数量と単価を掛けたamount列を追加し、以降の集計はこの1列を足すだけで済む形にしました。前処理をここに集約しておくと、集計側の関数がとても短くなります。
groupbyの名前付き集計で商品別に積み上げる
by_product = (
frame.groupby(["product_code", "product_name"], as_index=False)
.agg(quantity=("quantity", "sum"), amount=("amount", "sum"))
.sort_values("amount", ascending=False)
.reset_index(drop=True)
)商品コードと商品名の組み合わせでまとめ、数量と金額をそれぞれ合計しています。aggにquantity=("quantity", "sum")と書く名前付き集計を使うと、出来上がる列名が最初から決まるので後の整形が読みやすくなります。最後に金額の降順で並べ替え、売れ筋が上に来るようにしました。
share_pctで売上構成比を計算する
total_amount = int(frame["amount"].sum())
if total_amount > 0:
by_product["share_pct"] = (by_product["amount"] / total_amount * 100).round(1)
else:
by_product["share_pct"] = 0.0全体の売上合計を出し、商品ごとの金額を割って構成比を小数第1位まで求めています。合計が0のときに割り算するとゼロ除算で値が壊れるため、ifで分岐して0.0を入れる作りにしました。空データや金額0のCSVを渡されても落ちない、地味ですが効く保険です。
Pythonで売上定期レポート作成CLIのエラー対処
ここからは、この種のCLIを写経したときに一般に遭遇しやすいエラーを挙げます。実行中に赤い文字が出ても、最終行のエラー名とファイル名を読むだけで原因の見当はつくことが多いです。
特にファイル関連は、コードではなく実行した場所が原因のケースが目立ちます。慌ててコードを直す前に、今どのフォルダでコマンドを打ったのかを確認してみてください。
このセクションの用語
- トレースバック
- エラー発生時に表示される、呼び出し経路の一覧です。いちばん下の行にエラー名と原因が書かれています。
- 仮想環境
- プロジェクトごとにライブラリを分けて入れる仕組みです。別プロジェクトとのバージョン衝突を防げます。
| エラー例 | 原因 | 対処 |
|---|---|---|
| ModuleNotFoundError: No module named 'apscheduler' | ライブラリが未インストール、または別の環境に入っている | 有効化した仮想環境でapschedulerとpandasを入れ直す |
| FileNotFoundError: sales_data/sales.csv |
sampleを実行する前にreportを叩いた、または別フォルダで実行した |
先にsampleを実行し、同じ作業ディレクトリでコマンドを打つ |
| UnicodeDecodeError | CSVがShift_JISなど別の文字コードで保存されている |
read_csvのencodingを実ファイルに合わせるか、UTF-8で保存し直す |
| KeyError: 'quantity' | CSVの列名が想定と違う、またはヘッダー行が無い |
quantityやunit_priceなど必要な列名をCSV側でそろえる |
| ValueError(時刻指定の解析エラー) |
--atに7:30以外の形式や範囲外の値を渡した |
07:30のようにHH:MM形式で、時0〜23・分0〜59の範囲で指定する |
売上定期レポート作成CLIで注意したい点
いちばん多いのは、パスの感覚の食い違いです。sales_dataやreportsは相対パスなので、コマンドを打った場所が変わると別のフォルダを見に行きます。
文字コードと型も落とし穴になりがちです。Excelから出したCSVは数値が文字列として読まれることがあり、to_numericを通さないと掛け算が意図しない結果になります。
スケジュール実行では、プロセスが生きている間しかジョブが動かない点を忘れないでください。--run-secondsで終了させた後は、当然ながら次の実行も行われません。
ポイントとしては、ファイルの扱いとスケジュール常駐まわりで特に注意したい点を短くまとめます。
手を動かす前に確認しておくと安心な項目を挙げます。
-
python subject.py sampleを実行したフォルダにsales_dataができているか - 同じ日に
reportを2回実行したとき、レポートがどう扱われるかを自分の目で確認する - 実データを使う場合は、列名を
order_id・quantity・unit_priceにそろえる
実行場所:相対パスは起動位置基準
文字コード:CSVはutf-8で統一
型変換:to_numericで数値列を確定
日付付き保存:同日実行は上書き注意
Pythonで日次スケジュールを設定する
定期実行はOSのタスク管理に任せる手もありますが、今回はPythonのプロセス内で完結させたかったのでAPSchedulerを選びました。--at 07:30のように時刻を受け取り、CronTriggerで「毎日その時刻」という条件に変換します。
使ったのはBackgroundSchedulerです。裏側のスレッドでジョブを動かすタイプなので、メインの処理では次回実行時刻を表示したり、待機したりといった制御が書けます。
常駐しっぱなしだと動作確認がしづらいため、--run-secondsで指定した秒数だけ待って正常終了する作りにしました。ジョブIDはJOB_ID定数に固定し、同じジョブが二重登録されない形にしています。
スケジュール実行でつまずきやすい、Cron指定・常駐の仕組み・多重登録防止という3つのポイントを、以下に短くまとめます。
このセクションの用語
- BackgroundScheduler
- アプリの裏側のスレッドでジョブを実行する
APSchedulerのスケジューラです。呼び出し側の処理を止めません。 - CronTrigger
- 「毎日7時30分」のように、時刻や曜日で実行タイミングを指定するトリガーです。cronの書式に近い指定ができます。
- ジョブ
- スケジューラに登録する実行単位のことです。今回は集計とレポート保存を行う関数を1つ登録しています。
参考:
©APScheduler公式ドキュメント(User guide)BackgroundScheduler: use when you're not using any of the frameworks below, and want the scheduler to run in the background inside your application
Cron書式:--atは時と分に分割
常駐前提:プロセス終了でジョブ停止
多重登録:job_idで既存ジョブ確認
売上定期レポート作成CLIの動作確認
実際に手元で4回のコマンドを実行し、すべて終了コード0で正常に動作しました。順番はsampleでデータを作り、reportで集計し、listで保存結果を確認し、最後にscheduleという流れです。
scheduleは--run-seconds 6を付けたので、6秒待ったところで自分から終了します。待ち続けてCtrl+Cを押す必要がないため、動作確認のスクリーンショットも取りやすくなりました。
各コマンドの実行画面はキャプチャとして残しています。出力の形と終了コードを見比べると、どのコマンドがどこまで処理したのかがつかめるはずです。
このセクションの用語
- 標準出力
- プログラムがターミナルへ文字を書き出す先のことです。
printで表示した内容はここに流れます。
キャプチャとして記録した4回の実行コマンドは次のとおりです。
-
python subject.py sample --rows 60 --days 7 --seed 42でサンプル売上CSVを作成 -
python subject.py reportで即時集計と日付付きレポートの保存を実行 -
python subject.py listで保存済みレポートの一覧を表示 -
python subject.py schedule --at 07:30 --run-seconds 6で6秒だけ常駐して正常終了




売上レポート自動化が役立つ場面と応用
この構成は売上に限らず、「定期的にCSVを読んで、決まった形にまとめて保存する」仕事全般に応用できます。集計部分を差し替えるだけで別の指標にも転用しやすいのが利点です。
現場で使うなら、レポートの保存先を共有フォルダにする、通知を足すといった拡張が現実的でしょう。まずは手元で動かし、出力の形を固めてから広げるのが安全な進め方です。
このセクションの用語
- バッチ処理
- 決まったタイミングでまとめてデータを処理する方式です。夜間や早朝に自動実行させる使い方が一般的です。
| 使える場面 | 具体的な使い方 |
|---|---|
| 店舗の日次売上チェック | POSから出したCSVをsales_dataに置き、毎朝の時刻を--atに指定して商品別売上と構成比のレポートを自動生成する |
| ECサイトの商品別分析 | 注文明細を月ごとに読み込み、groupbyのキーをカテゴリ列に変えて、売れ筋と死に筋を一覧化する |
| 社内の定例報告資料づくり | 日付付きで貯まったレポートをlistで確認し、必要な日付のテキストをそのまま報告書へ貼り付ける |
| データ加工バッチの雛形 |
aggregateの中身を在庫数や問い合わせ件数の集計に差し替え、同じスケジュール処理を再利用する |
| 集計ロジックの検証環境 |
--seedを固定したサンプルCSVで、集計結果が変わらないことを確かめながら計算式を改修する |
売上定期レポート作成CLI開発のまとめ
pandasで集計し、APSchedulerで毎日の実行を任せるだけで、手作業だった売上集計はコマンド1つに置き換わりました。実装した機能を4つに絞ったおかげで、コードは1ファイルに収まっています。
実際の確認では4回のコマンドがすべて正常終了し、サンプル作成から日付付きレポートの保存、一覧表示、スケジュール起動までを一通り通せました。
次に手を入れるなら、集計軸を日別に変える、出力をCSVやHTMLにも対応させるあたりが取り組みやすいはずです。まずは自分の手元のCSVを読み込ませて、列名を合わせるところから試してみてください。
さらに詳しく学べるPython研修の詳細はこちら参考にした一次情報
- ^ APScheduler 3.x User guide. https://apscheduler.readthedocs.io/en/3.x/userguide.html, (参照26-08-18).
- ^ APScheduler 3.x apscheduler.triggers.cron. https://apscheduler.readthedocs.io/en/3.x/modules/triggers/cron.html, (参照26-08-18).
- ^ pandas: Group by: split-apply-combine. https://pandas.pydata.org/docs/user_guide/groupby.html, (参照26-08-18).
- ^ argparse — Parser for command-line options, arguments and subcommands. https://docs.python.org/3/library/argparse.html, (参照26-08-18).
※内容は執筆時点のものです。ライブラリやサイトの仕様は変わる可能性があるため、公式ドキュメントもあわせてご確認ください。













