Create utils.py

modules/utils.py

@@ -0,0 +1,41 @@
+import os, re, json, base64, hmac, hashlib
+from pathlib import Path
+from typing import Iterable, List
+
+DATA_DIR = Path("data")
+LOG_PATH = DATA_DIR / "events.jsonl"
+
+def ensure_dirs():
+    DATA_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)
+    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):
+    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")
+
+def verify_tracking_token(token: str):
+    try:
+        raw = base64.urlsafe_b64decode(token.encode())
+        data, sig = raw.rsplit(b".",1)
+        secret = os.getenv("TRACKING_SECRET", "dev").encode()
+        expected = hmac.new(secret, data, hashlib.sha256).digest()
+        if not hmac.compare_digest(sig, expected):
+            return None
+        return json.loads(data.decode())
+    except Exception:
+        return None