"""Jednorazowa ankieta in-app: skad uzytkownicy, czego uzywaja, za co by zaplacili. Pytania i etykiety sa w apce (mobile/src/components/SurveyModal.tsx, ten sam SURVEY_ID), backend trzyma tylko liste dozwolonych kluczy i limity rozmiaru, zeby nie przyjmowac smieci. - Jedna odpowiedz na urzadzenie (UNIQUE survey_id + device_id), ponowne wyslanie nadpisuje. - Bramka: pytamy tylko urzadzenia, ktore apki realnie uzywaja, czyli maja co najmniej SURVEY_MIN_SCENES wpisow w scene_play_progress. Klucz glowny tej tabeli to (device_id, scene_id), wiec liczenie z LIMIT to krotki skan indeksu. GET /survey/current wola kazdy launch, dopoki apka nie zapisze lokalnie "odpowiedziane/pominiete". - Podsumowanie (GET /survey/{id}/summary) liczy w Pythonie po wszystkich wierszach ankiety; przy tysiacach odpowiedzi to ulamek sekundy, bez osobnych agregatow w SQL. """ from __future__ import annotations import uuid from collections import Counter, defaultdict from datetime import datetime from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel, Field from sqlalchemy import func, select from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from app.api.device import LEGACY_DEVICE, get_device_id from app.auth import require_api_key from app.db import get_session from app.models.play_progress import ScenePlayProgress from app.models.survey_response import SurveyResponse router = APIRouter(prefix="/survey", tags=["survey"], dependencies=[Depends(require_api_key)]) CURRENT_SURVEY_ID = "2026-10" SURVEY_MIN_SCENES = 10 # Klucze pytan dozwolone w `answers` (wartosci: string albo lista stringow). # `*_text` to wolne dopowiedzenie do opcji "inne" w pytaniach jednokrotnego wyboru. _ALLOWED_KEYS: frozenset[str] = frozenset( { "source", "source_text", "use", "pain", "pain_text", "pay_for", "pay_how", "price", } ) _MAX_VALUE_LEN = 200 _MAX_LIST_ITEMS = 12 class SurveyStatusOut(BaseModel): id: str eligible: bool answered: bool class SurveySubmitIn(BaseModel): answers: dict[str, str | list[str]] free_text: str | None = Field(default=None, max_length=2000) app_version: str | None = Field(default=None, max_length=32) class SurveySummaryOut(BaseModel): survey_id: str total: int first_at: datetime | None last_at: datetime | None counts: dict[str, dict[str, int]] by_app_version: dict[str, int] texts: dict[str, list[str]] def _played_scenes(session: Session, device_id: str, cap: int) -> int: sub = ( select(ScenePlayProgress.scene_id) .where(ScenePlayProgress.device_id == device_id) .limit(cap) .subquery() ) return int(session.scalar(select(func.count()).select_from(sub)) or 0) def _answered(session: Session, survey_id: str, device_id: str) -> bool: return ( session.scalar( select(SurveyResponse.id).where( SurveyResponse.survey_id == survey_id, SurveyResponse.device_id == device_id, ) ) is not None ) @router.get("/current", response_model=SurveyStatusOut) def current_survey( session: Annotated[Session, Depends(get_session)], device_id: Annotated[str, Depends(get_device_id)], ) -> SurveyStatusOut: if device_id == LEGACY_DEVICE: return SurveyStatusOut(id=CURRENT_SURVEY_ID, eligible=False, answered=False) answered = _answered(session, CURRENT_SURVEY_ID, device_id) eligible = answered or _played_scenes(session, device_id, SURVEY_MIN_SCENES) >= SURVEY_MIN_SCENES return SurveyStatusOut(id=CURRENT_SURVEY_ID, eligible=eligible, answered=answered) def _clean_answers(raw: dict[str, str | list[str]]) -> dict[str, Any]: out: dict[str, Any] = {} for key, value in raw.items(): if key not in _ALLOWED_KEYS: raise HTTPException(status_code=422, detail=f"unknown question: {key}") if isinstance(value, str): v = value.strip() if len(v) > _MAX_VALUE_LEN: raise HTTPException(status_code=422, detail=f"value too long: {key}") if v: out[key] = v else: items = [s.strip() for s in value if isinstance(s, str) and s.strip()] if len(items) > _MAX_LIST_ITEMS or any(len(s) > _MAX_VALUE_LEN for s in items): raise HTTPException(status_code=422, detail=f"list too long: {key}") if items: out[key] = items if not out: raise HTTPException(status_code=422, detail="empty answers") return out @router.post("/{survey_id}", status_code=status.HTTP_201_CREATED) def submit_survey( survey_id: str, payload: SurveySubmitIn, session: Annotated[Session, Depends(get_session)], device_id: Annotated[str, Depends(get_device_id)], ) -> dict[str, str]: if survey_id != CURRENT_SURVEY_ID: raise HTTPException(status_code=404, detail="no such survey") if device_id == LEGACY_DEVICE: raise HTTPException(status_code=400, detail="device id required") answers = _clean_answers(payload.answers) free_text = (payload.free_text or "").strip() or None stmt = pg_insert(SurveyResponse).values( id=uuid.uuid4(), survey_id=survey_id, device_id=device_id, answers=answers, free_text=free_text, app_version=payload.app_version, ) stmt = stmt.on_conflict_do_update( constraint="uq_survey_responses_survey_device", set_={ "answers": stmt.excluded.answers, "free_text": stmt.excluded.free_text, "app_version": stmt.excluded.app_version, "updated_at": func.now(), }, ) session.execute(stmt) session.commit() return {"status": "ok"} @router.get("/{survey_id}/summary", response_model=SurveySummaryOut) def survey_summary( survey_id: str, session: Annotated[Session, Depends(get_session)], ) -> SurveySummaryOut: """Podsumowanie dla admina: liczniki per pytanie i opcja, wersje apki, wolne teksty.""" rows = session.scalars( select(SurveyResponse) .where(SurveyResponse.survey_id == survey_id) .order_by(SurveyResponse.created_at) ).all() counts: dict[str, Counter[str]] = defaultdict(Counter) versions: Counter[str] = Counter() texts: dict[str, list[str]] = defaultdict(list) for r in rows: versions[r.app_version or "?"] += 1 if r.free_text: texts["free_text"].append(r.free_text) for key, value in (r.answers or {}).items(): if key.endswith("_text"): if isinstance(value, str): texts[key].append(value) continue if isinstance(value, list): for v in value: counts[key][str(v)] += 1 else: counts[key][str(value)] += 1 return SurveySummaryOut( survey_id=survey_id, total=len(rows), first_at=rows[0].created_at if rows else None, last_at=rows[-1].created_at if rows else None, counts={k: dict(c.most_common()) for k, c in counts.items()}, by_app_version=dict(versions.most_common()), texts={k: v[-300:] for k, v in texts.items()}, )