""" kinmu PC Monitor Uploader ========================= 社員PCに常駐し、アクティブウィンドウ・マウス/キー入力数・定期スクショを kinmu サーバへ送信します。「全社オン」＋「社員個別オン」両方ONの時のみ稼働。 【プライバシー】 - ウィンドウタイトル・アプリ名・マウスクリック数・キー入力数・スクショ画像を取得します - 本人の同意なしに稼働させないでください - kinmu 上で OFF にされたら即座に送信停止します 【インストール】 1. このファイルを社員PCの任意の場所に保存（例: C:\\kinmu-monitor\\pc_monitor.py） 2. 依存パッケージインストール: Windows: pip install requests psutil pywin32 mss Pillow pynput Mac: pip install requests psutil pyobjc mss Pillow pynput 3. kinmu の「💻 PC計測設定」で発行された APIキーを下の API_KEY に貼付 4. 常駐起動: Windows: タスクスケジューラ「ログオン時」に `pythonw.exe pc_monitor.py` Mac: launchd plist で常駐 """ # ============================================================ # 設定（kinmu の「💻 PC計測設定」で発行された APIキーをここに貼付） # ============================================================ API_KEY = "PASTE_YOUR_API_KEY_HERE" SERVER_URL = "https://kinmu.insense.info" # ============================================================ # 以下、編集不要 # ============================================================ import sys import os import time import platform import threading import io import tempfile import logging try: import requests except ImportError: print("ERROR: pip install requests") sys.exit(1) IS_WINDOWS = platform.system() == "Windows" IS_MAC = platform.system() == "Darwin" # ログ LOG_DIR = os.path.join(tempfile.gettempdir(), "kinmu-monitor") os.makedirs(LOG_DIR, exist_ok=True) logging.basicConfig( filename=os.path.join(LOG_DIR, "monitor.log"), level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) log = logging.getLogger("kinmu-monitor") # ── アクティブウィンドウ取得（OS別） ── def get_active_window(): """(window_title, app_name) を返す。失敗時は ('', '')""" try: if IS_WINDOWS: import win32gui, win32process, psutil hwnd = win32gui.GetForegroundWindow() title = win32gui.GetWindowText(hwnd) or "" _, pid = win32process.GetWindowThreadProcessId(hwnd) try: proc = psutil.Process(pid) app = proc.name() except Exception: app = "" return title, app elif IS_MAC: from AppKit import NSWorkspace ws = NSWorkspace.sharedWorkspace() active = ws.frontmostApplication() app = active.localizedName() or "" return "", app # macOS でタイトル取るのは権限要なので最小実装 except Exception as e: log.warning("active window error: %s", e) return "", "" # ── マウス/キーカウント（pynputで集計） ── _counts = {"mouse": 0, "keys": 0} _counts_lock = threading.Lock() def _start_input_listeners(): try: from pynput import mouse, keyboard def on_click(*a): with _counts_lock: _counts["mouse"] += 1 def on_press(*a): with _counts_lock: _counts["keys"] += 1 mouse.Listener(on_click=on_click).start() keyboard.Listener(on_press=on_press).start() log.info("input listeners started") except Exception as e: log.warning("pynput unavailable, input counts will stay 0: %s", e) def _pop_counts(): with _counts_lock: m, k = _counts["mouse"], _counts["keys"] _counts["mouse"] = 0 _counts["keys"] = 0 return m, k # ── スクショ ── def take_screenshot_bytes(): try: import mss from PIL import Image with mss.mss() as sct: shot = sct.grab(sct.monitors[1]) # primary monitor img = Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX") # サイズ縮小（容量節約）: 横1280pxまで if img.width > 1280: r = 1280 / img.width img = img.resize((1280, int(img.height * r))) buf = io.BytesIO() img.save(buf, format="PNG", optimize=True) return buf.getvalue() except Exception as e: log.warning("screenshot error: %s", e) return None # ── サーバー通信 ── def fetch_config(): """サーバから稼働可否を取得。エラー時は None（稼働停止）。""" try: r = requests.get( f"{SERVER_URL}/api/monitoring/config", headers={"X-Monitoring-Key": API_KEY}, timeout=10, ) if r.status_code == 200: return r.json() log.warning("config fetch %s: %s", r.status_code, r.text[:200]) except Exception as e: log.warning("config fetch error: %s", e) return None def post_activity(title, app, mouse_n, key_n): try: r = requests.post( f"{SERVER_URL}/api/monitoring/activity", headers={"X-Monitoring-Key": API_KEY, "Content-Type": "application/json"}, json={"window_title": title, "app_name": app, "mouse_clicks": mouse_n, "key_presses": key_n}, timeout=10, ) if r.status_code != 200: log.warning("activity %s: %s", r.status_code, r.text[:200]) except Exception as e: log.warning("activity post error: %s", e) def post_screenshot(png_bytes): try: r = requests.post( f"{SERVER_URL}/api/monitoring/screenshot", headers={"X-Monitoring-Key": API_KEY}, files={"file": ("screen.png", png_bytes, "image/png")}, timeout=30, ) if r.status_code != 200: log.warning("screenshot %s: %s", r.status_code, r.text[:200]) except Exception as e: log.warning("screenshot post error: %s", e) # ── メインループ ── def main(): if API_KEY == "PASTE_YOUR_API_KEY_HERE": print("ERROR: API_KEY を設定してください（kinmu の PC計測設定で発行）") log.error("API_KEY not set") sys.exit(1) log.info("kinmu PC Monitor 起動 (os=%s)", platform.system()) _start_input_listeners() activity_interval = 300 # 5分（config で上書き可） screenshot_interval = 900 # 15分 last_activity = 0 last_screenshot = 0 config_refresh = 0 while True: now = time.time() # 設定を5分おきに再取得 if now - config_refresh >= 300: cfg = fetch_config() if cfg: activity_interval = int(cfg.get("activity_interval_sec", activity_interval)) screenshot_interval = int(cfg.get("screenshot_interval_sec", screenshot_interval)) enabled = bool(cfg.get("enabled")) if not enabled: log.info("monitoring disabled by server; sleeping 5min") time.sleep(300) continue else: # 設定取得失敗時は念のため停止 log.warning("config unavailable; sleeping 5min") time.sleep(300) continue config_refresh = now # 5分おきにアクティビティ送信 if now - last_activity >= activity_interval: title, app = get_active_window() m, k = _pop_counts() post_activity(title, app, m, k) last_activity = now # 15分おきにスクショ送信 if now - last_screenshot >= screenshot_interval: png = take_screenshot_bytes() if png: post_screenshot(png) last_screenshot = now time.sleep(10) if __name__ == "__main__": main()