Add device-key FCM debug route + update test_fcm.py to use device key

- New GET /api/wearable/device/fcm-debug (X-Device-Key auth) mirrors
/api/user/fcm-debug without needing JWT/email/password
- Refactored shared logic into _fcm_debug_for_user() helper
- test_fcm.py now uses device API key entirely — no credentials needed

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app.py

@@ -213,7 +213,28 @@ def fcm_debug():
     """
     user = request.current_user
     db = get_db()
-    fcm_token = db.get_fcm_token(user["id"])
+    return _fcm_debug_for_user(user["id"], db)
+
+
+@app.route("/api/wearable/device/fcm-debug", methods=["GET"])
+def device_fcm_debug():
+    """
+    Same as /api/user/fcm-debug but authenticated via X-Device-Key header.
+    Allows testing FCM without knowing the account email/password.
+    Query param: ?send_test=1 to send a test notification.
+    """
+    api_key = request.headers.get("X-Device-Key")
+    if not api_key:
+        return jsonify({"error": "X-Device-Key header required"}), 401
+    db = get_db()
+    user = db.get_user_by_device_key(api_key)
+    if not user:
+        return jsonify({"error": "Invalid or revoked device key"}), 401
+    return _fcm_debug_for_user(user["id"], db)
+
+
+def _fcm_debug_for_user(user_id, db):
+    fcm_token = db.get_fcm_token(user_id)
     result = {
         "fcm_backend_enabled": _fcm_enabled,
         "user_has_token": bool(fcm_token),

test_fcm.py

@@ -1,70 +1,43 @@
 """
 FCM Diagnostic Test
 ====================
-Logs in as a real user, checks FCM status on backend, sends a test push,
-then sends real stress sensor data as that user to trigger ML → FCM flow.
+Uses device API key to check FCM status and send a test push notification.
+No email/password needed.
 
 Usage:
     python test_fcm.py
 """
 
-import json
+import csv
 import sys
 import io
-import requests
-import csv
 import time
+import requests
 
 # Fix Windows console encoding
 sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
 
-BASE_URL = "https://santa47-cbt-companion-api.hf.space"
+BASE_URL      = "https://santa47-cbt-companion-api.hf.space"
+DEVICE_API_KEY = "0c4358009dbe75db461b760222a6e15b"
 
-# ---------- credentials ----------
-EMAIL    = "test@example.com"   # <-- your test account email
-PASSWORD = "test123"             # <-- your test account password
-# ---------------------------------
-
-DATA_DIR = "data"
+DATA_DIR   = "data"
 SEVERE_CSV = "sampled_data_DepressionLevel_Severe.csv"
 
-
-def login(email, password):
-    resp = requests.post(
-        f"{BASE_URL}/api/login",
-        json={"email": email, "password": password},
-        timeout=15,
-    )
-    if resp.status_code == 200 and resp.json().get("success"):
-        return resp.json()["token"]
-    raise RuntimeError(f"Login failed ({resp.status_code}): {resp.text}")
-
-
-def auth_headers(token):
-    return {"Content-Type": "application/json", "Authorization": f"Bearer {token}"}
+DEVICE_HEADERS = {
+    "Content-Type": "application/json",
+    "X-Device-Key": DEVICE_API_KEY,
+}
 
 
 def check_health():
-    resp = requests.get(f"{BASE_URL}/health", timeout=10)
-    return resp.json()
+    return requests.get(f"{BASE_URL}/health", timeout=10).json()
 
 
-def check_fcm_status(token):
-    resp = requests.get(
-        f"{BASE_URL}/api/user/fcm-debug",
-        headers=auth_headers(token),
-        timeout=10,
-    )
-    return resp.json()
-
-
-def send_test_push(token):
-    resp = requests.get(
-        f"{BASE_URL}/api/user/fcm-debug?send_test=1",
-        headers=auth_headers(token),
-        timeout=15,
-    )
-    return resp.json()
+def check_fcm_status(send_test=False):
+    url = f"{BASE_URL}/api/wearable/device/fcm-debug"
+    if send_test:
+        url += "?send_test=1"
+    return requests.get(url, headers=DEVICE_HEADERS, timeout=15).json()
 
 
 def load_severe_data():
@@ -81,118 +54,102 @@ def load_severe_data():
     return readings
 
 
-def send_sensor_reading(token, reading):
-    resp = requests.post(
-        f"{BASE_URL}/api/wearable/data",
-        headers=auth_headers(token),
+def send_sensor(reading):
+    return requests.post(
+        f"{BASE_URL}/api/wearable/device/data",
+        headers=DEVICE_HEADERS,
         json=reading,
         timeout=20,
-    )
-    return resp.json()
+    ).json()
 
 
 def main():
     print("\n" + "="*60)
     print("  CBT COMPANION — FCM DIAGNOSTIC TEST")
+    print(f"  Device key: {DEVICE_API_KEY[:8]}...")
     print("="*60)
 
-    # 1. Health check
+    # 1. Health
     print("\n[1] Health check...")
     health = check_health()
     print(f"  Status:          {health.get('status')}")
     print(f"  ML model loaded: {health.get('ml_model_loaded')}")
-    fcm_enabled = health.get("fcm_enabled")
-    print(f"  FCM enabled:     {fcm_enabled}")
-    if not fcm_enabled:
-        print("\n  ❌ firebase-admin is NOT loaded on the server.")
-        print("     → Check HuggingFace Space logs. The Space may still be building,")
-        print("       or FIREBASE_SERVICE_ACCOUNT secret may be missing/invalid.")
-
-    # 2. Login
-    print(f"\n[2] Logging in as {EMAIL}...")
-    try:
-        token = login(EMAIL, PASSWORD)
-        print("  ✅ Login OK — JWT acquired")
-    except RuntimeError as e:
-        print(f"  ❌ {e}")
+    print(f"  FCM enabled:     {health.get('fcm_enabled')}")
+    if not health.get("fcm_enabled"):
+        print("\n  ❌ firebase-admin NOT loaded on server.")
+        print("     Check HuggingFace Space logs + FIREBASE_SERVICE_ACCOUNT secret.")
         return
 
-    # 3. FCM token check
-    print("\n[3] Checking FCM token in database...")
-    status = check_fcm_status(token)
+    # 2. FCM token check (no test push yet)
+    print("\n[2] Checking if device owner has FCM token in DB...")
+    status = check_fcm_status(send_test=False)
     print(f"  FCM backend enabled: {status.get('fcm_backend_enabled')}")
     print(f"  User has token:      {status.get('user_has_token')}")
     print(f"  Token preview:       {status.get('token_preview', 'N/A')}")
 
     if not status.get("user_has_token"):
         print("\n  ❌ No FCM token stored for this user!")
-        print("     → Open the new APK on your phone and LOG IN.")
-        print("       The token uploads automatically when you log in.")
+        print("     → Open the app on your phone and LOG IN.")
+        print("       The token uploads automatically on login.")
         print("     → Then run this script again.")
-        print("\n  Aborting — cannot test FCM without a stored token.")
         return
 
-    # 4. Test push
-    print("\n[4] Sending test FCM push notification...")
-    push_result = send_test_push(token)
+    # 3. Test push
+    print("\n[3] Sending test FCM push notification...")
+    push_result = check_fcm_status(send_test=True)
     test_push = push_result.get("test_push", "unknown")
     print(f"  Result: {test_push}")
+
     if "SENT" in str(test_push):
-        print("\n  ✅ Test push SENT! Check your phone now.")
-        print("     If you don't see a notification:")
-        print("      • Disable battery optimization for the app")
-        print("      • Make sure notifications are allowed for the app")
-        print("      • Try with app in background (not foreground)")
-    elif "FAILED" in str(test_push):
-        print("\n  ❌ FCM send FAILED. Check HuggingFace Space logs for details.")
-        return
+        print("\n  ✅ Test push SENT — check your phone NOW.")
+        print("     (It may take a few seconds to arrive)")
+        if not health.get("ml_model_loaded"):
+            print("\n  ⚠️  ML model is NOT loaded — real alerts won't fire.")
+            print("     But the test push above confirms FCM itself is working.")
+            print("     The ML model file needs to be present on the server.")
+        else:
+            # 4. Real stress data
+            print("\n[4] Sending SEVERE stress sensor data to trigger ML → FCM alert...")
+            try:
+                readings = load_severe_data()
+            except FileNotFoundError:
+                print(f"  ⚠️  {SEVERE_CSV} not found in {DATA_DIR}/ — skipping")
+                print("     Run test_dry_run.py separately to trigger ML alerts.")
+                return
+
+            alert_seen = False
+            for i, reading in enumerate(readings):
+                result = send_sensor(reading)
+                ml = result.get("ml_prediction")
+                if ml and isinstance(ml, dict):
+                    if ml.get("status") == "waiting":
+                        cnt = ml.get("readings_count", "?")
+                        print(f"  [{i+1:2d}/{len(readings)}] waiting ({cnt}/25 readings)")
+                    else:
+                        pred = ml.get("prediction", "?")
+                        conf = ml.get("confidence", 0)
+                        risk = ml.get("risk_level", -1)
+                        lbl = {"NORMAL": "[NORMAL]",
+                               "MILD_STRESS": "[MILD_STRESS]",
+                               "HIGH_STRESS": "[HIGH_STRESS] !!!"}.get(pred, pred)
+                        print(f"  [{i+1:2d}/{len(readings)}] {lbl}  conf={conf:.0%}  risk={risk}")
+                        if risk >= 1:
+                            alert_seen = True
+                            print(f"           ↑ FCM push should fire to ...{status.get('token_preview')}")
+                time.sleep(0.3)
+
+            if not alert_seen:
+                print("\n  ⚠️  No stress detected in this data batch (model may need more readings).")
 
-    time.sleep(3)
-
-    # 5. Stress data trigger test
-    print("\n[5] Sending severe stress sensor data as THIS user (ML → FCM flow)...")
-    try:
-        readings = load_severe_data()
-    except FileNotFoundError:
-        print(f"  ⚠️  {SEVERE_CSV} not found in {DATA_DIR}/  — skipping stress data step")
-        print("     You can manually stress-test via the wearable or test_dry_run.py")
-        print("     But make sure to use YOUR device API key (registered under this account).")
-        print_summary()
-        return
+    elif "FAILED" in str(test_push):
+        print("\n  ❌ FCM send FAILED — check HuggingFace Space logs for the error message.")
+        print("     Common causes:")
+        print("      • FIREBASE_SERVICE_ACCOUNT secret is malformed")
+        print("      • FCM token is expired (log out and log back in on phone)")
+        print("      • Wrong Firebase project")
 
-    alert_triggered = False
-    for i, reading in enumerate(readings):
-        result = send_sensor_reading(token, reading)
-        ml = result.get("ml_prediction")
-        if ml and isinstance(ml, dict) and ml.get("status") != "waiting":
-            pred = ml.get("prediction", "?")
-            conf = ml.get("confidence", 0)
-            risk = ml.get("risk_level", -1)
-            label = {"NORMAL": "[NORMAL]", "MILD_STRESS": "[MILD_STRESS]",
-                     "HIGH_STRESS": "[HIGH_STRESS] !!!"}.get(pred, f"[{pred}]")
-            print(f"  [{i+1:2d}/{len(readings)}] {label}  conf={conf:.0%}  risk={risk}")
-            if risk >= 1:
-                alert_triggered = True
-                print(f"           ↑ FCM push should fire NOW to ...{status.get('token_preview', '?')}")
-        elif ml and isinstance(ml, dict):
-            cnt = ml.get("readings_count", "?")
-            print(f"  [{i+1:2d}/{len(readings)}] waiting ({cnt}/25 readings)")
-        time.sleep(0.3)
-
-    print_summary(alert_triggered)
-
-
-def print_summary(alert_triggered=None):
     print("\n" + "="*60)
-    print("  SUMMARY")
-    print("="*60)
-    if alert_triggered is True:
-        print("  ✅ Stress detected. If you got no push notification:")
-        print("     1. Check phone notifications are allowed for the app")
-        print("     2. Check HuggingFace Space logs for FCM errors")
-        print("     3. The token may have rotated — log out/in on phone and retry")
-    elif alert_triggered is False:
-        print("  ⚠️  No stress alert triggered (not enough/wrong data)")
+    print("  DONE")
     print("="*60 + "\n")