Files
impactflow_discovery/app/schemas.py
Claude 866bd60225 Persist discovery answers to CSV for reprocessing and recovery
Save each discovery conversation's prompts and answers to durable CSV
files (per-conversation + append-only master log) on both save and
completion, so answers survive an extraction error, can be re-fed to the
AI, and can be reviewed/resumed by the user.

- app/services/answer_store.py: canonical prompt list + atomic CSV writes,
  master append, and read-back helpers (DB stays system of record; CSV
  failures are logged, never fatal).
- discovery router: write CSV on /respond and /complete; new endpoints
  GET /answers, GET /answers.csv, POST /reprocess (shared extraction
  helper; locked profiles return 409).
- discovery.html: prefill/resume from saved answers after an error and a
  "Re-run analysis" button wired to /reprocess.
- scripts/reprocess_csv.py: offline CLI to re-run extraction from a CSV
  (print or --write-db).
- QUESTIONS_DIR / QUESTIONS_MASTER_CSV config, .gitignore, README, tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dg6XWUwprmP5QCL18HxssY
2026-06-19 01:06:40 +00:00

254 lines
6.2 KiB
Python

"""Pydantic request/response models for the discovery API."""
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
class StartResponse(BaseModel):
conversation_id: str
class RespondRequest(BaseModel):
prompt_alive: str = ""
prompt_friction: str = ""
prompt_pull: str = ""
prompt_recognition: str = ""
prompt_future: str = ""
prompt_goals_short: str = ""
prompt_goals_long: str = ""
class RespondResponse(BaseModel):
conversation_id: str
status: str
class AnswerItem(BaseModel):
prompt_key: str
prompt_title: str
answer: str = ""
class AnswersResponse(BaseModel):
"""The saved prompts and answers for a conversation, for review/resume."""
conversation_id: str
started_at: datetime
completed_at: Optional[datetime] = None
answers: list[AnswerItem] = []
class Confidence(BaseModel):
triad: Optional[str] = None
type: Optional[str] = None
variant: Optional[str] = None
ikigai: Optional[str] = None
class ProfileResponse(BaseModel):
id: str
user_id: str
conversation_id: str
generated_at: datetime
triad: Optional[str] = None
probable_type: Optional[int] = None
wing: Optional[int] = None
instinctual_variant: Optional[str] = None
instinctual_stack: Optional[str] = None
love_summary: Optional[str] = None
strength_summary: Optional[str] = None
mission_summary: Optional[str] = None
vocation_summary: Optional[str] = None
overlap_narrative: Optional[str] = None
short_term_goals: Optional[str] = None
long_term_goals: Optional[str] = None
confidence: Optional[Confidence] = None
locked: bool = False
extraction_notes: Optional[str] = None
class ProfileUpdate(BaseModel):
"""Partial edit of a profile's prose. Only fields explicitly provided are
updated (see exclude_unset in the router). The AI's structural inference
(triad/type/wing/variant) and confidence are not editable here."""
love_summary: Optional[str] = None
strength_summary: Optional[str] = None
mission_summary: Optional[str] = None
vocation_summary: Optional[str] = None
overlap_narrative: Optional[str] = None
short_term_goals: Optional[str] = None
long_term_goals: Optional[str] = None
class ConfirmResponse(BaseModel):
status: str
# -- Phase 2: AI coach reflection loop ---------------------------------------
class ReflectRequest(BaseModel):
# Empty/omitted starts the loop (the coach's opening reflection).
message: str = ""
# Phase 5: optional refinement focus, e.g. "goals" to steer the coach
# toward sharpening the near/long-term goals.
focus: str = ""
class ReflectionMessageOut(BaseModel):
role: str # "coach" or "person"
content: str
sequence: int
created_at: datetime
class ReflectTurnResponse(BaseModel):
"""One coach turn, plus the (possibly revised) profile."""
message: ReflectionMessageOut
profile: ProfileResponse
revised: bool = False
revision_note: Optional[str] = None
class ReflectionThreadResponse(BaseModel):
messages: list[ReflectionMessageOut]
profile: ProfileResponse
class ProfileRevisionOut(BaseModel):
id: str
source: str # "extraction" | "reflection" | "manual_edit"
fields: dict
note: Optional[str] = None
created_at: datetime
# -- Phase 3: coaching preferences + check-ins -------------------------------
class CoachingPreferencesOut(BaseModel):
coaching_frequency: str
coaching_style: str
misalignment_threshold: str
friction_tolerance: str
prefer_questions_over_directives: bool
time_of_day_preference: str
auto_generated: bool
updated_at: datetime
class CoachingPreferencesUpdate(BaseModel):
coaching_frequency: Optional[str] = None
coaching_style: Optional[str] = None
misalignment_threshold: Optional[str] = None
friction_tolerance: Optional[str] = None
prefer_questions_over_directives: Optional[bool] = None
time_of_day_preference: Optional[str] = None
class CheckinOut(BaseModel):
id: str
body: str
created_at: datetime
still_valid: Optional[bool] = None
response_note: Optional[str] = None
acknowledged_at: Optional[datetime] = None
class CheckinRespondRequest(BaseModel):
still_valid: bool
note: str = ""
class RunCheckinsResponse(BaseModel):
considered: int
generated: int
# -- Phase 4: task-to-goal integration ---------------------------------------
class FoundationOut(BaseModel):
key: str
label: str
text: Optional[str] = None
class TaskMappingCreate(BaseModel):
external_task_id: str
foundation: str
minutes: int = 0
task_label: str = ""
# When the work happened; defaults to now if omitted.
occurred_at: Optional[datetime] = None
class TaskMappingOut(BaseModel):
id: str
external_task_id: str
task_label: Optional[str] = None
foundation: str
minutes: int
occurred_at: datetime
created_at: datetime
class FoundationPattern(BaseModel):
foundation: str
label: str
minutes: int
task_count: int
last_at: Optional[datetime] = None
share: float
class WorkPatternsOut(BaseModel):
window_days: int
total_minutes: int
by_foundation: list[FoundationPattern]
neglected: list[str]
# -- Phase 5: iteration & polish ---------------------------------------------
class GoalHistoryEntry(BaseModel):
value: Optional[str] = None
source: Optional[str] = None
at: Optional[datetime] = None
class GoalHistoryOut(BaseModel):
short_term_goals: list[GoalHistoryEntry]
long_term_goals: list[GoalHistoryEntry]
class SuggestFoundationRequest(BaseModel):
task_label: str
class SuggestFoundationOut(BaseModel):
foundation: str
label: str
rationale: str = ""
confidence: str = "low"
class ConversationResponse(BaseModel):
id: str
user_id: str
started_at: datetime
completed_at: Optional[datetime] = None
prompt_alive: Optional[str] = None
prompt_friction: Optional[str] = None
prompt_pull: Optional[str] = None
prompt_recognition: Optional[str] = None
prompt_future: Optional[str] = None
prompt_goals_short: Optional[str] = None
prompt_goals_long: Optional[str] = None
model_config = {"from_attributes": True}