Spaces:
Build error
Build error
File size: 8,488 Bytes
67efe6c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 | from __future__ import annotations
from typing import Dict, List
import re
from .email_service import send_contact_email
from .suggestions import (
default_suggestions,
suggestions_for_intent,
suggestions_from_text,
)
class FlowManager:
"""
Manages lightweight session state for:
- Contact flow (collect message + email)
- Language flow (ask region -> ask language)
"""
def __init__(self):
self.sessions: Dict[str, Dict] = {}
# ---------- Suggestions ----------
def default_suggestions(self) -> List[str]:
return default_suggestions()
# ---------- Session helpers ----------
def _get(self, session_id: str) -> Dict:
if session_id not in self.sessions:
self.sessions[session_id] = {
"mode": "normal", # normal | contact_wait_msg | contact_wait_email | lang_wait_region | lang_wait_language
"contact_msg": None,
"lang": None, # preferred language (string), e.g. "Sinhala"
"region": None, # store region (string) if provided
}
return self.sessions[session_id]
# ---------- Intents ----------
def _detect_intents(self, text: str) -> List[str]:
t = (text or "").lower()
intents: List[str] = []
if any(k in t for k in ["contact", "support", "help desk", "reach", "email us", "contact us"]):
intents.append("contact")
if any(k in t for k in ["change response language", "change language", "language", "switch language", "translate"]):
intents.append("language")
if any(k in t for k in ["service", "services", "what do you do", "features", "what is syslink", "about"]):
intents.append("services")
return intents or ["rag"]
# ---------- Main entry ----------
def handle_message(self, session_id: str, user_message: str) -> Dict:
"""
Returns dict:
{
"action": "flow" | "rag",
"answer": "...",
"suggestions": [...],
"lang": optional preferred language for RAG
}
"""
state = self._get(session_id)
msg = (user_message or "").strip()
# 1) In-flow handling first
if state["mode"].startswith("contact_"):
return self._handle_contact_flow(state, msg)
if state["mode"].startswith("lang_"):
return self._handle_language_flow(state, msg)
# 2) Detect intent(s)
intents = self._detect_intents(msg)
# If user typed a custom prompt, provide new related suggestions
dynamic_suggestions = suggestions_from_text(msg)
# 3) Multi-intent handling
# If user asks "contact + language" together, do language first, then contact
if "contact" in intents and "language" in intents:
state["mode"] = "lang_wait_region"
state["next_after_lang"] = "contact"
return {
"action": "flow",
"answer": "Sure — first, tell me your region/country.",
"suggestions": [],
"lang": state.get("lang"),
}
# 4) Single intent handling
if "language" in intents:
state["mode"] = "lang_wait_region"
state["next_after_lang"] = None
return {
"action": "flow",
"answer": "Sure — tell me your region/country.",
"suggestions": [],
"lang": state.get("lang"),
}
if "contact" in intents:
state["mode"] = "contact_wait_msg"
return {
"action": "flow",
"answer": "Sure — please type your message for our team.",
"suggestions": suggestions_for_intent("contact"),
"lang": state.get("lang"),
}
if "services" in intents:
return {
"action": "rag",
"answer": "",
"suggestions": suggestions_for_intent("services"),
"lang": state.get("lang"),
}
# 5) Default: RAG
return {
"action": "rag",
"answer": "",
"suggestions": dynamic_suggestions,
"lang": state.get("lang"),
}
# ---------- Contact flow ----------
def _handle_contact_flow(self, state: Dict, msg: str) -> Dict:
if state["mode"] == "contact_wait_msg":
state["contact_msg"] = msg
state["mode"] = "contact_wait_email"
return {
"action": "flow",
"answer": "Thanks. Now please enter your email address.",
"suggestions": [],
"lang": state.get("lang"),
}
if state["mode"] == "contact_wait_email":
if not self._is_valid_email(msg):
return {
"action": "flow",
"answer": "That email doesn’t look valid. Please type a valid email (example: name@gmail.com).",
"suggestions": [],
"lang": state.get("lang"),
}
email = msg
message = state.get("contact_msg") or ""
result = send_contact_email(user_email=email, user_message=message)
# Reset flow state
state["mode"] = "normal"
state["contact_msg"] = None
if result.get("ok"):
return {
"action": "flow",
"answer": "✅ Sent! Thanks — our team will contact you soon.",
"suggestions": default_suggestions(),
"lang": state.get("lang"),
}
return {
"action": "flow",
"answer": (
"✅ I saved your message, but email sending isn’t configured yet on the server.\n"
"Our team can still contact you using the details you provided."
),
"suggestions": default_suggestions(),
"lang": state.get("lang"),
}
# fallback
state["mode"] = "normal"
return {"action": "rag", "answer": "", "suggestions": default_suggestions(), "lang": state.get("lang")}
def _is_valid_email(self, s: str) -> bool:
return bool(re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", (s or "").strip()))
# ---------- Language flow (NO mapping) ----------
def _handle_language_flow(self, state: Dict, msg: str) -> Dict:
# Step 1: ask region
if state["mode"] == "lang_wait_region":
state["region"] = msg
state["mode"] = "lang_wait_language"
return {
"action": "flow",
"answer": "Thanks. What language would you like me to respond in? (Type it, e.g., Sinhala / Tamil / English)",
"suggestions": ["Sinhala", "Tamil", "English"],
"lang": state.get("lang"),
}
# Step 2: set language
if state["mode"] == "lang_wait_language":
chosen = (msg or "").strip()
if not chosen:
return {
"action": "flow",
"answer": "Please type the language you want (example: Sinhala / Tamil / English).",
"suggestions": ["Sinhala", "Tamil", "English"],
"lang": state.get("lang"),
}
state["lang"] = chosen
state["mode"] = "normal"
# If user wanted contact after language, jump into contact flow
next_after = state.pop("next_after_lang", None)
if next_after == "contact":
state["mode"] = "contact_wait_msg"
return {
"action": "flow",
"answer": f"✅ Done. I’ll reply in {state['lang']} from now on.\nNow, please type your message for our team.",
"suggestions": [],
"lang": state.get("lang"),
}
return {
"action": "flow",
"answer": f"✅ Done. I’ll reply in {state['lang']} from now on.",
"suggestions": default_suggestions(),
"lang": state.get("lang"),
}
# fallback
state["mode"] = "normal"
return {"action": "rag", "answer": "", "suggestions": default_suggestions(), "lang": state.get("lang")} |