Update modules/utils.py

modules/utils.py

@@ -5,16 +5,69 @@ import base64
 import hmac
 import hashlib
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, List
 
-# ====== 共通ディレクトリの用意 ======
-DATA_DIR = Path(os.getenv("DATA_DIR", "data"))
-EXPORT_DIR = DATA_DIR / "exports"
+# ------------------------------------------------------------
+# 書き込みディレクトリの自動選択（/data → /cache → /tmp → 最後に ./data）
+# ------------------------------------------------------------
+_DATA_DIR: Optional[Path] = None
+_EXPORT_DIR: Optional[Path] = None
+
+def _is_writable(p: Path) -> bool:
+    try:
+        p.mkdir(parents=True, exist_ok=True)
+        testfile = p / ".w_test"
+        with open(testfile, "w", encoding="utf-8") as f:
+            f.write("ok")
+        testfile.unlink(missing_ok=True)
+        return True
+    except Exception:
+        return False
+
+def _pick_writable_dir(candidates: List[Path]) -> Path:
+    for p in candidates:
+        if _is_writable(p):
+            return p
+    # 全滅時は最終手段として /tmp に落とす
+    fallback = Path("/tmp/agent_studio")
+    fallback.mkdir(parents=True, exist_ok=True)
+    return fallback
 
 def ensure_dirs() -> None:
-    """必要なディレクトリを作成"""
-    DATA_DIR.mkdir(parents=True, exist_ok=True)
-    EXPORT_DIR.mkdir(parents=True, exist_ok=True)
+    """書き込み可能なデータディレクトリを決定して作成。"""
+    global _DATA_DIR, _EXPORT_DIR
+
+    if _DATA_DIR is not None and _EXPORT_DIR is not None:
+        return  # 既に確定済み
+
+    # 優先順: 環境変数 DATA_DIR → /data → /cache → /tmp → ./data
+    env_dir = os.getenv("DATA_DIR")
+    candidates = []
+    if env_dir:
+        candidates.append(Path(env_dir))
+    candidates.extend([
+        Path("/data/agent_studio"),
+        Path("/cache/agent_studio"),
+        Path("/tmp/agent_studio"),
+        Path("data"),  # 最後に相対パス
+    ])
+
+    chosen = _pick_writable_dir(candidates)
+    export = chosen / "exports"
+
+    chosen.mkdir(parents=True, exist_ok=True)
+    export.mkdir(parents=True, exist_ok=True)
+
+    _DATA_DIR = chosen
+    _EXPORT_DIR = export
+
+def data_dir() -> Path:
+    ensure_dirs()
+    return _DATA_DIR  # type: ignore
+
+def export_dir() -> Path:
+    ensure_dirs()
+    return _EXPORT_DIR  # type: ignore
 
 # ====== 簡易トラッキングトークン（追加依存なしのHMAC方式） ======
 _SECRET = os.getenv("TRACKING_SECRET", "dev-secret").encode("utf-8")
@@ -50,7 +103,6 @@ def verify_tracking_token(token: str) -> Optional[Dict[str, Any]]:
         part_json, part_sig = token.split(".", 1)
         b = _b64url_decode(part_json)
         expected = _sign(b)
-        # 時間差攻撃対策の恒等比較
         if not hmac.compare_digest(part_sig, expected):
             return None
         return json.loads(b.decode("utf-8"))
@@ -58,11 +110,13 @@ def verify_tracking_token(token: str) -> Optional[Dict[str, Any]]:
         return None
 
 # ====== クリック/イベントの簡易ログ ======
-EVENTS_PATH = DATA_DIR / "events.jsonl"
+def _events_path() -> Path:
+    return data_dir() / "events.jsonl"
 
 def log_event(event_type: str, payload: Dict[str, Any], meta: Optional[Dict[str, Any]] = None) -> None:
     """
     data/events.jsonl に1行追記。Spaceの Files タブから確認可能。
+    書き込み場所は ensure_dirs() により自動選択される。
     """
     ensure_dirs()
     rec = {
@@ -71,5 +125,5 @@ def log_event(event_type: str, payload: Dict[str, Any], meta: Optional[Dict[str,
         "payload": payload or {},
         "meta": meta or {},
     }
-    with open(EVENTS_PATH, "a", encoding="utf-8") as f:
+    with open(_events_path(), "a", encoding="utf-8") as f:
         f.write(json.dumps(rec, ensure_ascii=False) + "\n")