diff --git a/main.py b/main.py index c14abf6..a413143 100644 --- a/main.py +++ b/main.py @@ -105,16 +105,12 @@ def run_poll_loop(agent: EmailAgent, channel: EmailChannel, poll_interval: int) messages = channel.fetch_unread_messages() for msg in messages: - logger.info( - "Processing message from %s (thread: %s)", - msg["from"], - msg["thread_id"], - ) + logger.info("Processing message from %s", msg["from"]) try: reply = agent.process_message( - conversation_id=msg["thread_id"], parent_email=msg["from"], message_text=msg["body"], + inbound_message_id=msg["message_id"], ) if reply: channel.send_reply( diff --git a/src/agent/core.py b/src/agent/core.py index 564ea20..c2faee1 100644 --- a/src/agent/core.py +++ b/src/agent/core.py @@ -6,10 +6,10 @@ import re from datetime import datetime, timezone from ..models.conversation import ConversationState, ChatMessage -from ..models.registration import BookingDay +from ..models.registration import BookingDay, RegistrationData from ..providers.base import LLMProvider, LLMMessage from ..knowledge_base.loader import KnowledgeBase -from ..storage.json_store import ConversationStore +from ..storage.json_store import ConversationStore, normalize_email, _diff_registrations from ..notifications.notifier import AdminNotifier from .prompts import build_system_prompt @@ -17,11 +17,13 @@ logger = logging.getLogger(__name__) class EmailAgent: - """Processes one inbound message and returns the agent's reply text. + """Processes one inbound email and returns the agent's reply text. - All business logic (conversation flow, field extraction, validation, - storage, notifications) lives here. Channel-specific I/O is handled - by the caller. + Conversations are identified by the sender's normalized email address, so + a parent who composes a fresh email (instead of replying) continues their + existing conversation seamlessly. + + All business logic lives here; channel I/O is handled by the caller. """ def __init__( @@ -42,47 +44,67 @@ class EmailAgent: def process_message( self, - conversation_id: str, parent_email: str, message_text: str, + inbound_message_id: str = "", ) -> str: - """Process an inbound message and return the reply to send back. + """Process one inbound message and return the reply text. Args: - conversation_id: Unique thread identifier (from email headers). - parent_email: Sender's email address. + parent_email: Sender email address — used as conversation key. message_text: Stripped plain-text body of the inbound email. + inbound_message_id: Message-ID of the inbound email (stored for + reply threading headers; not used for conversation matching). Returns: - Reply text to send to the parent. + Reply text to send back to the parent. """ - # Load or create conversation state - state = self._store.load(conversation_id) + email_key = normalize_email(parent_email) + + # Load or create conversation state — keyed by email address + state = self._store.load(email_key) if state is None: state = ConversationState( - conversation_id=conversation_id, - parent_email=parent_email, + conversation_id=email_key, + parent_email=email_key, ) now = datetime.now(timezone.utc).isoformat() state.last_activity = now + if inbound_message_id: + state.last_inbound_message_id = inbound_message_id # Append the user's message to history state.messages.append(ChatMessage(role="user", content=message_text)) - # Build system prompt and conversation history for the LLM + # Route to the appropriate handler + if state.completed: + reply_text = self._handle_post_completion(state) + else: + reply_text = self._handle_registration(state) + + # Record the assistant reply and persist + state.messages.append(ChatMessage(role="assistant", content=reply_text)) + state.updated_at = now + self._store.save(state) + + return reply_text + + # ------------------------------------------------------------------ + # Registration flow + # ------------------------------------------------------------------ + + def _handle_registration(self, state: ConversationState) -> str: + """Drive the in-progress registration conversation.""" system = build_system_prompt(self._kb, state) - llm_messages = [ - LLMMessage(role=m.role, content=m.content) for m in state.messages - ] + llm_messages = [LLMMessage(role=m.role, content=m.content) for m in state.messages] - # Call the LLM try: response = self._provider.complete(system=system, messages=llm_messages) parsed = self._parse_llm_response(response.content) except Exception: - logger.exception("LLM call failed for conversation %s", conversation_id) - parsed = self._fallback_response(state) + logger.exception("LLM call failed for %s", state.conversation_id) + return self._fallback_message(state) reply_text: str = parsed.get("reply", "") updates: dict = parsed.get("updates", {}) or {} @@ -90,57 +112,100 @@ class EmailAgent: is_complete: bool = bool(parsed.get("registration_complete", False)) language: str = parsed.get("language", state.language) - # Apply extracted field updates self._apply_updates(state, updates) - - # Update conversation metadata state.flow_step = next_step state.language = language - state.updated_at = now - # Record the assistant reply - state.messages.append(ChatMessage(role="assistant", content=reply_text)) - - # Handle registration completion if is_complete and not state.completed: state.completed = True - registration_id = self._store.save_registration(state) + email_key, version = self._store.save_registration(state) try: self._notifier.notify_admin( registration=state.registration, - registration_id=registration_id, + registration_id=email_key, + version=version, conversation_id=state.conversation_id, channel="email", ) except Exception: - logger.exception("Failed to send admin notification for %s", registration_id) - logger.info("Registration complete: %s", registration_id) - - # Persist conversation state - self._store.save(state) + logger.exception("Failed to send admin notification for %s", email_key) + logger.info("Registration complete for %s", state.conversation_id) return reply_text # ------------------------------------------------------------------ - # Internal helpers + # Post-completion flow + # ------------------------------------------------------------------ + + def _handle_post_completion(self, state: ConversationState) -> str: + """Handle messages received after a registration is already complete.""" + system = build_system_prompt(self._kb, state) + llm_messages = [LLMMessage(role=m.role, content=m.content) for m in state.messages] + + try: + response = self._provider.complete(system=system, messages=llm_messages) + parsed = self._parse_llm_response(response.content) + except Exception: + logger.exception("LLM call failed (post-completion) for %s", state.conversation_id) + return self._fallback_message(state) + + reply_text: str = parsed.get("reply", "") + intent: str = parsed.get("intent", "question") + updates: dict = parsed.get("updates", {}) or {} + language: str = parsed.get("language", state.language) + state.language = language + + if intent == "update" and any(v is not None for v in updates.values()): + self._handle_registration_update(state, updates) + elif intent == "new_child": + # Reset registration so a fresh flow begins in the next message + state.registration = RegistrationData() + state.completed = False + state.flow_step = "child_name" + logger.info("Starting new child registration for %s", state.conversation_id) + + return reply_text + + def _handle_registration_update(self, state: ConversationState, updates: dict) -> None: + """Apply field updates, version the record, and notify the admin.""" + old_data = state.registration.to_dict() + self._apply_updates(state, updates) + new_data = state.registration.to_dict() + + change_summary = _diff_registrations(old_data, new_data) + if not change_summary: + return # Nothing actually changed + + email_key, version = self._store.save_registration_version(state, change_summary) + try: + self._notifier.notify_registration_update( + registration=state.registration, + registration_id=email_key, + version=version, + change_summary=change_summary, + conversation_id=state.conversation_id, + ) + except Exception: + logger.exception("Failed to send update notification for %s", email_key) + logger.info("Registration updated to v%d for %s", version, state.conversation_id) + + # ------------------------------------------------------------------ + # Shared helpers # ------------------------------------------------------------------ def _parse_llm_response(self, content: str) -> dict: - """Extract the JSON payload from the LLM's output.""" + """Extract the JSON payload from the LLM's raw output.""" text = content.strip() - # Strip markdown code fences if present - fence_pattern = re.compile(r"^```(?:json)?\s*\n(.*?)\n```\s*$", re.DOTALL) - match = fence_pattern.match(text) - if match: - text = match.group(1).strip() + fence_match = re.match(r"^```(?:json)?\s*\n(.*?)\n```\s*$", text, re.DOTALL) + if fence_match: + text = fence_match.group(1).strip() try: return json.loads(text) except json.JSONDecodeError: pass - # Fallback: find the first {...} block brace_match = re.search(r"\{.*\}", text, re.DOTALL) if brace_match: try: @@ -148,60 +213,51 @@ class EmailAgent: except json.JSONDecodeError: pass - logger.warning("Could not parse LLM response as JSON; using raw text as reply.") + logger.warning("Could not parse LLM response as JSON — using raw text as reply.") return { "reply": content, + "intent": "question", "updates": {}, "next_step": "greeting", "registration_complete": False, "language": "de", } - def _fallback_response(self, state: ConversationState) -> dict: - """Return a safe fallback when the LLM call fails.""" + def _fallback_message(self, state: ConversationState) -> str: if state.language == "en": - msg = ( + return ( "I'm sorry, I'm having a technical issue right now. " "Please try again in a moment or contact us directly." ) - else: - msg = ( - "Entschuldigung, ich habe gerade ein technisches Problem. " - "Bitte versuche es gleich nochmal oder kontaktiere uns direkt." - ) - return { - "reply": msg, - "updates": {}, - "next_step": state.flow_step, - "registration_complete": False, - "language": state.language, - } + return ( + "Entschuldigung, ich habe gerade ein technisches Problem. " + "Bitte versuche es gleich nochmal oder kontaktiere uns direkt." + ) def _apply_updates(self, state: ConversationState, updates: dict) -> None: """Write extracted field values into the RegistrationData object.""" reg = state.registration field_map = { - "child.fullName": lambda v: setattr(reg.child, "full_name", v), - "child.dateOfBirth": lambda v: setattr(reg.child, "date_of_birth", v), - "child.specialNeeds": lambda v: setattr(reg.child, "special_needs", v), - "parentGuardian.fullName": lambda v: ( + "child.fullName": lambda v: setattr(reg.child, "full_name", v), + "child.dateOfBirth": lambda v: setattr(reg.child, "date_of_birth", v), + "child.specialNeeds": lambda v: setattr(reg.child, "special_needs", v), + "parentGuardian.fullName": lambda v: ( setattr(reg.parent_guardian, "full_name", v), setattr(state, "parent_name", v), ), "parentGuardian.streetAddress": lambda v: setattr(reg.parent_guardian, "street_address", v), - "parentGuardian.postalCode": lambda v: setattr(reg.parent_guardian, "postal_code", str(v)), - "parentGuardian.city": lambda v: setattr(reg.parent_guardian, "city", v), - "parentGuardian.phone": lambda v: setattr(reg.parent_guardian, "phone", v), - "parentGuardian.email": lambda v: setattr(reg.parent_guardian, "email", v), - "emergencyContact.fullName": lambda v: setattr(reg.emergency_contact, "full_name", v), - "emergencyContact.phone": lambda v: setattr(reg.emergency_contact, "phone", v), + "parentGuardian.postalCode": lambda v: setattr(reg.parent_guardian, "postal_code", str(v)), + "parentGuardian.city": lambda v: setattr(reg.parent_guardian, "city", v), + "parentGuardian.phone": lambda v: setattr(reg.parent_guardian, "phone", v), + "parentGuardian.email": lambda v: setattr(reg.parent_guardian, "email", v), + "emergencyContact.fullName": lambda v: setattr(reg.emergency_contact, "full_name", v), + "emergencyContact.phone": lambda v: setattr(reg.emergency_contact, "phone", v), } for key, value in updates.items(): if value is None: continue - if key in field_map: field_map[key](value) elif key == "booking.playgroupTypes" and isinstance(value, list): diff --git a/src/agent/prompts.py b/src/agent/prompts.py index 8a43b92..e834379 100644 --- a/src/agent/prompts.py +++ b/src/agent/prompts.py @@ -6,7 +6,7 @@ from ..knowledge_base.loader import KnowledgeBase from ..models.conversation import ConversationState # --------------------------------------------------------------------------- -# Step descriptions help the model understand where it is in the flow. +# Step descriptions help the model understand where it is in the registration flow. # --------------------------------------------------------------------------- STEP_DESCRIPTIONS = { "greeting": "Greet the parent and detect their intent (registration vs. questions).", @@ -35,66 +35,28 @@ STEP_DESCRIPTIONS = { "complete": "Thank the parent, mention fees and next steps. Registration is done.", } - -def build_system_prompt(kb: KnowledgeBase, state: ConversationState) -> str: - """Return the full system prompt for the current conversation turn.""" - - kb_content = kb.get_all() - reg_json = json.dumps(state.registration.to_dict(), ensure_ascii=False, indent=2) - step_hint = STEP_DESCRIPTIONS.get(state.flow_step, "Continue the conversation.") - - return f"""You are the registration assistant for Spielgruppe Pumuckl, run by Familienverein Fällanden in Fällanden, Switzerland. You help parents register their children for the playgroup and answer questions about it. - -## Your Personality +_PERSONALITY = """## Your Personality - Warm, friendly, and helpful — like a caring playgroup staff member - Use informal "du" in German (never the formal "Sie") - Auto-detect the parent's language from their message; respond in the same language - Default language is German if unclear - Ask 1–2 questions at a time — never send an overwhelming form-like list -- Be patient and understanding; never make parents feel they made a mistake +- Be patient and understanding; never make parents feel they made a mistake""" -## Registration Flow (8 steps) -1. greeting — greet and detect intent -2. child_name — ask for child's full name -3. child_dob — ask for date of birth; validate age (indoor ≥2.5 yrs, outdoor ≥3 yrs) -4. playgroup_selection — present options, collect type(s) and day(s) -5. special_needs — ask about special needs / allergies / medical conditions -6. parent_contact — name, street address, postal code, city, phone, email -7. emergency_contact — emergency contact name and phone -8. confirmation — show full summary; ask to confirm; submit on confirmation -9. complete — thank parent, mention CHF 80 registration fee, monthly fees, and contacts +_CONTACTS = """## Admin Contacts +- Administration: Markus Graf — spielgruppen@familien-verein.ch — 079 261 16 37 +- Indoor leader: Andrea Sigrist — andrea.sigrist@gmx.net — 079 674 99 92 +- Outdoor leader: Barbara Gross — baba.laeubli@gmail.com — 078 761 19 64""" -**Current step: {state.flow_step}** -**What to do now: {step_hint}** - -At any point the parent may ask a question. Answer it from the knowledge base, then offer to continue the registration. - -## Current Registration Data (so far) -```json -{reg_json} -``` - -## Knowledge Base -Use the information below to answer parent questions accurately: - -{kb_content} - -## Playgroup Details +_PLAYGROUP_DETAILS = """## Playgroup Details - **Indoor (Innenspielgruppe)**: Mon / Wed / Thu, 09:00–11:30 | CHF 130/260/390 per month (1/2/3×/week) - **Outdoor Forest (Waldspielgruppe)**: Mon only, 09:00–14:00 (includes snack & lunch) | CHF 250/month - **One-time registration fee**: CHF 80 (first year); CHF 80 craft materials from second year - **Cleaning deposit (indoor only)**: CHF 50 (refundable) - **Sibling discount**: 10% per additional child -- **July & August**: fee-free +- **July & August**: fee-free""" -## Admin Contacts -- Administration: Markus Graf — spielgruppen@familien-verein.ch — 079 261 16 37 -- Indoor leader: Andrea Sigrist — andrea.sigrist@gmx.net — 079 674 99 92 -- Outdoor leader: Barbara Gross — baba.laeubli@gmail.com — 078 761 19 64 - ---- - -## CRITICAL: Response Format +_REGISTRATION_RESPONSE_FORMAT = """## CRITICAL: Response Format You MUST respond with **only** a valid JSON object — no markdown, no extra text outside the JSON. @@ -128,5 +90,138 @@ Rules: - Dates must be YYYY-MM-DD. Postal codes must be exactly 4 digits. - Valid days: "monday", "wednesday", "thursday" (indoor) or "monday" (outdoor). - `language` must be "de" or "en" based on the parent's message. +- The `reply` field must be natural, friendly, conversational text — not JSON and not a list of fields.""" + +_POST_COMPLETION_RESPONSE_FORMAT = """## CRITICAL: Response Format + +You MUST respond with **only** a valid JSON object — no markdown, no extra text outside the JSON. + +```json +{{ + "reply": "Your conversational message to the parent (plain text, NOT JSON)", + "intent": "question", + "updates": {{ + "child.fullName": "string or null", + "child.dateOfBirth": "YYYY-MM-DD or null", + "child.specialNeeds": "string or null", + "parentGuardian.fullName": "string or null", + "parentGuardian.streetAddress": "string or null", + "parentGuardian.postalCode": "4-digit string or null", + "parentGuardian.city": "string or null", + "parentGuardian.phone": "string or null", + "parentGuardian.email": "string or null", + "emergencyContact.fullName": "string or null", + "emergencyContact.phone": "string or null", + "booking.playgroupTypes": ["indoor", "outdoor"] or null, + "booking.selectedDays": [{{"day": "monday", "type": "indoor"}}] or null + }}, + "language": "de" +}} +``` + +`intent` values: +- `"question"` — parent is asking about fees, schedule, policies, etc. → answer from knowledge base; set `updates` to all nulls. +- `"update"` — parent explicitly wants to change their registration data → collect the new values in `updates`, confirm the change in `reply`. +- `"new_child"` — parent wants to register an additional child → treat as a new registration; begin from step child_name. + +Rules: +- Only set fields in `updates` when intent is `"update"` AND the parent has provided the new value in this message. +- Use `null` for all `updates` fields when intent is `"question"` or `"new_child"`. +- `language` must be "de" or "en" based on the parent's message. - The `reply` field must be natural, friendly, conversational text — not JSON and not a list of fields. +- If you are unsure of the parent's intent, ask a clarifying question and set intent to `"question"`.""" + + +def build_system_prompt(kb: KnowledgeBase, state: ConversationState) -> str: + """Return the system prompt appropriate for the current conversation state.""" + if state.completed: + return _build_post_completion_prompt(kb, state) + return _build_registration_prompt(kb, state) + + +def _build_registration_prompt(kb: KnowledgeBase, state: ConversationState) -> str: + """System prompt for an in-progress registration conversation.""" + kb_content = kb.get_all() + reg_json = json.dumps(state.registration.to_dict(), ensure_ascii=False, indent=2) + step_hint = STEP_DESCRIPTIONS.get(state.flow_step, "Continue the conversation.") + + return f"""You are the registration assistant for Spielgruppe Pumuckl, run by Familienverein Fällanden in Fällanden, Switzerland. You help parents register their children for the playgroup and answer questions about it. + +{_PERSONALITY} + +## Registration Flow (8 steps) +1. greeting — greet and detect intent +2. child_name — ask for child's full name +3. child_dob — ask for date of birth; validate age (indoor ≥2.5 yrs, outdoor ≥3 yrs) +4. playgroup_selection — present options, collect type(s) and day(s) +5. special_needs — ask about special needs / allergies / medical conditions +6. parent_contact — name, street address, postal code, city, phone, email +7. emergency_contact — emergency contact name and phone +8. confirmation — show full summary; ask to confirm; submit on confirmation +9. complete — thank parent, mention CHF 80 registration fee, monthly fees, and contacts + +**Current step: {state.flow_step}** +**What to do now: {step_hint}** + +At any point the parent may ask a question. Answer it from the knowledge base, then offer to continue the registration. + +## Current Registration Data (so far) +```json +{reg_json} +``` + +## Knowledge Base +Use the information below to answer parent questions accurately: + +{kb_content} + +{_PLAYGROUP_DETAILS} + +{_CONTACTS} + +--- + +{_REGISTRATION_RESPONSE_FORMAT} +""" + + +def _build_post_completion_prompt(kb: KnowledgeBase, state: ConversationState) -> str: + """System prompt for a conversation where registration is already complete.""" + kb_content = kb.get_all() + reg_json = json.dumps(state.registration.to_dict(), ensure_ascii=False, indent=2) + child_name = state.registration.child.full_name or "their child" + + return f"""You are the registration assistant for Spielgruppe Pumuckl, run by Familienverein Fällanden in Fällanden, Switzerland. + +{_PERSONALITY} + +## Context: Registration Already Complete +This parent has already completed registration for {child_name}. Their current registration data is: + +```json +{reg_json} +``` + +The parent is contacting you again. Your job is to: +1. Detect their **intent**: are they asking a question, requesting a change to their registration, or registering another child? +2. Respond helpfully and warmly. +3. If they want to **update** their registration, confirm exactly what they want to change and include the new values in `updates`. +4. If they are asking a **question**, answer from the knowledge base. +5. If they want to register a **new child**, let them know you'll start a new registration and guide them from the beginning. + +When handling update requests: +- Confirm the change explicitly before reporting it as done ("So you'd like to change X to Y — is that right?"). +- Once confirmed, include the new value in `updates` so it can be saved. +- Let the parent know the playgroup team will be informed of the change. + +## Knowledge Base +{kb_content} + +{_PLAYGROUP_DETAILS} + +{_CONTACTS} + +--- + +{_POST_COMPLETION_RESPONSE_FORMAT} """ diff --git a/src/channels/email_channel.py b/src/channels/email_channel.py index 778807a..cc0b5f9 100644 --- a/src/channels/email_channel.py +++ b/src/channels/email_channel.py @@ -2,9 +2,14 @@ Handles: - Polling the inbox for unread messages (IMAP) -- Thread tracking via Message-ID / In-Reply-To / References headers -- Sending reply emails (SMTP) with proper threading headers +- Conversation matching by sender email address (NOT by thread headers) +- Sending reply emails (SMTP) with proper threading headers for email clients - Stripping quoted reply text so the agent only sees the new content + +Threading headers (Message-ID, In-Reply-To, References) are preserved for +outbound replies so messages appear threaded in Gmail/Outlook, but they are +NOT used to identify which conversation an incoming message belongs to. +Conversation matching is exclusively by normalized sender email address. """ import email @@ -126,13 +131,15 @@ class EmailChannel: """Poll the inbox and return all unread messages as structured dicts. Each dict contains: - from — sender email address + from — sender email address (use this as conversation key) subject — decoded subject line - message_id — Message-ID header of this email - in_reply_to — In-Reply-To header (may be empty) - references — References header (may be empty) - thread_id — canonical ID for the email thread + message_id — Message-ID of this inbound email (for reply threading) + in_reply_to — In-Reply-To header (for reply threading, may be empty) + references — References header (for reply threading, may be empty) body — stripped plain-text body (quoted text removed) + + Note: ``thread_id`` is no longer returned. Conversation matching is done + by ``from`` (sender email address), not by threading headers. """ messages: list[dict] = [] try: @@ -161,8 +168,6 @@ class EmailChannel: imap.store(num, "+FLAGS", "\\Seen") continue - thread_id = self._resolve_thread_id(message_id, in_reply_to, references) - messages.append( { "from": from_addr, @@ -170,7 +175,6 @@ class EmailChannel: "message_id": message_id, "in_reply_to": in_reply_to, "references": references, - "thread_id": thread_id, "body": body, } ) @@ -274,16 +278,3 @@ class EmailChannel: conn.login(self._username, self._password) return conn - @staticmethod - def _resolve_thread_id(message_id: str, in_reply_to: str, references: str) -> str: - """Determine the canonical thread ID from email threading headers. - - The root of the thread is the first message ever sent — its ID is the - first token in the References header (oldest first convention). - """ - if references: - first_ref = references.strip().split()[0] - return first_ref - if in_reply_to: - return in_reply_to.strip() - return message_id.strip() or f"" diff --git a/src/models/conversation.py b/src/models/conversation.py index 0574030..db8a565 100644 --- a/src/models/conversation.py +++ b/src/models/conversation.py @@ -1,4 +1,4 @@ -"""Conversation state model — persisted per email thread.""" +"""Conversation state model — persisted per sender email address.""" from dataclasses import dataclass, field from datetime import datetime, timezone @@ -20,7 +20,7 @@ class ChatMessage: @dataclass class ConversationState: - conversation_id: str + conversation_id: str # normalized sender email address language: str = "de" # "de" or "en" flow_step: str = "greeting" # current step in registration flow registration: RegistrationData = field(default_factory=RegistrationData) @@ -32,6 +32,9 @@ class ConversationState: last_activity: str = field(default_factory=_now) completed: bool = False reminder_count: int = 0 + # Most recent inbound Message-ID — used for reply threading headers only, + # NOT for conversation matching (which is always by email address). + last_inbound_message_id: str = "" def to_dict(self) -> dict: return { @@ -50,6 +53,7 @@ class ConversationState: "last_activity": self.last_activity, "completed": self.completed, "reminder_count": self.reminder_count, + "last_inbound_message_id": self.last_inbound_message_id, } @classmethod @@ -73,4 +77,5 @@ class ConversationState: state.last_activity = data.get("last_activity", "") state.completed = data.get("completed", False) state.reminder_count = data.get("reminder_count", 0) + state.last_inbound_message_id = data.get("last_inbound_message_id", "") return state diff --git a/src/notifications/notifier.py b/src/notifications/notifier.py index 1bf9c78..b05126a 100644 --- a/src/notifications/notifier.py +++ b/src/notifications/notifier.py @@ -1,4 +1,4 @@ -"""Admin email notifications sent when a registration is completed.""" +"""Admin email notifications — new registrations and registration updates.""" import logging import smtplib @@ -17,11 +17,13 @@ _ADMIN_CC_EMAIL = "spielgruppen@familien-verein.ch" class AdminNotifier: - """Sends formatted admin notification emails upon registration completion. + """Sends formatted admin notification emails. - The SMTP credentials are re-used from the agent's outbound email config. - When *smtp_host* is empty the notifier logs the notification and skips - sending (useful for local development / testing). + Handles two notification types: + - New registration completed → "New Registration: …" + - Existing registration updated → "Registration Updated: …" (with field diff) + + When *smtp_host* is empty the notifier logs and skips sending (dev mode). """ def __init__( @@ -48,23 +50,42 @@ class AdminNotifier: self, registration: RegistrationData, registration_id: str, + version: int, conversation_id: str, channel: str, ) -> None: - """Send notification email(s) for a completed registration.""" + """Send notification for a newly completed registration (version 1).""" types = registration.booking.playgroup_types - - to_addresses: list[str] = [] - if "indoor" in types: - to_addresses.append(_INDOOR_EMAIL) - if "outdoor" in types: - to_addresses.append(_OUTDOOR_EMAIL) + to_addresses = self._recipients_for(types) subject = ( f"New Registration: {registration.child.full_name} " f"for {self._format_types(types)}" ) - body = self._build_body(registration, registration_id, channel) + body = self._build_new_body(registration, registration_id, version, channel) + + self._send( + to=to_addresses, + cc=[_ADMIN_CC_EMAIL], + subject=subject, + body=body, + reply_to=registration.parent_guardian.email or "", + ) + + def notify_registration_update( + self, + registration: RegistrationData, + registration_id: str, + version: int, + change_summary: dict, + conversation_id: str, + ) -> None: + """Send notification when an existing registration is updated.""" + types = registration.booking.playgroup_types + to_addresses = self._recipients_for(types) + + subject = f"Registration Updated: {registration.child.full_name}" + body = self._build_update_body(registration, registration_id, version, change_summary) self._send( to=to_addresses, @@ -78,6 +99,15 @@ class AdminNotifier: # Formatting helpers # ------------------------------------------------------------------ + @staticmethod + def _recipients_for(types: list[str]) -> list[str]: + recipients = [] + if "indoor" in types: + recipients.append(_INDOOR_EMAIL) + if "outdoor" in types: + recipients.append(_OUTDOOR_EMAIL) + return recipients + @staticmethod def _format_types(types: list[str]) -> str: has_indoor = "indoor" in types @@ -127,20 +157,32 @@ class AdminNotifier: @staticmethod def _format_days(registration: RegistrationData) -> str: - day_map = { - "monday": "Monday", - "wednesday": "Wednesday", - "thursday": "Thursday", - } + day_map = {"monday": "Monday", "wednesday": "Wednesday", "thursday": "Thursday"} return ", ".join( f"{day_map.get(d.day, d.day.capitalize())} ({d.type})" for d in registration.booking.selected_days ) - def _build_body( + @staticmethod + def _format_change_summary(change_summary: dict) -> str: + """Render field changes as a human-readable list.""" + lines = [] + for field_path, values in sorted(change_summary.items()): + old_val, new_val = values["old"], values["new"] + lines.append(f" {field_path}:") + lines.append(f" Old: {old_val}") + lines.append(f" New: {new_val}") + return "\n".join(lines) if lines else " (no changes detected)" + + # ------------------------------------------------------------------ + # Email body builders + # ------------------------------------------------------------------ + + def _build_new_body( self, registration: RegistrationData, registration_id: str, + version: int, channel: str, ) -> str: now = datetime.utcnow() @@ -154,7 +196,7 @@ class AdminNotifier: "\n" f"Submitted: {now.strftime('%d.%m.%Y')} at {now.strftime('%H:%M')} UTC\n" f"Channel: {channel.title()}\n" - f"Registration ID: {registration_id}\n" + f"Registration ID: {registration_id} (Version {version})\n" "\n" "-----------------------------------------------\n" "CHILD INFORMATION\n" @@ -193,6 +235,47 @@ class AdminNotifier: "This registration was submitted via the automated registration assistant.\n" ) + def _build_update_body( + self, + registration: RegistrationData, + registration_id: str, + version: int, + change_summary: dict, + ) -> str: + now = datetime.utcnow() + pg = registration.parent_guardian + + return ( + "===============================================\n" + "REGISTRATION UPDATE\n" + "===============================================\n" + "\n" + f"Updated: {now.strftime('%d.%m.%Y')} at {now.strftime('%H:%M')} UTC\n" + f"Registration ID: {registration_id} (Version {version})\n" + f"Child: {registration.child.full_name}\n" + f"Parent Email: {pg.email}\n" + "\n" + "-----------------------------------------------\n" + "WHAT CHANGED\n" + "-----------------------------------------------\n" + f"{self._format_change_summary(change_summary)}\n" + "\n" + "-----------------------------------------------\n" + "CURRENT REGISTRATION (after update)\n" + "-----------------------------------------------\n" + f"Playgroup: {self._format_types(registration.booking.playgroup_types)}\n" + f"Days: {self._format_days(registration)}\n" + f"Monthly Fee: {self._calculate_monthly_fee(registration)}\n" + "\n" + f"Parent: {pg.full_name}\n" + f"Address: {pg.street_address}, {pg.postal_code} {pg.city}\n" + f"Phone: {pg.phone}\n" + "\n" + "===============================================\n" + "\n" + "This update was submitted via the automated registration assistant.\n" + ) + # ------------------------------------------------------------------ # SMTP dispatch # ------------------------------------------------------------------ @@ -236,6 +319,6 @@ class AdminNotifier: server.login(self._username, self._password) server.sendmail(self._from_email, all_recipients, msg.as_string()) server.quit() - logger.info("Admin notification sent to %s", all_recipients) + logger.info("Notification sent to %s", all_recipients) except Exception: - logger.exception("Failed to send admin notification to %s", all_recipients) + logger.exception("Failed to send notification to %s", all_recipients) diff --git a/src/storage/json_store.py b/src/storage/json_store.py index 7917f8d..5012305 100644 --- a/src/storage/json_store.py +++ b/src/storage/json_store.py @@ -1,30 +1,95 @@ -"""File-based JSON storage for conversations and completed registrations.""" +"""File-based JSON storage for conversations and completed registrations. + +Conversations are keyed by the sender's normalized email address so that a +parent who sends a new email (instead of replying) continues the same +conversation. Completed registrations are stored with versioning so every +update produces a new numbered version rather than overwriting the original. + +Directory layout:: + + data/ + conversations/ + parent_at_example.com.json # one file per unique sender address + registrations/ + parent_at_example.com/ + v1_2024-09-15T10-30-00Z.json # initial registration + v2_2024-10-03T14-22-10Z.json # updated registration + current.json # copy of the latest version +""" import json import logging -import uuid +import re from datetime import datetime, timezone from pathlib import Path from ..models.conversation import ConversationState -from ..models.registration import RegistrationData logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def normalize_email(email: str) -> str: + """Return a canonical email address for matching and storage. + + Lowercases and strips whitespace. ``Maria@Example.com`` → ``maria@example.com``. + """ + return email.strip().lower() + + +def _email_to_filename(email: str) -> str: + """Convert a normalized email address to a safe filename stem. + + ``parent@example.com`` → ``parent_at_example.com`` + """ + return normalize_email(email).replace("@", "_at_") + + def _now() -> str: return datetime.now(timezone.utc).isoformat() +def _timestamp_for_filename() -> str: + """Return a filesystem-safe ISO-8601-ish timestamp (no colons).""" + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H-%M-%SZ") + + +def _diff_registrations(old: dict, new: dict) -> dict[str, tuple]: + """Return a mapping of field_path → (old_value, new_value) for changed fields.""" + changes: dict[str, tuple] = {} + + def _flatten(d: dict, prefix: str = "") -> dict: + out: dict = {} + for k, v in d.items(): + key = f"{prefix}.{k}" if prefix else k + if isinstance(v, dict): + out.update(_flatten(v, key)) + else: + out[key] = v + return out + + old_flat = _flatten(old) + new_flat = _flatten(new) + + all_keys = set(old_flat) | set(new_flat) + for key in sorted(all_keys): + o = old_flat.get(key) + n = new_flat.get(key) + if o != n: + changes[key] = (o, n) + + return changes + + +# --------------------------------------------------------------------------- +# ConversationStore +# --------------------------------------------------------------------------- + class ConversationStore: - """Persists ConversationState objects as JSON files on disk. - - Directory layout:: - - data/ - conversations/ # one file per email thread - registrations/ # one file per completed registration - """ + """Persists ConversationState and registration versions on disk.""" def __init__(self, data_dir: Path) -> None: self._conversations_dir = data_dir / "conversations" @@ -33,40 +98,43 @@ class ConversationStore: self._registrations_dir.mkdir(parents=True, exist_ok=True) # ------------------------------------------------------------------ - # Conversation CRUD + # Conversation CRUD — keyed by normalized email address # ------------------------------------------------------------------ - def load(self, conversation_id: str) -> ConversationState | None: - """Load a conversation by ID. Returns None if not found.""" - path = self._conversation_path(conversation_id) + def load(self, email_address: str) -> ConversationState | None: + """Load a conversation by sender email address. Returns None if not found.""" + path = self._conversation_path(email_address) if not path.exists(): return None try: data = json.loads(path.read_text(encoding="utf-8")) return ConversationState.from_dict(data) except Exception: - logger.exception("Failed to load conversation %s", conversation_id) + logger.exception("Failed to load conversation for %s", email_address) return None + # Alias for clarity in call sites that emphasise the email-lookup semantic + find_by_email = load + def save(self, state: ConversationState) -> None: """Persist a conversation state to disk.""" - path = self._conversation_path(state.conversation_id) + path = self._conversation_path(state.parent_email or state.conversation_id) try: path.write_text( json.dumps(state.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8", ) except Exception: - logger.exception("Failed to save conversation %s", state.conversation_id) + logger.exception("Failed to save conversation for %s", state.conversation_id) - def delete(self, conversation_id: str) -> None: - """Remove a conversation file (e.g., after retention period expires).""" - path = self._conversation_path(conversation_id) + def delete(self, email_address: str) -> None: + """Remove a conversation file.""" + path = self._conversation_path(email_address) if path.exists(): path.unlink() def list_incomplete(self) -> list[ConversationState]: - """Return all conversations that are not yet completed.""" + """Return all conversations that have not yet been completed.""" states: list[ConversationState] = [] for path in self._conversations_dir.glob("*.json"): try: @@ -79,65 +147,124 @@ class ConversationStore: return states # ------------------------------------------------------------------ - # Registration storage + # Versioned registration storage # ------------------------------------------------------------------ - def save_registration(self, state: ConversationState) -> str: - """Write the completed registration to disk and return its ID.""" - registration_id = str(uuid.uuid4()) - record = state.registration.to_dict() - record["metadata"] = { - "registrationId": registration_id, - "submittedAt": _now(), - "channel": "email", - "conversationId": state.conversation_id, + def save_registration(self, state: ConversationState) -> tuple[str, int]: + """Store the first version of a completed registration. + + Returns: + Tuple of (registration_dir_key, version_number). + """ + email_key = _email_to_filename(state.parent_email or state.conversation_id) + reg_dir = self._registrations_dir / email_key + reg_dir.mkdir(parents=True, exist_ok=True) + + version = 1 + record = self._build_record(state.registration.to_dict(), version, state) + + self._write_version(reg_dir, version, record) + logger.info("Saved initial registration v%d for %s", version, email_key) + return email_key, version + + def save_registration_version( + self, + state: ConversationState, + change_summary: dict[str, tuple], + ) -> tuple[str, int]: + """Store an updated registration as a new version. + + Args: + state: Current conversation state with updated registration data. + change_summary: Dict of field_path → (old_value, new_value). + + Returns: + Tuple of (registration_dir_key, new_version_number). + """ + email_key = _email_to_filename(state.parent_email or state.conversation_id) + reg_dir = self._registrations_dir / email_key + reg_dir.mkdir(parents=True, exist_ok=True) + + history = self.get_registration_history(state.parent_email or state.conversation_id) + version = len(history) + 1 + + record = self._build_record(state.registration.to_dict(), version, state) + record["metadata"]["changeSummary"] = { + k: {"old": v[0], "new": v[1]} for k, v in change_summary.items() } - path = self._registrations_dir / f"{registration_id}.json" - try: - path.write_text( - json.dumps(record, ensure_ascii=False, indent=2), - encoding="utf-8", - ) - logger.info("Saved registration %s", registration_id) - except Exception: - logger.exception("Failed to save registration %s", registration_id) + self._write_version(reg_dir, version, record) + logger.info("Saved registration v%d for %s", version, email_key) + return email_key, version - return registration_id + def get_registration_history(self, email_address: str) -> list[dict]: + """Return all registration versions for an email address, oldest first.""" + email_key = _email_to_filename(email_address) + reg_dir = self._registrations_dir / email_key + if not reg_dir.exists(): + return [] - def load_registration(self, registration_id: str) -> dict | None: - """Load a completed registration record by ID.""" - path = self._registrations_dir / f"{registration_id}.json" - if not path.exists(): - return None - try: - return json.loads(path.read_text(encoding="utf-8")) - except Exception: - logger.exception("Failed to load registration %s", registration_id) - return None - - def list_registrations(self) -> list[dict]: - """Return all completed registration records.""" records: list[dict] = [] - for path in sorted(self._registrations_dir.glob("*.json")): + for path in sorted(reg_dir.glob("v*.json")): try: records.append(json.loads(path.read_text(encoding="utf-8"))) except Exception: - logger.warning("Could not read registration file %s", path) + logger.warning("Could not read registration version %s", path) + return records + + def get_current_registration(self, email_address: str) -> dict | None: + """Return the latest registration version for an email address.""" + email_key = _email_to_filename(email_address) + current_path = self._registrations_dir / email_key / "current.json" + if not current_path.exists(): + return None + try: + return json.loads(current_path.read_text(encoding="utf-8")) + except Exception: + logger.exception("Failed to load current registration for %s", email_address) + return None + + def list_registrations(self) -> list[dict]: + """Return the current (latest) registration for every known email address.""" + records: list[dict] = [] + for email_dir in sorted(self._registrations_dir.iterdir()): + if not email_dir.is_dir(): + continue + current = email_dir / "current.json" + if current.exists(): + try: + records.append(json.loads(current.read_text(encoding="utf-8"))) + except Exception: + logger.warning("Could not read %s", current) return records # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ - def _conversation_path(self, conversation_id: str) -> Path: - """Sanitise the conversation ID to produce a safe filename.""" - safe = ( - conversation_id - .replace("/", "_") - .replace("\\", "_") - .replace("<", "") - .replace(">", "") - .replace("@", "_at_") - )[:200] - return self._conversations_dir / f"{safe}.json" + def _conversation_path(self, email_address: str) -> Path: + return self._conversations_dir / f"{_email_to_filename(email_address)}.json" + + @staticmethod + def _build_record(reg_data: dict, version: int, state: ConversationState) -> dict: + record = dict(reg_data) + record["metadata"] = { + "version": version, + "submittedAt": _now(), + "channel": "email", + "parentEmail": state.parent_email, + "conversationId": state.conversation_id, + } + return record + + @staticmethod + def _write_version(reg_dir: Path, version: int, record: dict) -> None: + ts = _timestamp_for_filename() + version_path = reg_dir / f"v{version}_{ts}.json" + version_path.write_text( + json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8" + ) + # Keep current.json as a plain copy of the latest version + (reg_dir / "current.json").write_text( + json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8" + )