feat: admin survey responses page with expandable detail and CSV export
- Backend: GET /admin/survey-responses (list with stats, invite join) - Backend: GET /admin/survey-responses/export (CSV download) - Frontend: SurveyResponsesPage with expandable row detail - Two-column Q&A grid with typed answer rendering (chips, ranked lists, quote blocks) - Stats cards (total responses, this week) - CSV export button with blob download - Sidebar nav + route wiring - Also: updated Q14 from product domain ranking to diagnostic prioritization Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,8 +1,12 @@
|
||||
"""Admin endpoints for managing survey invites."""
|
||||
"""Admin endpoints for managing survey invites and responses."""
|
||||
import csv
|
||||
import io
|
||||
import logging
|
||||
from typing import Annotated
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Annotated, Any
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from fastapi.responses import StreamingResponse
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
@@ -11,8 +15,14 @@ from app.core.config import settings
|
||||
from app.core.database import get_db
|
||||
from app.core.email import EmailService
|
||||
from app.models.survey_invite import SurveyInvite
|
||||
from app.models.survey_response import SurveyResponse
|
||||
from app.models.user import User
|
||||
from app.schemas.survey import SurveyInviteCreate, SurveyInviteResponse
|
||||
from app.schemas.survey import (
|
||||
SurveyInviteCreate,
|
||||
SurveyInviteResponse,
|
||||
SurveyResponseDetail,
|
||||
SurveyResponseListResponse,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -80,3 +90,148 @@ async def list_survey_invites(
|
||||
)
|
||||
invites = result.scalars().all()
|
||||
return [_build_invite_response(i) for i in invites]
|
||||
|
||||
|
||||
@router.get("/survey-responses", response_model=SurveyResponseListResponse)
|
||||
async def list_survey_responses(
|
||||
db: Annotated[AsyncSession, Depends(get_db)],
|
||||
current_user: Annotated[User, Depends(require_admin)],
|
||||
):
|
||||
"""List all survey responses with summary stats."""
|
||||
stmt = (
|
||||
select(SurveyResponse, SurveyInvite.recipient_name)
|
||||
.outerjoin(SurveyInvite, SurveyResponse.invite_id == SurveyInvite.id)
|
||||
.order_by(SurveyResponse.created_at.desc())
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
one_week_ago = datetime.now(timezone.utc) - timedelta(days=7)
|
||||
this_week = 0
|
||||
responses: list[SurveyResponseDetail] = []
|
||||
|
||||
for survey_resp, invite_name in rows:
|
||||
source = "invite" if survey_resp.invite_id else "direct"
|
||||
responses.append(
|
||||
SurveyResponseDetail(
|
||||
id=str(survey_resp.id),
|
||||
respondent_name=survey_resp.respondent_name,
|
||||
responses=survey_resp.responses,
|
||||
source=source,
|
||||
invite_name=invite_name,
|
||||
created_at=survey_resp.created_at,
|
||||
)
|
||||
)
|
||||
if survey_resp.created_at >= one_week_ago:
|
||||
this_week += 1
|
||||
|
||||
return SurveyResponseListResponse(
|
||||
responses=responses,
|
||||
total=len(responses),
|
||||
this_week=this_week,
|
||||
)
|
||||
|
||||
|
||||
# Question IDs in survey order, used for CSV export columns.
|
||||
QUESTION_IDS = [
|
||||
"prereqs",
|
||||
"verify_fix",
|
||||
"steps_at_a_time",
|
||||
"first_step",
|
||||
"junior_mistake",
|
||||
"pivot",
|
||||
"scenario_approach",
|
||||
"scenario_deeper",
|
||||
"doc_pct",
|
||||
"go_to_commands",
|
||||
"secret_weapon",
|
||||
"gotcha",
|
||||
"hard_rules",
|
||||
"prioritization",
|
||||
"detail_level",
|
||||
"ai_personality",
|
||||
]
|
||||
|
||||
QUESTION_LABELS = {
|
||||
"prereqs": "Q1: Prereqs",
|
||||
"verify_fix": "Q2: Verify Fix",
|
||||
"steps_at_a_time": "Q3: Steps at a Time",
|
||||
"first_step": "Q4: First Step",
|
||||
"junior_mistake": "Q5: Junior Mistake",
|
||||
"pivot": "Q6: Pivot",
|
||||
"scenario_approach": "Q7: Scenario Approach",
|
||||
"scenario_deeper": "Q8: Scenario Deeper",
|
||||
"doc_pct": "Q9: Doc Pct",
|
||||
"go_to_commands": "Q10: Go-To Commands",
|
||||
"secret_weapon": "Q11: Secret Weapon",
|
||||
"gotcha": "Q12: Gotcha",
|
||||
"hard_rules": "Q13: Hard Rules",
|
||||
"prioritization": "Q14: Prioritization",
|
||||
"detail_level": "Q15: Detail Level",
|
||||
"ai_personality": "Q16: AI Personality",
|
||||
}
|
||||
|
||||
|
||||
def _format_answer(value: Any) -> str:
|
||||
"""Format a survey answer for CSV export."""
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, list):
|
||||
# Check if this looks like a ranked list (list of strings to number)
|
||||
# Rank answers: "1. X; 2. Y"
|
||||
# Multi-select: "X; Y"
|
||||
# Heuristic: if it's a list, treat as multi-select joined with "; "
|
||||
return "; ".join(str(v) for v in value)
|
||||
return str(value)
|
||||
|
||||
|
||||
def _format_ranked_answer(value: Any) -> str:
|
||||
"""Format a ranked answer: '1. X; 2. Y'."""
|
||||
if isinstance(value, list):
|
||||
return "; ".join(f"{i + 1}. {v}" for i, v in enumerate(value))
|
||||
return _format_answer(value)
|
||||
|
||||
|
||||
# Question IDs that use ranked answers
|
||||
RANKED_QUESTIONS = {"prioritization"}
|
||||
|
||||
|
||||
@router.get("/survey-responses/export")
|
||||
async def export_survey_responses_csv(
|
||||
db: Annotated[AsyncSession, Depends(get_db)],
|
||||
current_user: Annotated[User, Depends(require_admin)],
|
||||
):
|
||||
"""Export all survey responses as CSV."""
|
||||
stmt = (
|
||||
select(SurveyResponse)
|
||||
.order_by(SurveyResponse.created_at.desc())
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
rows = result.scalars().all()
|
||||
|
||||
output = io.StringIO()
|
||||
writer = csv.writer(output)
|
||||
|
||||
# Header row
|
||||
headers = ["Respondent", "Date"] + [QUESTION_LABELS.get(qid, qid) for qid in QUESTION_IDS]
|
||||
writer.writerow(headers)
|
||||
|
||||
for resp in rows:
|
||||
row = [
|
||||
resp.respondent_name or "Anonymous",
|
||||
resp.created_at.strftime("%Y-%m-%d %H:%M") if resp.created_at else "",
|
||||
]
|
||||
for qid in QUESTION_IDS:
|
||||
answer = resp.responses.get(qid) if resp.responses else None
|
||||
if qid in RANKED_QUESTIONS:
|
||||
row.append(_format_ranked_answer(answer))
|
||||
else:
|
||||
row.append(_format_answer(answer))
|
||||
writer.writerow(row)
|
||||
|
||||
output.seek(0)
|
||||
return StreamingResponse(
|
||||
iter([output.getvalue()]),
|
||||
media_type="text/csv",
|
||||
headers={"Content-Disposition": "attachment; filename=survey-responses.csv"},
|
||||
)
|
||||
|
||||
@@ -44,3 +44,20 @@ class SurveyInviteStatus(BaseModel):
|
||||
"""Public invite status check — minimal info."""
|
||||
name: str
|
||||
status: str
|
||||
|
||||
|
||||
class SurveyResponseDetail(BaseModel):
|
||||
"""Full survey response returned to admin."""
|
||||
id: str
|
||||
respondent_name: Optional[str]
|
||||
responses: dict[str, Any]
|
||||
source: str
|
||||
invite_name: Optional[str]
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class SurveyResponseListResponse(BaseModel):
|
||||
"""List of survey responses with summary stats."""
|
||||
responses: list[SurveyResponseDetail]
|
||||
total: int
|
||||
this_week: int
|
||||
|
||||
@@ -113,3 +113,59 @@ async def test_create_and_list_invites(client, admin_auth_headers):
|
||||
assert list_res.status_code == 200
|
||||
invites = list_res.json()
|
||||
assert len(invites) >= 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_survey_responses_admin(client, admin_auth_headers):
|
||||
"""Admin can list survey responses."""
|
||||
await client.post(
|
||||
"/api/v1/survey/submit",
|
||||
json={"respondent_name": "Tester", "responses": {"q1": "answer"}},
|
||||
)
|
||||
res = await client.get("/api/v1/admin/survey-responses", headers=admin_auth_headers)
|
||||
assert res.status_code == 200
|
||||
data = res.json()
|
||||
assert "responses" in data
|
||||
assert "total" in data
|
||||
assert "this_week" in data
|
||||
assert data["total"] >= 1
|
||||
assert data["responses"][0]["respondent_name"] == "Tester"
|
||||
assert data["responses"][0]["source"] == "direct"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_survey_responses_requires_admin(client, auth_headers):
|
||||
"""Non-admin cannot list survey responses."""
|
||||
res = await client.get("/api/v1/admin/survey-responses", headers=auth_headers)
|
||||
assert res.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_export_survey_responses_csv(client, admin_auth_headers):
|
||||
"""Admin can export survey responses as CSV."""
|
||||
await client.post(
|
||||
"/api/v1/survey/submit",
|
||||
json={
|
||||
"respondent_name": "CSV Tester",
|
||||
"responses": {
|
||||
"prereqs": ["Who's affected", "What changed recently"],
|
||||
"verify_fix": "Have the user confirm",
|
||||
"steps_at_a_time": "5 steps",
|
||||
"prioritization": ["Likelihood", "Speed", "Blast radius"],
|
||||
},
|
||||
},
|
||||
)
|
||||
res = await client.get("/api/v1/admin/survey-responses/export", headers=admin_auth_headers)
|
||||
assert res.status_code == 200
|
||||
assert "text/csv" in res.headers["content-type"]
|
||||
assert "attachment" in res.headers.get("content-disposition", "")
|
||||
body = res.text
|
||||
assert "CSV Tester" in body
|
||||
assert "Respondent" in body
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_export_survey_responses_requires_admin(client, auth_headers):
|
||||
"""Non-admin cannot export survey responses."""
|
||||
res = await client.get("/api/v1/admin/survey-responses/export", headers=auth_headers)
|
||||
assert res.status_code == 403
|
||||
|
||||
Reference in New Issue
Block a user