import json import logging import re import sqlite3 import threading import unicodedata from contextlib import closing from datetime import datetime, timezone from pathlib import Path from typing import Any logger = logging.getLogger(__name__) _LOCK = threading.Lock() POSTOUCH_KEYWORDS = { "postouch", "pos touch", "pos", "facturacion", "facturar", "factura", "facturas", "caja", "ticket", "tickets", "ventas mostrador", "punto de venta", } BACKOFFICE_KEYWORDS = { "backoffice", "backofice", "back office", "back", "bo", "inventario", "inventarios", "mbinv", "mb inv", "mb inventario", "bodega", "stock", "existencias", } FLOW_TITLES = { "emergencia_cobro": "Emergencia de Cobro", "soporte": "Soporte", "implementacion": "Implementacion", "ventas": "Ventas", "administracion": "Administracion", } PROFILE_FIELDS = { "empresa", "nombre_empresa", "nombre", "telefono", "correo", "industria", "nit", "anydesk", } FIELD_LABELS = { "nombre_empresa": "Nombre y empresa", "empresa": "Empresa", "sistema": "Sistema", "tienda": "Tienda", "caja": "Caja", "descripcion_inconveniente": "Descripcion del inconveniente", "descripcion_solicitud": "Descripcion de la solicitud", "descripcion_duda": "Descripcion de la duda", "descripcion_consulta": "Descripcion de la consulta", "anydesk": "AnyDesk", "nombre": "Nombre", "telefono": "Telefono", "correo": "Correo electronico", "industria": "Industria", "nit": "NIT", } FIELD_PROMPTS = { "nombre_empresa": "Indicanos tu nombre y el nombre de tu empresa.", "empresa": "Indicanos el nombre de tu empresa.", "sistema": "Selecciona el sistema donde necesitas apoyo.", "tienda": "Indicanos la tienda donde ocurre el problema.", "caja": "Indicanos la caja o terminal donde ocurre el problema.", "descripcion_inconveniente": ( "Por favor describenos el inconveniente que se esta presentando. " "Si cuentas con una imagen o video, puedes enviarlo en este momento; " "nos ayudara a comprender mejor el caso y apoyarte de forma mas agil." ), "descripcion_solicitud": ( "Por favor cuentanos que ocurre o que necesitas que revisemos. " "Si tienes una imagen o video del problema, puedes enviarlo en este momento; " "eso nos ayuda a darte un mejor apoyo." ), "descripcion_duda": "Describe la duda que tienes.", "descripcion_consulta": "Describe tu consulta administrativa.", "anydesk": "Indicanos el numero de AnyDesk. Si no tienes, responde: No tengo AnyDesk.", "nombre": "Indicanos tu nombre.", "telefono": "Indicanos tu numero de telefono.", "correo": "Indicanos tu correo electronico.", "industria": "Indicanos tu tipo de industria.", "nit": "Indicanos el NIT de tu empresa.", } OPTION_TO_FLOW = { "1": "soporte", "soporte": "soporte", "support": "soporte", "2": "emergencia_cobro", "emergencia": "emergencia_cobro", "emergencia cobro": "emergencia_cobro", "emergencia de cobro": "emergencia_cobro", "3": "implementacion", "implementacion": "implementacion", "4": "ventas", "venta": "ventas", "ventas": "ventas", "sales": "ventas", "5": "administracion", "administracion": "administracion", "admin": "administracion", } def build_customer_intake_reply( store_path: str, customer_key: str, message_text: str, backend: str = "sqlite", mysql_config: dict[str, Any] | None = None, ) -> dict[str, Any] | None: """Build a stateful intake reply or return None to use the normal bot flow.""" if not customer_key: return None service = CustomerStateService( backend=backend, sqlite_path=store_path, mysql_config=mysql_config, ) message = (message_text or "").strip() normalized = _normalize(message) state = service.load(customer_key) data = _normalize_data_shape(_ensure_data_dict(state.get("data"))) if normalized in {"cancelar", "salir"}: if state.get("active_flow"): _clear_active_request(data) service.save(customer_key, None, None, data) return { "type": "text", "text": "He cancelado la captura de datos. Si necesitas apoyo, escribe menu.", "intake": True, "flow": state.get("active_flow"), "field": None, } return None active_flow = _valid_flow(state.get("active_flow")) if active_flow: return _continue_active_flow(service, customer_key, active_flow, state, data, message) selected_flow = detect_selected_flow(message) if not selected_flow: return None data["active_request"] = _new_request(selected_flow) missing_field = _first_missing_field(selected_flow, data) if missing_field: service.save(customer_key, selected_flow, missing_field, data) return _ask_field_payload( selected_flow, missing_field, intro=f"Para {FLOW_TITLES[selected_flow]} voy a pedirte algunos datos, uno por uno.", ) return _finish_flow(service, customer_key, selected_flow, data, already_available=True) def detect_selected_flow(message_text: str) -> str | None: normalized = _normalize(message_text) if normalized in OPTION_TO_FLOW: return OPTION_TO_FLOW[normalized] if "soporte" in normalized: return "soporte" if "emergencia" in normalized: return "emergencia_cobro" if "implementacion" in normalized: return "implementacion" if "ventas" in normalized: return "ventas" if "administracion" in normalized: return "administracion" return None class CustomerStateService: def __init__( self, backend: str = "sqlite", sqlite_path: str = "data/customer_state.db", mysql_config: dict[str, Any] | None = None, ) -> None: self.backend = "mysql" if backend == "mysql" else "sqlite" self.sqlite_path = Path(sqlite_path) self.mysql_config = mysql_config or {} self.table_name = _safe_table_name(str(self.mysql_config.get("table") or "customer_states")) if self.backend == "sqlite": self.sqlite_path.parent.mkdir(parents=True, exist_ok=True) self._ensure_schema() def load(self, customer_key: str) -> dict[str, Any]: if self.backend == "mysql": return self._load_mysql(customer_key) return self._load_sqlite(customer_key) def save( self, customer_key: str, active_flow: str | None, current_field: str | None, data: dict[str, Any], ) -> None: if self.backend == "mysql": self._save_mysql(customer_key, active_flow, current_field, data) return self._save_sqlite(customer_key, active_flow, current_field, data) def _load_sqlite(self, customer_key: str) -> dict[str, Any]: with _LOCK, closing(self._connect_sqlite()) as conn: row = conn.execute( ( "SELECT active_flow, current_field, data_json " "FROM customer_states WHERE customer_key = ?" ), (customer_key,), ).fetchone() return _row_to_state(customer_key, row) def _save_sqlite( self, customer_key: str, active_flow: str | None, current_field: str | None, data: dict[str, Any], ) -> None: updated_at = _utc_now_for_db() data_json = json.dumps(data, ensure_ascii=False, sort_keys=True) with _LOCK, closing(self._connect_sqlite()) as conn: conn.execute( ( "INSERT INTO customer_states " "(customer_key, active_flow, current_field, data_json, updated_at) " "VALUES (?, ?, ?, ?, ?) " "ON CONFLICT(customer_key) DO UPDATE SET " "active_flow = excluded.active_flow, " "current_field = excluded.current_field, " "data_json = excluded.data_json, " "updated_at = excluded.updated_at" ), (customer_key, active_flow, current_field, data_json, updated_at), ) conn.commit() def _load_mysql(self, customer_key: str) -> dict[str, Any]: with _LOCK, closing(self._connect_mysql()) as conn: with conn.cursor() as cursor: cursor.execute( ( f"SELECT active_flow, current_field, data_json " f"FROM `{self.table_name}` WHERE customer_key = %s" ), (customer_key,), ) row = cursor.fetchone() return _row_to_state(customer_key, row) def _save_mysql( self, customer_key: str, active_flow: str | None, current_field: str | None, data: dict[str, Any], ) -> None: updated_at = _utc_now_for_db() data_json = json.dumps(data, ensure_ascii=False, sort_keys=True) with _LOCK, closing(self._connect_mysql()) as conn: with conn.cursor() as cursor: cursor.execute( ( f"INSERT INTO `{self.table_name}` " "(customer_key, active_flow, current_field, data_json, updated_at) " "VALUES (%s, %s, %s, %s, %s) " "ON DUPLICATE KEY UPDATE " "active_flow = VALUES(active_flow), " "current_field = VALUES(current_field), " "data_json = VALUES(data_json), " "updated_at = VALUES(updated_at)" ), (customer_key, active_flow, current_field, data_json, updated_at), ) conn.commit() def _ensure_schema(self) -> None: if self.backend == "mysql": self._ensure_mysql_schema() return self._ensure_sqlite_schema() def _ensure_sqlite_schema(self) -> None: with _LOCK, closing(self._connect_sqlite()) as conn: conn.execute( ( "CREATE TABLE IF NOT EXISTS customer_states (" "customer_key TEXT PRIMARY KEY," "active_flow TEXT," "current_field TEXT," "data_json TEXT NOT NULL DEFAULT '{}'," "updated_at TEXT NOT NULL" ")" ) ) conn.commit() def _ensure_mysql_schema(self) -> None: with _LOCK, closing(self._connect_mysql()) as conn: with conn.cursor() as cursor: cursor.execute( ( f"CREATE TABLE IF NOT EXISTS `{self.table_name}` (" "customer_key VARCHAR(191) PRIMARY KEY," "active_flow VARCHAR(64) NULL," "current_field VARCHAR(64) NULL," "data_json JSON NOT NULL," "updated_at DATETIME NOT NULL," "INDEX idx_updated_at (updated_at)" ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci" ) ) conn.commit() def _connect_sqlite(self) -> sqlite3.Connection: return sqlite3.connect(self.sqlite_path) def _connect_mysql(self) -> Any: try: import pymysql except ImportError as exc: raise RuntimeError( "PyMySQL no esta instalado. Ejecuta: pip install -r requirements.txt" ) from exc return pymysql.connect( host=str(self.mysql_config.get("host") or "127.0.0.1"), port=int(self.mysql_config.get("port") or 3306), user=str(self.mysql_config.get("user") or ""), password=str(self.mysql_config.get("password") or ""), database=str(self.mysql_config.get("database") or ""), charset="utf8mb4", autocommit=False, ) def _continue_active_flow( service: CustomerStateService, customer_key: str, active_flow: str, state: dict[str, Any], data: dict[str, Any], message: str, ) -> dict[str, Any]: current_field = _current_or_first_missing(active_flow, state.get("current_field"), data) if not current_field: return _finish_flow(service, customer_key, active_flow, data, already_available=True) if not _has_answer(message): service.save(customer_key, active_flow, current_field, data) return _ask_field_payload( active_flow, current_field, intro="Necesito este dato para continuar.", ) if current_field == "sistema": system_value = _parse_system_value(message) if not system_value: service.save(customer_key, active_flow, current_field, data) return _ask_field_payload( active_flow, current_field, intro=( "No logre identificar el sistema. Puedes tocar una opcion o " "responder con palabras como POS, facturas, inventario o MBINV." ), ) _set_field_value(data, current_field, system_value) else: _set_field_value(data, current_field, message.strip()) next_field = _first_missing_field(active_flow, data) if next_field: service.save(customer_key, active_flow, next_field, data) return _ask_field_payload(active_flow, next_field) return _finish_flow(service, customer_key, active_flow, data, already_available=False) def _ask_field_payload(flow: str, field: str, intro: str | None = None) -> dict[str, Any]: prompt = FIELD_PROMPTS[field] text = f"{intro}\n\n{prompt}" if intro else prompt payload: dict[str, Any] = { "type": "text", "text": text, "intake": True, "flow": flow, "field": field, } if field == "sistema": payload.update( { "type": "quick_replies", "help_text": "Selecciona el sistema", "buttons": ["POSTouch", "BackOffice"], "fallback_text": ( f"{text}\n\n" "Opciones: POSTouch (facturacion/cajas) o BackOffice (inventario/MBINV)." ), } ) return payload def _finish_flow( service: CustomerStateService, customer_key: str, flow: str, data: dict[str, Any], already_available: bool, ) -> dict[str, Any]: fields = _fields_for_flow(flow, data) request_snapshot = _request_fields(data).copy() request_snapshot["flow"] = flow request_snapshot["completed_at"] = _utc_now_for_db() if not already_available: requests = data.setdefault("requests", []) if isinstance(requests, list): requests.append(request_snapshot) data["requests"] = requests[-20:] payload = _summary_payload(flow, data, fields, already_available=already_available) _clear_active_request(data) service.save(customer_key, None, None, data) return payload def _summary_payload( flow: str, data: dict[str, Any], fields: list[str], already_available: bool, ) -> dict[str, Any]: title = FLOW_TITLES[flow] prefix = "Ya tengo los datos registrados para" if already_available else "Datos recibidos para" lines = [f"{prefix} {title}:"] for field in fields: value = _get_field_value(data, field) lines.append(f"- {FIELD_LABELS[field]}: {value}") lines.append("") lines.append("En breve un asesor continuara con el apoyo.") return { "type": "text", "text": "\n".join(lines), "intake": True, "flow": flow, "field": None, "complete": True, "already_available": already_available, } def _first_missing_field(flow: str, data: dict[str, Any]) -> str | None: for field in _fields_for_flow(flow, data): if not _has_answer(_get_field_value(data, field)): return field return None def _fields_for_flow(flow: str, data: dict[str, Any]) -> list[str]: if flow == "soporte": fields = ["empresa", "sistema"] if _normalize(_get_field_value(data, "sistema")) == "postouch": fields.extend(["tienda", "caja"]) fields.extend(["descripcion_solicitud", "anydesk"]) return fields if flow == "emergencia_cobro": return [ "nombre_empresa", "tienda", "caja", "descripcion_inconveniente", "anydesk", ] if flow == "implementacion": return ["nombre_empresa", "descripcion_duda"] if flow == "ventas": return ["nombre", "telefono", "correo", "empresa", "industria"] if flow == "administracion": return ["nombre_empresa", "nit", "descripcion_consulta"] return [] def _current_or_first_missing(flow: str, current_field: Any, data: dict[str, Any]) -> str | None: fields = _fields_for_flow(flow, data) if isinstance(current_field, str) and current_field in fields: if not _has_answer(_get_field_value(data, current_field)): return current_field return _first_missing_field(flow, data) def _get_field_value(data: dict[str, Any], field: str) -> str: if field in PROFILE_FIELDS: return str(_profile(data).get(field) or "").strip() return str(_request_fields(data).get(field) or "").strip() def _set_field_value(data: dict[str, Any], field: str, value: str) -> None: if field in PROFILE_FIELDS: _profile(data)[field] = value return _request_fields(data)[field] = value def _new_request(flow: str) -> dict[str, Any]: return { "flow": flow, "fields": {}, "started_at": _utc_now_for_db(), } def _clear_active_request(data: dict[str, Any]) -> None: data["active_request"] = {} def _profile(data: dict[str, Any]) -> dict[str, Any]: profile = data.setdefault("profile", {}) if not isinstance(profile, dict): profile = {} data["profile"] = profile return profile def _request_fields(data: dict[str, Any]) -> dict[str, Any]: request = data.setdefault("active_request", {}) if not isinstance(request, dict): request = {} data["active_request"] = request fields = request.setdefault("fields", {}) if not isinstance(fields, dict): fields = {} request["fields"] = fields return fields def _normalize_data_shape(data: dict[str, Any]) -> dict[str, Any]: profile = _profile(data) for field in PROFILE_FIELDS: if not profile.get(field) and data.get(field): profile[field] = data[field] active_request = data.setdefault("active_request", {}) if not isinstance(active_request, dict): data["active_request"] = {} requests = data.setdefault("requests", []) if not isinstance(requests, list): data["requests"] = [] return data def _parse_system_value(message: str) -> str | None: normalized = _normalize(message) postouch_score = _keyword_score(normalized, POSTOUCH_KEYWORDS) backoffice_score = _keyword_score(normalized, BACKOFFICE_KEYWORDS) if postouch_score > backoffice_score: return "POSTouch" if backoffice_score > postouch_score: return "BackOffice" return None def _keyword_score(value: str, keywords: set[str]) -> int: score = 0 padded = f" {value} " for keyword in keywords: normalized_keyword = _normalize(keyword) if not normalized_keyword: continue if normalized_keyword == value: score = max(score, 3) continue if f" {normalized_keyword} " in padded: score = max(score, 2) continue if normalized_keyword in value: score = max(score, 1) return score def _row_to_state(customer_key: str, row: Any) -> dict[str, Any]: if not row: return {"active_flow": None, "current_field": None, "data": {}} active_flow, current_field, data_json = row[0], row[1], row[2] if isinstance(data_json, bytes): data_json = data_json.decode("utf-8", errors="replace") try: data = json.loads(data_json or "{}") except ValueError: logger.warning("Invalid customer state JSON for customer_key=%s", customer_key) data = {} return { "active_flow": active_flow, "current_field": current_field, "data": data if isinstance(data, dict) else {}, } def _valid_flow(value: Any) -> str | None: return value if isinstance(value, str) and value in FLOW_TITLES else None def _ensure_data_dict(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} def _has_answer(value: str) -> bool: return bool((value or "").strip()) def _normalize(value: str) -> str: normalized = unicodedata.normalize("NFKD", value or "") normalized = "".join(char for char in normalized if not unicodedata.combining(char)) return " ".join(normalized.lower().strip().split()) def _safe_table_name(value: str) -> str: if not re.fullmatch(r"[A-Za-z0-9_]+", value or ""): raise ValueError("CUSTOMER_STATE_MYSQL_TABLE solo puede contener letras, numeros y guion bajo.") return value def _utc_now_for_db() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")