Update modules/utils.py

modules/utils.py

@@ -1,42 +1,75 @@
-import os, re, json, base64, hmac, hashlib
+import os
+import json
+import time
+import base64
+import hmac
+import hashlib
 from pathlib import Path
-from typing import Iterable
+from typing import Any, Dict, Optional
 
-DATA_DIR = Path("data")
-LOG_PATH = DATA_DIR / "event.jsonl"
+# ====== 共通ディレクトリの用意 ======
+DATA_DIR = Path(os.getenv("DATA_DIR", "data"))
+EXPORT_DIR = DATA_DIR / "exports"
 
-def ensure_dirs():
+def ensure_dirs() -> None:
+    """必要なディレクトリを作成"""
     DATA_DIR.mkdir(parents=True, exist_ok=True)
+    EXPORT_DIR.mkdir(parents=True, exist_ok=True)
 
-def chunk_text(text: str, max_len: int=600) -> Iterable[str]:
-    lines = re.split(r"(\n+)", text)
-    buff, count = [], 0
-    for ln in lines:
-        if count + len(ln) > max_len and buff:
-            yield"".join(buff).strip()
-            buff, count = [], 0
-        buff.append(ln)
-        count += len(ln)
-    if buff:
-        yield "".join(buff).strip()
-
-def truncate(text: str, n: int) -> str:
-    return text if len(text) <= n else text [:n] + "..."
-
-def log_event(event_type: str, payload: dict, meta:dict | None=None):
+# ====== 簡易トラッキングトークン（追加依存なしのHMAC方式） ======
+_SECRET = os.getenv("TRACKING_SECRET", "dev-secret").encode("utf-8")
+
+def _b64url_encode(b: bytes) -> str:
+    return base64.urlsafe_b64encode(b).rstrip(b"=").decode("ascii")
+
+def _b64url_decode(s: str) -> bytes:
+    pad = "=" * (-len(s) % 4)
+    return base64.urlsafe_b64decode((s + pad).encode("ascii"))
+
+def _sign(payload_bytes: bytes) -> str:
+    mac = hmac.new(_SECRET, payload_bytes, hashlib.sha256).digest()
+    return _b64url_encode(mac)
+
+def make_tracking_token(payload: Dict[str, Any]) -> str:
+    """
+    payload を JSON にして HMAC 署名し、'<b64json>.<sig>' 形式で返す。
+    例: {"company":"Test Inc.","ts":1690000000,"redirect":"/"}
+    """
     ensure_dirs()
-    rec = {"type": event_type, "payload": payload, "meta": meta or {}}
-    with open(LOG_PATH, "a", encoding="utf-8") as f:
-        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+    payload = dict(payload or {})
+    if "ts" not in payload:
+        payload["ts"] = int(time.time())
+    b = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    return f"{_b64url_encode(b)}.{_sign(b)}"
 
-def verify_tracking_token(token: str)
+def verify_tracking_token(token: str) -> Optional[Dict[str, Any]]:
+    """
+    トークンの署名検証に成功したら payload を返す。失敗したら None。
+    """
     try:
-        raw = base64.urlsage_b64decode(token.encode())
-        data, sig = raw.resplit(b".", 1)
-        secret = os.getenv("TRACKING_SECRET", "dev").encode()
-        expected = hmac.new(secret, data, hashlib.sha256).digest()
-        if not hmac.compare_digest(sig, expected):
+        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.load(data.decode())
+        return json.loads(b.decode("utf-8"))
     except Exception:
         return None
+
+# ====== クリック/イベントの簡易ログ ======
+EVENTS_PATH = 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()
+    rec = {
+        "ts": int(time.time()),
+        "type": event_type,
+        "payload": payload or {},
+        "meta": meta or {},
+    }
+    with open(EVENTS_PATH, "a", encoding="utf-8") as f:
+        f.write(json.dumps(rec, ensure_ascii=False) + "\n")