Aswin Biju
3640b2d785
Shops can now send handwritten slips as WhatsApp photos. srs-api reads them through OpenRouter in the background. The model only proposes text: the parser decides what is an entry, and nothing is saved without the user's Save.
279 lines
10 KiB
Python
279 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
import logging
|
|
import mimetypes
|
|
import os
|
|
import stat
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
log = logging.getLogger("srs_gate")
|
|
|
|
SKIP = {"action": "skip", "reason": "srs-gate"}
|
|
CONFIG_NAME = "srs-gate.env"
|
|
TRIES = 3
|
|
TRY_TIMEOUT_S = 3.0
|
|
BACKOFF_S = 0.3
|
|
OVERALL_TIMEOUT_S = 9.0
|
|
PHOTO_TRY_TIMEOUT_S = 15.0
|
|
PHOTO_OVERALL_TIMEOUT_S = 40.0
|
|
MAX_PHOTO_BYTES = 8_000_000
|
|
PHOTO_PLACEHOLDER = "[image received]"
|
|
|
|
_config_warned = False
|
|
|
|
|
|
class ConfigError(Exception):
|
|
"""srs-gate.env is missing or incomplete. Messages carry fixed wording only, never values."""
|
|
|
|
|
|
class _NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
"""Never follow a redirect: it could carry the x-srs-secret header (and the text) to another host."""
|
|
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
_OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect)
|
|
|
|
|
|
# ---------------------------------------------------------------- config
|
|
|
|
|
|
def _hermes_home() -> Path:
|
|
home = os.environ.get("HERMES_HOME")
|
|
return Path(home) if home else Path.home() / ".hermes"
|
|
|
|
|
|
def _config_path() -> Path:
|
|
return _hermes_home() / CONFIG_NAME
|
|
|
|
|
|
def _unquote(value: str) -> str:
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
|
|
return value[1:-1]
|
|
return value
|
|
|
|
|
|
def _load_config() -> dict:
|
|
try:
|
|
raw = _config_path().read_text(encoding="utf-8")
|
|
except FileNotFoundError:
|
|
raise ConfigError("config file not found") from None
|
|
except (OSError, UnicodeError):
|
|
raise ConfigError("config file unreadable") from None
|
|
cfg = {}
|
|
for line in raw.splitlines():
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
key, _, value = line.partition("=")
|
|
cfg[key.strip()] = _unquote(value.strip())
|
|
url = cfg.get("SRS_URL", "")
|
|
secret = cfg.get("SRS_SECRET", "")
|
|
if not secret:
|
|
raise ConfigError("config has no SRS_SECRET")
|
|
if not url.lower().startswith(("http://", "https://")):
|
|
raise ConfigError("config SRS_URL missing or not http(s)")
|
|
return {"SRS_URL": url, "SRS_SECRET": secret}
|
|
|
|
|
|
# ---------------------------------------------------------------- event filter
|
|
|
|
|
|
def _route(event):
|
|
"""The ids of a forwardable direct message of any type, else None."""
|
|
if event is None:
|
|
return None
|
|
src = getattr(event, "source", None)
|
|
if src is None or getattr(event, "internal", False) or getattr(src, "is_bot", False):
|
|
return None
|
|
if getattr(src, "chat_type", None) != "dm":
|
|
return None
|
|
if getattr(event, "edit_revision", None) not in (None, ""):
|
|
return None
|
|
message_id = getattr(event, "message_id", None)
|
|
sender = getattr(src, "user_id", None)
|
|
chat_id = getattr(src, "chat_id", None)
|
|
if not (message_id and sender and chat_id):
|
|
return None
|
|
if str(chat_id).endswith("@g.us"):
|
|
return None
|
|
return {"message_id": str(message_id), "sender_jid": str(sender), "chat_id": str(chat_id)}
|
|
|
|
|
|
def _is_type(event, name: str) -> bool:
|
|
mtype = getattr(event, "message_type", None)
|
|
return mtype is not None and str(getattr(mtype, "value", mtype)) == name
|
|
|
|
|
|
def _payload(event):
|
|
"""The /ingest body for a forwardable direct text message, else None."""
|
|
route = _route(event)
|
|
if route is None or not _is_type(event, "text"):
|
|
return None
|
|
text = getattr(event, "text", None)
|
|
if not (isinstance(text, str) and text.strip()):
|
|
return None
|
|
return {"kind": "text", **route, "text": text}
|
|
|
|
|
|
def _photo_work(event):
|
|
"""What to send for a forwardable direct photo, else None. Reads no files, so it is safe on the gateway's loop.
|
|
|
|
Text sent with the photo is what the user typed, so it goes on as a typed message. The placeholder is Hermes' own.
|
|
"""
|
|
route = _route(event)
|
|
if route is None or not _is_type(event, "photo"):
|
|
return None
|
|
urls = getattr(event, "media_urls", None)
|
|
paths = [u for u in urls if isinstance(u, str) and u] if isinstance(urls, (list, tuple)) else []
|
|
text = getattr(event, "text", None)
|
|
typed = text.replace(PHOTO_PLACEHOLDER, "").strip() if isinstance(text, str) else ""
|
|
if not paths and not typed:
|
|
return None
|
|
return {"route": route, "paths": paths, "text": typed}
|
|
|
|
|
|
# ---------------------------------------------------------------- HTTP (runs in a worker thread)
|
|
|
|
|
|
def _error_name(exc: BaseException) -> str:
|
|
reason = getattr(exc, "reason", None) if isinstance(exc, urllib.error.URLError) else None
|
|
return type(reason).__name__ if isinstance(reason, BaseException) else type(exc).__name__
|
|
|
|
|
|
def _post(cfg: dict, payload: dict, try_timeout=None):
|
|
"""Blocking POST with retries. Returns (http_status or None, error type name or None). Never raises.
|
|
|
|
Nothing may cross the thread boundary: on Python 3.9-3.11 asyncio.wait_for runs this in its own task, and a
|
|
BaseException such as SystemExit would escape through the event loop instead of the callback's except.
|
|
"""
|
|
status = None
|
|
error = None
|
|
timeout = TRY_TIMEOUT_S if try_timeout is None else try_timeout
|
|
try:
|
|
req = urllib.request.Request(
|
|
cfg["SRS_URL"].rstrip("/") + "/ingest",
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
method="POST",
|
|
headers={"content-type": "application/json", "x-srs-secret": cfg["SRS_SECRET"]},
|
|
)
|
|
for attempt in range(TRIES):
|
|
try:
|
|
with _OPENER.open(req, timeout=timeout) as resp:
|
|
return resp.status, None
|
|
except urllib.error.HTTPError as err:
|
|
status, error = err.code, None
|
|
err.close()
|
|
if err.code < 500:
|
|
return status, None
|
|
except Exception as exc:
|
|
status, error = None, _error_name(exc)
|
|
if attempt + 1 < TRIES:
|
|
time.sleep(BACKOFF_S)
|
|
return status, error
|
|
except BaseException as exc:
|
|
return status, type(exc).__name__
|
|
|
|
|
|
def _read_photo(path: str):
|
|
"""Returns (bytes, mime, None), or (None, None, a fixed reason). Only a regular file inside Hermes' cache is read."""
|
|
try:
|
|
resolved = Path(path).expanduser().resolve()
|
|
if not resolved.is_relative_to((_hermes_home() / "cache").resolve()):
|
|
return None, None, "outside_cache"
|
|
if not stat.S_ISREG(resolved.stat().st_mode):
|
|
return None, None, "not_a_file"
|
|
with open(resolved, "rb") as handle:
|
|
data = handle.read(MAX_PHOTO_BYTES + 1)
|
|
if len(data) > MAX_PHOTO_BYTES:
|
|
return None, None, "too_large"
|
|
return data, mimetypes.guess_type(resolved.name)[0] or "image/jpeg", None
|
|
except (OSError, ValueError, RuntimeError):
|
|
return None, None, "unreadable"
|
|
|
|
|
|
def _forward_photo(cfg: dict, work: dict):
|
|
"""Sends the photos, then any typed text, in that order. Returns (id, status, error, reason) rows. Never raises."""
|
|
rows = []
|
|
route = work["route"]
|
|
try:
|
|
for index, path in enumerate(work["paths"]):
|
|
message_id = route["message_id"] if index == 0 else "%s#%d" % (route["message_id"], index + 1)
|
|
data, mime, reason = _read_photo(path)
|
|
if reason is not None:
|
|
rows.append((message_id, None, None, reason))
|
|
continue
|
|
payload = {
|
|
"kind": "image",
|
|
**route,
|
|
"message_id": message_id,
|
|
"image_base64": base64.b64encode(data).decode("ascii"),
|
|
"mime_type": mime,
|
|
}
|
|
status, error = _post(cfg, payload, PHOTO_TRY_TIMEOUT_S)
|
|
rows.append((message_id, status, error, None))
|
|
if work["text"]:
|
|
# Sent even if a photo failed: typed tickets must not vanish because of the picture beside them.
|
|
message_id = route["message_id"] + "#text"
|
|
status, error = _post(cfg, {"kind": "text", **route, "message_id": message_id, "text": work["text"]})
|
|
rows.append((message_id, status, error, None))
|
|
except BaseException as exc:
|
|
rows.append((route["message_id"], None, type(exc).__name__, None))
|
|
return rows
|
|
|
|
|
|
def _log_result(message_id, status, error, reason=None):
|
|
if reason is not None:
|
|
log.warning("srs-gate: id=%s photo not sent (%s)", message_id, reason)
|
|
elif status is not None and 200 <= status < 300:
|
|
log.info("srs-gate: id=%s status=%s", message_id, status)
|
|
else:
|
|
log.warning("srs-gate: id=%s status=%s error=%s", message_id, status, error)
|
|
|
|
|
|
# ---------------------------------------------------------------- the hook
|
|
|
|
|
|
async def on_inbound(event=None, gateway=None, session_store=None, **kwargs):
|
|
global _config_warned
|
|
message_id = "-"
|
|
try:
|
|
payload = _payload(event)
|
|
photo = _photo_work(event) if payload is None else None
|
|
if payload is not None or photo is not None:
|
|
message_id = payload["message_id"] if payload is not None else photo["route"]["message_id"]
|
|
try:
|
|
cfg = _load_config()
|
|
except ConfigError as exc:
|
|
cfg = None
|
|
if not _config_warned:
|
|
_config_warned = True
|
|
log.warning("srs-gate: %s; nothing is forwarded", exc)
|
|
if cfg is not None:
|
|
_config_warned = False
|
|
if payload is not None:
|
|
status, error = await asyncio.wait_for(
|
|
asyncio.to_thread(_post, cfg, payload), OVERALL_TIMEOUT_S
|
|
)
|
|
_log_result(message_id, status, error)
|
|
else:
|
|
rows = await asyncio.wait_for(
|
|
asyncio.to_thread(_forward_photo, cfg, photo), PHOTO_OVERALL_TIMEOUT_S
|
|
)
|
|
for row in rows:
|
|
_log_result(*row)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except (Exception, SystemExit) as exc:
|
|
log.warning("srs-gate: id=%s not forwarded (%s)", message_id, type(exc).__name__)
|
|
return dict(SKIP)
|
|
|
|
|
|
def register(ctx):
|
|
ctx.register_hook("pre_gateway_dispatch", on_inbound)
|