"""Evolution API — the development provider (ADR-0006). Self-hosted, free, no approval process. It drives WhatsApp through an unofficial path, which carries a real risk of the number being banned, so **this provider is for development only and must never be pointed at the firm's live number.** Its job here is to make the messaging path exercisable end to end from Phase 3 onward, rather than integrating WhatsApp last when integration problems are most expensive to find. """ from __future__ import annotations import base64 import binascii import logging from pathlib import Path from app.support.http import HttpTransport, UrllibTransport from app.whatsapp.messages import ( InboundMessage, MediaReference, MessageType, OutboundMessage, SendResult, ) from app.whatsapp.session import LinkRequest, SessionState, SessionStatus logger = logging.getLogger(__name__) #: Evolution's connection states, mapped onto ours. #: #: "close" is reported both for an instance that was never linked and for one #: the phone logged out. Evolution does not distinguish them, so neither does #: this: reporting LOGGED_OUT would tell somebody their working connection had #: dropped when in fact they had simply never made one. _STATES = { "open": SessionState.CONNECTED, "connecting": SessionState.AWAITING_SCAN, "close": SessionState.DISCONNECTED, "closed": SessionState.DISCONNECTED, # Evolution stopped offering codes because nobody scanned one. Found in its # source, not its documentation — and neither obvious mapping is honest: # UNAVAILABLE blames a provider that is working, LOGGED_OUT reports a # dropped connection that never existed. "refused": SessionState.LINK_EXPIRED, } #: How many codes Evolution offers before abandoning the attempt. #: #: Its QRCODE_LIMIT, which defaults to 30 and is not exposed over the API — so #: it is configured here to match whatever the instance is actually running. #: Wrong only makes the progress indicator wrong; the state still comes from #: the provider. DEFAULT_QR_ROTATION_LIMIT = 30 _TYPE_MAP = { "conversation": MessageType.TEXT, "extendedTextMessage": MessageType.TEXT, "imageMessage": MessageType.IMAGE, "documentMessage": MessageType.DOCUMENT, "audioMessage": MessageType.AUDIO, "videoMessage": MessageType.VIDEO, } #: Message keys that carry no content of their own — each holds the real #: message one level down, under "message". _ENVELOPES = ( "documentWithCaptionMessage", "ephemeralMessage", "viewOnceMessage", "viewOnceMessageV2", "viewOnceMessageV2Extension", ) #: Envelopes nest (a disappearing message containing a captioned document), but #: not deeply. The cap is what stops a malformed or hostile payload looping. _MAX_ENVELOPE_DEPTH = 5 def _unwrap(body: dict) -> dict: """The real message inside however many envelopes WhatsApp wrapped it in. WhatsApp does not add a caption to a document by setting a field on it; it replaces the whole message with a `documentWithCaptionMessage` holding the original one level down. The same is true of disappearing and view-once messages. This matters more than a parsing nicety. A key that is not in `_TYPE_MAP` yields `kind = None`, so the message has no media, `has_document` is False, and the daemon skips it as "nothing to read" — at DEBUG, which is below the level these deployments run at. A document sent with a caption would be dropped in silence, and the caption is precisely how somebody says which client a file belongs to. """ for _ in range(_MAX_ENVELOPE_DEPTH): if not isinstance(body, dict): return {} key = next((k for k in _ENVELOPES if k in body), None) if key is None: return body envelope = body.get(key) inner = envelope.get("message") if isinstance(envelope, dict) else None if not isinstance(inner, dict): # An envelope with nothing in it. Returning what we have lets the # caller decide it is unsupported, rather than raising here. return body body = inner return body class EvolutionProvider: """Speaks Evolution's webhook and send formats.""" name = "evolution" def __init__( self, base_url: str = "", instance: str = "", api_key: str = "", transport: HttpTransport | None = None, qr_rotation_limit: int = DEFAULT_QR_ROTATION_LIMIT, ) -> None: self._base_url = base_url.rstrip("/") self._instance = instance self._api_key = api_key self._qr_rotation_limit = qr_rotation_limit self._http = transport or UrllibTransport() self.sent: list[OutboundMessage] = [] """Recorded rather than transmitted when no base URL is configured, so the whole path is testable without a live Evolution instance.""" def send(self, message: OutboundMessage) -> SendResult: if not self._base_url: self.sent.append(message) return SendResult(ok=True, provider_message_id=f"local-{len(self.sent)}") if message.is_template: # Evolution has no template concept — it drives a personal client, # where every message is free-form. MessageSender still enforces # approval so development cannot build something that only works # here, but the wire format has nowhere to put it. logger.debug("Sending template '%s' as plain text.", message.template) response = self._http.request( "POST", f"{self._base_url}/message/sendText/{self._instance}", headers=self._auth(), json={"number": message.recipient, "text": message.text}, ) if not response.ok: logger.error("Evolution send failed (%s).", response.status) return SendResult(ok=False, error=f"HTTP {response.status}") body = response.json() if response.body else {} return SendResult( ok=True, provider_message_id=str((body.get("key") or {}).get("id") or "") or None, ) def parse_webhook(self, payload: dict) -> list[InboundMessage]: """Read Evolution's webhook shape. Tolerant by necessity: this is an unofficial API whose payloads vary between versions. Anything unrecognised is skipped rather than raised on — a webhook that errors is a webhook that gets retried forever. """ raw_messages = payload.get("data") or [] if isinstance(raw_messages, dict): # Some versions send a single object rather than a list. raw_messages = [raw_messages] # The envelope names the connected account: Evolution emits # `sender: this.wuid` with every webhook (channel.service.ts). # # Used ONLY to fill in a recipient, never to decide anything. It is a # claim made by whoever posted to this endpoint, and a forged payload # could assert any owner it liked. The self-chat policy resolves the # owner separately, from the authenticated session, so a lie here # changes what a message is labelled and not whether it is processed. account = str(payload.get("sender") or "") parsed: list[InboundMessage] = [] for item in raw_messages: if not isinstance(item, dict): continue key = item.get("key") or {} message_id = key.get("id") remote_jid = key.get("remoteJid") or "" # In a group, remoteJid is the group and `participant` is whoever # spoke. In a direct message there is no participant and the two are # the same. The allow-list needs them apart: watching a group is a # different decision from trusting one member of it. participant = key.get("participant") or "" sender = (participant or remote_jid).split("@")[0] if not message_id or not sender: continue # `fromMe` is NOT a reason to skip, and that is the whole point of # the note-to-self inbox: a document the account owner sends to # their own number is from them. Dropping these would ignore exactly # the messages this feature exists to process. # # Our own outgoing replies are also fromMe, and are filtered instead # by having already been recorded as seen when they were sent — an # id we generated cannot be mistaken for one the user typed. from_me = bool(key.get("fromMe")) # Unwrapped first: a captioned, disappearing or view-once message # keeps the document one or more levels down, and everything below # this line looks for the document at the top. body = _unwrap(item.get("message") or {}) kind = next((k for k in body if k in _TYPE_MAP), None) message_type = _TYPE_MAP.get(kind or "", MessageType.UNSUPPORTED) # Who it was addressed to. Outgoing, that is the far end of the # conversation; incoming, it is this account. In the self-chat they # are the same value, which is the whole shape of the thing. recipient = remote_jid if from_me else (account or remote_jid) parsed.append( InboundMessage( provider_message_id=message_id, sender=sender, recipient=recipient.split("@")[0] if "@g.us" not in recipient else recipient, chat=remote_jid.split("@")[0] if "@g.us" not in remote_jid else remote_jid, type=message_type, text=self._text_of(body), media=self._media_of(body, kind, item), from_me=from_me, raw=item, ) ) return parsed def download_media(self, media: MediaReference, destination: Path) -> Path: """Fetch a document. Evolution returns the file base64-encoded in a JSON body rather than as a URL, so there is only one call — and the whole original message is what it wants back, not the media handle and not just the key. THE WHOLE MESSAGE, DELIBERATELY Evolution decides between two paths (whatsapp.baileys.service.ts): const msg = m?.message ? m : await this.getMessage(m.key, true); if (!msg) throw 'Message not found'; Given the message it decrypts from what it was handed. Given only a key it goes looking in its own database — and this deployment runs with DATABASE_SAVE_DATA_NEW_MESSAGE=false, because ADR-0002 says a client's documents and conversations do not get a second home in Evolution's Postgres. So sending only the key asked Evolution to find a message it had been told never to keep. Every real document failed with HTTP 400 "Message not found" while the whole path in front of it worked perfectly. Sending the message keeps both properties: the download works, and Evolution still stores nothing. """ if not self._base_url: raise MediaDownloadError( "No Evolution base URL configured; there is nothing to download from." ) raw = media.raw or {} # The key alone is the fallback for a reference built without its # original fragment. It only succeeds where Evolution is persisting # messages, which is not how this is deployed — but a worse request is # better than no request. payload = raw if raw.get("message") else {"key": raw.get("key", {})} response = self._http.request( "POST", f"{self._base_url}/chat/getBase64FromMediaMessage/{self._instance}", headers=self._auth(), json={"message": payload}, ) if not response.ok: raise MediaDownloadError(f"Evolution refused the media request (HTTP {response.status}).") encoded = (response.json() or {}).get("base64") if not encoded: raise MediaDownloadError("Evolution returned no media content.") # strict: an invalid character means the payload was corrupted, and # decoding leniently would write a truncated document that OCR would # then read as a bad scan. try: content = base64.b64decode(encoded, validate=True) except (binascii.Error, ValueError) as exc: # Re-raised as this module's error. binascii.Error escaping would # slip past every caller catching the documented exception, and a # corrupted document would surface as an unhandled crash instead of # a message that says what happened. raise MediaDownloadError("Evolution returned media that is not valid base64.") from exc destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(content) return destination # ── Linking an account (LinkableProvider) ─────────────────── # # Evolution drives a personal WhatsApp client, so an account links the way a # laptop does: Linked Devices, scan a code. That is only possible because # the protocol is unofficial — which is also why it risks the number being # banned. That warning belongs in front of the customer before they scan, # not buried in a docstring here. def start_link(self): """Create the instance if needed and return a code to scan. An account that is already connected gets its status back instead of a QR. Issuing one would begin a fresh link and drop the working session: a customer pressing Connect twice must not disconnect themselves. """ if not self._base_url: return SessionStatus( SessionState.UNAVAILABLE, "no Evolution URL is configured for this deployment", ) current = self.session_status() if current.state is SessionState.CONNECTED: return current created = self._session_request("POST", "/instance/create", json={ "instanceName": self._instance, "qrcode": True, # Baileys is the personal-client integration. Named explicitly # because Evolution's default has changed between versions, and an # instance quietly created on a different one would never link. "integration": "WHATSAPP-BAILEYS", }) # "Already exists" is the ordinary case on a reconnect, not a failure. if created is None or not created.ok: created = self._session_request("GET", f"/instance/connect/{self._instance}") if created is None or not created.ok: return SessionStatus(SessionState.UNAVAILABLE, "Evolution would not start a session") found = self._qr_payload(created) if found is None: # Either it connected between the check and here, or this Evolution # returns a shape we do not know. Ask, rather than guess. return self.session_status() payload, rotation = found return LinkRequest( payload=payload, rotation=rotation, rotation_limit=self._qr_rotation_limit, ) def session_status(self) -> SessionStatus: if not self._base_url: return SessionStatus( SessionState.UNAVAILABLE, "no Evolution URL is configured for this deployment", ) response = self._session_request("GET", f"/instance/connectionState/{self._instance}") if response is None: return SessionStatus(SessionState.UNAVAILABLE, "Evolution could not be reached") if response.status == 404: # No instance yet. Nothing is wrong — nobody has connected. return SessionStatus(SessionState.DISCONNECTED, "no session has been created") if not response.ok: return SessionStatus( SessionState.UNAVAILABLE, f"Evolution answered HTTP {response.status}" ) try: body = response.json() except Exception: # noqa: BLE001 - an HTML error page is not a state return SessionStatus(SessionState.UNAVAILABLE, "Evolution returned an unreadable state") raw = str(((body or {}).get("instance") or {}).get("state") or "").lower() return SessionStatus( state=_STATES.get(raw, SessionState.UNAVAILABLE), detail=raw or "no state reported", number=self._connected_number() if raw == "open" else None, ) def unlink(self) -> SessionStatus: """Log out, so the phone stops listing this deployment as a device.""" if not self._base_url: return SessionStatus(SessionState.UNAVAILABLE, "no Evolution URL is configured") response = self._session_request("DELETE", f"/instance/logout/{self._instance}") # 404 means there was nothing to log out of, which is the state asked for. if response is None or not (response.ok or response.status == 404): return SessionStatus( SessionState.UNAVAILABLE, "Evolution would not log the session out" ) return SessionStatus(SessionState.DISCONNECTED, "logged out") def _connected_number(self) -> str | None: """The linked number, shown back so a customer can confirm the account.""" response = self._session_request( "GET", f"/instance/fetchInstances?instanceName={self._instance}" ) if response is None or not response.ok: return None try: body = response.json() except Exception: # noqa: BLE001 return None for entry in (body if isinstance(body, list) else [body]): if not isinstance(entry, dict): continue instance = entry.get("instance") if isinstance(entry.get("instance"), dict) else entry owner = instance.get("owner") or instance.get("ownerJid") or instance.get("number") if owner: # 923001234567@s.whatsapp.net -> 923001234567 return str(owner).split("@")[0].split(":")[0] return None @staticmethod def _qr_payload(response) -> tuple[str, int | None] | None: """Dig the code and its rotation number out of the response. Read by key, and across both shapes seen in the wild, because a version bump that moves it must not silently produce a blank screen and no error. `count` comes from Evolution's own qrCode getter, which returns {pairingCode, code, base64, count}. It is what lets the screen say "code 4 of 30" instead of leaving the customer to discover the limit by hitting it. """ try: body = response.json() or {} except Exception: # noqa: BLE001 return None for holder in (body.get("qrcode"), body): if not isinstance(holder, dict): continue for key in ("base64", "code", "pairingCode"): value = holder.get(key) if isinstance(value, str) and value.strip(): count = holder.get("count") return value, count if isinstance(count, int) else None return None def _session_request(self, method: str, path: str, json: dict | None = None): """One call to Evolution. Returns None when it could not be reached. Session operations answer a person waiting at a screen, so an unreachable provider is a state to report rather than an exception to throw into a poll loop. """ try: return self._http.request( method, f"{self._base_url}{path}", headers=self._auth(), json=json, timeout=20.0, ) except Exception as exc: # noqa: BLE001 logger.warning("Evolution %s %s failed: %s", method, path, exc) return None # ── Internals ───────────────────────────────────────────────────────── def _auth(self) -> dict[str, str]: return {"apikey": self._api_key} if self._api_key else {} @staticmethod def _text_of(body: dict) -> str: if "conversation" in body: return str(body["conversation"]) extended = body.get("extendedTextMessage") or {} if isinstance(extended, dict) and extended.get("text"): return str(extended["text"]) # A caption on an image or document is the client telling us what it is. for key in ("imageMessage", "documentMessage", "videoMessage"): part = body.get(key) or {} if isinstance(part, dict) and part.get("caption"): return str(part["caption"]) return "" @staticmethod def _media_of(body: dict, kind: str | None, item: dict | None = None) -> MediaReference | None: if kind not in {"imageMessage", "documentMessage", "audioMessage", "videoMessage"}: return None part = body.get(kind) or {} if not isinstance(part, dict): return None handle = part.get("url") or part.get("mediaKey") if not handle: return None return MediaReference( handle=str(handle), mime_type=part.get("mimetype"), filename=part.get("fileName"), size_bytes=part.get("fileLength"), # The download endpoint wants the original message key, not the # handle — so the fragment travels with the reference. raw=item or {}, ) class MediaDownloadError(RuntimeError): """A document could not be fetched."""