全スクリプトで使い回せる！Bootstrap × smarthomelog × InfluxWriter で作るPythonスマートホームの共通基盤設計

smarthome/                         ← PROJECT_ROOT（ここを自動検出）
├── config/
│   ├── settings.json              ← 全スクリプト共通の設定ファイル
│   └── services/                  ← systemd サービスファイル置き場
│       ├── smh-rain-forecast-alert.service
│       ├── smh-process-status-record.service
│       └── ...（全サービス分）
├── libs/
│   ├── bootstrap.py               ← ① PROJECT_ROOT検出・設定読込・sys.path設定
│   ├── smarthomelog.py            ← ② ログ出力（ファイル＋InfluxDB）
│   └── influx/
│       └── v2/
│           ├── client.py          ← ③ InfluxDB への HTTP API クライアント
│           └── writer.py          ← ④ measurement 別の書き込みメソッド集
└── core/
    ├── alert/
    │   └── rain_forecast/
    │       └── rain_forecast_alert.py   ← 個別スクリプト（共通基盤を使う側）
    └── record/
        └── raspi/
            └── process_status_record.py

# -*- coding: utf-8 -*-
__group__ = '情報記録'           # Grafanaでのグループ名（任意）
__description__ = 'このスクリプトの説明'  # Grafanaでの説明（任意）
# ------ 全Pythonプログラム共通の宣言 ----------------------------
from pathlib import Path
import sys
# プロジェクトのルートフォルダを取得
PROJECT_ROOT = Path(__file__).resolve()
while PROJECT_ROOT.name != "smarthome":
    if PROJECT_ROOT.parent == PROJECT_ROOT:
        raise RuntimeError("smarthome project root not found")
    PROJECT_ROOT = PROJECT_ROOT.parent
# Bootstrapをimportするためだけにプロジェクトルートをパスに追加
sys.path.insert(0, str(PROJECT_ROOT))
# BootstrapでPathの設定と設定ファイルの取得
from libs.bootstrap import Bootstrap
bootstrap = Bootstrap(__file__)   # Bootstrapをインスタンス化
settings = bootstrap.settings    # 設定ファイルを取得
# ------------------------------------------------------------
# このファイル固有のインポート（以下にスクリプト独自の import を書く）

from pathlib import Path
import sys
import os
import json
from typing import Iterable

class Bootstrap:
    def __init__(self, entry_file: str):
        self.entry_file: Path = Path(entry_file).resolve()
        self.project_root: Path = self._detect_project_root()
        self.settings: dict = self._load_common_settings()
        self._setup_base_sys_path()

    def _detect_project_root(self) -> Path:
        """config/settings.json を持つディレクトリをさかのぼって PROJECT_ROOT とする"""
        start_dir = self.entry_file.parent.parent  # 1つ上から開始

        for parent in [start_dir, *start_dir.parents]:
            if (parent / "config" / "settings.json").exists():
                return parent

        raise RuntimeError("PROJECT_ROOT not found (config/settings.json not found)")

    def _add_sys_paths(self, paths: Iterable[Path]):
        for p in paths:
            p = Path(p).resolve()
            p_str = str(p)
            if p_str not in sys.path:
                sys.path.insert(0, p_str)

    def _setup_base_sys_path(self):
        """libs/ と interfaces/outbound/ を sys.path に追加"""
        paths = [
            self.project_root / "libs",
            self.project_root / "interfaces" / "outbound",
        ]
        self._add_sys_paths(paths)

    def _load_common_settings(self) -> dict:
        """PROJECT_ROOT/config/settings.json を読む"""
        settings_path = self.project_root / "config" / "settings.json"
        if not settings_path.exists():
            raise FileNotFoundError(f"settings.json not found: {settings_path}")
        with settings_path.open("r", encoding="utf-8") as f:
            return json.load(f)

    def add_sys_path(self, *paths: Path):
        """スクリプト固有の sys.path 追加（個別スクリプトから呼び出す）"""
        self._add_sys_paths(paths)

    def chdir_local(self):
        """実行スクリプトのあるディレクトリに cwd を変更する"""
        os.chdir(self.entry_file.parent)

    def load_local_settings(self, filename: str = "config/settings.json") -> dict:
        """実行スクリプトと同じディレクトリにあるローカル設定を読む（なければ空dict）"""
        local_path = self.entry_file.parent / filename
        if not local_path.exists():
            return {}
        with local_path.open("r", encoding="utf-8") as f:
            return json.load(f)

    @property
    def local_dir(self) -> Path:
        return self.entry_file.parent

class LogType:
    SUCCESS           = settings['log']['path']['success']
    SYSTEM_ERROR      = settings['log']['path']['system_error']
    APPLICATION_ERROR = settings['log']['path']['application_error']
    GOOGLE_LIVING     = settings['log']['path']['google_living']
    LINE              = settings['log']['path']['line']
    # ...（その他のログパス）

class Log:
    def __init__(self, program_path: str, logtype: LogType):
        self.log_path    = logtype
        self.program_path = program_path
        self.logtype_name = [k for k, v in LogType.__dict__.items() if v == logtype][0]
        self.influx_writer = influxwriter.Writer()

    def write(self, message: str, program_path: str = None, use_separator: bool = False):
        if self.logtype_name == 'SUCCESS':
            return   # SUCCESSはファイルに書かない（InfluxDBのみ）

        now = datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
        filename = os.path.basename(program_path) if program_path else ''
        log_message = (filename + ' ' if filename else '') + str(message)

        with open(self.log_path, 'a') as f:
            f.write(now + ' ' + log_message + '\n')

        # APPLICATION_ERROR は InfluxDB にも記録
        if self.logtype_name == 'APPLICATION_ERROR':
            self.influx_writer.insert_execution_status_log(
                self.program_path, influxwriter.LogStatusType.APPLICATION_ERROR, log_message
            )

    def write_error(self, exc_info):
        """システムエラーをファイル＋InfluxDB に記録（トレースバック付き）"""
        now = datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
        t, v, tb = exc_info
        messages = traceback.format_exception(t, v, tb)
        with open(self.log_path, 'a') as f:
            f.write(now + ' ' + self.program_path + '\n')
            for msg in messages:
                f.write(now + ' ' + msg + '\n')

        self.influx_writer.insert_execution_status_log(
            self.program_path, influxwriter.LogStatusType.SYSTEM_ERROR, str(messages)
        )

    def get_seconds_since_the_file_was_updated(self) -> int:
        """最終ログ更新からの経過秒数を返す（死活監視に利用）"""
        if not os.path.exists(self.log_path):
            return 999999999
        diff = datetime.datetime.now() - datetime.datetime.fromtimestamp(
            os.path.getmtime(self.log_path)
        )
        return int(diff.seconds)

import smarthomelog

# インスタンス作成（どのログファイルに書くかを LogType で指定）
success_log = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)
app_err_log = smarthomelog.Log(__file__, smarthomelog.LogType.APPLICATION_ERROR)

try:
    # ... メイン処理 ...

    # 正常終了時
    success_log.write('処理が正常に終了しました。', __file__)

except Exception as e:
    # システムエラー時（トレースバック付きで記録）
    log = smarthomelog.Log(__file__, smarthomelog.LogType.SYSTEM_ERROR)
    log.write_error(sys.exc_info())

import requests

API_URL      = 'http://192.168.xx.xx/internal/influx'
INFLUX_BUCKET = settings.get('influxdb2').get("bucket")

def write_dict(measurement: str, tags: dict, fields: dict, time: str = None):
    """1レコードを InfluxDB に書き込む"""
    record = {
        "measurement": measurement,
        "tags": tags,
        "fields": fields,
    }
    if time:
        record["time"] = time
    resp = requests.post(f"{API_URL}/write", json=record)
    return resp.status_code, resp.text

def write_dict_batch(measurement: str, tags_list: list, fields_list: list, time_list: list = None):
    """複数レコードを一括で InfluxDB に書き込む"""
    if len(tags_list) != len(fields_list):
        raise ValueError("tags_list と fields_list は同じ長さが必要です")

    payloads = {
        "measurement": measurement,
        "tags_list": tags_list,
        "fields_list": fields_list,
        "time_list": time_list
    }
    return write_batch(payloads)

class LogStatusType:
    SUCCESS           = 'SUCCESS'
    APPLICATION_ERROR = 'APPLICATION_ERROR'
    SYSTEM_ERROR      = 'SYSTEM_ERROR'

class Writer:
    def __init__(self):
        self.conn = influxclient   # client モジュールをそのまま使用

    def insert_execution_log(self, program_path: str, message: str):
        """スクリプトの実行開始ログを記録"""
        measurement = 'execution_log'
        tags   = {'program_path': program_path}
        fields = {'message': message,
                  'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S')}
        self.conn.write_dict(measurement, tags, fields)

    def insert_execution_status_log(self, program_path: str, status: LogStatusType, message: str):
        """スクリプトの実行状態（正常/エラー）を記録"""
        measurement = 'execution_status_log'
        tags   = {'program_path': program_path, 'status': status}
        fields = {'message': message,
                  'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S')}
        self.conn.write_dict(measurement, tags, fields)

    def insert_nginx_log(self, log_type: str, message: str):
        """NGINX ログを記録"""
        # ... （各 measurement 対応のメソッドが並ぶ）

    def insert_nucbox_system_log(self, hostname: str, fields: dict):
        """NUCBOX のシステム情報を記録"""
        # ...

# ボイラープレート（省略）...

try:
    # ① InfluxDB 接続・ログインスタンス生成
    influx = influxwriter.Writer()
    success_log = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)
    app_err_log = smarthomelog.Log(__file__, smarthomelog.LogType.APPLICATION_ERROR)

    # ② 実行開始をInfluxDBに記録
    influx.insert_execution_log(__file__, 'Start!')

    # ③ メイン処理
    while True:
        # ... 処理内容 ...

        # ④ 正常終了ログ
        success_log.write('処理が正常に終了しました。', __file__)
        time.sleep(5)

except Exception as e:
    # ⑤ システムエラー：トレースバック付きでファイル＋InfluxDB に記録
    log = smarthomelog.Log(__file__, smarthomelog.LogType.SYSTEM_ERROR)
    log.write_error(sys.exc_info())

[Unit]
Description=Weather Alert
After=nginx.service influxdb.service webapi.service
Wants=nginx.service influxdb.service webapi.service

[Service]
ExecStart=/home/<username>/Projects/venv/bin/python \
    /home/<username>/Projects/smarthome/core/alert/rain_forecast/rain_forecast_alert.py
Restart=always
RestartSec=60
Type=simple
User=<username>

[Install]
WantedBy=multi-user.target

# 全サービスファイルを /etc/systemd/system/ にコピー
sudo cp /home/<username>/Projects/smarthome/config/services/* /etc/systemd/system/

# systemd に新しいサービスファイルを認識させる
sudo systemctl daemon-reload

# 各サービスを有効化・起動（例）
sudo systemctl enable smh-rain-forecast-alert
sudo systemctl start  smh-rain-forecast-alert

# 状態確認（直近のログも表示される）
sudo systemctl status smh-rain-forecast-alert

# 起動 / 停止 / 再起動
sudo systemctl start   smh-rain-forecast-alert
sudo systemctl stop    smh-rain-forecast-alert
sudo systemctl restart smh-rain-forecast-alert

# 自動起動の有効化 / 無効化
sudo systemctl enable  smh-rain-forecast-alert
sudo systemctl disable smh-rain-forecast-alert

# サービスファイルを編集した後は必ず daemon-reload
sudo systemctl daemon-reload

# 全サービスの状態を一覧表示（smh- 始まりのサービスだけ絞り込む）
systemctl list-units --type=service | grep smh-

# リアルタイムでログを流す（-f = follow）
sudo journalctl -u smh-rain-forecast-alert -f

# 直近100行だけ表示
sudo journalctl -u smh-rain-forecast-alert -n 100

from pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve()
while PROJECT_ROOT.name != "smarthome":
    PROJECT_ROOT = PROJECT_ROOT.parent
sys.path.insert(0, str(PROJECT_ROOT))

from libs.bootstrap import Bootstrap
bootstrap = Bootstrap(__file__)
print(bootstrap.project_root)   # → /home//Projects/smarthome
print(bootstrap.settings.keys()) # → settings.json のキー一覧