【Python】Richとre、collectionsでログ解析ツールを作ってみた

サーバーのアクセスログは、そのまま眺めても傾向がつかみにくいものです。そこで、ログを正規表現で1行ずつ分解し、レベルやステータス、経路ごとの件数を集計して、Richで色分けした表とパネルに表示するCLIツールを作ってみました。

手元にログが無くても試せるようにサンプル生成の機能も用意し、genとreportという2つのサブコマンドだけで完結する構成です。正規表現の名前付きグループ、Counterやdefaultdictによる集計、Richでの見やすい出力という、初心者にも役立つ道具立てを1つの題材でまとめて追体験できます。

Richの基本概念、要件定義、実装、動作確認までを順番に学べる構成です。動画は目次から確認したい場面へ移動でき、本文だけでも手順と考え方が完結します。

今回使用する主要なライブラリについて、役割と使い分けを順番に確認します。

ターミナル出力を彩るRich

Richは、ターミナル出力に色や罫線、レイアウトを加えられるPythonのライブラリです。標準のprintが文字を並べるだけなのに対し、Richなら表やパネルを組み立てて情報を構造的に見せられます。このツールでは、Consoleが画面への書き出しを担い、Panelがサマリを枠で囲み、Tableが集計結果を段組みで整えます。

件数や割合が並ぶログデータでも、色分けと右揃えによって傾向を一目でつかめるでしょう。

Richの主要な部品と、このツールでの使いどころは次のとおりです

• Consoleのprintで色付きテキストを画面へ出力する
• Panelで読み込み件数などのサマリを枠で囲む
• Tableで列見出しと右揃えを指定して集計を並べる
• 角かっこのマークアップで文字色や太字を切り替える
• LEVEL_STYLESでレベルごとの配色をまとめて管理する

ログ解析を支えるreとcollections

reは正規表現を扱う標準ライブラリで、決まった書式の文字列から必要な部分を取り出すのに向いています。ログの1行は日付・レベル・IP・パスなどが並んだ定型文なので、名前付きグループを使うと各項目に名前を付けて辞書として取り出せます。collectionsは集計に便利なデータ構造を提供する標準ライブラリです。

Counterは要素の出現回数を自動で数え、defaultdictは存在しないキーへ初期値を用意してくれるため、件数集計や経路ごとのグループ分けが短く書けます。どちらも追加インストールなしで使えるのが利点でしょう。

解析と集計に使う道具の役割は次のとおりです

• re.compileで正規表現をあらかじめ用意する
• 名前付きグループで抽出項目に名前を付ける
• groupdictでマッチ結果を辞書として取り出す
• Counterでレベルやステータスの件数を数える
• defaultdictで経路ごとに応答時間を集める
• most_commonで多い順に並べ替える

この記事のセットアップ手順と掲載コードは、Windows 11 Pro、PowerShell 5.1、Python 3.13.3、Rich 15.0.0で動作確認しています。仮想環境を有効化せず、その中のPythonを直接指定するため、以下のコマンドはPowerShellとコマンドプロンプト（cmd）の両方で使えます。

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install rich

macOS・Linuxでは仮想環境内のPythonパスが異なります。今回の動作確認環境とは異なるため、以下は環境差分を補う参考手順です。

python3 -m venv .venv
./.venv/bin/python -m pip install rich

• genサブコマンドでサンプルログを作ってから、同じフォルダでreportサブコマンドを実行する流れです。
• Richの色付き表示はWindows TerminalやPowerShellのコンソールでそのまま確認できます。

目的は、アクセスログを正規表現で1行ずつ解析し、Counterやdefaultdictで集計した結果をRichの表とパネルで見やすく表示するCLIツールを作れるようになることです。

対象者として、Pythonの標準ライブラリとRichを組み合わせて、実用的なコマンドラインツールの作り方を学びたい人を想定しています。

完成物は、genとreportの2つのサブコマンドで、サンプル生成からログ解析と集計表示までを行えるコマンドラインツールです。

実装へ入る前に、機能・品質・受け入れ条件を分けて確認します。

機能要件

• genサブコマンドで指定した行数のサンプルログを生成する
• --linesで生成するログの行数を指定する
• reportサブコマンドでログを解析して集計結果を表示する
• 名前付きグループの正規表現でログ1行を各項目へ分解する
• 正規表現にマッチしない行を未対応として数える
• Counterでレベル別とステータス別の件数を集計する
• defaultdictで経路ごとの応答時間をまとめる
• most_commonでアクセスの多い経路を上位から並べる
• --levelで指定したレベルだけを集計対象に絞る
• --topで経路ランキングの表示件数を変える
• Richのパネルとテーブルで集計を色分け表示する

非機能要件

• サブコマンドの構成に標準ライブラリのargparseを使う
• 外部依存をRichだけにし解析と集計は標準ライブラリで行う
• generate_sample_logはseed=42で乱数を固定する
• ログファイルの読み書きをUTF-8に統一する
• 空行は集計対象から除外する
• サブコマンド未指定をrequired=Trueでエラーにする
• --linesと --topをtype=intで整数として受け取る

実装方針

今回はRichとre、collectionsの基本動作を追いやすくするため、アクセスログ解析CLI本体を1つのPythonファイルへまとめます。

入力、判定、結果表示の役割を分け、実行結果を確認しながら機能を積み上げます。

アクセスログ解析CLIを安全に組み立てるための実装方針は次のとおりです。

• サブコマンドの構成に標準ライブラリのargparseを使う
• 外部依存をRichだけにし解析と集計は標準ライブラリで行う
• generate_sample_logはseed=42で乱数を固定する
• ログファイルの読み書きをUTF-8に統一する
• 空行は集計対象から除外する
• サブコマンド未指定をrequired=Trueでエラーにする
• --linesと --topをtype=intで整数として受け取る

完成と判断する条件

1. gen実行で指定したパスにログファイルが作成される
2. 生成された各行がLOG_PATTERNの書式に一致する
3. report実行でサマリのパネルが表示される
4. レベル別・経路別・ステータス別の3つの表が並ぶ
5. --level ERRORでERROR行だけが集計される
6. --top 3で経路ランキングが3件に絞られる
7. 未対応行の件数がサマリに表示される

このツールは、argparseによってgenとreportの2つのサブコマンドに処理を分けています。genはサンプルログを生成するコマンドで、本物のログが手元に無くても動作を試せるようにする役割です。reportが解析の本体であり、ファイルの読み込み、1行ずつの正規表現マッチ、集計、Richでの表示までを順にたどります。

--levelを指定した場合は解析後のレコードを1つのレベルへ絞り込み、--topで経路ランキングの件数も変えられます。処理が小さな関数に分かれているぶん、生成・解析・集計・表示のどこで何をしているかを追いやすい構成でしょう。

genとreportを実行したときの処理の流れは次のとおりです

• genはgenerate_sample_logで乱数ログを書き出す
• reportはparse_logでファイルを解析する
• 読み込み全体の行数をtotal_readとして保持する
• --level指定時は該当レベルへレコードを絞る
• aggregateで集計結果の辞書を組み立てる
• render_reportがパネルと3つの表を描画する

genでサンプルログを生成する

generate_sample_logは、レベルやパス、ステータスの候補から1行分の値を選び、指定した行数だけログを書き出します。INFOを多め、ERRORを少なめにするなど、出現の偏りは候補リストの重複数で調整する仕組みです。

seedを固定しているため、同じ行数を指定すれば毎回同じログが得られます。生成後は書き出した行数を返し、コマンド側で生成完了のメッセージとともに件数を表示する流れです。

genが生成するログの特徴は次のとおりです

• 1行がLOG_PATTERNの書式に一致する
• レベルやステータスに現実的な偏りを持たせる
• --linesで生成する行数を指定できる
• seed固定で毎回同じ内容になる

reportで解析から集計までを行う

parse_logは各行をLOG_PATTERNで照合し、マッチした行だけを辞書化してリストにためる関数です。マッチしなかった行は未対応として数え、書式の乱れも把握できるようにしています。

解析後、--levelが指定されていればレコードをそのレベルだけへ絞り込みます。続いてaggregateがレベル別・ステータス別・経路別の集計をまとめた辞書を返す仕組みです。

reportが解析時に行う処理は次のとおりです

• 空行を飛ばして各行を正規表現に掛ける
• マッチ行を辞書化してレコードに加える
• マッチしない行を未対応として数える
• --level指定時にレベルで絞り込む

集計結果をレポートとして表示する

render_reportは、まず対象ファイル名と行数の内訳をPanelで示し、その後レベル・経路・ステータスの3つの表を順に出力します。読み込み全体・集計対象・未対応の行数が並ぶため、どれだけのログを扱えたのかが分かる構成です。

経路の表には、リクエスト数と平均応答時間を同時に載せています。数値の列は右揃えにし、平均応答は小数第1位までそろえて、多い経路や遅い経路を見比べやすくしました。

レポートに表示される要素は次のとおりです

• 対象ファイルと行数内訳のサマリパネル
• ログレベル別の件数テーブル
• アクセスの多い経路と平均応答のテーブル
• ステータスコード別の件数テーブル

このツールは1つのファイルにまとめた小さなCLIです。大きく分けて、サンプルログの生成、正規表現による解析、集計、Richでの表示という4つの役割で構成しました。

サブコマンドの受け付けには標準ライブラリのargparseを使い、genとreportを1つのコマンドにまとめています。

処理の中心は1本の正規表現LOG_PATTERNです。ログ1行をこのパターンに掛けると、日時やレベル、IP、パス、ステータス、応答時間までを名前付きで一度に取り出せます。

集計はcollectionsのCounterとdefaultdictに任せ、most_commonでランキングに整えました。

最後にrichのTableとPanelへ結果を流し込み、レベルやステータスを色分けして表示します。

"""Webアクセスログを正規表現で解析し、Richで集計レポートを表示するCLIツール。"""

import argparse
import random
import re
from collections import Counter, defaultdict

from rich.console import Console
from rich.panel import Panel
from rich.table import Table


# ログ1行の書式を表す正規表現(名前付きグループで各フィールドを抽出)
LOG_PATTERN = re.compile(
    r"(?P<date>\d{4}-\d{2}-\d{2})\s+"
    r"(?P<time>\d{2}:\d{2}:\d{2})\s+"
    r"\[(?P<level>[A-Z]+)\]\s+"
    r"(?P<ip>\d{1,3}(?:\.\d{1,3}){3})\s+"
    r'"(?P<method>[A-Z]+)\s+(?P<path>\S+)"\s+'
    r"(?P<status>\d{3})\s+"
    r"(?P<latency>\d+)ms"
)

LEVEL_STYLES = {"ERROR": "bold red", "WARN": "yellow", "INFO": "green", "DEBUG": "dim cyan"}


# サンプルログの生成(手元にログが無くても試せるようにする)
def generate_sample_log(path, lines=400, seed=42):
    random.seed(seed)
    levels = ["INFO"] * 70 + ["WARN"] * 15 + ["ERROR"] * 10 + ["DEBUG"] * 5
    paths = ["/", "/login", "/products", "/products?page=2",
             "/product/12", "/cart", "/checkout", "/api/search"]
    methods = ["GET"] * 8 + ["POST"] * 2
    statuses = [200] * 75 + [301] * 5 + [404] * 12 + [500] * 8
    with open(path, "w", encoding="utf-8") as f:
        for _ in range(lines):
            level = random.choice(levels)
            ip = f"192.168.{random.randint(0, 5)}.{random.randint(1, 254)}"
            method = random.choice(methods)
            route = random.choice(paths)
            status = random.choice(statuses)
            latency = random.randint(8, 850)
            hh, mm, ss = random.randint(0, 23), random.randint(0, 59), random.randint(0, 59)
            f.write(f'2026-07-23 {hh:02d}:{mm:02d}:{ss:02d} [{level}] {ip} '
                    f'"{method} {route}" {status} {latency}ms\n')
    return lines


# ログの解析(1行ずつ正規表現に掛け、マッチした行だけ辞書化する)
def parse_log(path):
    records = []
    unmatched = 0
    with open(path, encoding="utf-8") as f:
        for line in f:
            if not line.strip():
                continue
            match = LOG_PATTERN.search(line)
            if match:
                records.append(match.groupdict())
            else:
                unmatched += 1
    return records, unmatched


# 集計(Counterでレベル・ステータスの件数、defaultdictで経路別の応答時間をまとめる)
def aggregate(records, top=10):
    level_counts = Counter(r["level"] for r in records)
    status_counts = Counter(r["status"] for r in records)
    path_counts = Counter(r["path"] for r in records)

    latency_by_path = defaultdict(list)
    for r in records:
        latency_by_path[r["path"]].append(int(r["latency"]))

    top_paths = []
    for path, count in path_counts.most_common(top):
        latencies = latency_by_path[path]
        top_paths.append((path, count, sum(latencies) / len(latencies)))

    return {"level_counts": level_counts, "status_counts": status_counts, "top_paths": top_paths}


# Richでの整形表示(サマリのパネルと3種類の集計テーブル)
def render_report(console, path, records, unmatched, total_read, summary):
    header = (f"[bold]{path}[/bold]\n"
              f"読み込み {total_read} 行 / 集計対象 {len(records)} 行 / 未対応 {unmatched} 行")
    console.print(Panel(header, title="ログ解析サマリ", border_style="cyan"))

    level_table = Table(title="ログレベル別の件数")
    level_table.add_column("レベル")
    level_table.add_column("件数", justify="right")
    for level, count in summary["level_counts"].most_common():
        style = LEVEL_STYLES.get(level, "white")
        level_table.add_row(f"[{style}]{level}[/{style}]", str(count))
    console.print(level_table)

    path_table = Table(title="アクセスの多い経路")
    path_table.add_column("パス")
    path_table.add_column("リクエスト数", justify="right")
    path_table.add_column("平均応答(ms)", justify="right")
    for p, count, avg in summary["top_paths"]:
        path_table.add_row(p, str(count), f"{avg:.1f}")
    console.print(path_table)

    status_table = Table(title="ステータスコード別の件数")
    status_table.add_column("ステータス")
    status_table.add_column("件数", justify="right")
    for status, count in summary["status_counts"].most_common():
        style = "green" if status.startswith("2") else "red" if status.startswith("5") else "yellow"
        status_table.add_row(f"[{style}]{status}[/{style}]", str(count))
    console.print(status_table)


# コマンドライン引数の解釈と実行
def main():
    parser = argparse.ArgumentParser(description="正規表現とRichで作るログ解析ツール")
    sub = parser.add_subparsers(dest="command", required=True)

    gen_parser = sub.add_parser("gen", help="サンプルログを生成する")
    gen_parser.add_argument("path", help="出力するログファイルのパス")
    gen_parser.add_argument("--lines", type=int, default=400, help="生成する行数")

    report_parser = sub.add_parser("report", help="ログを解析して集計結果を表示する")
    report_parser.add_argument("path", help="解析するログファイルのパス")
    report_parser.add_argument("--level", help="指定したレベルだけを集計する")
    report_parser.add_argument("--top", type=int, default=10, help="経路ランキングの表示件数")

    args = parser.parse_args()
    console = Console()

    if args.command == "gen":
        count = generate_sample_log(args.path, lines=args.lines)
        console.print(f"[green]サンプルログを生成しました[/green]: {args.path}({count} 行)")
        return

    records, unmatched = parse_log(args.path)
    total_read = len(records) + unmatched
    if args.level:
        records = [r for r in records if r["level"] == args.level.upper()]
    summary = aggregate(records, top=args.top)
    render_report(console, args.path, records, unmatched, total_read, summary)


if __name__ == "__main__":
    main()

コード全文は上の折り畳みに入れてあるので、全部を上から読む必要はありません。ここでは特に重要な部分だけを抜き出して、何をしているのか順番に見ていきます。

LOG_PATTERNの名前付きグループ

LOG_PATTERN = re.compile(
    r"(?P<date>\d{4}-\d{2}-\d{2})\s+"
    r"(?P<time>\d{2}:\d{2}:\d{2})\s+"
    r"\[(?P<level>[A-Z]+)\]\s+"
    r"(?P<ip>\d{1,3}(?:\.\d{1,3}){3})\s+"
    r'"(?P<method>[A-Z]+)\s+(?P<path>\S+)"\s+'
    r"(?P<status>\d{3})\s+"
    r"(?P<latency>\d+)ms"
)

re.compileで作る1本の正規表現に、(?P<date>...)のように名前を付けたグループを並べています。こうすると、ログ1行から日時やレベル、IPやパスを名前付きでまとめて取り出せます。\d{4}-\d{2}-\d{2}は数字4桁-2桁-2桁の日付を表す部分です。

generate_sample_logでサンプルを作る

level = random.choice(levels)
ip = f"192.168.{random.randint(0, 5)}.{random.randint(1, 254)}"
method = random.choice(methods)
route = random.choice(paths)
status = random.choice(statuses)
latency = random.randint(8, 850)

random.choiceでレベルやパス、ステータスをランダムに選び、1行分のログの材料を組み立てています。手元に本物のログが無くても、この関数でサンプルを作れば動作確認をすぐ始められます。件数や乱数の種は引数で変えられる仕組みです。

parse_logで1行ずつ照合する

match = LOG_PATTERN.search(line)
if match:
    records.append(match.groupdict())
else:
    unmatched += 1

LOG_PATTERN.searchで1行ずつ照合し、マッチした行だけをgroupdict()で辞書に変換しています。マッチしない行はunmatchedとして数え、あとで取りこぼしに気づけるようにしました。1行1レコードの素直な作りです。

Counterで件数を数える

level_counts = Counter(r["level"] for r in records)
status_counts = Counter(r["status"] for r in records)
path_counts = Counter(r["path"] for r in records)

Counterにレベルやステータスの値を次々に渡すだけで、種類ごとの件数が自動で集計されます。自分で数え上げるループを書かずに済むのが利点です。同じ書き方で経路別の件数もまとめています。

defaultdictで経路別に応答時間をためる

latency_by_path = defaultdict(list)
for r in records:
    latency_by_path[r["path"]].append(int(r["latency"]))

defaultdict(list)は、初めて登場したキーにも空のリストを自動で用意してくれます。おかげで存在確認をせずに、経路ごとの応答時間をそのまま追加できるのが便利です。集計対象をキー別にためたい場面で重宝します。

most_commonでランキングを取り出す

for path, count in path_counts.most_common(top):
    latencies = latency_by_path[path]

most_common(top)は出現回数の多い順に経路を並べ、上位N件だけを取り出します。ランキング表示のための並べ替えを、自分で書かずに任せられるのが強みです。取り出した経路ごとに応答時間を集計していきます。

参考：

©Python公式ドキュメントcollections

Return a list of the n most common elements and their counts from the most common to the least.

実際にこのCLIツールを4回のコマンドで実行し、すべて正常終了(終了コード0)で動作しました。まずサンプルログを作り、次に集計レポートを表示するという順で確認しています。

最初にpython subject.py gen sample.logでサンプルログを生成しました。続けてpython subject.py report sample.logを実行すると、レベルやステータスごとの集計がRichの表と枠で表示されます。

絞り込みも試しました。python subject.py report sample.log --level ERRORではERROR行だけに、python subject.py report sample.log --top 5では上位5件に表示を絞れます。

いずれのコマンドも終了コード0で、エラーなく最後まで動きました。下のキャプチャは、その4回の実行結果です。

ここからは、この種のCLIツールで一般に遭遇しやすい実行エラーを整理します。実行環境やログの書式が想定と違うと、前処理の段階でつまずきがちです。

とはいえ、多くは原因がはっきりしていて、対処もほぼ決まっています。下の表に代表的な例をまとめました。

アクセスログ解析CLIで注意したい点

正規表現とRich表示は便利ですが、初心者がつまずきやすい箇所も決まっています。特に多いのは、正規表現の書式ずれとRichのスタイル指定まわりです。

正規表現は1文字違うだけでマッチしなくなり、解析結果が0件になることがあります。ログの書式を1行コピーして、パターンと1項目ずつ照らし合わせると原因を見つけやすいです。

Richの色分けは、bold redのようなスタイル文字列のつづりが正しいことが前提になります。定義したLEVEL_STYLESのキーと、実際のレベル名が一致しているかも確認しておきたい点です。

ポイントとしては、正規表現の書式合わせとRichのスタイル指定という2点に絞って確認しておくと安心です。

このツールは題材こそ小さいですが、ログを集計する骨組みはそのまま実務にも応用できます。正規表現と集計、表示を分けて作ってあるので、対象を差し替えるだけで使い回しやすい構成です。

特に、定期的に同じ形式のログを見たい場面と相性がよいです。

Webアクセスログを正規表現で解析し、Richで色分けレポートを出すCLIツールを作りました。サンプル生成から解析、集計、表示までを1ファイルにまとめ、4回のコマンドすべてが正常終了で動作しています。

肝は、名前付きグループの正規表現1本で各項目を取り出し、Counterとdefaultdictで手早く集計する流れです。表示をRichに任せるだけで、ターミナルの出力が一気に読みやすくなります。

同じ骨組みは、対象のログや集計軸を変えるだけで幅広く使い回せます。まずはgenでサンプルを作り、手元で表示を確かめながら育てていくのがおすすめです。

参考にした一次情報

1. ^ Rich公式ドキュメント. https://rich.readthedocs.io/en/stable/, (参照26-07-23).
2. ^ re — 正規表現操作(Python公式ドキュメント). https://docs.python.org/ja/3/library/re.html, (参照26-07-23).
3. ^ collections — コンテナデータ型(Python公式ドキュメント). https://docs.python.org/ja/3/library/collections.html, (参照26-07-23).
4. ^ argparse — コマンドラインオプション解析(Python公式ドキュメント). https://docs.python.org/ja/3/library/argparse.html, (参照26-07-23).

※内容は執筆時点のものです。ライブラリやサイトの仕様は変わる可能性があるため、公式ドキュメントもあわせてご確認ください。

ITやプログラミングに関するコラム

• 
  【CSS】notで複数の件を除外する方法
• 
  【Git】remote設定を変更する方法
• 
  【VBA】コメントアウトを設定する方法
• 
  x86とx64の違いを分かりやすく解説
• 
  マークダウンで改行する方法
• 
  CapsLockキーを解除する方法
• 
  GitLabとGitHubの違いを解説
• 
  Linuxで環境変数を確認する方法
• 
  UbuntuのIPアドレスを確認する方法
• 
  パソコンのメモリの目安を用途別に選ぶ方法

ITやプログラミングに関するニュース

• 
  VercelがAI GatewayにSeedream 5.0 Proを追加、AI SDKのモデル指定で画像生成と編集が可能に
• 
  AWSがAmazon LocationのPlaces APIを強化、住所表記の指定と移動手段別の検索が可能に
• 
  VercelがトレースにTree・Waterfallビューを追加、ログ画面で処理の階層と所要時間を確認可能に
• 
  Googleがエージェント評価の再考を提唱、難易度を情報量で測るDiscovery Benchを解説
• 
  Google CloudがCloud Runサンドボックスを公開プレビューで提供、サービスヘルスは一般提供に
• 
  Google Cloud EMEAが英国金融の重要第三者に指定、イングランド銀行・PRA・FCAの直接監督下に
• 
  AWS DMS Schema ConversionがSQL Serverのオフライン変換に対応、ソースDBへ接続せずスキーマを変換可能に
• 
  EC2 G7インスタンスが米国東部（バージニア北部）で利用可能に、G6比でAI推論性能が最大4.6倍
• 
  SageMaker HyperPodが継続プロビジョニングでのAMIベース構成に対応、S3のスクリプト管理なしでSlurmクラスターを作成可能に
• 
  AWSがEMR on EKSでSparkトラブルシューティングエージェントに対応、失敗ジョブの原因分析を自然言語で依頼可能に