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