""" notifier.py — Lv2 Jarvis 常駐デーモン 通知モジュール dashboard書き込み + iMessage 警告 通知トリガー（呼び出し側で判定）： - モード切替忘れ：1時間に呼名5回超 → 会議モード切替提案 - モード自動降格直前：会議モード残り10分警告 - 誤発火多発：30分で誤発火3回 → 学習モード切替提案 - トークン80%到達：警告 - トークン上限到達：全停止 + iMessage """ import subprocess import json from datetime import datetime from pathlib import Path from typing import Optional # ────────────────────────────────────────────── # パス定義 # ────────────────────────────────────────────── # 起業フォルダー/dashboard DASHBOARD_DIR = Path(__file__).parent.parent.parent.parent.parent / "dashboard" # live_session/live_meetings/ LIVE_MEETING_DIR = Path(__file__).parent.parent / "live_meetings" # live_session/notification.md NOTIFICATION_PATH = Path(__file__).parent.parent / "notification.md" # ────────────────────────────────────────────── # 定数 # ────────────────────────────────────────────── # ユーザーの電話番号（+81形式） # NOTE: 推定値。実値に要修正（feedback_imessage_send.md 準拠） USER_PHONE = "+819014046219" # 通知レベル絵文字 LEVEL_INFO = "ℹ️" LEVEL_WARN = "⚠️" LEVEL_CRITICAL = "🔴" # モード絵文字マップ MODE_EMOJI = { "WAITING": "⏳", "MEETING": "🎙️", "LEARNING": "📚", "PAUSED": "⏸️", } # ────────────────────────────────────────────── # dashboard カード書き込み # ────────────────────────────────────────────── def write_dashboard_card( transcript_excerpt: str, jarvis_response: str, mode: str = "WAITING", marked_false_positive: bool = False, ) -> Path: """ dashboard カード（Markdown）を live_meeting_YYYY-MM-DD.md に追記。 形式： ## HH:MM:SS [{mode_emoji}] **発話：** {transcript_excerpt} **ジャービス：** {jarvis_response} {⚠️ 誤発火・撤回 （marked_false_positive=True なら）} --- ファイル無ければ新規作成（ヘッダー付き）。 戻り値：書き込み先のパス """ LIVE_MEETING_DIR.mkdir(parents=True, exist_ok=True) date_str = datetime.now().strftime("%Y-%m-%d") time_str = datetime.now().strftime("%H:%M:%S") filename = f"live_meeting_{date_str}.md" card_path = LIVE_MEETING_DIR / filename mode_emoji = MODE_EMOJI.get(mode, "❓") # ファイル初回 → ヘッダー付きで作成 if not card_path.exists(): header = f"# ライブミーティングログ {date_str}\n\n" card_path.write_text(header, encoding="utf-8") # カード本文 fp_mark = "\n⚠️ 誤発火・撤回\n" if marked_false_positive else "" card = ( f"## {time_str} [{mode_emoji}]\n" f"**発話：** {transcript_excerpt}\n\n" f"**ジャービス：** {jarvis_response}\n" f"{fp_mark}\n" "---\n\n" ) with card_path.open("a", encoding="utf-8") as f: f.write(card) return card_path def mark_false_positive(timestamp: str) -> bool: """ 既存の dashboard カードに「⚠️ 誤発火・撤回」マークを追加。 timestamp で対象エントリを特定（HH:MM:SS）。 戻り値：成功したか """ date_str = datetime.now().strftime("%Y-%m-%d") card_path = LIVE_MEETING_DIR / f"live_meeting_{date_str}.md" if not card_path.exists(): return False content = card_path.read_text(encoding="utf-8") target_heading = f"## {timestamp} [" if target_heading not in content: return False # セクションに ⚠️ 誤発火・撤回 が未追加なら挿入（--- の手前に差し込む） # 対象ブロックの最初の "---" の前に追記 idx = content.index(target_heading) section = content[idx:] divider_pos = section.find("\n---\n") if divider_pos == -1: return False mark = "\n⚠️ 誤発火・撤回" if mark in section[:divider_pos]: return True # 既にマーク済み insert_at = idx + divider_pos new_content = content[:insert_at] + mark + "\n" + content[insert_at:] card_path.write_text(new_content, encoding="utf-8") return True # ────────────────────────────────────────────── # 通知バナー書き込み # ────────────────────────────────────────────── def write_notification( level: str, message: str, action_hint: Optional[str] = None, ) -> Path: """ dashboard 上部通知バナー用の notification.md に書き込み（上書き・最新1件のみ）。 形式： # 通知 {level} **{datetime}** {message} → {action_hint} （あれば） """ NOTIFICATION_PATH.parent.mkdir(parents=True, exist_ok=True) dt_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") lines = [ "# 通知\n", f"\n{level} **{dt_str}** {message}\n", ] if action_hint: lines.append(f"\n→ {action_hint}\n") NOTIFICATION_PATH.write_text("".join(lines), encoding="utf-8") return NOTIFICATION_PATH def clear_notification() -> None: """notification.md を空にする（通知解除）""" if NOTIFICATION_PATH.exists(): NOTIFICATION_PATH.write_text("", encoding="utf-8") # ────────────────────────────────────────────── # iMessage 送信 # ────────────────────────────────────────────── def send_imessage(message: str, phone: Optional[str] = None) -> dict: """ osascript 経由で iMessage 送信。 feedback_imessage_send.md の方式を踏襲（+81形式）。 注意：osascript exit 0 でも実際の送信成否は別問題 （Messages アプリが起動・ログイン済みであること前提） 戻り値：{"ok": bool, "stdout": str, "stderr": str, "returncode": int} """ target = phone or USER_PHONE # " を \" にエスケープ escaped = message.replace('"', '\\"') script = ( f'tell application "Messages" to send "{escaped}" ' f'to buddy "{target}" of ' f'(1st service whose service type is iMessage)' ) try: result = subprocess.run( ["osascript", "-e", script], capture_output=True, text=True, timeout=15, ) return { "ok": result.returncode == 0, "stdout": result.stdout.strip(), "stderr": result.stderr.strip(), "returncode": result.returncode, } except subprocess.TimeoutExpired: return {"ok": False, "stdout": "", "stderr": "timeout", "returncode": -1} except Exception as e: return {"ok": False, "stdout": "", "stderr": str(e), "returncode": -1} # ────────────────────────────────────────────── # 通知トリガー関数 # ────────────────────────────────────────────── def notify_mode_switch_suggestion(call_count: int, window_min: int = 60) -> None: """呼名検出が1時間に5回超 → 会議モード切替提案""" msg = f"呼名検出が直近{window_min}分で{call_count}回。会議モードに切替する？" write_notification(LEVEL_INFO, msg, action_hint="「ジャービス会議モード」と発話で切替") def notify_mode_demote_warning(remaining_min: int = 10) -> None: """モード自動降格 残り10分警告""" msg = f"会議モード終了まで残り{remaining_min}分。継続するなら呼んでください。" write_notification(LEVEL_WARN, msg) def notify_false_positive_burst(count: int, window_min: int = 30) -> None: """誤発火30分で3回 → 学習モード切替提案""" msg = f"誤発火が直近{window_min}分で{count}回。学習モードに切替して誤発火収集する？" write_notification(LEVEL_WARN, msg, action_hint="「ジャービス学習モード」と発話で切替") def notify_token_threshold(current: int, limit: int) -> None: """トークン80%警告""" pct = current / limit * 100 msg = ( f"トークン消費 {pct:.0f}% 到達（{current:,} / {limit:,}）。" "あと20%で停止します。" ) write_notification(LEVEL_WARN, msg) def notify_token_exhausted(current: int, limit: int) -> None: """トークン上限到達 → 全停止 + iMessage""" msg = ( f"🔴 Jarvis停止：トークン上限到達（{current:,} / {limit:,}）。" "手動再起動が必要。" ) write_notification( LEVEL_CRITICAL, msg, action_hint="ターミナルで jarvis_daemon.py を再起動", ) imsg = f"⚠️ Jarvis Lv2デーモン停止：トークン上限到達（{current:,}/{limit:,}）" send_imessage(imsg) # ────────────────────────────────────────────── # self-test # ────────────────────────────────────────────── if __name__ == "__main__": import sys print("=== notifier self-test ===") # dashboard カード書き込みテスト if "--card" in sys.argv: path = write_dashboard_card( transcript_excerpt="ジャービス、今日の天気は？", jarvis_response="未確認です。確認には外部API連携が必要。", mode="WAITING", ) print(f"カード書き込み: {path}") # 誤発火マークテスト ts = datetime.now().strftime("%H:%M:%S") # 直前に書いたカードのタイムスタンプでマーク（数秒ズレあり、手動確認推奨） ok = mark_false_positive(ts) print(f"誤発火マーク（{ts}）: {'成功' if ok else '対象未検出（タイムスタンプズレ）'}") # 通知テスト if "--notify" in sys.argv: notify_token_threshold(24000, 30000) print(f"通知書き込み: {NOTIFICATION_PATH}") # 通知クリアテスト clear_notification() print("通知クリア完了") # 各トリガーテスト notify_mode_switch_suggestion(call_count=6) print("会議モード提案通知 書き込み完了") notify_mode_demote_warning(remaining_min=10) print("モード降格警告通知 書き込み完了") notify_false_positive_burst(count=3) print("誤発火多発通知 書き込み完了") # iMessage テスト（明示指定時のみ・実送信される） if "--imessage" in sys.argv: result = send_imessage("Jarvis Lv2 notifier self-test") print(f"iMessage結果: {result}") else: print("iMessage実送信は --imessage 指定時のみ")