"""Durable CSV record of discovery questions and answers. The SQLite database remains the system of record. This module keeps a secondary, portable copy of every conversation's prompts and answers on disk so that: * answers survive even if the AI extraction step errors (they are written on save, before extraction runs); * a saved conversation can be re-fed to the extractor (``reprocess``); * a person can review or resume from their original answers. Two artifacts are written for every save: * a *per-conversation* file (``data/questions/{conversation_id}.csv``) that is rewritten in full on each save — always the latest answers for that conversation, easy to hand to an AI or download; and * a *master* append-only log (``data/questions_master.csv``) that records every save/complete event across all conversations, for batch re-processing. CSV writes must never break an API request: the caller's DB commit has already succeeded, so any I/O failure here is logged and swallowed. """ from __future__ import annotations import csv import logging import os from datetime import datetime, timezone from typing import TYPE_CHECKING if TYPE_CHECKING: # pragma: no cover - typing only from app.models import DiscoveryConversation, User logger = logging.getLogger(__name__) # Canonical list of discovery prompts: the short key used by the extractor and # CSV, the DB column on DiscoveryConversation, and the human-facing title. # This is the single source of truth shared by the store, the router, and the # reprocess paths. DISCOVERY_PROMPTS = [ {"key": "alive", "column": "prompt_alive", "title": "The Alive Moment"}, {"key": "friction", "column": "prompt_friction", "title": "The Friction Moment"}, {"key": "pull", "column": "prompt_pull", "title": "The Natural Pull"}, {"key": "recognition", "column": "prompt_recognition", "title": "The Recognition Moment"}, {"key": "future", "column": "prompt_future", "title": "The Future Pull"}, {"key": "goals_short", "column": "prompt_goals_short", "title": "Near-Term Goals (6–12 months)"}, {"key": "goals_long", "column": "prompt_goals_long", "title": "Long-Term Goals (3–5 years)"}, ] CSV_FIELDS = [ "conversation_id", "user_id", "user_email", "status", "saved_at", "prompt_key", "prompt_title", "answer", ] def _questions_dir() -> str: return os.getenv("QUESTIONS_DIR", "./data/questions") def _master_path() -> str: return os.getenv("QUESTIONS_MASTER_CSV", "./data/questions_master.csv") def conversation_csv_path(conversation_id: str) -> str: """Filesystem path of the per-conversation CSV for ``conversation_id``.""" return os.path.join(_questions_dir(), f"{conversation_id}.csv") def _rows_for( conversation: "DiscoveryConversation", user_email: str, status: str, saved_at: str, ) -> list[dict]: """One row per prompt, in canonical prompt order.""" rows = [] for prompt in DISCOVERY_PROMPTS: answer = getattr(conversation, prompt["column"], None) or "" rows.append( { "conversation_id": conversation.id, "user_id": conversation.user_id, "user_email": user_email, "status": status, "saved_at": saved_at, "prompt_key": prompt["key"], "prompt_title": prompt["title"], "answer": answer, } ) return rows def write_conversation_csv( conversation: "DiscoveryConversation", user: "User | None", status: str, ) -> bool: """Rewrite the per-conversation CSV with the conversation's current answers. The write is atomic (temp file + ``os.replace``) so a crash mid-write can never leave a half-written file. Returns True on success, False if the write failed (failures are logged, never raised). """ saved_at = datetime.now(timezone.utc).isoformat() email = getattr(user, "email", "") or "" rows = _rows_for(conversation, email, status, saved_at) path = conversation_csv_path(conversation.id) tmp_path = f"{path}.tmp" try: os.makedirs(os.path.dirname(path), exist_ok=True) with open(tmp_path, "w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=CSV_FIELDS) writer.writeheader() writer.writerows(rows) os.replace(tmp_path, path) return True except OSError as exc: logger.warning( "Failed to write conversation CSV for %s: %s", conversation.id, exc ) try: if os.path.exists(tmp_path): os.remove(tmp_path) except OSError: pass return False def append_master( conversation: "DiscoveryConversation", user: "User | None", status: str, ) -> bool: """Append this save/complete event's rows to the master log. Writes the header row once, when the file is first created. Returns True on success, False on a logged (never raised) failure. """ saved_at = datetime.now(timezone.utc).isoformat() email = getattr(user, "email", "") or "" rows = _rows_for(conversation, email, status, saved_at) path = _master_path() try: directory = os.path.dirname(path) if directory: os.makedirs(directory, exist_ok=True) is_new = not os.path.exists(path) or os.path.getsize(path) == 0 with open(path, "a", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=CSV_FIELDS) if is_new: writer.writeheader() writer.writerows(rows) return True except OSError as exc: logger.warning( "Failed to append master CSV for %s: %s", conversation.id, exc ) return False def save_answers( conversation: "DiscoveryConversation", user: "User | None", status: str, ) -> None: """Persist both CSV artifacts for a conversation. Never raises.""" write_conversation_csv(conversation, user, status) append_master(conversation, user, status) def conversation_csv_text( conversation: "DiscoveryConversation", user: "User | None", status: str = "responses_saved", ) -> str: """Render a conversation's answers as CSV text, built from the DB row. Used by the download endpoint so it works even if the on-disk file was never written (DB stays the system of record). """ import io saved_at = datetime.now(timezone.utc).isoformat() email = getattr(user, "email", "") or "" rows = _rows_for(conversation, email, status, saved_at) buf = io.StringIO() writer = csv.DictWriter(buf, fieldnames=CSV_FIELDS) writer.writeheader() writer.writerows(rows) return buf.getvalue() def read_conversation_answers(conversation_id: str) -> dict[str, str]: """Read saved answers from the per-conversation CSV. Returns a ``{prompt_key: answer}`` dict, or ``{}`` if the file is missing or unreadable. The per-conversation file is rewritten in full on each save, so it always holds the latest answers. """ path = conversation_csv_path(conversation_id) answers: dict[str, str] = {} try: with open(path, newline="", encoding="utf-8") as fh: for row in csv.DictReader(fh): key = row.get("prompt_key") if key: answers[key] = row.get("answer", "") or "" except FileNotFoundError: return {} except OSError as exc: logger.warning( "Failed to read conversation CSV for %s: %s", conversation_id, exc ) return {} return answers