brettsp/proteomics-feedback
0
1"""2UC Davis Proteomics Core — Feedback Collection App3Hosted on Hugging Face Spaces (Docker/FastAPI)4 5Data is saved both locally (CSV) and to a HF Dataset for persistence.6"""7 8import csv9import os10import json11from datetime import datetime12from pathlib import Path13from threading import Thread14 15from fastapi import FastAPI, Request16from fastapi.responses import HTMLResponse, JSONResponse, FileResponse17 18app = FastAPI()19 20# Local storage (works even without persistent storage)21DATA_DIR = Path("/tmp/feedback-data")22CSV_PATH = DATA_DIR / "feedback_responses.csv"23 24# HF Dataset for durable storage25HF_TOKEN = os.environ.get("HF_TOKEN", "")26HF_REPO = os.environ.get("HF_DATASET_REPO", "") # e.g. "brettsp/proteomics-feedback-data"27 28CSV_FIELDS = [29 "timestamp",30 "services",31 "frequency",32 "satisfaction",33 "criticality",34 "recommend",35 "well",36 "improve",37 "testimonial_yn",38 "testimonial_text",39 "name",40 "dept",41 "institution",42]43 44 45def ensure_csv():46 """Create CSV with headers if it doesn't exist."""47 DATA_DIR.mkdir(parents=True, exist_ok=True)48 if not CSV_PATH.exists():49 # Try to restore from HF Dataset50 restored = restore_from_hf()51 if not restored:52 with open(CSV_PATH, "w", newline="", encoding="utf-8") as f:53 writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)54 writer.writeheader()55 56 57def restore_from_hf():58 """Try to download existing CSV from HF Dataset on startup."""59 if not HF_TOKEN or not HF_REPO:60 return False61 try:62 from huggingface_hub import hf_hub_download63 path = hf_hub_download(64 repo_id=HF_REPO,65 filename="feedback_responses.csv",66 repo_type="dataset",67 token=HF_TOKEN,68 local_dir=str(DATA_DIR),69 )70 print(f"Restored CSV from HF Dataset: {path}")71 return True72 except Exception as e:73 print(f"No existing data in HF Dataset (normal for first run): {e}")74 return False75 76 77def sync_to_hf():78 """Upload current CSV to HF Dataset (runs in background thread)."""79 if not HF_TOKEN or not HF_REPO:80 return81 try:82 from huggingface_hub import HfApi83 api = HfApi(token=HF_TOKEN)84 api.upload_file(85 path_or_fileobj=str(CSV_PATH),86 path_in_repo="feedback_responses.csv",87 repo_id=HF_REPO,88 repo_type="dataset",89 )90 print(f"Synced CSV to HF Dataset: {HF_REPO}")91 except Exception as e:92 print(f"Warning: could not sync to HF Dataset: {e}")93 94 95@app.on_event("startup")96async def startup():97 ensure_csv()98 99 100@app.post("/api/submit")101async def submit_feedback(request: Request):102 """Receive feedback form submission."""103 try:104 data = await request.json()105 except Exception:106 return JSONResponse({"error": "Invalid JSON"}, status_code=400)107 108 row = {109 "timestamp": datetime.now().isoformat(),110 "services": "; ".join(data.get("services", [])) if isinstance(data.get("services"), list) else data.get("services", ""),111 "frequency": data.get("frequency", ""),112 "satisfaction": data.get("satisfaction", ""),113 "criticality": data.get("criticality", ""),114 "recommend": data.get("recommend", ""),115 "well": data.get("well", ""),116 "improve": data.get("improve", ""),117 "testimonial_yn": data.get("testimonial_yn", ""),118 "testimonial_text": data.get("testimonial_text", ""),119 "name": data.get("name", ""),120 "dept": data.get("dept", ""),121 "institution": data.get("institution", ""),122 }123 124 # Append to local CSV125 with open(CSV_PATH, "a", newline="", encoding="utf-8") as f:126 writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)127 writer.writerow(row)128 129 # Sync to HF Dataset in background (non-blocking)130 Thread(target=sync_to_hf, daemon=True).start()131 132 return JSONResponse({"status": "ok", "message": "Feedback recorded. Thank you!"})133 134 135@app.get("/api/export")136async def export_csv(request: Request):137 """Download responses as CSV (protected by a simple token)."""138 token = request.query_params.get("token", "")139 expected = os.environ.get("ADMIN_TOKEN", "changeme")140 if token != expected:141 return JSONResponse({"error": "Unauthorized"}, status_code=401)142 if not CSV_PATH.exists():143 return JSONResponse({"error": "No data yet"}, status_code=404)144 return FileResponse(CSV_PATH, filename="feedback_responses.csv", media_type="text/csv")145 146 147@app.get("/api/stats")148async def get_stats(request: Request):149 """Quick summary stats (protected)."""150 token = request.query_params.get("token", "")151 expected = os.environ.get("ADMIN_TOKEN", "changeme")152 if token != expected:153 return JSONResponse({"error": "Unauthorized"}, status_code=401)154 155 if not CSV_PATH.exists():156 return JSONResponse({"n": 0})157 158 responses = []159 with open(CSV_PATH, "r", encoding="utf-8") as f:160 reader = csv.DictReader(f)161 for row in reader:162 responses.append(row)163 164 n = len(responses)165 if n == 0:166 return JSONResponse({"n": 0})167 168 sat_scores = []169 for r in responses:170 try:171 sat_scores.append(int(r.get("satisfaction", "")))172 except (ValueError, TypeError):173 pass174 175 recommend_yes = sum(1 for r in responses if r.get("recommend", "").lower() == "yes")176 testimonials = sum(1 for r in responses if r.get("testimonial_yn", "").lower() == "yes" and r.get("testimonial_text", "").strip())177 178 return JSONResponse({179 "n": n,180 "mean_satisfaction": round(sum(sat_scores) / len(sat_scores), 1) if sat_scores else None,181 "pct_4_or_5": round(sum(1 for s in sat_scores if s >= 4) / len(sat_scores) * 100) if sat_scores else None,182 "pct_recommend": round(recommend_yes / n * 100) if n else None,183 "testimonials_available": testimonials,184 })185 186 187# Serve the form as the root page188@app.get("/", response_class=HTMLResponse)189async def root():190 html_path = Path(__file__).parent / "static" / "index.html"191 return HTMLResponse(html_path.read_text(encoding="utf-8"))192