"""Marlimar (Veloce API) connector.
Marlimar Interactive's Veloce platform provides SMS/messaging automation
used by home improvement contractors for lead engagement and customer
communication.
Auth: session token — POST form-encoded ``username``/``password`` to
``{base}/Login/``. A successful response returns JSON with ``status`` (1 =
success), ``token``, ``id`` (the logged-in id), and a ``companies`` list
whose first entry carries ``companies_id``. Sessions are closed via
POST ``loggedin_id`` + ``token`` to ``{base}/Login/Logout/``.
Credentials the user provides:
* ``username`` — Veloce API username
* ``password`` — Veloce API password
* ``base_url`` — optional override (default ``https://api.marlimar.com``)
* ``company_ref``— optional Marlimar companies_id (auto-detected on login)
Synced messages are upserted into ``ExternalSyncRecord`` keyed by
``(company_id, external_id)`` so repeated syncs are idempotent.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
from . import BaseConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata ----------------------------------------------------
register_connector(
"marlimar",
{
"service": "marlimar",
"display_name": "Marlimar",
"name": "Marlimar",
"category": "telephony",
"description": "Marlimar Veloce SMS/messaging automation — send texts, sync message history, and engage leads for home improvement contractors.",
"auth_type": "api_key",
"auth_fields": ["username", "password"],
"config_fields": [
{"name": "username", "label": "Veloce API Username", "type": "text", "required": True},
{"name": "password", "label": "Veloce API Password", "type": "password", "required": True},
{"name": "base_url", "label": "API Base URL", "type": "text", "required": False},
{"name": "company_ref", "label": "Marlimar Company ID (auto-detected)", "type": "text", "required": False},
],
"capabilities": ["sms_send", "messages", "contacts"],
"data_types": ["messages"],
"rate_limit": "Not documented; conservative back-off applied.",
"docs_url": "https://api.marlimar.com/Documentation/Center/Login/",
},
)
_DEFAULT_BASE = "https://api.marlimar.com"
class MarlimarConnector(BaseConnector):
"""Marlimar Veloce integration — SMS/messaging for contractor lead engagement."""
_SERVICE = "marlimar"
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self._token: str = ""
self._loggedin_id: str = ""
self._companies_id: str = ""
# -- Config helpers ---------------------------------------------------------
@property
def _base_url(self) -> str:
return (self.config.get("base_url") or _DEFAULT_BASE).rstrip("/")
# -- Auth (session token lifecycle) ------------------------------------------
def _login(self) -> Dict[str, Any]:
"""Authenticate against the Veloce Login endpoint.
Response JSON: ``{status: 1, token, id, companies: [{companies_id, ...}]}``
"""
username = self.config.get("username", "")
password = self.config.get("password", "")
if not username or not password:
raise ValueError("Marlimar credentials incomplete — username and password required.")
resp = requests.post(
f"{self._base_url}/Login/",
data={"username": username, "password": password},
headers={"Accept": "application/json"},
timeout=30,
)
resp.raise_for_status()
data = resp.json()
if int(data.get("status", 0)) != 1:
raise ValueError(
f"Marlimar login failed: {data.get('message') or data.get('error') or 'invalid credentials'}"
)
self._token = str(data.get("token", ""))
self._loggedin_id = str(data.get("id", ""))
companies = data.get("companies") or []
if companies and isinstance(companies, list) and isinstance(companies[0], dict):
self._companies_id = str(companies[0].get("companies_id", ""))
elif self.config.get("company_ref"):
self._companies_id = str(self.config["company_ref"])
if not self._token:
raise ValueError("Marlimar login succeeded but no session token was returned")
return data
def _logout(self) -> None:
"""Close the Veloce session (best-effort)."""
if not self._token or not self._loggedin_id:
return
try:
requests.post(
f"{self._base_url}/Login/Logout/",
data={"loggedin_id": self._loggedin_id, "token": self._token},
timeout=15,
)
except Exception:
logger.debug("Marlimar logout failed (non-fatal)", exc_info=True)
finally:
self._token = ""
self._loggedin_id = ""
def _ensure_session(self) -> None:
if not self._token:
self._retry(self._login)
def _api_post(self, path: str, payload: Optional[Dict[str, Any]] = None) -> Any:
"""POST to a Veloce endpoint with the session token attached."""
self._ensure_session()
data = dict(payload or {})
data.setdefault("loggedin_id", self._loggedin_id)
data.setdefault("token", self._token)
if self._companies_id:
data.setdefault("companies_id", self._companies_id)
resp = requests.post(
f"{self._base_url}{path}",
data=data,
headers={"Accept": "application/json"},
timeout=30,
)
if resp.status_code in (401, 403):
# Session expired — re-login once and retry.
self._token = ""
self._ensure_session()
data["loggedin_id"] = self._loggedin_id
data["token"] = self._token
resp = requests.post(
f"{self._base_url}{path}",
data=data,
headers={"Accept": "application/json"},
timeout=30,
)
if resp.status_code == 429:
retry_after = resp.headers.get("Retry-After")
logger.warning("Marlimar rate limited: 429, Retry-After=%s", retry_after)
resp.raise_for_status()
if not resp.content:
return {}
try:
return resp.json()
except ValueError:
return {"raw": resp.text}
# -- connect / disconnect ------------------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate credentials by performing a login/logout round-trip."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
self._retry(self._login)
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={"companies_id": self._companies_id},
)
return {
"status": "connected",
"service": "marlimar",
"companies_id": self._companies_id,
"duration_ms": duration_ms,
}
except Exception as exc:
self._log(event_type="connect_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc)}
def test_connection(self) -> Dict[str, Any]:
"""Validate credentials without side effects — used by the test route.
Performs login then logs the session back out so we don't hold a
session open.
"""
result = self.connect()
self._logout()
return result
def disconnect(self) -> Dict[str, Any]:
"""Log out of the Veloce session and clear the stored password."""
self._logout()
self._connected = False
self.config.pop("password", None)
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "marlimar"}
# -- sync -------------------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Sync message history from the Veloce Message Center.
Idempotent — messages are upserted by (company_id, external_id).
"""
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total = 0
try:
self._ensure_session()
messages = self._sync_messages()
total += messages
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=total,
duration_ms=duration_ms,
details={"messages": messages},
)
return {
"status": "success",
"record_count": total,
"duration_ms": duration_ms,
"details": {"messages": messages},
"last_sync_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as exc:
self._log(event_type="sync_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc), "record_count": total}
finally:
self._logout()
def _sync_messages(self) -> int:
"""Pull message history and upsert into ExternalSyncRecord."""
from app.models import db, ExternalSyncRecord
self._check_rate_limit()
try:
data = self._retry(self._api_post, "/Center/Message/List/", {})
except requests.HTTPError as exc:
code = getattr(getattr(exc, "response", None), "status_code", 0)
if code == 404:
# Message list endpoint not enabled for this account —
# connector still functions for outbound sends.
logger.info("Marlimar message list endpoint unavailable — skipping sync")
return 0
raise
if isinstance(data, dict):
items = data.get("messages") or data.get("items") or data.get("data") or []
else:
items = data or []
if not isinstance(items, list):
items = []
count = 0
for item in items:
if not isinstance(item, dict):
continue
raw_id = item.get("messages_id") or item.get("message_id") or item.get("id")
if raw_id in (None, ""):
continue
external_id = f"marlimar:message:{raw_id}"
record = ExternalSyncRecord.query.filter_by(
company_id=self.company_id, external_id=external_id
).first()
if record is None:
record = ExternalSyncRecord(
company_id=self.company_id,
external_id=external_id,
source_service="marlimar",
entity_type="message",
)
db.session.add(record)
record.name = str(item.get("phone") or item.get("to") or item.get("recipient") or raw_id)[:255]
record.status = str(item.get("status") or item.get("direction") or "")[:100]
record.properties_json = {
"body": item.get("message") or item.get("body", ""),
"direction": item.get("direction", ""),
"sent_at": item.get("created") or item.get("sent_at", ""),
"raw": item,
}
count += 1
db.session.commit()
return count
# -- messaging ---------------------------------------------------------------------
def send_message(self, phone: str, message: str, **extra) -> Dict[str, Any]:
"""Send an SMS via the Veloce Message Center.
Args:
phone: destination phone number (E.164 or 10-digit).
message: message body.
extra: any additional Veloce-native form fields.
"""
self._log(event_type="message_send_start", status="pending")
try:
payload = {"phone": phone, "message": message, **extra}
result = self._retry(self._api_post, "/Center/Message/", payload)
self._log(
event_type="message_send_success",
status="success",
record_count=1,
details={"result": result if isinstance(result, dict) else {}},
)
return {"status": "success", "result": result}
except Exception as exc:
self._log(event_type="message_send_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc)}
# -- status ------------------------------------------------------------------------
def status(self) -> Dict[str, Any]:
"""Report connection health."""
return {
"service": "marlimar",
"connected": self._connected or bool(self._token),
"has_credentials": bool(self.config.get("username") and self.config.get("password")),
"companies_id": self._companies_id or self.config.get("company_ref", ""),
"base_url": self._base_url,
}
_REGISTRY["marlimar"] = MarlimarConnector