diff --git a/channels/base.py b/channels/base.py new file mode 100644 index 00000000..d2d30c51 --- /dev/null +++ b/channels/base.py @@ -0,0 +1,75 @@ +import abc +import threading + +import auth + + +class BaseChannel(abc.ABC): + def __init__(self): + self._last_message = "" + self._msg_lock = threading.Lock() + + self._authenticated_id = None + self._auth_lock = threading.Lock() + + self._running = False + self._connected = False + self._thread = None + + def _set_last(self, msg: str) -> None: + with self._msg_lock: + if self._last_message == "": + self._last_message = msg + else: + self._last_message = self._last_message + " | " + msg + + def getLastMessage(self) -> str: + with self._msg_lock: + tmp = self._last_message + self._last_message = "" + return tmp + + @staticmethod + def _parse_auth_candidate(msg: str) -> str: + text = msg.strip() + lower = text.lower() + if lower.startswith("auth "): + return text[5:].strip() + if lower.startswith("/auth "): + return text[6:].strip() + return text + + @staticmethod + def _is_auth_command(msg: str) -> bool: + lower = msg.strip().lower() + return lower.startswith("auth ") or lower.startswith("/auth ") + + def _is_allowed_message(self, sender_id: str, msg: str) -> str: + with self._auth_lock: + if not auth.is_auth_enabled(): + return "allow" + if self._authenticated_id is not None: + return "allow" if sender_id == self._authenticated_id else "ignore" + if not self._is_auth_command(msg): + return "ignore" + candidate = self._parse_auth_candidate(msg) + if auth.verify_token(candidate): + self._authenticated_id = sender_id + return "auth_bound" + return "ignore" + + def start(self) -> threading.Thread: + self._running = True + self._connected = False + self._thread = threading.Thread(target=self._run_loop, daemon=True) + self._thread.start() + return self._thread + + def stop(self) -> None: + self._running = False + + @abc.abstractmethod + def _run_loop(self) -> None: ... + + @abc.abstractmethod + def send_message(self, text: str) -> None: ... diff --git a/channels/channels_registry.py b/channels/channels_registry.py new file mode 100644 index 00000000..18165720 --- /dev/null +++ b/channels/channels_registry.py @@ -0,0 +1,66 @@ +import importlib +import os + +_REGISTRY = { + "irc": ( + "channels.irc", "start_irc", + lambda token, channel_id, poll_interval, server_url: ( + channel_id or "##omegaclaw", + server_url or "irc.quakenet.org", + int(os.environ.get("IRC_PORT", "6667")), + os.environ.get("IRC_USER", "omegaclaw"), + ), + ), + "telegram": ( + "channels.telegram", "start_telegram", + lambda token, channel_id, poll_interval, server_url: ( + token or os.environ.get("TG_BOT_TOKEN", ""), + channel_id, + poll_interval, + ), + ), + "slack": ( + "channels.slack", "start_slack", + lambda token, channel_id, poll_interval, server_url: ( + token or os.environ.get("SL_BOT_TOKEN", ""), + channel_id, + poll_interval, + ), + ), + "mattermost": ( + "channels.mattermost", "start_mattermost", + lambda token, channel_id, poll_interval, server_url: ( + server_url or "https://chat.singularitynet.io", + channel_id or "8fjrmabjx7gupy7e5kjznpt5qh", + token or os.environ.get("MM_BOT_TOKEN", ""), + ), + ), + "mock": ( + "channels.mock", "start_mock", + lambda token, channel_id, poll_interval, server_url: (), + ), +} + +_active_module = None + + +def start(channel_name: str, token="", channel_id="", poll_interval=20, server_url=""): + global _active_module + channel_name = str(channel_name) + entry = _REGISTRY.get(channel_name) + if not entry: + raise ValueError(f"Unknown channel: {channel_name}") + module_path, start_fn, args_fn = entry + args = args_fn(str(token), str(channel_id), int(poll_interval), str(server_url)) + mod = importlib.import_module(module_path) + _active_module = mod + return getattr(mod, start_fn)(*args) + + +def getLastMessage() -> str: + return _active_module.getLastMessage() if _active_module else "" + + +def send_message(text: str) -> None: + if _active_module: + _active_module.send_message(text) diff --git a/channels/irc.py b/channels/irc.py index 89566db7..e538c73f 100644 --- a/channels/irc.py +++ b/channels/irc.py @@ -1,163 +1,141 @@ -import os import random import socket +import textwrap import threading import time -import textwrap -import auth -_running = False -_sock = None -_sock_lock = threading.Lock() -_last_message = "" -_msg_lock = threading.Lock() -_channel = None -_connected = False -_auth_lock = threading.Lock() -_authenticated_nick = None - -def _send(cmd): - with _sock_lock: - if _sock: - _sock.sendall((cmd + "\r\n").encode()) - time.sleep(1) - -def _set_last(msg): - global _last_message - with _msg_lock: - if _last_message == "": - _last_message = msg - else: - _last_message = _last_message + " | " + msg - -def getLastMessage(): - global _last_message - with _msg_lock: - tmp = _last_message - _last_message = "" - return tmp - - -def _normalize_nick(nick): - return nick.strip().lower() - - -def _parse_auth_candidate(msg): - text = msg.strip() - lower = text.lower() - if lower.startswith("auth "): - return text[5:].strip() - if lower.startswith("/auth "): - return text[6:].strip() - return text - -def _is_auth_command(msg): - lower = msg.strip().lower() - return lower.startswith("auth ") or lower.startswith("/auth ") - -def _is_allowed_message(nick, msg): - global _authenticated_nick - norm_nick = _normalize_nick(nick) - with _auth_lock: - if not auth.is_auth_enabled(): - return "allow" - if _authenticated_nick is not None: - return "allow" if norm_nick == _authenticated_nick else "ignore" - if not _is_auth_command(msg): +import auth +from channels.base import BaseChannel + + +class IRCChannel(BaseChannel): + def __init__(self, channel, server, port, nick): + super().__init__() + if not channel.startswith("#"): + channel = f"#{channel}" + self._channel = channel + self._server = server + self._port = int(port) + self._nick = f"{nick}{random.randint(1000, 9999)}" + self._sock = None + self._sock_lock = threading.Lock() + + def _is_allowed_message(self, sender_id: str, msg: str) -> str: + norm_nick = sender_id.strip().lower() + with self._auth_lock: + if not auth.is_auth_enabled(): + return "allow" + if self._authenticated_id is not None: + return "allow" if norm_nick == self._authenticated_id else "ignore" + if not self._is_auth_command(msg): + return "ignore" + candidate = self._parse_auth_candidate(msg) + if auth.verify_token(candidate): + self._authenticated_id = norm_nick + return "auth_bound" return "ignore" - candidate = _parse_auth_candidate(msg) - if auth.verify_token(candidate): - _authenticated_nick = norm_nick - return "auth_bound" - return "ignore" - -def _irc_loop(channel, server, port, nick): - global _running, _sock, _connected - print(f"[IRC] Connecting to {server}:{port} as {nick} for channel {channel}") - try: - sock = socket.create_connection((server, int(port)), timeout=15) - sock.settimeout(60) - print("[IRC] TCP connected") - except OSError as e: - print(f"[IRC] Connect failed: {e}") - return - _sock = sock - _send(f"NICK {nick}") - _send(f"USER {nick} 0 * :{nick}") - #_send(f"JOIN {channel}") - read_buffer = "" - while _running: + + def _send_raw(self, cmd: str) -> None: + with self._sock_lock: + if self._sock: + self._sock.sendall((cmd + "\r\n").encode()) + time.sleep(1) + + def send_message(self, text: str) -> None: + segments = text.replace("\r", "").split("\\n") + lines = [] + for seg in segments: + lines.extend(textwrap.wrap(seg, width=400, break_long_words=True, break_on_hyphens=False)) + for chunk in lines: + try: + if self._connected and self._channel: + self._send_raw(f"PRIVMSG {self._channel} :{chunk}") + except Exception as e: + print(f"[IRC] send error: {e}") + + def _run_loop(self) -> None: + print(f"[IRC] Connecting to {self._server}:{self._port} as {self._nick}") try: - data = sock.recv(4096).decode(errors="ignore") - if not data: - break - except socket.timeout: - continue - except OSError: - break - read_buffer += data - while "\r\n" in read_buffer: - line, read_buffer = read_buffer.split("\r\n", 1) - if not line: + sock = socket.create_connection((self._server, self._port), timeout=15) + sock.settimeout(60) + except OSError as e: + print(f"[IRC] Connect failed: {e}") + return + + self._sock = sock + self._send_raw(f"NICK {self._nick}") + self._send_raw(f"USER {self._nick} 0 * :{self._nick}") + + buf = "" + while self._running: + try: + data = sock.recv(4096).decode(errors="ignore") + if not data: + break + except socket.timeout: continue - if line.startswith("PING"): - _send(f"PONG {line.split()[1]}") - parts = line.split() - if len(parts) > 1 and parts[1] == "001": - _connected = True - print(f"[IRC] Registered. Joining {_channel}") - _send(f"JOIN {_channel}") - elif len(parts) > 1 and parts[1] in {"403", "405", "471", "473", "474", "475"}: - print(f"[IRC] Join failed: {line}") - elif len(parts) > 1 and parts[1] == "433": - print(f"[IRC] Nickname in use: {line}") - elif line.startswith(":") and " PRIVMSG " in line: - try: - prefix, trailing = line[1:].split(" PRIVMSG ", 1) - nick = prefix.split("!", 1)[0] - - if " :" not in trailing: - continue # malformed, ignore safely - - msg = trailing.split(" :", 1)[1] - state = _is_allowed_message(nick, msg) - if state == "allow": - _set_last(f"{nick}: {msg}") - elif state == "auth_bound": - _send(f"PRIVMSG {_channel} :Authentication successful for {nick}.") - except Exception as e: - print(f"[IRC]: exception caught {repr(e)}") - _connected = False - with _sock_lock: - _sock = None - sock.close() - print("[IRC] Disconnected") - -def start_irc(channel, server="irc.libera.chat", port=6667, nick="omegaclaw"): - global _running, _channel, _connected - nick = f"{nick}{random.randint(1000, 9999)}" - if not channel.startswith("#"): - channel = f"#{channel}" - _running = True - _connected = False - _channel = channel - t = threading.Thread(target=_irc_loop, args=(channel, server, port, nick), daemon=True) - t.start() - return t + except OSError: + break + + buf += data + while "\r\n" in buf: + line, buf = buf.split("\r\n", 1) + if not line: + continue + if line.startswith("PING"): + self._send_raw(f"PONG {line.split()[1]}") + continue + parts = line.split() + if len(parts) < 2: + continue + if parts[1] == "001": + self._connected = True + print(f"[IRC] Registered. Joining {self._channel}") + self._send_raw(f"JOIN {self._channel}") + elif parts[1] in {"403", "405", "471", "473", "474", "475"}: + print(f"[IRC] Join failed: {line}") + elif parts[1] == "433": + print(f"[IRC] Nickname in use: {line}") + elif line.startswith(":") and " PRIVMSG " in line: + try: + prefix, trailing = line[1:].split(" PRIVMSG ", 1) + nick = prefix.split("!", 1)[0] + if " :" not in trailing: + continue # malformed, ignore safely + msg = trailing.split(" :", 1)[1] + state = self._is_allowed_message(nick, msg) + if state == "allow": + self._set_last(f"{nick}: {msg}") + elif state == "auth_bound": + self._send_raw(f"PRIVMSG {self._channel} :Authentication successful for {nick}.") + except Exception: + pass + + self._connected = False + with self._sock_lock: + self._sock = None + sock.close() + print("[IRC] Disconnected") + + +_instance = None + + +def start_irc(channel, server="irc.quakenet.org", port=6667, nick="omegaclaw"): + global _instance + _instance = IRCChannel(channel, server, port, nick) + return _instance.start() + def stop_irc(): - global _running - _running = False - -def send_message(text): - max_len = 400 - segments = text.replace("\r", "").split("\\n") - lines = [] - for segment in segments: - lines.extend(textwrap.wrap(segment, width=max_len, break_long_words=True, break_on_hyphens=False)) - for chunk in lines: - try: - if _connected and _channel: - _send(f"PRIVMSG {_channel} :{chunk}") - except Exception as e: - print(f"[IRC] error in send_message on channel {_channel}: {e}") + if _instance: + _instance.stop() + + +def getLastMessage() -> str: + return _instance.getLastMessage() if _instance else "" + + +def send_message(text: str) -> None: + if _instance: + _instance.send_message(text) diff --git a/channels/mattermost.py b/channels/mattermost.py index e13fe647..8f738159 100644 --- a/channels/mattermost.py +++ b/channels/mattermost.py @@ -1,169 +1,103 @@ import json -import os -import threading import time - +import threading import requests import websocket -import auth - -_running = False -_ws = None -_ws_lock = threading.Lock() -_last_message = "" -_msg_lock = threading.Lock() -_connected = False -_auth_lock = threading.Lock() -_authenticated_user_id = None -_use_proxy = True - -# ---- Mattermost config (dummy token ok) ---- -MM_URL = "https://chat.singularitynet.io" -CHANNEL_ID = "8fjrmabjx7gupy7e5kjznpt5qh" #NOT AN ID JUST NAME: "omegaclaw"x -BOT_TOKEN = "" - -def _get_bot_user_id(): - global headers - r = requests.get( - f"{MM_URL}/api/v4/users/me", - headers=_headers - ) - return r.json()["id"] - -def _set_last(msg): - global _last_message - with _msg_lock: - if _last_message == "": - _last_message = msg - else: - _last_message = _last_message + " | " + msg - -def getLastMessage(): - global _last_message - with _msg_lock: - tmp = _last_message - _last_message = "" - return tmp - -def _parse_auth_candidate(msg): - text = msg.strip() - lower = text.lower() - if lower.startswith("auth "): - return text[5:].strip() - if lower.startswith("/auth "): - return text[6:].strip() - return text - -def _is_auth_command(msg): - lower = msg.strip().lower() - return lower.startswith("auth ") or lower.startswith("/auth ") - -def _is_allowed_message(user_id, msg): - global _authenticated_user_id - with _auth_lock: - if not auth.is_auth_enabled(): - return "allow" - if _authenticated_user_id is not None: - return "allow" if user_id == _authenticated_user_id else "ignore" - if not _is_auth_command(msg): - return "ignore" - candidate = _parse_auth_candidate(msg) - if auth.verify_token(candidate): - _authenticated_user_id = user_id - return "auth_bound" - return "ignore" - -def _get_display_name(user_id): - r = requests.get( - f"{MM_URL}/api/v4/users/{user_id}", - headers=_headers - ) - u = r.json() - - # Mimic common Mattermost display setting - if u.get("first_name") or u.get("last_name"): - return f"{u.get('first_name','')} {u.get('last_name','')}".strip() - return u["username"] - -def _ws_loop(): - global _ws, _connected, BOT_USER_ID, _use_proxy - - if _use_proxy: - ws_url = MM_URL.replace("http", "ws") - else: - ws_url = MM_URL.replace("https", "wss") - ws_url = ws_url + "/api/v4/websocket" - ws = websocket.WebSocket() - ws.connect(ws_url, header=[f"Authorization: Bearer {BOT_TOKEN}"]) - - BOT_USER_ID = _get_bot_user_id() - _ws = ws - _connected = True - - last_ping = time.time() - - while _running: - try: - # send ping every 25s - if time.time() - last_ping > 25: - ws.ping() - last_ping = time.time() - - ws.settimeout(1) - event = json.loads(ws.recv()) - - if event.get("event") == "posted": - post = json.loads(event["data"]["post"]) - if post["channel_id"] == CHANNEL_ID and post["user_id"] != BOT_USER_ID: - user_id = post["user_id"] - message = post.get("message", "") - state = _is_allowed_message(user_id, message) - if state == "allow": - name = _get_display_name(user_id) - _set_last(f"{name}: {message}") - elif state == "auth_bound": - name = _get_display_name(user_id) - send_message(f"Authentication successful for {name}.") - - except websocket.WebSocketTimeoutException: - continue - except Exception: - break - - ws.close() - _connected = False - -def start_mattermost(MM_URL_, CHANNEL_ID_): - global _running, MM_URL, CHANNEL_ID, BOT_TOKEN, _headers, _connected, _use_proxy +import auth +from channels.base import BaseChannel + +class MattermostChannel(BaseChannel): + def __init__(self, url, channel_id, bot_token): + super().__init__() + self._url = url.rstrip("/") + self._channel_id = channel_id + self._bot_token = bot_token + self._headers = {"Authorization": f"Bearer {bot_token}"} + self._bot_user_id = "" + + def _get(self, path): + return requests.get(f"{self._url}/api/v4{path}", headers=self._headers).json() + + def _get_bot_user_id(self): + return self._get("/users/me")["id"] + + def _get_display_name(self, user_id): + u = self._get(f"/users/{user_id}") + if u.get("first_name") or u.get("last_name"): + return f"{u.get('first_name','')} {u.get('last_name','')}".strip() + return u["username"] + + def _run_loop(self) -> None: + ws_url = (self._url.replace("http", "ws") if self._bot_token == "proxy" + else self._url.replace("https", "wss")) + "/api/v4/websocket" + ws = websocket.WebSocket() + ws_headers = ([f"Authorization: Bearer {self._bot_token}"] + if self._bot_token and self._bot_token != "proxy" else []) + ws.connect(ws_url, header=ws_headers) + self._bot_user_id = self._get_bot_user_id() + self._connected = True + last_ping = time.time() + print("[MATTERMOST] Connected") + + while self._running: + try: + if time.time() - last_ping > 25: + ws.ping() + last_ping = time.time() + ws.settimeout(1) + event = json.loads(ws.recv()) + if event.get("event") == "posted": + post = json.loads(event["data"]["post"]) + if post["channel_id"] == self._channel_id \ + and post["user_id"] != self._bot_user_id: + user_id = post["user_id"] + msg = post.get("message", "") + state = self._is_allowed_message(user_id, msg) + if state == "allow": + name = self._get_display_name(user_id) + self._set_last(f"{name}: {msg}") + elif state == "auth_bound": + name = self._get_display_name(user_id) + self.send_message(f"Authentication successful for {name}.") + except websocket.WebSocketTimeoutException: + continue + except Exception: + break + + ws.close() + self._connected = False + print("[MATTERMOST] Disconnected") + + def send_message(self, text: str) -> None: + text = text.replace("\\n", "\n") + if not self._connected: + return + requests.post(f"{self._url}/api/v4/posts", headers=self._headers, + json={"channel_id": self._channel_id, "message": text}) + +_instance = None + +def start_mattermost(url, channel_id, bot_token): + global _instance proxy = auth.get_proxy_url() if proxy: - MM_URL = f"{proxy}/mattermost" - BOT_TOKEN = "proxy" - _headers = {} - _use_proxy = True + url = f"{proxy}/mattermost" + bot_token = "proxy" + _instance = MattermostChannel(url, channel_id, bot_token) + _instance._headers = {} else: - MM_URL = MM_URL_ - BOT_TOKEN = os.environ.get("MM_BOT_TOKEN", "").strip() - _headers = {"Authorization": f"Bearer {BOT_TOKEN}"} if BOT_TOKEN else {} - _use_proxy = False - CHANNEL_ID = CHANNEL_ID_ - _running = True - _connected = False - t = threading.Thread(target=_ws_loop, daemon=True) - t.start() - return t + _instance = MattermostChannel(url, channel_id, bot_token) + return _instance.start() def stop_mattermost(): - global _running - _running = False + if _instance: + _instance.stop() + +def getLastMessage() -> str: + return _instance.getLastMessage() if _instance else "" + -def send_message(text): - text = text.replace("\\n", "\n") - if not _connected: - return - requests.post( - f"{MM_URL}/api/v4/posts", - headers=_headers, - json={"channel_id": CHANNEL_ID, "message": text} - ) +def send_message(text: str) -> None: + if _instance: + _instance.send_message(text) diff --git a/channels/slack.py b/channels/slack.py index e148641c..10cccd95 100644 --- a/channels/slack.py +++ b/channels/slack.py @@ -1,48 +1,21 @@ import json -import os import re import threading import time import urllib.error import urllib.parse import urllib.request -import auth - -_running = False -_last_message = "" -_msg_lock = threading.Lock() -_state_lock = threading.Lock() -_bot_token = "" -_channel_id = "" -_poll_interval = 10 -_bot_user_id = "" -_connected = False -_user_cache = {} -_channel_offsets = {} -_channel_name_cache = {} -_auto_bind_channels = [] -_auto_bind_index = 0 -_auto_bind_last_refresh = 0.0 +import auth +from channels.base import BaseChannel -_authenticated_user_id = None -_rate_limit_until = 0.0 _AUTO_BIND_REFRESH_INTERVAL = 300 -_SL_URL = "https://slack.com" - -class _SlackRateLimitError(Exception): - def __init__(self, retry_after): - super().__init__(f"Slack rate limited (retry after {retry_after}s)") - self.retry_after = retry_after - - _URL_DISPLAY_RE = re.compile(r"<[^|>\s]+\|([^>]*)>") _URL_BARE_RE = re.compile(r"<([^>\s]+)>") def _slack_unwrap(text): - """Reverse Slack's outgoing auto-formatting so downstream layers see the original text.""" if not text: return text text = _URL_DISPLAY_RE.sub(r"\1", text) @@ -50,425 +23,266 @@ def _slack_unwrap(text): text = text.replace("<", "<").replace(">", ">").replace("&", "&") return text +class _RateLimitError(Exception): + def __init__(self, retry_after): + super().__init__(f"Rate limited (retry after {retry_after}s)") + self.retry_after = retry_after -def _set_last(msg): - global _last_message - with _msg_lock: - if _last_message == "": - _last_message = msg - else: - _last_message = _last_message + " | " + msg - - -def getLastMessage(): - global _last_message - with _msg_lock: - tmp = _last_message - _last_message = "" - return tmp - - -def _parse_auth_candidate(msg): - text = msg.strip() - lower = text.lower() - if lower.startswith("auth "): - return text[5:].strip() - if lower.startswith("/auth "): - return text[6:].strip() - return text - -def _is_auth_command(msg): - lower = msg.strip().lower() - return lower.startswith("auth ") or lower.startswith("/auth ") - -def _channel_label(channel_id): - with _state_lock: - label = _channel_name_cache.get(channel_id) - return label or channel_id - - -def _is_allowed_message(channel_id, user_id, msg): - global _channel_id, _authenticated_user_id - with _state_lock: - if _channel_id and channel_id != _channel_id: +class SlackChannel(BaseChannel): + def __init__(self, bot_token, channel_id="", poll_interval=60): + super().__init__() + self._bot_token = str(bot_token).strip() + self._api_base = "https://slack.com" + self._channel_id = str(channel_id).strip() + self._poll_interval = max(1, int(poll_interval)) + self._bot_user_id = "" + self._state_lock = threading.Lock() + self._user_cache = {} + self._channel_offsets = {} + self._channel_name_cache = {} + self._auto_bind_channels = [] + self._auto_bind_index = 0 + self._auto_bind_last_refresh = 0.0 + self._rate_limit_until = 0.0 + + def _is_allowed_message(self, sender_id: str, msg: str, channel_id: str = "") -> str: + with self._auth_lock: + if self._channel_id and channel_id != self._channel_id: + return "ignore" + if not auth.is_auth_enabled(): + if not self._channel_id: + with self._state_lock: + label = self._channel_name_cache.get(channel_id, channel_id) + print(f"[SLACK] Auto-bound to channel {label}") + self._channel_id = channel_id + return "allow" + if self._authenticated_id is not None: + return "allow" if sender_id == self._authenticated_id else "ignore" + if not self._is_auth_command(msg): + return "ignore" + candidate = self._parse_auth_candidate(msg) + if auth.verify_token(candidate): + self._authenticated_id = sender_id + self._channel_id = channel_id + return "auth_bound" return "ignore" - if not auth.is_auth_enabled(): - if not _channel_id: - _bind_label = _channel_name_cache.get(channel_id, channel_id) - print(f"[SLACK] Auto-bound to channel {_bind_label}") - _channel_id = channel_id - return "allow" - if _authenticated_user_id is not None: - return "allow" if user_id == _authenticated_user_id else "ignore" - candidate = _parse_auth_candidate(msg) - if auth.verify_token(candidate): - _authenticated_user_id = user_id - _channel_id = channel_id - return "auth_bound" - return "ignore" - - -def _parse_retry_after(value): - try: - seconds = int(str(value).strip()) - return max(1, seconds) - except Exception: - return 60 - - -def _set_rate_limit_backoff(seconds): - global _rate_limit_until - until = time.time() + max(1, int(seconds)) - with _state_lock: - if until > _rate_limit_until: - _rate_limit_until = until - - -def _wait_for_rate_limit_window(): - with _state_lock: - wait = _rate_limit_until - time.time() - if wait > 0: - time.sleep(wait) - - -def _api_call(method, params=None, timeout=30): - global _SL_URL, _bot_token - - if not _bot_token: - raise RuntimeError("Slack adapter not initialized") - - params = params or {} - body = urllib.parse.urlencode(params).encode("utf-8") - req = urllib.request.Request( - f"{_SL_URL}/api/{method}", - data=body, - headers={ - "Authorization": f"Bearer {_bot_token}", - "Content-Type": "application/x-www-form-urlencoded", - }, - method="POST", - ) - - _wait_for_rate_limit_window() - try: - with urllib.request.urlopen(req, timeout=timeout) as response: - payload = json.loads(response.read().decode("utf-8", errors="ignore")) - response_headers = response.headers - except urllib.error.HTTPError as exc: - if exc.code == 429: - retry_after = _parse_retry_after(exc.headers.get("Retry-After")) - _set_rate_limit_backoff(retry_after) - raise _SlackRateLimitError(retry_after) from exc - raise - - if not payload.get("ok"): - err = payload.get("error", f"{method} failed") - if err == "ratelimited": - retry_after = _parse_retry_after(response_headers.get("Retry-After")) - _set_rate_limit_backoff(retry_after) - raise _SlackRateLimitError(retry_after) - raise RuntimeError(err) - - return payload - - -def _get_display_name(user_id): - with _state_lock: - cached = _user_cache.get(user_id) - if cached: - return cached - - name = user_id - try: - payload = _api_call("users.info", {"user": user_id}, timeout=15) - user = payload.get("user") or {} - profile = user.get("profile") or {} - - display_name = str(profile.get("display_name", "")).strip() - real_name = str(profile.get("real_name", "")).strip() - username = str(user.get("name", "")).strip() - - if display_name: - name = display_name - elif real_name: - name = real_name - elif username: - name = username - except Exception as exc: - print(f"[SLACK] Could not resolve user {user_id}: {exc}") - - with _state_lock: - _user_cache[user_id] = name - return name - - -def _cache_channel(channel): - channel_id = str(channel.get("id", "")).strip() - if not channel_id: - return - channel_name = str(channel.get("name", "")).strip() - label = f"#{channel_name}" if channel_name else channel_id - with _state_lock: - _channel_name_cache[channel_id] = label - - -def _initialize_identity(): - global _bot_user_id - payload = _api_call("auth.test", timeout=15) - bot_user_id = str(payload.get("user_id", "")).strip() - with _state_lock: - _bot_user_id = bot_user_id - - -def _validate_channel(): - payload = _api_call("conversations.info", {"channel": _channel_id}, timeout=15) - channel = payload.get("channel") or {} - _cache_channel(channel) - channel_name = str(channel.get("name", "")).strip() - if channel_name: - print(f"[SLACK] Channel ready: #{channel_name}") - else: - print(f"[SLACK] Channel ready: {_channel_id}") - - -def _list_joined_channels(): - channels = [] - cursor = "" - while True: - params = { - "types": "public_channel,private_channel", - "exclude_archived": "true", - "limit": 200, - } - if cursor: - params["cursor"] = cursor - - payload = _api_call("conversations.list", params=params, timeout=20) - for channel in payload.get("channels") or []: - if not channel.get("is_member"): - continue - channel_id = str(channel.get("id", "")).strip() - if not channel_id: - continue - _cache_channel(channel) - channels.append(channel_id) - metadata = payload.get("response_metadata") or {} - cursor = str(metadata.get("next_cursor", "")).strip() - if not cursor: - break - return channels - - -def _initialize_cursor_for_channel(channel_id): - try: - payload = _api_call( - "conversations.history", - {"channel": channel_id, "limit": 1}, - timeout=15, + def _api_call(self, method, params=None, timeout=30): + params = params or {} + body = urllib.parse.urlencode(params).encode("utf-8") + req = urllib.request.Request( + f"{self._api_base}/api/{method}", data=body, + headers={"Authorization": f"Bearer {self._bot_token}", + "Content-Type": "application/x-www-form-urlencoded"}, + method="POST", ) - messages = payload.get("messages") or [] - ts = "" - if messages: - ts = str(messages[0].get("ts", "")).strip() - with _state_lock: - _channel_offsets[channel_id] = ts - except Exception as exc: - print(f"[SLACK] Could not initialize cursor for {_channel_label(channel_id)}: {exc}") - - -def _initialize_auto_bind_cursors(): - channels = _refresh_auto_bind_channels(force=True) - if not channels: - print("[SLACK] Auto-bind waiting: no joined channels visible yet.") - else: - print(f"[SLACK] Auto-bind watching {len(channels)} joined channel(s).") - - -def _refresh_auto_bind_channels(force=False): - global _auto_bind_channels, _auto_bind_last_refresh, _auto_bind_index - now = time.time() - with _state_lock: - cached = list(_auto_bind_channels) - last_refresh = _auto_bind_last_refresh - - if (not force) and cached and (now - last_refresh) < _AUTO_BIND_REFRESH_INTERVAL: - return cached - - channels = _list_joined_channels() - with _state_lock: - _auto_bind_channels = channels - _auto_bind_last_refresh = now - if _auto_bind_index >= len(channels): - _auto_bind_index = 0 - return channels - - -def _next_auto_bind_channel(): - global _auto_bind_index - with _state_lock: - if not _auto_bind_channels: - return "" - channel_id = _auto_bind_channels[_auto_bind_index] - _auto_bind_index = (_auto_bind_index + 1) % len(_auto_bind_channels) - return channel_id - - -def _poll_channel(channel_id): - params = {"channel": channel_id, "limit": 15} - with _state_lock: - oldest = _channel_offsets.get(channel_id, "") - if oldest: - params["oldest"] = oldest - params["inclusive"] = "false" - - payload = _api_call("conversations.history", params=params, timeout=30) - messages = payload.get("messages") or [] - if not messages: - return - - ordered = sorted(messages, key=lambda m: float(m.get("ts", 0.0))) - max_ts = oldest - for message in ordered: - ts = str(message.get("ts", "")).strip() - if ts: - max_ts = ts - - # Ignore bot/system messages and process regular user text. - if message.get("subtype"): - continue - - text = _slack_unwrap(str(message.get("text", "")).strip()) - user_id = str(message.get("user", "")).strip() - if not text or not user_id: - continue - - with _state_lock: - bot_user_id = _bot_user_id - if bot_user_id and user_id == bot_user_id: - continue - - state = _is_allowed_message(channel_id, user_id, text) - display_name = _get_display_name(user_id) - if state == "allow": - _set_last(f"{display_name}: {text}") - elif state == "auth_bound": - send_message(f"Authentication successful for {display_name}.") - - if max_ts != oldest: - with _state_lock: - _channel_offsets[channel_id] = max_ts - - -def _poll_loop(): - global _connected - print("[SLACK] Polling started") - - while _running: + with self._state_lock: + wait = self._rate_limit_until - time.time() + if wait > 0: + time.sleep(wait) try: - with _state_lock: - bound_channel = _channel_id - - if bound_channel: - with _state_lock: - known = bound_channel in _channel_offsets - if not known: - _initialize_cursor_for_channel(bound_channel) - _connected = True - else: - _poll_channel(bound_channel) - else: - channels = _refresh_auto_bind_channels() - if not channels: - channels = _refresh_auto_bind_channels(force=True) - channel_id = _next_auto_bind_channel() - if channel_id: - with _state_lock: - known = channel_id in _channel_offsets - if not known: - _initialize_cursor_for_channel(channel_id) - _connected = True - else: - _poll_channel(channel_id) - - _connected = True - except _SlackRateLimitError as exc: - _connected = False - print(f"[SLACK] Rate limited. Backing off for {exc.retry_after}s.") + with urllib.request.urlopen(req, timeout=timeout) as resp: + payload = json.loads(resp.read().decode("utf-8", errors="ignore")) + headers = resp.headers + except urllib.error.HTTPError as exc: + if exc.code == 429: + retry = max(1, int(exc.headers.get("Retry-After", 60))) + with self._state_lock: + self._rate_limit_until = time.time() + retry + raise _RateLimitError(retry) from exc + raise + if not payload.get("ok"): + err = payload.get("error", f"{method} failed") + if err == "ratelimited": + retry = max(1, int(headers.get("Retry-After", 60))) + with self._state_lock: + self._rate_limit_until = time.time() + retry + raise _RateLimitError(retry) + raise RuntimeError(err) + return payload + + def _get_display_name(self, user_id): + with self._state_lock: + cached = self._user_cache.get(user_id) + if cached: + return cached + name = user_id + try: + payload = self._api_call("users.info", {"user": user_id}, timeout=15) + profile = (payload.get("user") or {}).get("profile") or {} + name = (profile.get("display_name") or profile.get("real_name") or + (payload["user"].get("name")) or user_id).strip() except Exception as exc: - _connected = False - print(f"[SLACK] Poll error: {exc}") - - time.sleep(max(1, int(_poll_interval))) - - _connected = False - print("[SLACK] Polling stopped") - - -def start_slack(channel_id, poll_interval=60): - global _running, _bot_token, _channel_id, _poll_interval, _connected - global _rate_limit_until, _auto_bind_channels, _auto_bind_index, _auto_bind_last_refresh - global _SL_URL - - proxy = auth.get_proxy_url() - if proxy: - _SL_URL = f"{proxy}/slack" - _bot_token = "proxy" - else: - _bot_token = os.environ.get("SL_BOT_TOKEN", "").strip() - if not _bot_token: + print(f"[SLACK] Could not resolve user {user_id}: {exc}") + with self._state_lock: + self._user_cache[user_id] = name + return name + + def _cache_channel(self, channel): + cid = str(channel.get("id", "")).strip() + name = str(channel.get("name", "")).strip() + if cid: + self._channel_name_cache[cid] = f"#{name}" if name else cid + + def _list_joined_channels(self): + channels, cursor = [], "" + while True: + params = {"types": "public_channel,private_channel", + "exclude_archived": "true", "limit": 200} + if cursor: + params["cursor"] = cursor + payload = self._api_call("conversations.list", params=params, timeout=20) + for ch in payload.get("channels") or []: + if ch.get("is_member"): + cid = str(ch.get("id", "")).strip() + if cid: + self._cache_channel(ch) + channels.append(cid) + cursor = str((payload.get("response_metadata") or {}).get("next_cursor", "")).strip() + if not cursor: + break + return channels + + def _init_cursor(self, channel_id): + try: + payload = self._api_call("conversations.history", + {"channel": channel_id, "limit": 1}, timeout=15) + msgs = payload.get("messages") or [] + ts = str(msgs[0].get("ts", "")).strip() if msgs else "" + self._channel_offsets[channel_id] = ts + except Exception as exc: + print(f"[SLACK] Could not initialize cursor for {channel_id}: {exc}") + + def _refresh_auto_bind(self, force=False): + now = time.time() + if (not force) and self._auto_bind_channels and \ + (now - self._auto_bind_last_refresh) < _AUTO_BIND_REFRESH_INTERVAL: + return self._auto_bind_channels + self._auto_bind_channels = self._list_joined_channels() + self._auto_bind_last_refresh = now + if self._auto_bind_index >= len(self._auto_bind_channels): + self._auto_bind_index = 0 + return self._auto_bind_channels + + def _next_auto_bind_channel(self): + if not self._auto_bind_channels: + return "" + cid = self._auto_bind_channels[self._auto_bind_index] + self._auto_bind_index = (self._auto_bind_index + 1) % len(self._auto_bind_channels) + return cid + + def _poll_channel(self, channel_id): + oldest = self._channel_offsets.get(channel_id, "") + params = {"channel": channel_id, "limit": 15} + if oldest: + params["oldest"] = oldest + params["inclusive"] = "false" + payload = self._api_call("conversations.history", params=params, timeout=30) + messages = sorted(payload.get("messages") or [], + key=lambda m: float(m.get("ts", 0.0))) + max_ts = oldest + for msg in messages: + ts = str(msg.get("ts", "")).strip() + if ts: + max_ts = ts + if msg.get("subtype"): + continue + text = _slack_unwrap(str(msg.get("text", "")).strip()) + user_id = str(msg.get("user", "")).strip() + if not text or not user_id or user_id == self._bot_user_id: + continue + state = self._is_allowed_message(user_id, text, channel_id) + name = self._get_display_name(user_id) + if state == "allow": + self._set_last(f"{name}: {text}") + elif state == "auth_bound": + self.send_message(f"Authentication successful for {name}.") + if max_ts != oldest: + self._channel_offsets[channel_id] = max_ts + + def _run_loop(self) -> None: + print("[SLACK] Polling started") + while self._running: + try: + with self._state_lock: + bound = self._channel_id + if bound: + if bound not in self._channel_offsets: + self._init_cursor(bound) + else: + self._poll_channel(bound) + else: + channels = self._refresh_auto_bind() or self._refresh_auto_bind(force=True) + cid = self._next_auto_bind_channel() + if cid: + if cid not in self._channel_offsets: + self._init_cursor(cid) + else: + self._poll_channel(cid) + self._connected = True + except _RateLimitError as exc: + self._connected = False + print(f"[SLACK] Rate limited. Backing off {exc.retry_after}s.") + except Exception as exc: + self._connected = False + print(f"[SLACK] Poll error: {exc}") + time.sleep(max(1, self._poll_interval)) + self._connected = False + print("[SLACK] Polling stopped") + + def send_message(self, text: str) -> None: + text = str(text).replace("\\n", "\n").replace("\r", "") + if not text: + return + with self._state_lock: + target = self._channel_id + if not target: + return + for i in range(0, len(text), 3900): + chunk = text[i:i + 3900] + try: + self._api_call("chat.postMessage", {"channel": target, "text": chunk}, timeout=15) + except Exception as exc: + print(f"[SLACK] Send failed: {exc}") + return + + def start(self) -> threading.Thread: + proxy = auth.get_proxy_url() + if proxy: + self._bot_token = "proxy" + self._api_base = f"{proxy}/slack" + elif not self._bot_token: raise ValueError("SL_BOT_TOKEN is required") + payload = self._api_call("auth.test", timeout=15) + self._bot_user_id = str(payload.get("user_id", "")).strip() + if self._channel_id: + info = self._api_call("conversations.info", + {"channel": self._channel_id}, timeout=15) + self._cache_channel(info.get("channel") or {}) + self._init_cursor(self._channel_id) + print(f"[SLACK] Channel ready: " + f"{self._channel_name_cache.get(self._channel_id, self._channel_id)}") + else: + print("[SLACK] Starting in auto-bind mode.") + self._refresh_auto_bind(force=True) + print(f"[SLACK] Starting adapter with channel target: {self._channel_id or 'auto-bind'}") + return super().start() - _channel_id = str(channel_id).strip() - - try: - _poll_interval = min(60, int(poll_interval)) - except Exception: - _poll_interval = 60 - - with _state_lock: - _user_cache.clear() - _channel_offsets.clear() - _channel_name_cache.clear() - _auto_bind_channels = [] - _auto_bind_index = 0 - _auto_bind_last_refresh = 0.0 - _rate_limit_until = 0.0 - _connected = False - _initialize_identity() - if _channel_id: - _validate_channel() - _initialize_cursor_for_channel(_channel_id) - else: - print("[SLACK] Starting adapter in auto-bind mode (channel not configured).") - _initialize_auto_bind_cursors() - - _running = True - print(f"[SLACK] Starting adapter with channel target: {_channel_id or 'auto-bind'}") - t = threading.Thread(target=_poll_loop, daemon=True) - t.start() - return t +_instance = None +def start_slack(bot_token, channel_id="", poll_interval=60): + global _instance + _instance = SlackChannel(bot_token, channel_id, poll_interval) + return _instance.start() def stop_slack(): - global _running - _running = False + if _instance: + _instance.stop() +def getLastMessage() -> str: + return _instance.getLastMessage() if _instance else "" -def send_message(text): - text = str(text).replace("\\n", "\n").replace("\r", "") - if not text: - return - if not _channel_id: - return - max_len = 3900 - for i in range(0, len(text), max_len): - chunk = text[i:i + max_len] - if not chunk: - continue - try: - _api_call("chat.postMessage", {"channel": _channel_id, "text": chunk}, timeout=15) - except Exception as exc: - print(f"[SLACK] Send failed: {exc}") - return +def send_message(text: str) -> None: + if _instance: + _instance.send_message(text) diff --git a/channels/telegram.py b/channels/telegram.py index a91f8326..b28c5a58 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -1,253 +1,167 @@ import json -import os import threading import time import urllib.parse import urllib.request -import auth - -_running = False -_last_message = "" -_msg_lock = threading.Lock() -_state_lock = threading.Lock() - -_bot_token = "" -_api_base = "" -_chat_id = "" -_poll_timeout = 20 -_offset = None -_connected = False - -_authenticated_user_id = None - - -def _set_last(msg): - global _last_message - with _msg_lock: - if _last_message == "": - _last_message = msg - else: - _last_message = _last_message + " | " + msg - - -def getLastMessage(): - global _last_message - with _msg_lock: - tmp = _last_message - _last_message = "" - return tmp - - -def _parse_auth_candidate(msg): - text = msg.strip() - lower = text.lower() - if lower.startswith("auth "): - return text[5:].strip() - if lower.startswith("/auth "): - return text[6:].strip() - return text - - -def _display_name(user, chat): - username = str(user.get("username", "")).strip() - if username: - return f"@{username}" - - first = str(user.get("first_name", "")).strip() - last = str(user.get("last_name", "")).strip() - full = f"{first} {last}".strip() - if full: - return full - - title = str(chat.get("title", "")).strip() - if title: - return title - - return "telegram_user" - - -def _api_call(method, params=None, timeout=30, use_post=False): - if not _api_base: - raise RuntimeError("Telegram adapter not initialized") - - params = params or {} - encoded = urllib.parse.urlencode(params).encode("utf-8") - url = f"{_api_base}/{method}" - - if use_post: - req = urllib.request.Request(url, data=encoded) - else: - if params: - url = f"{url}?{urllib.parse.urlencode(params)}" - req = urllib.request.Request(url) - - with urllib.request.urlopen(req, timeout=timeout) as response: - payload = json.loads(response.read().decode("utf-8", errors="ignore")) - - if not payload.get("ok"): - raise RuntimeError(payload.get("description", f"{method} failed")) - - return payload.get("result") - - -def _initialize_offset(): - global _offset - try: - updates = _api_call("getUpdates", {"timeout": 0}, timeout=10) or [] - except Exception as exc: - print(f"[TELEGRAM] Could not read initial offset: {exc}") - return - - max_update = -1 - for update in updates: - update_id = update.get("update_id") - if isinstance(update_id, int): - max_update = max(max_update, update_id) - - if max_update >= 0: - with _state_lock: - _offset = max_update + 1 - - -def _is_auth_command(msg): - lower = msg.strip().lower() - return lower.startswith("auth ") or lower.startswith("/auth ") - -def _is_allowed_message(chat_id, user_id, msg): - global _chat_id, _authenticated_user_id - - with _state_lock: - if _chat_id and chat_id != _chat_id: - return "ignore" - if not auth.is_auth_enabled(): - if not _chat_id: - _chat_id = chat_id - return "allow" - if _authenticated_user_id is not None: - if chat_id != _chat_id: +import auth +from channels.base import BaseChannel + +class TelegramChannel(BaseChannel): + def __init__(self, bot_token, chat_id="", poll_timeout=20): + super().__init__() + self._bot_token = str(bot_token).strip() + self._api_base = f"https://api.telegram.org/bot{self._bot_token}" + self._chat_id = str(chat_id).strip() + self._poll_timeout = max(1, int(poll_timeout)) + self._offset = None + self._state_lock = threading.Lock() + self._authenticated_chat_id = None + + def _is_allowed_message(self, sender_id: str, msg: str, chat_id: str = "") -> str: + with self._auth_lock: + if self._chat_id and chat_id != self._chat_id: + return "ignore" + if not auth.is_auth_enabled(): + if not self._chat_id: + self._chat_id = chat_id + return "allow" + if self._authenticated_id is not None: + if chat_id != self._chat_id: + return "ignore" + return "allow" if sender_id == self._authenticated_id else "ignore" + if not self._is_auth_command(msg): return "ignore" - return "allow" if user_id == _authenticated_user_id else "ignore" - if not _is_auth_command(msg): + candidate = self._parse_auth_candidate(msg) + if auth.verify_token(candidate): + self._authenticated_id = sender_id + self._authenticated_chat_id = chat_id + self._chat_id = chat_id + return "auth_bound" return "ignore" - candidate = _parse_auth_candidate(msg) - if auth.verify_token(candidate): - _authenticated_user_id = user_id - _chat_id = chat_id - return "auth_bound" - return "ignore" - -def _poll_loop(): - global _connected, _offset - print("[TELEGRAM] Polling started") - - while _running: + def _api_call(self, method, params=None, timeout=30, use_post=False): + params = params or {} + url = f"{self._api_base}/{method}" + encoded = urllib.parse.urlencode(params).encode("utf-8") + if use_post: + req = urllib.request.Request(url, data=encoded) + else: + if params: + url = f"{url}?{urllib.parse.urlencode(params)}" + req = urllib.request.Request(url) + with urllib.request.urlopen(req, timeout=timeout) as resp: + payload = json.loads(resp.read().decode("utf-8", errors="ignore")) + if not payload.get("ok"): + raise RuntimeError(payload.get("description", f"{method} failed")) + return payload.get("result") + + def _initialize_offset(self): try: - params = {"timeout": int(_poll_timeout)} - with _state_lock: - if _offset is not None: - params["offset"] = _offset - - updates = _api_call("getUpdates", params=params, timeout=int(_poll_timeout) + 10) or [] - _connected = True - - for update in updates: - update_id = update.get("update_id") - if isinstance(update_id, int): - with _state_lock: - if _offset is None or (update_id + 1) > _offset: - _offset = update_id + 1 - - message = update.get("message") or update.get("edited_message") - if not isinstance(message, dict): - continue - - text = message.get("text") - if not text: - continue - - chat = message.get("chat") or {} - user = message.get("from") or {} - chat_id = str(chat.get("id", "")).strip() - user_id = str(user.get("id", "")).strip() - if not chat_id or not user_id: - continue - - state = _is_allowed_message(chat_id, user_id, text) - display_name = _display_name(user, chat) - if state == "allow": - _set_last(f"{display_name}: {text}") - elif state == "auth_bound": - send_message(f"Authentication successful for {display_name}.") + updates = self._api_call("getUpdates", {"timeout": 0}, timeout=10) or [] except Exception as exc: - _connected = False - print(f"[TELEGRAM] Poll error: {exc}") - time.sleep(2) - - _connected = False - print("[TELEGRAM] Polling stopped") - - -def start_telegram(chat_id="", poll_timeout=20): - global _running, _bot_token, _api_base, _chat_id, _poll_timeout, _offset, _connected - - proxy = auth.get_proxy_url() - if proxy: - _bot_token = "proxy" - _api_base = f"{proxy}/telegram" - else: - _bot_token = os.environ.get("TG_BOT_TOKEN", "").strip() - if not _bot_token: + print(f"[TELEGRAM] Could not read initial offset: {exc}") + return + max_update = max((u.get("update_id", -1) for u in updates), default=-1) + if max_update >= 0: + with self._state_lock: + self._offset = max_update + 1 + + @staticmethod + def _display_name(user, chat): + username = str(user.get("username", "")).strip() + if username: + return f"@{username}" + full = f"{user.get('first_name','')} {user.get('last_name','')}".strip() + if full: + return full + title = str(chat.get("title", "")).strip() + return title or "telegram_user" + + def _run_loop(self) -> None: + print("[TELEGRAM] Polling started") + while self._running: + try: + params = {"timeout": self._poll_timeout} + with self._state_lock: + if self._offset is not None: + params["offset"] = self._offset + updates = self._api_call("getUpdates", params=params, + timeout=self._poll_timeout + 10) or [] + self._connected = True + for update in updates: + uid = update.get("update_id") + if isinstance(uid, int): + with self._state_lock: + if self._offset is None or uid + 1 > self._offset: + self._offset = uid + 1 + message = update.get("message") or update.get("edited_message") + if not isinstance(message, dict): + continue + text = message.get("text") + if not text: + continue + chat = message.get("chat") or {} + user = message.get("from") or {} + chat_id = str(chat.get("id", "")).strip() + user_id = str(user.get("id", "")).strip() + if not chat_id or not user_id: + continue + state = self._is_allowed_message(user_id, text, chat_id) + name = self._display_name(user, chat) + if state == "allow": + self._set_last(f"{name}: {text}") + elif state == "auth_bound": + self.send_message(f"Authentication successful for {name}.") + except Exception as exc: + self._connected = False + print(f"[TELEGRAM] Poll error: {exc}") + time.sleep(2) + self._connected = False + print("[TELEGRAM] Polling stopped") + + def send_message(self, text: str) -> None: + text = str(text).replace("\\n", "\n").replace("\r", "") + if not text: + return + with self._auth_lock: + target = self._chat_id + if not self._connected or not target: + return + for i in range(0, len(text), 3900): + chunk = text[i:i + 3900] + try: + self._api_call("sendMessage", {"chat_id": target, "text": chunk}, + timeout=15, use_post=True) + except Exception as exc: + print(f"[TELEGRAM] Send failed: {exc}") + return + + def start(self) -> threading.Thread: + proxy = auth.get_proxy_url() + if proxy: + self._bot_token = "proxy" + self._api_base = f"{proxy}/telegram" + elif not self._bot_token: raise ValueError("TG_BOT_TOKEN is required") - _api_base = f"https://api.telegram.org/bot{_bot_token}" - - _chat_id = str(chat_id).strip() - - try: - _poll_timeout = max(1, int(poll_timeout)) - except Exception: - _poll_timeout = 20 + print(f"[TELEGRAM] Starting adapter with chat target: {self._chat_id or 'auto-bind'}") + self._initialize_offset() + return super().start() - _offset = None - _running = True - _connected = False - print(f"[TELEGRAM] Starting adapter with chat target: {_chat_id or 'auto-bind'}") - _initialize_offset() - - t = threading.Thread(target=_poll_loop, daemon=True) - t.start() - return t +_instance = None +def start_telegram(bot_token, chat_id="", poll_timeout=20): + global _instance + _instance = TelegramChannel(bot_token, chat_id, poll_timeout) + return _instance.start() def stop_telegram(): - global _running - _running = False - - -def send_message(text): - text = str(text).replace("\\n", "\n").replace("\r", "") - if not text: - return - - with _state_lock: - target_chat = _chat_id + if _instance: + _instance.stop() - if not _connected or not target_chat: - return +def getLastMessage() -> str: + return _instance.getLastMessage() if _instance else "" - max_len = 3900 - for i in range(0, len(text), max_len): - chunk = text[i:i + max_len] - if not chunk: - continue - try: - _api_call( - "sendMessage", - {"chat_id": target_chat, "text": chunk}, - timeout=15, - use_post=True, - ) - except Exception as exc: - print(f"[TELEGRAM] Send failed: {exc}") - return +def send_message(text: str) -> None: + if _instance: + _instance.send_message(text) diff --git a/lib_omegaclaw.metta b/lib_omegaclaw.metta index 7c5ca790..76341d15 100644 --- a/lib_omegaclaw.metta +++ b/lib_omegaclaw.metta @@ -8,11 +8,13 @@ !(import! &self (library OmegaClaw-Core ./src/utils)) !(import! &self (library OmegaClaw-Core ./src/helper.py)) !(import! &self (library OmegaClaw-Core ./src/agentverse.py)) +!(import! &self (library OmegaClaw-Core ./channels/base.py)) !(import! &self (library OmegaClaw-Core ./channels/irc.py)) !(import! &self (library OmegaClaw-Core ./channels/mattermost.py)) !(import! &self (library OmegaClaw-Core ./channels/telegram.py)) !(import! &self (library OmegaClaw-Core ./channels/slack.py)) !(import! &self (library OmegaClaw-Core ./channels/websearch.py)) +!(import! &self (library OmegaClaw-Core ./channels/channels_registry.py)) !(import! &self (library OmegaClaw-Core ./src/channels)) !(import! &self (library OmegaClaw-Core ./src/skills)) !(import! &self (library OmegaClaw-Core ./src/memory)) diff --git a/scripts/omegaclaw b/scripts/omegaclaw index 1f0145e4..4e278749 100755 --- a/scripts/omegaclaw +++ b/scripts/omegaclaw @@ -280,6 +280,7 @@ def config_run_omegaclaw(config_output_path): channel_config = _choose_channel() provider, embeddingprovider, api_token_var, llm_server_local_url = _choose_provider() + import_kb_on_start = _choose_import_kb() token = _prompt_llm_token() import_kb_on_start = _choose_import_kb() diff --git a/src/channels.metta b/src/channels.metta index 95e642e4..d99a6d41 100644 --- a/src/channels.metta +++ b/src/channels.metta @@ -1,68 +1,38 @@ ;configured at runtime: -(= (IRC_channel) (empty)) -(= (IRC_server) (empty)) -(= (IRC_port) (empty)) -(= (IRC_user) (empty)) -(= (commchannel) (empty)) -(= (MM_URL) (empty)) -(= (MM_CHANNEL_ID) (empty)) -(= (TG_CHAT_ID) (empty)) -(= (TG_POLL_TIMEOUT) (empty)) -(= (SL_CHANNEL_ID) (empty)) -(= (SL_POLL_INTERVAL) (empty)) +(= (commchannel) (empty)) +(= (bot_token) (empty)) +(= (channel_id) (empty)) +(= (poll_interval) (empty)) +(= (server_url) (empty)) -;Connect all the channels: +;Connect the configured channel — MeTTa resolves config, Python starts it: (= (initChannels) (progn (println! "Initializing channels") - (configure commchannel irc) - (if (== (commchannel) irc) - (progn (configure IRC_channel ##omegaclaw) - (configure IRC_server "irc.quakenet.org") - (configure IRC_port 6667) - (configure IRC_user omegaclaw) - (py-call (irc.start_irc (IRC_channel) (IRC_server) (IRC_port) (IRC_user)))) - (if (== (commchannel) telegram) - (progn (configure TG_CHAT_ID "") - (configure TG_POLL_TIMEOUT 20) - (py-call (telegram.start_telegram (TG_CHAT_ID) (TG_POLL_TIMEOUT)))) - (if (== (commchannel) slack) - (progn (configure SL_CHANNEL_ID "") - (configure SL_POLL_INTERVAL 60) - (py-call (slack.start_slack (SL_CHANNEL_ID) (SL_POLL_INTERVAL)))) - (if (== (commchannel) mattermost) - (progn (configure MM_URL "https://chat.singularitynet.io") - (configure MM_CHANNEL_ID "8fjrmabjx7gupy7e5kjznpt5qh") - (py-call (mattermost.start_mattermost (MM_URL) (MM_CHANNEL_ID)))) - (py-call (mock.start_mock)))))))) + (configure commchannel irc) + (configure bot_token "") + (configure channel_id "") ; channel_id and chat_id can be used interchangably, depending on the channel type + (configure poll_interval 20) + (configure server_url "") + (py-call (channels_registry.start + (commchannel) + (bot_token) + (channel_id) + (poll_interval) + (server_url))))) -;Receive the latest user message considering all communication channels: +;Receive the latest user message: (= (receive) - (if (== (commchannel) irc) - (py-call (irc.getLastMessage)) - (if (== (commchannel) telegram) - (py-call (telegram.getLastMessage)) - (if (== (commchannel) slack) - (py-call (slack.getLastMessage)) - (if (== (commchannel) mattermost) - (py-call (mattermost.getLastMessage)) - (py-call (mock.getLastMessage))))))) + (py-call (channels_registry.getLastMessage))) -;Send a message to all communication channels: +;Send a message: !(change-state! &lastsend "") (= (send $msg) (if (!= $msg (get-state &lastsend)) (progn (change-state! &lastsend $msg) - (let $safemsg (string-replace $msg "\n" "\\n") - (if (== (commchannel) irc) - (let $temp (cut) (py-call (irc.send_message $safemsg))) - (if (== (commchannel) telegram) - (let $temp (cut) (py-call (telegram.send_message $safemsg))) - (if (== (commchannel) slack) - (let $temp (cut) (py-call (slack.send_message $safemsg))) - (if (== (commchannel) mattermost) - (let $temp (cut) (py-call (mattermost.send_message $safemsg))) - (let $temp (cut) (py-call (mock.send_message $safemsg))))))))) _)) + (let $safemsg (string-replace $msg "\n" "\\n") + (py-call (channels_registry.send_message $safemsg)))) + _)) ;Search the internet for some information: (= (search $msg) - (py-call (websearch.search $msg))) + (py-call (websearch.search $msg))) \ No newline at end of file