Telemetry collector API

app.py

@@ -0,0 +1,98 @@
+import json
+import os
+from datetime import datetime
+from pathlib import Path
+from threading import Lock
+
+import gradio as gr
+from huggingface_hub import HfApi
+
+DATASET_REPO = "bobber/routangseng-telemetry"
+HF_TOKEN = os.environ.get("HF_TOKEN")
+BUFFER = []
+BUFFER_LOCK = Lock()
+FLUSH_SIZE = 5  # flush every 5 entries
+
+api = HfApi(token=HF_TOKEN)
+
+
+def flush_buffer():
+    """Append buffered entries to the dataset."""
+    global BUFFER
+    with BUFFER_LOCK:
+        if not BUFFER:
+            return 0
+        entries = BUFFER.copy()
+        BUFFER = []
+
+    # Download existing data, append, re-upload
+    try:
+        path = api.hf_hub_download(DATASET_REPO, "data/telemetry.jsonl", repo_type="dataset")
+        existing = open(path).read().strip()
+    except Exception:
+        existing = ""
+
+    new_lines = "\n".join(json.dumps(e, ensure_ascii=False) for e in entries)
+    content = (existing + "\n" + new_lines).strip() + "\n"
+
+    api.upload_file(
+        path_or_fileobj=content.encode("utf-8"),
+        path_in_repo="data/telemetry.jsonl",
+        repo_id=DATASET_REPO,
+        repo_type="dataset",
+        commit_message=f"Add {len(entries)} telemetry entries",
+    )
+    return len(entries)
+
+
+def collect(data: str):
+    """Receive telemetry JSON from the WebGPU Space."""
+    try:
+        entry = json.loads(data)
+        # Validate required fields
+        if "question" not in entry or "answer" not in entry:
+            return {"status": "error", "message": "missing question or answer"}
+        
+        entry["received_at"] = datetime.utcnow().isoformat() + "Z"
+        
+        with BUFFER_LOCK:
+            BUFFER.append(entry)
+            buf_size = len(BUFFER)
+
+        if buf_size >= FLUSH_SIZE:
+            n = flush_buffer()
+            return {"status": "ok", "flushed": n}
+        
+        return {"status": "ok", "buffered": buf_size}
+    except json.JSONDecodeError:
+        return {"status": "error", "message": "invalid JSON"}
+
+
+def manual_flush():
+    """Manually flush buffer to dataset."""
+    n = flush_buffer()
+    return f"Flushed {n} entries"
+
+
+def status():
+    """Check buffer status."""
+    return f"Buffer: {len(BUFFER)} entries"
+
+
+# Gradio interface (minimal UI + API)
+with gr.Blocks(title="肉糖生 Telemetry") as demo:
+    gr.Markdown("# 肉糖生 Telemetry Collector\nReceives Q&A data from the WebGPU Space.")
+    
+    with gr.Row():
+        status_btn = gr.Button("Check Status")
+        flush_btn = gr.Button("Flush to Dataset")
+    output = gr.Textbox(label="Status")
+    
+    status_btn.click(fn=status, outputs=output)
+    flush_btn.click(fn=manual_flush, outputs=output)
+
+# API endpoint: POST /api/collect
+demo.queue()
+
+if __name__ == "__main__":
+    demo.launch()