import tkinter as tk from tkinter import scrolledtext import requests import json import datetime import webbrowser import os import io import time import threading from urllib.parse import urlsplit import ipaddress CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".jigyodan_config.json") def load_config(): try: if os.path.exists(CONFIG_PATH): with open(CONFIG_PATH, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return {} def save_config(cfg): try: with open(CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(cfg, f, ensure_ascii=False, indent=2) except Exception: pass class SettingsDialog(tk.Toplevel): def __init__(self, master, cfg): super().__init__(master) self.title("設定") self.resizable(False, False) self.grab_set() self.cfg = cfg.copy() frm = tk.Frame(self) frm.pack(padx=12, pady=12) row = 0 # 接続先 tk.Label(frm, text="接続先").grid(row=row, column=0, sticky='e', padx=6, pady=6) self.base_choices = [ "https://jigyodan.sakura.ne.jp/zaitakukanri_honbu/", "http://49.212.200.138/zaitakukanri_honbu/public/", "http://49.212.200.138/zaitakukanri_honbu/", ] self.var_base = tk.StringVar(value=self.cfg.get("base_url", self.base_choices[0])) self.opt_base = tk.OptionMenu(frm, self.var_base, *self.base_choices) self.opt_base.config(width=38) self.opt_base.grid(row=row, column=1, padx=6, pady=6, sticky='w') row += 1 tk.Label(frm, text="ユーザー名").grid(row=row, column=0, sticky='e', padx=6, pady=6) self.ent_user = tk.Entry(frm, width=28) self.ent_user.grid(row=row, column=1, padx=6, pady=6) self.ent_user.insert(0, cfg.get("user_id", "")) row += 1 tk.Label(frm, text="パスワード").grid(row=row, column=0, sticky='e', padx=6, pady=6) self.ent_pass = tk.Entry(frm, show='*', width=28) self.ent_pass.grid(row=row, column=1, padx=6, pady=6) self.ent_pass.insert(0, cfg.get("password", "")) row += 1 tk.Label(frm, text="ボリューム番号").grid(row=row, column=0, sticky='e', padx=6, pady=6) self.ent_vol = tk.Entry(frm, width=28) self.ent_vol.grid(row=row, column=1, padx=6, pady=6) self.ent_vol.insert(0, cfg.get("volume_no", "")) row += 1 tk.Label(frm, text="シリアル番号").grid(row=row, column=0, sticky='e', padx=6, pady=6) self.ent_ser = tk.Entry(frm, width=28) self.ent_ser.grid(row=row, column=1, padx=6, pady=6) self.ent_ser.insert(0, cfg.get("serial_no", "")) row += 1 # 任意: ホストヘッダ（IP直指定時に仮想ホストを指定するため） tk.Label(frm, text="Hostヘッダ(任意)").grid(row=row, column=0, sticky='e', padx=6, pady=6) self.ent_host = tk.Entry(frm, width=28) self.ent_host.grid(row=row, column=1, padx=6, pady=6) self.ent_host.insert(0, cfg.get("host_header", "")) row += 1 # 任意: カメラ番号（既定: 0） tk.Label(frm, text="カメラ番号(任意)").grid(row=row, column=0, sticky='e', padx=6, pady=6) self.ent_cam = tk.Entry(frm, width=28) self.ent_cam.grid(row=row, column=1, padx=6, pady=6) self.ent_cam.insert(0, str(cfg.get("camera_index", 0))) row += 1 btns = tk.Frame(self) btns.pack(padx=12, pady=(0, 12), fill='x') tk.Button(btns, text="保存", command=self.on_save).pack(side='right', padx=4) tk.Button(btns, text="閉じる", command=self.destroy).pack(side='right', padx=4) def on_save(self): self.cfg["base_url"] = self.var_base.get().strip() self.cfg["user_id"] = self.ent_user.get().strip() self.cfg["password"] = self.ent_pass.get().strip() self.cfg["volume_no"] = self.ent_vol.get().strip() self.cfg["serial_no"] = self.ent_ser.get().strip() self.cfg["host_header"] = self.ent_host.get().strip() try: self.cfg["camera_index"] = int(self.ent_cam.get().strip()) except Exception: self.cfg["camera_index"] = 0 # 即時反映 self.master._temp_password = self.cfg["password"] self.master.base_url = self.cfg["base_url"] self.master.host_header = self.cfg.get("host_header", "") self.master.camera_index = self.cfg.get("camera_index", 0) save_config(self.cfg) self.destroy() class MainWindow(tk.Tk): def __init__(self): super().__init__() self.title("Jigyodan Main Application") self.geometry("980x720") cfg0 = load_config() self.base_url = cfg0.get("base_url", "https://jigyodan.sakura.ne.jp/zaitakukanri_honbu/") self.token = None self.last_chat_date = "" self.logged_in = False self._poll_after_id = None self._temp_password = "" self.host_header = cfg0.get("host_header", "") self.camera_index = int(cfg0.get("camera_index", 0)) # サーバから受け取る中間キャプチャ間隔（分）。未取得時は30分。 self.middle_interval_minutes = 30 self._middle_after_id = None # キーボード/マウス反応監視 self.keyboard_interval_minutes = 30 self.mouse_interval_minutes = 30 self._kbd_after_id = None self._mouse_after_id = None self.keyboard_flag = '2' # '1':反応あり, '2':反応なし self.mouse_flag = '2' self._kbd_listener = None self._mouse_listener = None # WebSocket self._ws = None self._ws_thread = None # 上部: グループ・操作ボタン self.top_frame = tk.Frame(self) self.top_frame.pack(padx=10, pady=(10, 5), fill='x') self.group_label = tk.Label(self.top_frame, text="", font=("Meiryo", 14, "bold")) self.group_label.pack(side='left') # 右上: 設定 / ログイン / 状態 self.lbl_login = tk.Label(self.top_frame, text="未ログイン", relief='solid', width=10) self.lbl_login.pack(side='right', padx=(5, 0)) self.btn_login = tk.Button(self.top_frame, text="ログイン", command=self.toggle_login) self.btn_login.pack(side='right', padx=(5, 0)) self.btn_setting = tk.Button(self.top_frame, text="設定", command=self.open_settings) self.btn_setting.pack(side='right', padx=(5, 0)) # 操作: 日報 / 問い合せ / 業務 self.report_button = tk.Button(self.top_frame, text="日報", command=self.open_report) self.report_button.pack(side='right', padx=(5, 0)) self.inquiry_button = tk.Button(self.top_frame, text="問い合せ", command=self.send_inquiry) self.inquiry_button.pack(side='right', padx=(5, 0)) self.training_button = tk.Button(self.top_frame, text="作業開始", command=self.toggle_training) self.training_button.pack(side='right', padx=(5, 0)) # 通知表示 self.notice_text = scrolledtext.ScrolledText(self, height=6, state='disabled', wrap='word') self.notice_text.pack(padx=10, pady=(0, 10), fill='x') # 中段: 外部リンクとステータス self.mid_frame = tk.Frame(self) self.mid_frame.pack(padx=10, pady=(0, 10), fill='x') links_frame = tk.LabelFrame(self.mid_frame, text="外部リンク") links_frame.pack(side='left', padx=(0, 10), fill='y') self.links_list = tk.Listbox(links_frame, height=8) self.links_list.pack(padx=6, pady=6) self.links_list.bind('', self.open_selected_link) self._links_map = {} status_frame = tk.LabelFrame(self.mid_frame, text="業務ステータス") status_frame.pack(side='left', fill='both', expand=True) self.training_started = False self.training_start_time = None self.training_end_time = None row = 0 tk.Label(status_frame, text="開始時刻").grid(row=row, column=0, sticky='w', padx=6, pady=4) self.lbl_start_val = tk.Label(status_frame, text="-") self.lbl_start_val.grid(row=row, column=1, sticky='w', padx=6, pady=4) row += 1 tk.Label(status_frame, text="終了時刻").grid(row=row, column=0, sticky='w', padx=6, pady=4) self.lbl_end_val = tk.Label(status_frame, text="-") self.lbl_end_val.grid(row=row, column=1, sticky='w', padx=6, pady=4) row += 1 tk.Label(status_frame, text="業務時間").grid(row=row, column=0, sticky='w', padx=6, pady=4) self.lbl_time_val = tk.Label(status_frame, text="-") self.lbl_time_val.grid(row=row, column=1, sticky='w', padx=6, pady=4) # チャット見出し + 更新 chat_head = tk.Frame(self) chat_head.pack(fill='x', padx=10) tk.Label(chat_head, text="メッセージ（チャット）", font=("Meiryo", 12, "bold")).pack(side='left') self.update_button = tk.Button(chat_head, text="更新", command=self.fetch_messages_once) self.update_button.pack(side='right') # チャット表示 self.chat_display = scrolledtext.ScrolledText(self, state='disabled', wrap='word') self.chat_display.pack(padx=10, pady=(0, 10), fill='both', expand=True) # チャット入力 self.input_frame = tk.Frame(self) self.input_frame.pack(padx=10, pady=(0, 10), fill='x') self.message_entry = tk.Entry(self.input_frame, state='disabled') self.message_entry.pack(side='left', fill='x', expand=True) self.message_entry.bind("", self.send_message_event) self.send_button = tk.Button(self.input_frame, text="送信", command=self.send_message, state='disabled') self.send_button.pack(side='right', padx=5) # 起動時は未ログイン状態 self.set_logged_out_state() def post_multipart(self, url: str, fields: dict, files: dict | None = None): files_param = {} # 文字列フィールドも multipart/form-data で送る for k, v in fields.items(): files_param[k] = (None, "" if v is None else str(v)) if files: for k, v in files.items(): files_param[k] = v # (filename, fileobj, content_type) 形式を想定 # 仮想ホスト対策: IP直指定時に Host ヘッダを付与 headers = {} try: hostname = urlsplit(self.base_url).hostname or "" # host_header が設定されていれば優先 if self.host_header: headers["Host"] = self.host_header else: # base_url が IP ならデフォルトで既存ドメインを付与 try: ipaddress.ip_address(hostname) headers["Host"] = "jigyodan.sakura.ne.jp" except Exception: pass except Exception: pass return requests.post(url, files=files_param, headers=headers) def capture_camera(self): """カメラから1枚撮影し、requestsのfiles用タプルを返す。 返り値: (filename, fileobj, content_type) or None """ try: import cv2 except Exception as e: self.display_message(f"[注意] カメラ未対応: opencv-python をインストールしてください ({e})") return None try: idx = int(self.camera_index) if isinstance(self.camera_index, int) or str(self.camera_index).isdigit() else 0 cap = cv2.VideoCapture(idx, cv2.CAP_DSHOW) if not cap.isOpened(): cap = cv2.VideoCapture(idx) cap.set(cv2.CAP_PROP_FRAME_WIDTH, 320) cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 240) ok, frame = cap.read() for _ in range(10): if ok and frame is not None: break time.sleep(0.2) ok, frame = cap.read() cap.release() if not ok or frame is None: self.display_message("[注意] カメラから画像を取得できませんでした。") return None ok2, buf = cv2.imencode('.jpg', frame) if not ok2: self.display_message("[注意] 画像のエンコードに失敗しました。") return None bio = io.BytesIO(buf.tobytes()) bio.seek(0) return ("camera.jpg", bio, "image/jpeg") except Exception as e: self.display_message(f"[注意] カメラ撮影に失敗: {e}") return None def open_settings(self): cfg = load_config() SettingsDialog(self, cfg) def toggle_login(self): if not self.logged_in: self.login() else: self.logout() def login(self): cfg = load_config() user_id = (cfg.get("user_id") or "").strip() volume_no = (cfg.get("volume_no") or "").strip() serial_no = (cfg.get("serial_no") or "").strip() password = (self._temp_password or cfg.get("password") or "").strip() if not user_id or not password or not volume_no or not serial_no: self.display_message("設定からログイン情報を入力してください。") return try: url = self.base_url + "training_login.php" data = { "user_id": user_id, "user_password": password, "volume_no": volume_no, "serial_no": serial_no, } res = self.post_multipart(url, data) res.raise_for_status() js = res.json() if isinstance(js, dict) and isinstance(js.get("res"), str) and len(js["res"]) == 32: self.token = js["res"] # 旧互換の仕様: s=画面キャプチャ(分) が存在すれば取得 try: if 's' in js: self.middle_interval_minutes = max(1, int(str(js['s']))) if 'k' in js: self.keyboard_interval_minutes = max(1, int(str(js['k']))) if 'm' in js: self.mouse_interval_minutes = max(1, int(str(js['m']))) except Exception: pass self.logged_in = True self.lbl_login.config(text="ログイン中") self.btn_login.config(text="ログアウト") self.load_first() self.message_entry.config(state='normal') self.send_button.config(state='normal') # WebSocketでリアルタイム受信（HTTPポーリングは使わない） self.start_ws() self.display_message("[システム] ログインしました。") self.last_chat_date = "" # チャット日時は空で開始 else: err = js.get("error", "ログインに失敗しました。") if isinstance(js, dict) else "ログインに失敗しました。" self.display_message(f"[エラー] {err}") except requests.exceptions.RequestException as e: self.display_message(f"[エラー] ログイン通信失敗: {e}") except json.JSONDecodeError: self.display_message("[エラー] 応答が不正です。") def logout(self): try: if self.token: url = self.base_url + "training_logout.php" self.post_multipart(url, {"value": self.token}) except Exception: pass self.token = None self.logged_in = False self.lbl_login.config(text="未ログイン") self.btn_login.config(text="ログイン") self.group_label.config(text="") self.notice_text.config(state='normal') self.notice_text.delete("1.0", tk.END) self.notice_text.config(state='disabled') self.links_list.delete(0, tk.END) self._links_map.clear() self.last_chat_date = "" self.stop_polling() self.message_entry.config(state='disabled') self.send_button.config(state='disabled') self.training_started = False self.training_button.config(text="作業開始") self.lbl_start_val.config(text="-") self.lbl_end_val.config(text="-") self.lbl_time_val.config(text="-") self.display_message("[システム] ログアウトしました。") # 念のため停止 self.stop_middle_timer() self.stop_keyboard_timer() self.stop_mouse_timer() self.stop_kbd_mouse_listeners() # WebSocket切断 self.stop_ws() def set_logged_out_state(self): # 起動時にUIを未ログイン状態に整える self.token = None self.logged_in = False self.lbl_login.config(text="未ログイン") self.btn_login.config(text="ログイン") self.message_entry.config(state='disabled') self.send_button.config(state='disabled') # 初期表示（グループ/通知/リンク） def load_first(self): if not self.token: return try: url = self.base_url + "first.php" data = {"value": self.token} res = self.post_multipart(url, data) try: res.raise_for_status() except requests.exceptions.HTTPError as e: # 新プロジェクトでは first.php が存在しない構成（public ルータ運用）のため # 404 は無視して初期表示をスキップし、チャットのみ動かす if res.status_code == 404: return raise js = res.json() group = js.get("group") if isinstance(group, str): self.group_label.config(text=group) notice = js.get("notice") if notice is not None: self.notice_text.config(state='normal') self.notice_text.delete("1.0", tk.END) self.notice_text.insert(tk.END, str(notice)) self.notice_text.config(state='disabled') self.links_list.delete(0, tk.END) self._links_map.clear() links = js.get("link") if isinstance(links, list): for idx, item in enumerate(links): name = item.get("link_name") if isinstance(item, dict) else None url2 = item.get("link_url") if isinstance(item, dict) else None if name and url2: self.links_list.insert(tk.END, name) self._links_map[idx] = url2 except requests.exceptions.RequestException as e: self.display_message(f"エラー（初期表示）: {e}") except json.JSONDecodeError: self.display_message("エラー: 初期データの応答が不正です。") def open_report(self): if not self.token: self.display_message("[注意] ログインしてください。") return url = self.base_url + f"report_edit.php?t={self.token}" webbrowser.open(url) def open_selected_link(self, event=None): sel = self.links_list.curselection() if not sel: return idx = sel[0] url = self._links_map.get(idx) if url: webbrowser.open(url) self.links_list.selection_clear(0, tk.END) def send_inquiry(self): if not self.token: self.display_message("[注意] ログインしてください。") return try: url = self.base_url + "training_inquiry.php" data = {"value": self.token} cam = self.capture_camera() files = {"file": cam} if cam else None res = self.post_multipart(url, data, files) res.raise_for_status() self.display_message("[システム] 問い合せを送信しました。") except requests.exceptions.RequestException as e: self.display_message(f"エラー（問い合せ）: {e}") def toggle_training(self): if not self.token: self.display_message("[注意] ログインしてください。") return try: if not self.training_started: url = self.base_url + "training_start.php" data = {"value": self.token} cam = self.capture_camera() files = {"file": cam} if cam else None res = self.post_multipart(url, data, files) res.raise_for_status() # 応答でトークン更新 js = {} try: js = res.json() except Exception: pass if isinstance(js, dict) and isinstance(js.get("res"), str) and len(js["res"]) == 32: self.token = js["res"] self.training_started = True self.training_start_time = datetime.datetime.now() self.lbl_start_val.config(text=self.training_start_time.strftime("%Y-%m-%d %H:%M:%S")) self.training_button.config(text="作業終了") self.lbl_end_val.config(text="-") self.lbl_time_val.config(text="-") # 画面キャプチャの定期送信を開始 self.start_middle_timer() # 入力反応監視を開始 self.start_kbd_mouse_listeners() self.start_keyboard_timer() self.start_mouse_timer() else: url = self.base_url + "training_end.php" data = {"value": self.token} cam = self.capture_camera() files = {"file": cam} if cam else None res = self.post_multipart(url, data, files) res.raise_for_status() js = {} try: js = res.json() except Exception: pass if isinstance(js, dict) and isinstance(js.get("res"), str) and len(js["res"]) == 32: self.token = js["res"] # サーバー反映前に UI だけ進めないよう、成功を前提 self.training_started = False self.training_end_time = datetime.datetime.now() self.lbl_end_val.config(text=self.training_end_time.strftime("%Y-%m-%d %H:%M:%S")) self.training_button.config(text="作業開始") if self.training_start_time and self.training_end_time: delta = self.training_end_time - self.training_start_time mins = int(delta.total_seconds() // 60) self.lbl_time_val.config(text=f"{mins} 分") # 定期送信を停止 self.stop_middle_timer() self.stop_keyboard_timer() self.stop_mouse_timer() self.stop_kbd_mouse_listeners() except requests.exceptions.RequestException as e: self.display_message(f"エラー（業務）: {e}") def send_message_event(self, event): self.send_message() def send_message(self): message = self.message_entry.get() if not message: return self.message_entry.delete(0, tk.END) if not self.token: self.display_message("[注意] ログインしてください。") return # WebSocketで送信し、永続化のためHTTPにも送信（サーバのRatchetは中継のみのため） sent_ws = False try: if hasattr(self, '_ws') and self._ws is not None: payload = { "type": "chat_insert", "value": self.token, "value2": self.last_chat_date, "value3": message, } self._ws.send(json.dumps(payload)) sent_ws = True except Exception as e: self.display_message(f"[WS送信エラー] {e}") sent_ws = False # HTTPでも保存（管理画面等での一貫性確保） try: url = self.base_url + "training_insert.php" data = { "value": self.token, "value2": self.last_chat_date, "value3": message, } res = self.post_multipart(url, data) res.raise_for_status() except requests.exceptions.RequestException as e: self.display_message(f"エラー（送信）: {e}") def fetch_messages_once(self): if not self.token: return try: url = self.base_url + "training_update.php" data = {"value": self.token, "value2": self.last_chat_date, "value3": ""} res = self.post_multipart(url, data) # デバッグ用途: ステータス確認 try: res.raise_for_status() except requests.exceptions.HTTPError as e: self.display_message(f"[HTTP] {e} body={res.text[:200]}") raise response_data = res.json() if isinstance(response_data, list): for item in response_data: user_name = item.get("user_name") admin_name = item.get("admin_name") insert_date = item.get("insert_date") chat_text = item.get("chat_text") display_text = "" if user_name: display_text += f"{user_name}>" elif admin_name: display_text += f"{admin_name}>" if insert_date: display_text += f"{insert_date}\n" if chat_text: display_text += f"{chat_text}" if display_text: self.display_message(display_text) # C# 版は常に空文字で更新しているため、同様の挙動に合わせる # self.last_chat_date は更新しない（＝常に全件取得） except requests.exceptions.RequestException as e: self.display_message(f"エラー（受信）: {e}") except json.JSONDecodeError: self.display_message("エラー: サーバーからの応答が不正です。") def start_polling(self): # フォールバックのHTTPポーリングは使用しない self.stop_polling() def stop_polling(self): if self._poll_after_id is not None: try: self.after_cancel(self._poll_after_id) except Exception: pass self._poll_after_id = None def display_message(self, message): self.chat_display.config(state='normal') self.chat_display.insert(tk.END, message + "\n") self.chat_display.yview(tk.END) self.chat_display.config(state='disabled') # --- WebSocket リアルタイム受信 --- def _derive_ws_url(self): """base_url から ws://:8080/ を推定（Ratchet IoServer 8080）""" try: from urllib.parse import urlparse u = urlparse(self.base_url) host = u.hostname or 'localhost' except Exception: host = 'localhost' # Ratchetサーバは平文WS: ポート8080 return f"ws://{host}:8080/" def start_ws(self): # 既存の接続を停止 self.stop_ws() # 依存の確認 try: import websocket # websocket-client except Exception as e: self.display_message(f"[注意] WebSocket未対応: websocket-client をインストールしてください ({e})") return ws_url = self._derive_ws_url() def on_message(ws, message): def handle(): try: body = json.loads(message) except Exception: self.display_message(str(message)) return # サーバのRatchetは受信したJSONをそのまま中継する想定 # 2系統を許容: 既存API形式 or WS送信用の簡易形式 items = body if isinstance(body, list) else [body] for item in items: if not isinstance(item, dict): self.display_message(str(item)) continue # 既存API形式 user_name = item.get('user_name') admin_name = item.get('admin_name') insert_date = item.get('insert_date') chat_text = item.get('chat_text') # 簡易形式（WS送信時の自前ペイロード） if not chat_text and item.get('type') == 'chat_insert': chat_text = item.get('value3') # 表示名は不明のためユーザー名があれば流用 if not user_name and 'user' in item: user_name = item.get('user') display_text = '' if user_name: display_text += f"{user_name}>" elif admin_name: display_text += f"{admin_name}>" if insert_date: display_text += f"{insert_date}\n" if chat_text: display_text += f"{chat_text}" if display_text: self.display_message(display_text) try: self.after(0, handle) except Exception: pass def on_error(ws, error): try: self.after(0, lambda: self.display_message(f"[WSエラー] {error}")) except Exception: pass def on_close(ws, *args): try: self.after(0, lambda: self.display_message("[WS] 接続が閉じられました")) except Exception: pass def on_open(ws): try: self.after(0, lambda: self.display_message("[WS] 接続しました")) except Exception: pass def run_ws(): try: self._ws = websocket.WebSocketApp( ws_url, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open, ) self._ws.run_forever() except Exception as e: try: self.after(0, lambda: self.display_message(f"[WS例外] {e}")) except Exception: pass self._ws_thread = threading.Thread(target=run_ws, daemon=True) self._ws_thread.start() def stop_ws(self): try: if self._ws is not None: self._ws.close() except Exception: pass self._ws = None t = self._ws_thread self._ws_thread = None if t and t.is_alive(): try: t.join(timeout=0.5) except Exception: pass # --- 画面キャプチャ（中間） --- def capture_screen_file(self): """デスクトップ全体をキャプチャしてJPEG bytesIOを返す。失敗時はNone。""" # できれば mss、無ければ Pillow(ImageGrab) を試す # 320x240 に縮小 try: import mss import mss.tools from PIL import Image except Exception: try: from PIL import ImageGrab, Image except Exception as e: self.display_message(f"[注意] 画面キャプチャ未対応: mss または pillow をインストールしてください ({e})") return None try: img = ImageGrab.grab() img = img.resize((320, 240)) bio = io.BytesIO() img.save(bio, format='JPEG') bio.seek(0) return ("capture.jpg", bio, "image/jpeg") except Exception as e: self.display_message(f"[注意] 画面キャプチャ失敗: {e}") return None # mss がある場合 try: with mss.mss() as sct: monitor = sct.monitors[0] img = sct.grab(monitor) from PIL import Image im = Image.frombytes('RGB', img.size, img.rgb) im = im.resize((320, 240)) bio = io.BytesIO() im.save(bio, format='JPEG') bio.seek(0) return ("capture.jpg", bio, "image/jpeg") except Exception as e: self.display_message(f"[注意] 画面キャプチャ失敗(mss): {e}") return None def tick_middle(self): if not self.logged_in or not self.training_started or not self.token: return try: url = self.base_url + "training_middle.php" cap = self.capture_screen_file() files = {"file": cap} if cap else None self.post_multipart(url, {"value": self.token}, files) except Exception: pass finally: # 次回スケジュール millis = max(1, int(self.middle_interval_minutes)) * 60 * 1000 self._middle_after_id = self.after(millis, self.tick_middle) def start_middle_timer(self): self.stop_middle_timer() # 初回はすぐ実行 self.tick_middle() def stop_middle_timer(self): if self._middle_after_id is not None: try: self.after_cancel(self._middle_after_id) except Exception: pass self._middle_after_id = None # --- 入力反応監視（キーボード/マウス） --- def start_kbd_mouse_listeners(self): try: from pynput import keyboard, mouse except Exception as e: self.display_message(f"[注意] 入力監視未対応: pynput をインストールしてください ({e})") return # ハンドラ def on_key_press(key): self.keyboard_flag = '1' def on_key_release(key): self.keyboard_flag = '1' def on_move(x, y): self.mouse_flag = '1' def on_click(x, y, button, pressed): self.mouse_flag = '1' def on_scroll(x, y, dx, dy): self.mouse_flag = '1' try: self._kbd_listener = keyboard.Listener(on_press=on_key_press, on_release=on_key_release) self._mouse_listener = mouse.Listener(on_move=on_move, on_click=on_click, on_scroll=on_scroll) self._kbd_listener.daemon = True self._mouse_listener.daemon = True self._kbd_listener.start() self._mouse_listener.start() except Exception as e: self.display_message(f"[注意] 入力監視開始に失敗: {e}") def stop_kbd_mouse_listeners(self): try: if self._kbd_listener is not None: self._kbd_listener.stop() except Exception: pass try: if self._mouse_listener is not None: self._mouse_listener.stop() except Exception: pass self._kbd_listener = None self._mouse_listener = None def tick_keyboard(self): if not self.logged_in or not self.training_started or not self.token: return try: url = self.base_url + "training_keyboard.php" self.post_multipart(url, {"value": self.token, "value2": self.keyboard_flag}) except Exception: pass finally: # 反応フラグをリセット self.keyboard_flag = '2' # 次回スケジュール millis = max(1, int(self.keyboard_interval_minutes)) * 60 * 1000 self._kbd_after_id = self.after(millis, self.tick_keyboard) def start_keyboard_timer(self): self.stop_keyboard_timer() self.tick_keyboard() def stop_keyboard_timer(self): if self._kbd_after_id is not None: try: self.after_cancel(self._kbd_after_id) except Exception: pass self._kbd_after_id = None def tick_mouse(self): if not self.logged_in or not self.training_started or not self.token: return try: url = self.base_url + "training_mouse.php" self.post_multipart(url, {"value": self.token, "value2": self.mouse_flag}) except Exception: pass finally: self.mouse_flag = '2' millis = max(1, int(self.mouse_interval_minutes)) * 60 * 1000 self._mouse_after_id = self.after(millis, self.tick_mouse) def start_mouse_timer(self): self.stop_mouse_timer() self.tick_mouse() def stop_mouse_timer(self): if self._mouse_after_id is not None: try: self.after_cancel(self._mouse_after_id) except Exception: pass self._mouse_after_id = None if __name__ == "__main__": app = MainWindow() app.mainloop()