teraterm + esp-idf log parser utility

gistfile1.txt

"""
TeraTerm のログ機能で ESP-IDF のコンソールログを記録したファイルをパースするユーティリティ。
時刻、ログレベル、タグ、ログメッセージを分解して sqlite3 のデータベースファイルに記録する。
特定の日付に絞り込んで参照する場合は sqlite3 コマンド経由でクエリを投げること。

使い方

>>> # 初期化
>>> python3 log_parser.py --db or_esp32.sqlite --init
>>> python3 log_parser.py --db or_esp32.sqlite log_file.txt

クエリ

$ sqlite3 or_esp32.sqlite
SQLite version 3.43.2 2023-10-10 13:08:14
Enter ".help" for usage hints.
sqlite> SELECT DISTINCT log_severity FROM logs WHERE tag = 'fan_control_task' AND message like 'Disable%%';
"""
import re
import sqlite3
import argparse

def parse_log(log_line):
    # 正規表現でログを分解
    pattern = re.compile(r"\[(.*?)\] (\w) \((\d+)\) ([^:]+): (.*)")
    match = pattern.match(log_line)
    
    if match:
        return {
            "datetime": match.group(1),
            "monotonic": match.group(3),
            "log_severity": match.group(2),
            "tag": match.group(4),
            "message": match.group(5)
        }
    else:
        raise ValueError("ログ形式が一致しません: {}".format(log_line))

def initialize_database(db_file):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS logs (
            datetime TEXT,
            monotonic TEXT,
            log_severity TEXT,
            tag TEXT,
            message TEXT,
            UNIQUE(datetime, monotonic, log_severity, tag, message)
        )
    """)
    conn.commit()
    conn.close()

def process_log_file(log_file, db_file):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()

    try:
        with open(log_file, 'r', errors='ignore') as file:
            for line in file:
                line = line.strip()
                if line:
                    try:
                        parsed_log = parse_log(line)
                        cursor.execute("""
                            INSERT OR IGNORE INTO logs (datetime, monotonic, log_severity, tag, message)
                            VALUES (?, ?, ?, ?, ?)
                        """, (parsed_log['datetime'], parsed_log['monotonic'], parsed_log['log_severity'], parsed_log['tag'], parsed_log['message']))
                    except ValueError as e:
                        print(f"Error parsing line: {e}")
        conn.commit()
    finally:
        conn.close()

def main():
    parser = argparse.ArgumentParser(description="Process log files and store them in a SQLite database.")
    parser.add_argument("--db", type=str, required=True, help="Path to the SQLite database file.")
    parser.add_argument("--init", action="store_true", help="Initialize the database (create tables).")
    parser.add_argument("log_file", type=str, nargs="?", help="Path to the log file to process.")

    args = parser.parse_args()

    if args.init:
        initialize_database(args.db)
        if not args.log_file:
            print("Database initialized.")
            return

    if not args.log_file:
        print("Error: log_file must be specified unless --init is used alone.")
        return

    process_log_file(args.log_file, args.db)

if __name__ == "__main__":
    main()