Spaces:
Runtime error
Runtime error
| """ | |
| Hugging Face Spaces永続ストレージ対応ユーザー管理システム | |
| Cookieベースのユーザー識別と/mnt/dataでの状態永続化を提供 | |
| """ | |
| import os | |
| import json | |
| import uuid | |
| import logging | |
| from datetime import datetime, timedelta | |
| from typing import Optional, Dict, Any, List | |
| import streamlit as st | |
| from streamlit_cookies_manager import EncryptedCookieManager | |
| logger = logging.getLogger(__name__) | |
| class PersistentUserManager: | |
| """永続ストレージ対応ユーザー管理クラス""" | |
| def __init__(self, storage_base_path: str = "/mnt/data"): | |
| """ | |
| 初期化 | |
| Args: | |
| storage_base_path: 永続ストレージのベースパス(HF Spacesでは/mnt/data) | |
| """ | |
| # Hugging Face Spacesの永続ストレージパス | |
| self.storage_base_path = storage_base_path | |
| self.user_data_dir = os.path.join(storage_base_path, "mari_users") | |
| self.session_data_dir = os.path.join(storage_base_path, "mari_sessions") | |
| # 永続Cookie管理設定(完全にユニークなキー名) | |
| self.cookie_name = "hf_spaces_persistent_session_id" # HF Spaces専用の完全にユニークなキー | |
| self.cookie_expiry_days = 365 # 1年間有効な永続Cookie | |
| # ディレクトリ作成 | |
| self._ensure_directories() | |
| # Cookie管理の遅延初期化(初回使用時に初期化) | |
| self.cookies = None | |
| self._cookie_initialized = False | |
| logger.info(f"永続ユーザー管理システム初期化: {self.user_data_dir}") | |
| def _ensure_directories(self): | |
| """必要なディレクトリを作成""" | |
| try: | |
| os.makedirs(self.user_data_dir, exist_ok=True) | |
| os.makedirs(self.session_data_dir, exist_ok=True) | |
| logger.info(f"ストレージディレクトリ確認完了: {self.user_data_dir}") | |
| except Exception as e: | |
| logger.error(f"ディレクトリ作成エラー: {e}") | |
| # フォールバック: ローカルディレクトリを使用 | |
| self.user_data_dir = "local_mari_users" | |
| self.session_data_dir = "local_mari_sessions" | |
| os.makedirs(self.user_data_dir, exist_ok=True) | |
| os.makedirs(self.session_data_dir, exist_ok=True) | |
| logger.warning(f"フォールバック: ローカルディレクトリを使用 {self.user_data_dir}") | |
| def _ensure_cookie_manager(self, force_init: bool = False): | |
| """Cookie管理システムの初期化(永続Cookie対応)""" | |
| # 既に初期化済みで、強制初期化でない場合はスキップ | |
| if self._cookie_initialized and not force_init and self.cookies is not None: | |
| logger.debug("Cookie管理システム既に初期化済み - スキップ") | |
| return | |
| # 固定キーでCookie初期化チェック(永続化対応) | |
| browser_session_key = "hf_persistent_cookie_initialized" | |
| # より厳密な重複チェック | |
| if (not force_init and | |
| st.session_state.get(browser_session_key, False) and | |
| self.cookies is not None and | |
| hasattr(self.cookies, 'ready') and | |
| self.cookies.ready()): | |
| self._cookie_initialized = True | |
| logger.debug("永続Cookie管理システム既に準備完了 - スキップ") | |
| return | |
| # 初期化処理中フラグを設定(重複初期化防止) | |
| init_in_progress_key = "hf_persistent_cookie_initializing" | |
| if st.session_state.get(init_in_progress_key, False): | |
| logger.warning("永続Cookie初期化処理中 - 重複初期化を防止") | |
| return | |
| st.session_state[init_in_progress_key] = True | |
| try: | |
| logger.info("HF Spaces永続Cookie管理システム初期化開始(固定prefix使用)...") | |
| # セキュアなパスワードを生成(環境変数から取得、なければ生成) | |
| cookie_password = os.getenv("MARI_COOKIE_PASSWORD", "mari_chat_secure_key_2024") | |
| # 固定のprefixを使用(永続Cookie読み込みのため) | |
| # リロード後も同じCookieを読み込めるよう、固定のprefixを使用 | |
| fixed_prefix = "hf_persistent_mari_" | |
| logger.debug(f"Cookie初期化: prefix={fixed_prefix} (固定prefix使用)") | |
| # EncryptedCookieManagerを初期化(永続Cookie対応) | |
| cookies = EncryptedCookieManager( | |
| prefix=fixed_prefix, # 固定prefixで永続Cookie読み込み対応 | |
| password=cookie_password | |
| ) | |
| # Cookieが準備できるまで待機(より長い待機時間で既存Cookie読み込み対応) | |
| max_attempts = 10 # 試行回数を増加 | |
| total_wait_time = 0 | |
| max_wait_time = 3.0 # 最大3秒待機 | |
| for attempt in range(max_attempts): | |
| if cookies.ready(): | |
| logger.info(f"Cookie準備完了 ({attempt + 1}回目の試行で成功)") | |
| break | |
| wait_time = 0.2 + (attempt * 0.1) # 段階的に待機時間を延長 | |
| total_wait_time += wait_time | |
| if total_wait_time > max_wait_time: | |
| logger.warning(f"Cookie準備タイムアウト ({total_wait_time:.1f}秒経過)") | |
| break | |
| logger.debug(f"Cookie準備待機中... ({attempt + 1}/{max_attempts}, {wait_time:.1f}秒待機)") | |
| import time | |
| time.sleep(wait_time) | |
| # Cookie準備ができていない場合でも、一度試行してみる | |
| if not cookies.ready(): | |
| logger.warning("Cookie準備未完了だが、既存Cookie読み込みを試行") | |
| # フォールバックモードにせず、Cookieオブジェクトは保持 | |
| # 既存のCookieがある場合は読み込める可能性がある | |
| self.cookies = cookies | |
| self._cookie_initialized = True | |
| # 永続化フラグを設定 | |
| st.session_state[browser_session_key] = True | |
| # 初期化処理中フラグをクリア | |
| if init_in_progress_key in st.session_state: | |
| del st.session_state[init_in_progress_key] | |
| # 既存Cookieの確認 | |
| try: | |
| existing_cookies = dict(self.cookies) | |
| logger.info(f"HF Spaces永続Cookie管理システム初期化完了 - prefix: {fixed_prefix}, 既存Cookie数: {len(existing_cookies)}") | |
| # 既存のユーザーIDがあるかチェック | |
| if self.cookie_name in existing_cookies: | |
| existing_user_id = existing_cookies[self.cookie_name] | |
| logger.info(f"既存の永続Cookie発見: ユーザーID {existing_user_id[:8]}...") | |
| else: | |
| logger.info("新規ユーザー: 永続Cookieなし") | |
| except Exception as cookie_check_e: | |
| logger.debug(f"Cookie確認エラー(通常動作に影響なし): {cookie_check_e}") | |
| logger.info(f"HF Spaces永続Cookie管理システム初期化完了 - prefix: {fixed_prefix}") | |
| except Exception as e: | |
| logger.error(f"永続Cookie管理初期化エラー: {e}") | |
| # 初期化処理中フラグをクリア | |
| if init_in_progress_key in st.session_state: | |
| del st.session_state[init_in_progress_key] | |
| # フォールバック: セッション状態のみ使用 | |
| self.cookies = None | |
| self._cookie_initialized = True | |
| st.session_state[browser_session_key] = True | |
| def get_or_create_user_id(self, force_reset: bool = False) -> str: | |
| """ | |
| Cookie認証ベースのユーザーID取得(ブラウザごとに一つのCookie) | |
| Args: | |
| force_reset: フルリセット時のみTrue | |
| Returns: | |
| ユーザーID | |
| """ | |
| try: | |
| # 重複呼び出し防止: 処理中フラグをチェック | |
| if st.session_state.get('user_id_processing', False): | |
| logger.debug("ユーザーID取得処理中 - 待機") | |
| # 既存のuser_idがあればそれを返す | |
| if 'user_id' in st.session_state: | |
| return st.session_state.user_id | |
| # なければ一時的なIDを返す | |
| return f"temp_{id(st.session_state)}" | |
| # 処理中フラグを設定 | |
| st.session_state.user_id_processing = True | |
| try: | |
| # Cookie管理システムを初期化(最優先) | |
| self._ensure_cookie_manager(force_init=force_reset) | |
| # 1. CookieからユーザーIDを取得(最優先) | |
| user_id = self._get_user_id_from_cookie() | |
| if user_id and self._is_valid_user_id(user_id): | |
| # 有効なCookieベースのユーザーIDが存在 | |
| self._update_user_access_time(user_id) | |
| st.session_state.user_id = user_id # セッション状態に保存 | |
| logger.info(f"HF Spaces永続Cookie認証成功: {user_id[:8]}...") | |
| return user_id | |
| # 2. フルリセット時以外で、セッション状態にuser_idがある場合はCookieと照合 | |
| if not force_reset and 'user_id' in st.session_state: | |
| existing_id = st.session_state.user_id | |
| if existing_id and self._is_valid_uuid(existing_id) and self._is_valid_user_id(existing_id): | |
| # セッション状態のIDが有効な場合、Cookieにも保存 | |
| self._set_user_id_cookie(existing_id) | |
| logger.info(f"セッション状態からユーザーID復元: {existing_id[:8]}...") | |
| return existing_id | |
| # 2. Cookieが無効または存在しない場合は新規作成 | |
| if force_reset or not user_id: | |
| user_id = self._create_new_user_with_cookie() | |
| logger.info(f"新規Cookie認証ユーザー作成: {user_id[:8]}...") | |
| return user_id | |
| # 3. 最終フォールバック(Cookieが無効だが何らかのIDがある場合) | |
| if user_id and self._is_valid_uuid(user_id): | |
| # UUIDとして有効だが、ユーザーファイルが存在しない場合は新規作成 | |
| user_id = self._create_new_user_with_cookie() | |
| logger.info(f"フォールバック新規ユーザー作成: {user_id[:8]}...") | |
| return user_id | |
| # 4. 完全フォールバック(Cookie無効時) | |
| logger.warning("Cookie認証失敗 - 一時的なセッションIDを使用") | |
| temp_id = str(uuid.uuid4()) | |
| st.session_state.user_id = temp_id | |
| return temp_id | |
| finally: | |
| # 処理中フラグをクリア | |
| st.session_state.user_id_processing = False | |
| except Exception as e: | |
| logger.error(f"Cookie認証エラー: {e}") | |
| # 処理中フラグをクリア | |
| st.session_state.user_id_processing = False | |
| # 完全フォールバック: 一時的なIDを生成(Cookieに保存しない) | |
| temp_id = str(uuid.uuid4()) | |
| st.session_state.user_id = temp_id | |
| logger.warning(f"一時的なセッションID使用: {temp_id[:8]}...") | |
| return temp_id | |
| def _get_user_id_from_cookie(self) -> Optional[str]: | |
| """CookieからユーザーIDを取得(永続Cookie対応・準備未完了でも試行)""" | |
| try: | |
| # Cookie管理の初期化(初回のみ) | |
| self._ensure_cookie_manager() | |
| if self.cookies is None: | |
| logger.debug("Cookie管理システム無効 - None返却") | |
| return None | |
| # Cookie準備状態をチェックするが、準備未完了でも読み込みを試行 | |
| is_ready = self.cookies.ready() | |
| if not is_ready: | |
| logger.debug("Cookie準備未完了だが、既存Cookie読み込みを試行") | |
| else: | |
| logger.debug("Cookie準備完了 - 通常読み込み") | |
| # 既存Cookieの読み込みを試行(準備状態に関係なく) | |
| try: | |
| # 全てのCookieを確認(デバッグ用) | |
| all_cookies = dict(self.cookies) | |
| logger.debug(f"利用可能なCookie: {list(all_cookies.keys())} (準備状態: {is_ready})") | |
| user_id = self.cookies.get(self.cookie_name) | |
| if user_id: | |
| logger.debug(f"Cookie値取得: {user_id[:8] if len(user_id) >= 8 else user_id}...") | |
| if self._is_valid_uuid(user_id): | |
| logger.info(f"HF Spaces永続CookieからユーザーID取得成功: {user_id[:8]}... (キー: {self.cookie_name}, 準備状態: {is_ready})") | |
| return user_id | |
| else: | |
| logger.warning(f"Cookie内のユーザーIDが無効な形式: {user_id}") | |
| return None | |
| logger.debug(f"Cookie内に有効なユーザーIDなし (準備状態: {is_ready})") | |
| return None | |
| except Exception as cookie_read_e: | |
| logger.warning(f"Cookie読み込み試行エラー: {cookie_read_e}") | |
| return None | |
| except Exception as e: | |
| logger.warning(f"Cookie取得エラー: {e}") | |
| return None | |
| def _set_user_id_cookie(self, user_id: str): | |
| """ユーザーIDを永続的なクッキーに設定(準備未完了でも試行)""" | |
| try: | |
| # Cookie管理システムが初期化されていない場合はスキップ | |
| if not self._cookie_initialized: | |
| logger.debug("Cookie管理システム未初期化 - Cookie設定スキップ") | |
| return | |
| if self.cookies is None: | |
| logger.debug("Cookie管理システム無効 - Cookie設定スキップ") | |
| return | |
| # Cookie準備状態をチェックするが、準備未完了でも設定を試行 | |
| is_ready = self.cookies.ready() | |
| if not is_ready: | |
| logger.debug("Cookie準備未完了だが、Cookie設定を試行") | |
| # 永続的なCookieを設定 | |
| self.cookies[self.cookie_name] = user_id | |
| # Cookieを保存(streamlit-cookies-managerは自動的に永続化される) | |
| self.cookies.save() | |
| logger.info(f"HF Spaces永続クッキーに保存: {user_id[:8]}... (キー: {self.cookie_name}, 準備状態: {is_ready})") | |
| except Exception as e: | |
| logger.warning(f"永続Cookie設定エラー: {e}") | |
| # エラーが発生してもアプリケーションは継続 | |
| def _is_valid_uuid(self, uuid_string: str) -> bool: | |
| """UUIDの形式チェック""" | |
| try: | |
| uuid.UUID(uuid_string, version=4) | |
| return True | |
| except (ValueError, TypeError): | |
| return False | |
| def _is_valid_user_id(self, user_id: str) -> bool: | |
| """ユーザーIDの有効性チェック""" | |
| try: | |
| if not self._is_valid_uuid(user_id): | |
| return False | |
| user_file = os.path.join(self.user_data_dir, f"{user_id}.json") | |
| return os.path.exists(user_file) | |
| except Exception as e: | |
| logger.warning(f"ユーザーID検証エラー: {e}") | |
| return False | |
| def _create_new_user_with_cookie(self) -> str: | |
| """Cookie認証ベースの新規ユーザーを作成""" | |
| try: | |
| user_id = str(uuid.uuid4()) | |
| # ユーザーデータを作成 | |
| user_data = { | |
| "user_id": user_id, | |
| "created_at": datetime.now().isoformat(), | |
| "last_access": datetime.now().isoformat(), | |
| "version": "1.0", | |
| "browser_fingerprint": self._generate_browser_fingerprint(), | |
| "game_data": { | |
| "affection": 30, | |
| "messages": [{"role": "assistant", "content": "何の用?遊びに来たの?", "is_initial": True}], | |
| "scene_params": {"theme": "default"}, | |
| "ura_mode": False | |
| }, | |
| "settings": { | |
| "notifications_enabled": True, | |
| "auto_save": True | |
| } | |
| } | |
| # ファイルに保存 | |
| user_file = os.path.join(self.user_data_dir, f"{user_id}.json") | |
| with open(user_file, 'w', encoding='utf-8') as f: | |
| json.dump(user_data, f, ensure_ascii=False, indent=2) | |
| # Cookieに保存(ブラウザごとに一つ) | |
| self._set_user_id_cookie(user_id) | |
| # セッション状態に保存 | |
| st.session_state.user_id = user_id | |
| logger.info(f"Cookie認証ベース新規ユーザー作成完了: {user_id[:8]}...") | |
| return user_id | |
| except Exception as e: | |
| logger.error(f"Cookie認証ベース新規ユーザー作成エラー: {e}") | |
| # フォールバック: 一時的なIDを生成(Cookieに保存しない) | |
| temp_id = str(uuid.uuid4()) | |
| st.session_state.user_id = temp_id | |
| logger.warning(f"フォールバック一時ID: {temp_id[:8]}...") | |
| return temp_id | |
| def _create_new_user(self) -> str: | |
| """従来の新規ユーザー作成(後方互換性のため残す)""" | |
| return self._create_new_user_with_cookie() | |
| def _generate_browser_fingerprint(self) -> str: | |
| """ブラウザフィンガープリントを生成(簡易版)""" | |
| try: | |
| # Streamlitのセッション情報を使用してフィンガープリントを生成 | |
| import hashlib | |
| # セッション固有の情報を組み合わせ | |
| session_info = f"{id(st.session_state)}_{datetime.now().strftime('%Y%m%d')}" | |
| fingerprint = hashlib.md5(session_info.encode()).hexdigest()[:16] | |
| logger.debug(f"ブラウザフィンガープリント生成: {fingerprint}") | |
| return fingerprint | |
| except Exception as e: | |
| logger.warning(f"ブラウザフィンガープリント生成エラー: {e}") | |
| return "unknown_browser" | |
| def _update_user_access_time(self, user_id: str): | |
| """ユーザーの最終アクセス時刻を更新""" | |
| try: | |
| user_file = os.path.join(self.user_data_dir, f"{user_id}.json") | |
| if os.path.exists(user_file): | |
| with open(user_file, 'r', encoding='utf-8') as f: | |
| user_data = json.load(f) | |
| user_data["last_access"] = datetime.now().isoformat() | |
| with open(user_file, 'w', encoding='utf-8') as f: | |
| json.dump(user_data, f, ensure_ascii=False, indent=2) | |
| logger.debug(f"ユーザーアクセス時刻更新: {user_id[:8]}...") | |
| except Exception as e: | |
| logger.warning(f"アクセス時刻更新エラー: {e}") | |
| def load_user_game_data(self, user_id: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| ユーザーのゲームデータを読み込み | |
| Args: | |
| user_id: ユーザーID | |
| Returns: | |
| ゲームデータ(存在しない場合はNone) | |
| """ | |
| try: | |
| user_file = os.path.join(self.user_data_dir, f"{user_id}.json") | |
| if not os.path.exists(user_file): | |
| logger.info(f"ユーザーファイルが存在しません: {user_id[:8]}...") | |
| return None | |
| with open(user_file, 'r', encoding='utf-8') as f: | |
| user_data = json.load(f) | |
| game_data = user_data.get("game_data", {}) | |
| logger.info(f"ゲームデータ読み込み完了: {user_id[:8]}... (データサイズ: {len(str(game_data))}文字)") | |
| return game_data | |
| except Exception as e: | |
| logger.error(f"ゲームデータ読み込みエラー: {e}") | |
| return None | |
| def save_user_game_data(self, user_id: str, game_data: Dict[str, Any]) -> bool: | |
| """ | |
| ユーザーのゲームデータを永続ストレージ(/mnt/data)に保存 | |
| Args: | |
| user_id: ユーザーID | |
| game_data: 保存するゲームデータ | |
| Returns: | |
| 保存成功時True | |
| """ | |
| try: | |
| user_file = os.path.join(self.user_data_dir, f"{user_id}.json") | |
| # 保存先パスをログ出力(永続ストレージ確認用) | |
| logger.info(f"ゲームデータ保存先: {user_file}") | |
| logger.info(f"永続ストレージベースパス: {self.storage_base_path}") | |
| # 既存データを読み込み | |
| if os.path.exists(user_file): | |
| with open(user_file, 'r', encoding='utf-8') as f: | |
| user_data = json.load(f) | |
| logger.info(f"既存ユーザーデータを更新: {user_id[:8]}...") | |
| else: | |
| # 新規ユーザーデータを作成 | |
| user_data = { | |
| "user_id": user_id, | |
| "created_at": datetime.now().isoformat(), | |
| "version": "1.0" | |
| } | |
| logger.info(f"新規ユーザーデータを作成: {user_id[:8]}...") | |
| # ゲームデータを更新 | |
| user_data["game_data"] = game_data | |
| user_data["last_access"] = datetime.now().isoformat() | |
| user_data["last_save"] = datetime.now().isoformat() | |
| # ファイルに保存 | |
| with open(user_file, 'w', encoding='utf-8') as f: | |
| json.dump(user_data, f, ensure_ascii=False, indent=2) | |
| # 保存後のファイルサイズを確認 | |
| file_size = os.path.getsize(user_file) | |
| logger.info(f"永続ストレージにゲームデータ保存完了: {user_id[:8]}... (ファイルサイズ: {file_size}バイト)") | |
| return True | |
| except Exception as e: | |
| logger.error(f"永続ストレージへのゲームデータ保存エラー: {e}") | |
| return False | |
| def get_user_info(self, user_id: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| ユーザー情報を取得 | |
| Args: | |
| user_id: ユーザーID | |
| Returns: | |
| ユーザー情報 | |
| """ | |
| try: | |
| user_file = os.path.join(self.user_data_dir, f"{user_id}.json") | |
| if os.path.exists(user_file): | |
| with open(user_file, 'r', encoding='utf-8') as f: | |
| return json.load(f) | |
| return None | |
| except Exception as e: | |
| logger.error(f"ユーザー情報取得エラー: {e}") | |
| return None | |
| def delete_user_data(self, user_id: str) -> bool: | |
| """ | |
| ユーザーデータを削除 | |
| Args: | |
| user_id: 削除するユーザーID | |
| Returns: | |
| 削除成功時True | |
| """ | |
| try: | |
| user_file = os.path.join(self.user_data_dir, f"{user_id}.json") | |
| if os.path.exists(user_file): | |
| os.remove(user_file) | |
| logger.info(f"ユーザーデータ削除: {user_id[:8]}...") | |
| # 永続Cookieも削除 | |
| try: | |
| # Cookie管理の強制初期化(フルリセット時) | |
| self._ensure_cookie_manager(force_init=True) | |
| if self.cookies: | |
| if self.cookie_name in self.cookies: | |
| # 永続Cookieを削除(有効期限を過去に設定) | |
| del self.cookies[self.cookie_name] | |
| self.cookies.save() | |
| logger.info(f"HF Spaces永続Cookie削除完了 (キー: {self.cookie_name})") | |
| except Exception as e: | |
| logger.warning(f"HF Spaces永続Cookie削除エラー: {e}") | |
| # セッション状態からも削除 | |
| if 'persistent_user_id' in st.session_state: | |
| del st.session_state.persistent_user_id | |
| if 'persistent_user_id_checked' in st.session_state: | |
| del st.session_state.persistent_user_id_checked | |
| return True | |
| except Exception as e: | |
| logger.error(f"ユーザーデータ削除エラー: {e}") | |
| return False | |
| def full_reset_user_session(self) -> str: | |
| """ | |
| フルリセット(Cookie初期化を含む) | |
| Returns: | |
| 新しいユーザーID | |
| """ | |
| try: | |
| logger.info("フルリセット開始 - Cookie初期化を含む") | |
| # セッション状態をクリア | |
| if 'persistent_user_id' in st.session_state: | |
| del st.session_state.persistent_user_id | |
| if 'persistent_user_id_checked' in st.session_state: | |
| del st.session_state.persistent_user_id_checked | |
| if 'cookie_manager_initialized' in st.session_state: | |
| del st.session_state.cookie_manager_initialized | |
| # Cookie初期化フラグをリセット | |
| self._cookie_initialized = False | |
| self.cookies = None | |
| # 新しいユーザーIDを作成(強制リセット) | |
| new_user_id = self.get_or_create_user_id(force_reset=True) | |
| logger.info(f"フルリセット完了: {new_user_id[:8]}...") | |
| return new_user_id | |
| except Exception as e: | |
| logger.error(f"フルリセットエラー: {e}") | |
| # フォールバック | |
| import uuid | |
| fallback_id = str(uuid.uuid4()) | |
| st.session_state.persistent_user_id = fallback_id | |
| return fallback_id | |
| def list_all_users(self) -> List[Dict[str, Any]]: | |
| """ | |
| 全ユーザーの一覧を取得(管理用) | |
| Returns: | |
| ユーザー情報のリスト | |
| """ | |
| try: | |
| users = [] | |
| for filename in os.listdir(self.user_data_dir): | |
| if filename.endswith('.json'): | |
| user_file = os.path.join(self.user_data_dir, filename) | |
| try: | |
| with open(user_file, 'r', encoding='utf-8') as f: | |
| user_data = json.load(f) | |
| # 基本情報のみ抽出 | |
| user_info = { | |
| "user_id": user_data.get("user_id", "unknown")[:8] + "...", | |
| "created_at": user_data.get("created_at", "unknown"), | |
| "last_access": user_data.get("last_access", "unknown"), | |
| "has_game_data": "game_data" in user_data, | |
| "file_size": os.path.getsize(user_file) | |
| } | |
| users.append(user_info) | |
| except Exception as e: | |
| logger.warning(f"ユーザーファイル読み込みエラー {filename}: {e}") | |
| return users | |
| except Exception as e: | |
| logger.error(f"ユーザー一覧取得エラー: {e}") | |
| return [] | |
| def cleanup_old_users(self, days_threshold: int = 30) -> int: | |
| """ | |
| 古いユーザーデータをクリーンアップ | |
| Args: | |
| days_threshold: 削除対象の日数閾値 | |
| Returns: | |
| 削除されたユーザー数 | |
| """ | |
| try: | |
| current_time = datetime.now() | |
| deleted_count = 0 | |
| for filename in os.listdir(self.user_data_dir): | |
| if filename.endswith('.json'): | |
| user_file = os.path.join(self.user_data_dir, filename) | |
| try: | |
| with open(user_file, 'r', encoding='utf-8') as f: | |
| user_data = json.load(f) | |
| last_access = datetime.fromisoformat(user_data.get("last_access", "")) | |
| if (current_time - last_access).days > days_threshold: | |
| os.remove(user_file) | |
| deleted_count += 1 | |
| logger.info(f"古いユーザーデータ削除: {filename}") | |
| except Exception as e: | |
| logger.warning(f"クリーンアップ処理エラー {filename}: {e}") | |
| logger.info(f"ユーザーデータクリーンアップ完了: {deleted_count}件削除") | |
| return deleted_count | |
| except Exception as e: | |
| logger.error(f"クリーンアップエラー: {e}") | |
| return 0 | |
| def get_storage_stats(self) -> Dict[str, Any]: | |
| """ | |
| ストレージ使用状況を取得 | |
| Returns: | |
| ストレージ統計情報 | |
| """ | |
| try: | |
| stats = { | |
| "user_count": 0, | |
| "total_size": 0, | |
| "storage_path": self.user_data_dir, | |
| "cookie_enabled": self.cookies is not None | |
| } | |
| if os.path.exists(self.user_data_dir): | |
| for filename in os.listdir(self.user_data_dir): | |
| if filename.endswith('.json'): | |
| user_file = os.path.join(self.user_data_dir, filename) | |
| stats["user_count"] += 1 | |
| stats["total_size"] += os.path.getsize(user_file) | |
| return stats | |
| except Exception as e: | |
| logger.error(f"ストレージ統計取得エラー: {e}") | |
| return {"error": str(e)} |